From 9cfb38fdbdfe677c24b99946b197e07176609b32 Mon Sep 17 00:00:00 2001 From: Jinjing <6427696+AmethystLiang@users.noreply.github.com> Date: Wed, 12 Aug 2026 22:00:07 -0700 Subject: [PATCH] consolidate scan issues and gate session scanner I/O (STA-4049) Both local and remote session scans hit stalled WSL distros identically: one failed probe per discovered path. Unifying issue recording and gate refusal handling prevents duplication and ensures consistent behavior. - Gate all file operations (stat, readdir, read, open) through WSL stall detection instead of scattered or missing gates - Serve cached transcripts when stat stalls; distinguish gate refusals from missing files - Serialize UNC close operations to prevent thread pool exhaustion - Incremental chunk decoding in streams handles codepoint boundaries correctly --- src/main/ai-vault/remote-session-file-stat.ts | 4 +- .../ai-vault/remote-session-scan-issues.ts | 24 ----- .../remote-session-scanner-discovery.ts | 4 +- src/main/ai-vault/remote-session-scanner.ts | 6 +- .../session-first-user-prompt-read.ts | 6 +- src/main/ai-vault/session-scan-issues.ts | 27 ++++++ .../session-scanner-antigravity-history.ts | 35 ++++++- .../session-scanner-claude-subagents.ts | 49 +++++++--- ...sion-scanner-core-parser-wsl-stall.test.ts | 26 ++++++ .../ai-vault/session-scanner-discovery.ts | 13 ++- .../ai-vault/session-scanner-kimi-paths.ts | 21 ++--- .../session-scanner-omp-subagent-listing.ts | 23 ++++- .../session-scanner-scope-discovery.ts | 5 +- src/main/ai-vault/session-scanner.ts | 17 ++-- .../transcript-read-cache-wsl-stall.test.ts | 91 +++++++++++++++++++ src/main/native-chat/transcript-read-cache.ts | 11 ++- .../wsl-transcript-fs-access.test.ts | 24 +++++ .../native-chat/wsl-transcript-fs-access.ts | 70 +++++++++++++- .../wsl-transcript-fs-gate.test.ts | 23 +++++ .../native-chat/wsl-transcript-fs-gate.ts | 23 ++--- 20 files changed, 411 insertions(+), 91 deletions(-) delete mode 100644 src/main/ai-vault/remote-session-scan-issues.ts create mode 100644 src/main/ai-vault/session-scan-issues.ts create mode 100644 src/main/native-chat/transcript-read-cache-wsl-stall.test.ts diff --git a/src/main/ai-vault/remote-session-file-stat.ts b/src/main/ai-vault/remote-session-file-stat.ts index aa5809a776d..57e6d2c8ed6 100644 --- a/src/main/ai-vault/remote-session-file-stat.ts +++ b/src/main/ai-vault/remote-session-file-stat.ts @@ -2,7 +2,7 @@ import type { AiVaultAgent, AiVaultScanIssue } from '../../shared/ai-vault-types import type { ExecutionHostId } from '../../shared/execution-host' import type { FileStat } from '../providers/types' import { throwIfAiVaultScanCancelled } from './ai-vault-scan-cancellation' -import { recordRemoteSessionScanIssue } from './remote-session-scan-issues' +import { recordSessionScanIssue } from './session-scan-issues' import type { RemoteSessionFilesystemProvider } from './remote-session-scanner-types' import type { FileWithMtime } from './session-scanner-types' import { errorMessage } from './session-scanner-values' @@ -32,7 +32,7 @@ export async function statRemoteSessionFile( } catch (error) { throwIfAiVaultScanCancelled(options?.signal) if (!options?.missingIsExpected || !isMissingRemoteSessionPathError(error)) { - recordRemoteSessionScanIssue(issues, { + recordSessionScanIssue(issues, { executionHostId, agent, path, diff --git a/src/main/ai-vault/remote-session-scan-issues.ts b/src/main/ai-vault/remote-session-scan-issues.ts deleted file mode 100644 index aee229f443c..00000000000 --- a/src/main/ai-vault/remote-session-scan-issues.ts +++ /dev/null @@ -1,24 +0,0 @@ -import type { AiVaultScanIssue } from '../../shared/ai-vault-types' - -const REMOTE_SCAN_ISSUE_LIMIT = 500 - -export function recordRemoteSessionScanIssue( - issues: AiVaultScanIssue[], - issue: AiVaultScanIssue -): void { - if (issues.length < REMOTE_SCAN_ISSUE_LIMIT - 1) { - issues.push(issue) - return - } - if (issues.length === REMOTE_SCAN_ISSUE_LIMIT - 1) { - // Kinded: this row is a scan notice, not a skipped transcript — the panel - // counts unkinded issues as skipped transcript files. - issues.push({ - executionHostId: issue.executionHostId, - agent: issue.agent, - kind: 'notice', - path: 'Agent Session History scan', - message: 'Additional scan issues were omitted.' - }) - } -} diff --git a/src/main/ai-vault/remote-session-scanner-discovery.ts b/src/main/ai-vault/remote-session-scanner-discovery.ts index 91f21eab720..329bdb2e584 100644 --- a/src/main/ai-vault/remote-session-scanner-discovery.ts +++ b/src/main/ai-vault/remote-session-scanner-discovery.ts @@ -7,7 +7,7 @@ import type { FileWithMtime } from './session-scanner-types' import { errorMessage } from './session-scanner-values' import { mapRemoteScanBatches } from './remote-session-scan-batching' import { throwIfAiVaultScanCancelled } from './ai-vault-scan-cancellation' -import { recordRemoteSessionScanIssue } from './remote-session-scan-issues' +import { recordSessionScanIssue } from './session-scan-issues' import type { RemoteScannerContext, RemoteSessionCandidate, @@ -124,7 +124,7 @@ function recordRemoteDirectoryIssue( err: unknown ): void { if (!isMissingRemoteSessionPathError(err)) { - recordRemoteSessionScanIssue(issues, { + recordSessionScanIssue(issues, { executionHostId, agent: source.agent, kind: 'host', diff --git a/src/main/ai-vault/remote-session-scanner.ts b/src/main/ai-vault/remote-session-scanner.ts index 2523cd22a43..568b42c27cd 100644 --- a/src/main/ai-vault/remote-session-scanner.ts +++ b/src/main/ai-vault/remote-session-scanner.ts @@ -24,7 +24,7 @@ import { createAntigravityWorkspaceResolver } from './session-scanner-antigravit import { errorMessage } from './session-scanner-values' import { mapRemoteScanBatches } from './remote-session-scan-batching' import { throwIfAiVaultScanCancelled } from './ai-vault-scan-cancellation' -import { recordRemoteSessionScanIssue } from './remote-session-scan-issues' +import { recordSessionScanIssue } from './session-scan-issues' import { limitRemoteScanFilesystemConcurrency } from './remote-session-scan-concurrency' import { aiVaultScanLimit } from '../../shared/ai-vault-session-depth' @@ -205,7 +205,7 @@ async function scanRemoteInScopeSessions(args: { } if (index < candidates.length && sessions.length < args.limit) { - recordRemoteSessionScanIssue(args.issues, { + recordSessionScanIssue(args.issues, { executionHostId: args.context.executionHostId, agent: 'codex', kind: 'scope', @@ -241,7 +241,7 @@ async function parseRemoteSessionCandidate( return session } catch (err) { throwIfAiVaultScanCancelled(context.signal) - recordRemoteSessionScanIssue(issues, { + recordSessionScanIssue(issues, { executionHostId: context.executionHostId, agent: candidate.source.agent, path: candidate.file.path, diff --git a/src/main/ai-vault/session-first-user-prompt-read.ts b/src/main/ai-vault/session-first-user-prompt-read.ts index f250846a20e..ac5a35f2a53 100644 --- a/src/main/ai-vault/session-first-user-prompt-read.ts +++ b/src/main/ai-vault/session-first-user-prompt-read.ts @@ -1,10 +1,10 @@ -import { stat } from 'node:fs/promises' import type { AiVaultAgent, AiVaultFirstUserPromptResult, AiVaultSession } from '../../shared/ai-vault-types' import { LOCAL_EXECUTION_HOST_ID, type ExecutionHostId } from '../../shared/execution-host' +import { wslGatedStat } from '../native-chat/wsl-transcript-fs-access' import { parseAgentSessionFile } from './session-scanner-agent-parser' import { withFullFirstUserPromptCapture } from './session-scanner-first-user-prompt-capture' import { parseOpenCodeSqliteSession } from './session-scanner-opencode-sqlite' @@ -118,7 +118,9 @@ async function fileWithMtimeForPath( } try { - const info = await stat(filePath) + // 'scan' matches the parser this feeds, so the two halves of one re-parse + // share a lane instead of the stat jumping the live-transcript queue. + const info = await wslGatedStat(filePath, 'scan') return { path: filePath, mtimeMs: info.mtimeMs, diff --git a/src/main/ai-vault/session-scan-issues.ts b/src/main/ai-vault/session-scan-issues.ts new file mode 100644 index 00000000000..b3e21f9c5a8 --- /dev/null +++ b/src/main/ai-vault/session-scan-issues.ts @@ -0,0 +1,27 @@ +import type { AiVaultScanIssue } from '../../shared/ai-vault-types' + +// Why: a stalled WSL distro or an unreachable remote host fails one probe per +// discovered path, so an uncapped list ships one row per transcript file in the +// tree over IPC. Local and remote scans share the bound — the failure shape is +// the same and both feed the same panel. +const SCAN_ISSUE_LIMIT = 500 + +/** Append a scan issue, collapsing everything past the cap into one notice. */ +export function recordSessionScanIssue(issues: AiVaultScanIssue[], issue: AiVaultScanIssue): void { + if (issues.length < SCAN_ISSUE_LIMIT - 1) { + issues.push(issue) + return + } + if (issues.length === SCAN_ISSUE_LIMIT - 1) { + // Kinded: this row is a scan notice, not a skipped transcript — the panel + // counts unkinded issues as skipped transcript files. Spread first so a + // remote issue keeps its executionHostId and a local one never gains an + // explicit `undefined` (which would outrank the scanner's own stamping). + issues.push({ + ...issue, + kind: 'notice', + path: 'Agent Session History scan', + message: 'Additional scan issues were omitted.' + }) + } +} diff --git a/src/main/ai-vault/session-scanner-antigravity-history.ts b/src/main/ai-vault/session-scanner-antigravity-history.ts index 989ac7b4179..ccfb506b7f7 100644 --- a/src/main/ai-vault/session-scanner-antigravity-history.ts +++ b/src/main/ai-vault/session-scanner-antigravity-history.ts @@ -1,8 +1,28 @@ import type { AiVaultSession } from '../../shared/ai-vault-types' +import { wslGatedReadFile } from '../native-chat/wsl-transcript-fs-access' +import { WslTranscriptFsError } from '../native-chat/wsl-transcript-fs-gate' import { normalizeTitleText, parseJsonObject, timestampMs } from './session-scanner-values' const HISTORY_MATCH_WINDOW_MS = 2_000 +/** + * The local-scan `readHistory`; remote scans inject their own transport. A + * missing or unreadable history file is genuinely "no enrichment", but a gate + * refusal must propagate so the caller records a scan issue — degrading it to + * null lists the session with a missing cwd and no retry signal, and the + * resolver's memo relies on the rejection to evict rather than pin a stall. + */ +export async function readLocalAntigravityHistory(path: string): Promise { + try { + return await wslGatedReadFile(path, 'utf-8', 'scan') + } catch (error) { + if (error instanceof WslTranscriptFsError) { + throw error + } + return null + } +} + type AntigravityHistoryEntry = { timestampMs: number workspace: string @@ -26,8 +46,19 @@ export function createAntigravityWorkspaceResolver( } let index = indexes.get(historyPath) if (!index) { - index = readHistory(historyPath).then(indexAntigravityHistory) - indexes.set(historyPath, index) + // Why: a read failure is transient (a stalled WSL distro refuses here), + // so it must not be memoized — every later session under this history + // file would inherit the rejection for the process lifetime. + const pending: Promise = readHistory(historyPath) + .then(indexAntigravityHistory) + .catch((error: unknown) => { + if (indexes.get(historyPath) === pending) { + indexes.delete(historyPath) + } + throw error + }) + index = pending + indexes.set(historyPath, pending) } const workspace = findAntigravityWorkspace(session, await index) return workspace ? { ...session, cwd: workspace } : session diff --git a/src/main/ai-vault/session-scanner-claude-subagents.ts b/src/main/ai-vault/session-scanner-claude-subagents.ts index 5c269d91cf3..a2dc5745948 100644 --- a/src/main/ai-vault/session-scanner-claude-subagents.ts +++ b/src/main/ai-vault/session-scanner-claude-subagents.ts @@ -1,5 +1,3 @@ -import { createReadStream } from 'node:fs' -import { readdir, readFile, stat } from 'node:fs/promises' import { basename, extname, join } from 'node:path' import { createInterface } from 'node:readline' import type { @@ -8,6 +6,14 @@ import type { AiVaultSubagentListResult, AiVaultSubagentRunStatus } from '../../shared/ai-vault-types' +import { + openTranscriptReadStream, + wslGatedReaddir, + wslGatedReadFile, + wslGatedStat +} from '../native-chat/wsl-transcript-fs-access' +import { WslTranscriptFsError } from '../native-chat/wsl-transcript-fs-gate' +import { recordSessionScanIssue } from './session-scan-issues' import { sessionIdFromFileName, sessionSortTime } from './session-scanner-accumulator' import { parseClaudeSessionFile } from './session-scanner-primary-parsers' import { @@ -29,6 +35,10 @@ const SUBAGENT_RUNNING_RECENCY_MS = 5 * 60_000 // Match the main scanner's deliberate parse batching (SESSION_PARSE_CONCURRENCY): // opening every subagent read stream at once stalls over SSH/WSL UNC paths. const SUBAGENT_PARSE_CONCURRENCY = 8 +// 'scan', not 'exact', even though a user expands this on demand: it is bulk +// directory work feeding the same parsers as the main scan, and the exact lane +// stays reserved for live transcript probes. +const SUBAGENT_FS_PRIORITY = 'scan' const TASK_NOTIFICATION_MARKER = '' const TOOL_USE_RESULT_MARKER = '"toolUseResult"' @@ -71,8 +81,13 @@ export async function listClaudeSubagentSessions(args: { let entries try { - entries = await readdir(subagentsDir, { withFileTypes: true }) - } catch { + entries = await wslGatedReaddir(subagentsDir, SUBAGENT_FS_PRIORITY) + } catch (err) { + // A gate refusal is a stalled distro, not a session without subagents — + // report it so the panel offers a retry instead of showing an empty list. + if (err instanceof WslTranscriptFsError) { + recordSessionScanIssue(issues, { agent: 'claude', path: subagentsDir, message: err.message }) + } return { sessions: [], issues } } @@ -130,7 +145,7 @@ async function parseSubagentTranscript(args: { issues: AiVaultScanIssue[] }): Promise { try { - const fileStat = await stat(args.filePath) + const fileStat = await wslGatedStat(args.filePath, SUBAGENT_FS_PRIORITY) const session = await parseClaudeSessionFile( { path: args.filePath, @@ -161,7 +176,11 @@ async function parseSubagentTranscript(args: { } } } catch (err) { - args.issues.push({ agent: 'claude', path: args.filePath, message: errorMessage(err) }) + recordSessionScanIssue(args.issues, { + agent: 'claude', + path: args.filePath, + message: errorMessage(err) + }) return null } } @@ -187,11 +206,13 @@ function resolveSubagentStatus(args: { // 'async_launched', notification 'running') are superseded by terminal ones. async function collectSubagentTaskStatuses(parentFilePath: string): Promise> { const statuses = new Map() + const input = openTranscriptReadStream( + parentFilePath, + { encoding: 'utf-8' }, + SUBAGENT_FS_PRIORITY + ) + const lines = createInterface({ input, crlfDelay: Infinity }) try { - const lines = createInterface({ - input: createReadStream(parentFilePath, { encoding: 'utf-8' }), - crlfDelay: Infinity - }) for await (const line of lines) { const hasNotification = line.includes(TASK_NOTIFICATION_MARKER) const hasTaskResult = @@ -232,6 +253,11 @@ async function collectSubagentTaskStatuses(parentFilePath: string): Promise { const metaPath = `${transcriptPath.slice(0, -extname(transcriptPath).length)}.meta.json` try { - const record = asRecord(JSON.parse(await readFile(metaPath, 'utf-8')) as unknown) + const raw = await wslGatedReadFile(metaPath, 'utf-8', SUBAGENT_FS_PRIORITY) + const record = asRecord(JSON.parse(raw) as unknown) return { description: normalizeTitleText(extractString(record?.description) ?? ''), agentType: extractString(record?.agentType) diff --git a/src/main/ai-vault/session-scanner-core-parser-wsl-stall.test.ts b/src/main/ai-vault/session-scanner-core-parser-wsl-stall.test.ts index 72f767ec09a..44fe71c80b2 100644 --- a/src/main/ai-vault/session-scanner-core-parser-wsl-stall.test.ts +++ b/src/main/ai-vault/session-scanner-core-parser-wsl-stall.test.ts @@ -175,6 +175,18 @@ describe('Kimi session parse against a stalled WSL transcript', () => { expect(recovered?.messageCount).toBe(1) }) + + // Pins the narrowness of the parser's `instanceof WslTranscriptFsError` + // rethrow: widening it to every error turns this into a hard scan issue. + it('still returns a metadata-only session when the wire file is simply missing', async () => { + const state = uncPath('KimiMissing', ...KIMI_HOME, 'state.json') + mocks.readFile.mockResolvedValue(KIMI_STATE) + mocks.open.mockRejectedValue(missing()) + + const session = await parseAgentSessionFileCached(candidate('kimi', state), 'linux') + + expect(session).toMatchObject({ agent: 'kimi', title: 'Kimi session', messageCount: 0 }) + }) }) describe('Grok session parse against a stalled WSL transcript', () => { @@ -238,4 +250,18 @@ describe('OpenCode session parse against a stalled WSL transcript', () => { expect(recovered?.messageCount).toBe(1) }) + + // Same guard as Kimi's: a live process mid-write must stay a parseable + // session, not a refusal. + it('still returns the session when a live process left a half-written message', async () => { + const file = sessionPath('OpenCodeHalfWritten') + mocks.readdir.mockResolvedValue([dirent('msg-1.json')]) + mocks.readFile.mockImplementation((path: string) => + Promise.resolve(path === file ? OPENCODE_SESSION : '{"role":"user",') + ) + + const session = await parseAgentSessionFileCached(candidate('opencode', file), 'linux') + + expect(session).toMatchObject({ agent: 'opencode', title: 'OpenCode session', messageCount: 0 }) + }) }) diff --git a/src/main/ai-vault/session-scanner-discovery.ts b/src/main/ai-vault/session-scanner-discovery.ts index 7a6e824e661..9ec569bc547 100644 --- a/src/main/ai-vault/session-scanner-discovery.ts +++ b/src/main/ai-vault/session-scanner-discovery.ts @@ -3,6 +3,7 @@ import { extname, join } from 'node:path' import type { AiVaultAgent, AiVaultScanIssue } from '../../shared/ai-vault-types' import { wslGatedReaddir, wslGatedStat } from '../native-chat/wsl-transcript-fs-access' import { WslTranscriptFsError } from '../native-chat/wsl-transcript-fs-gate' +import { recordSessionScanIssue } from './session-scan-issues' import type { FileWithMtime, SessionFileDiscovery } from './session-scanner-types' import { errorMessage } from './session-scanner-values' @@ -29,7 +30,11 @@ export async function discoverFiles(args: { if (!(err instanceof WslTranscriptFsError)) { throw err } - args.issues.push({ agent: args.agent, path: args.rootDir, message: err.message }) + recordSessionScanIssue(args.issues, { + agent: args.agent, + path: args.rootDir, + message: err.message + }) return { agent: args.agent, rootDir: args.rootDir, files: [] } } const files: FileWithMtime[] = [] @@ -46,7 +51,11 @@ export async function discoverFiles(args: { nlink: fileStat.nlink }) } catch (err) { - args.issues.push({ agent: args.agent, path, message: errorMessage(err) }) + recordSessionScanIssue(args.issues, { + agent: args.agent, + path, + message: errorMessage(err) + }) } } return { diff --git a/src/main/ai-vault/session-scanner-kimi-paths.ts b/src/main/ai-vault/session-scanner-kimi-paths.ts index ef19a8b45a3..590e4aee814 100644 --- a/src/main/ai-vault/session-scanner-kimi-paths.ts +++ b/src/main/ai-vault/session-scanner-kimi-paths.ts @@ -87,8 +87,7 @@ export async function readKimiWorkDirBySessionId(indexPath: string): Promise - parseKimiSessionIndex(indexPath).then((parsed) => { - refused = parsed.refused - return parsed.map + parseKimiSessionIndex(indexPath).then(({ map, refused }) => { + // Why: a gate refusal is a stalled distro, not a cwd-less index — evict + // so the partial map is not served until the index's identity changes. + // Inside the loader, not at the call site: only the caller that missed + // runs this, so a concurrent hit would otherwise serve and keep it. + if (refused) { + workDirCacheByIndexPath.delete(indexPath, generation) + } + return map }) ) - // Why: a gate refusal is a stalled distro, not a cwd-less index — evicting it - // stops the partial map being served until the index's identity changes. - if (refused) { - workDirCacheByIndexPath.delete(indexPath, generation) - } - return workDirs } async function parseKimiSessionIndex( diff --git a/src/main/ai-vault/session-scanner-omp-subagent-listing.ts b/src/main/ai-vault/session-scanner-omp-subagent-listing.ts index 0efcdbdc92c..3ecfade1813 100644 --- a/src/main/ai-vault/session-scanner-omp-subagent-listing.ts +++ b/src/main/ai-vault/session-scanner-omp-subagent-listing.ts @@ -1,10 +1,12 @@ -import { readdir, stat } from 'node:fs/promises' import { basename, extname, join } from 'node:path' import type { AiVaultScanIssue, AiVaultSession, AiVaultSubagentListResult } from '../../shared/ai-vault-types' +import { wslGatedReaddir, wslGatedStat } from '../native-chat/wsl-transcript-fs-access' +import { WslTranscriptFsError } from '../native-chat/wsl-transcript-fs-gate' +import { recordSessionScanIssue } from './session-scan-issues' import { sessionIdFromFileName, sessionSortTime } from './session-scanner-accumulator' import { parseMessageGraphSessionFile } from './session-scanner-graph-parsers' import { @@ -16,6 +18,8 @@ import { errorMessage } from './session-scanner-values' // Match the Claude subagent lister's deliberate parse batching: opening every // read stream at once stalls over WSL UNC paths. const OMP_SUBAGENT_PARSE_CONCURRENCY = 8 +// Bulk directory work, so 'scan' — same reasoning as the Claude lister. +const OMP_SUBAGENT_FS_PRIORITY = 'scan' /** * List the task-subagent transcripts of one OMP session, on demand. The main @@ -35,8 +39,13 @@ export async function listOmpSubagentSessions(args: { let entries try { - entries = await readdir(artifactDir, { withFileTypes: true }) - } catch { + entries = await wslGatedReaddir(artifactDir, OMP_SUBAGENT_FS_PRIORITY) + } catch (err) { + // A gate refusal is a stalled distro, not a session without subagents — + // report it so the panel offers a retry instead of showing an empty list. + if (err instanceof WslTranscriptFsError) { + recordSessionScanIssue(issues, { agent: 'omp', path: artifactDir, message: err.message }) + } return { sessions: [], issues } } @@ -79,7 +88,7 @@ async function parseOmpSubagentTranscript(args: { }): Promise { const filePath = join(args.artifactDir, args.name) try { - const fileStat = await stat(filePath) + const fileStat = await wslGatedStat(filePath, OMP_SUBAGENT_FS_PRIORITY) // The shared OMP parser decorates every parse with an artifact-dir count, so // a child row carries its own grandchild count. It is accurate but has no // renderer — subagent rows don't expand — and this lister is local-only, so @@ -107,7 +116,11 @@ async function parseOmpSubagentTranscript(args: { } } } catch (err) { - args.issues.push({ agent: 'omp', path: filePath, message: errorMessage(err) }) + recordSessionScanIssue(args.issues, { + agent: 'omp', + path: filePath, + message: errorMessage(err) + }) return null } } diff --git a/src/main/ai-vault/session-scanner-scope-discovery.ts b/src/main/ai-vault/session-scanner-scope-discovery.ts index 3ae8a75abfa..e787aef628e 100644 --- a/src/main/ai-vault/session-scanner-scope-discovery.ts +++ b/src/main/ai-vault/session-scanner-scope-discovery.ts @@ -12,6 +12,7 @@ import { } from '../../shared/cross-platform-path' import type { AiVaultScanIssue } from '../../shared/ai-vault-types' import { parseWslUncPath } from '../../shared/wsl-paths' +import { recordSessionScanIssue } from './session-scan-issues' import type { FileWithMtime } from './session-scanner-types' import { errorMessage, extractString, parseJsonObject } from './session-scanner-values' @@ -34,7 +35,7 @@ export function resetProjectDirCwdCacheForTests(): void { // degrade-to-empty containment but make the gap visible in the scan issues. function recordRefusal(issues: AiVaultScanIssue[], path: string, error: unknown): void { if (error instanceof WslTranscriptFsError) { - issues.push({ agent: 'claude', path, message: error.message }) + recordSessionScanIssue(issues, { agent: 'claude', path, message: error.message }) } } @@ -282,7 +283,7 @@ async function collectClaudeFiles(args: { sizeBytes: fileStat.size }) } catch (err) { - args.issues.push({ agent: 'claude', path, message: errorMessage(err) }) + recordSessionScanIssue(args.issues, { agent: 'claude', path, message: errorMessage(err) }) } } } diff --git a/src/main/ai-vault/session-scanner.ts b/src/main/ai-vault/session-scanner.ts index 68f4af09ddf..b4ef6f06ce7 100644 --- a/src/main/ai-vault/session-scanner.ts +++ b/src/main/ai-vault/session-scanner.ts @@ -1,4 +1,3 @@ -import { wslGatedReadFile } from '../native-chat/wsl-transcript-fs-access' import type { AiVaultListResult, AiVaultScanIssue, @@ -14,6 +13,7 @@ import { } from './codex-session-root-dedup' import { createAntigravityWorkspaceResolver, + readLocalAntigravityHistory, type AntigravityWorkspaceResolver } from './session-scanner-antigravity-history' import { antigravityHistoryPathForBrainDir } from './session-scanner-antigravity-paths' @@ -27,6 +27,7 @@ import { parseAgentSessionFileCached, type SessionParseStats } from './session-scanner-parse-cache' +import { recordSessionScanIssue } from './session-scan-issues' import { discoverInScopeClaudeFiles } from './session-scanner-scope-discovery' import { DEFAULT_CODEX_HOME_DIR, @@ -71,7 +72,9 @@ export async function scanAiVaultSessions( const executionHostId = options.executionHostId ?? LOCAL_EXECUTION_HOST_ID const issues: AiVaultScanIssue[] = [] const parseStats = createSessionParseStats() - const antigravityWorkspaceResolver = createAntigravityWorkspaceResolver(readOptionalTextFile) + const antigravityWorkspaceResolver = createAntigravityWorkspaceResolver( + readLocalAntigravityHistory + ) // Why: persisted entries must be seeded before any candidate is parsed, or // the cold scan gains nothing from the cache file (#9210). throwIfAiVaultScanCancelled(options.signal) @@ -255,7 +258,7 @@ async function parseSessionCandidates(args: { for (const result of results) { if (result.issue) { - args.issues.push(result.issue) + recordSessionScanIssue(args.issues, result.issue) } if (result.session) { sessions.push(result.session) @@ -305,14 +308,6 @@ async function parseSessionCandidate( } } -async function readOptionalTextFile(path: string): Promise { - try { - return await wslGatedReadFile(path, 'utf-8', 'scan') - } catch { - return null - } -} - function withSessionExecutionHost( session: AiVaultSession, executionHostId: ExecutionHostId diff --git a/src/main/native-chat/transcript-read-cache-wsl-stall.test.ts b/src/main/native-chat/transcript-read-cache-wsl-stall.test.ts new file mode 100644 index 00000000000..1ce0a22e455 --- /dev/null +++ b/src/main/native-chat/transcript-read-cache-wsl-stall.test.ts @@ -0,0 +1,91 @@ +import { beforeEach, describe, expect, it, vi } from 'vitest' +import type * as NodeFsPromisesModule from 'node:fs/promises' + +const UNC_PATH = '\\\\wsl.localhost\\Ubuntu\\home\\ada\\.claude\\projects\\p\\session.jsonl' + +const mocks = vi.hoisted(() => ({ + resolve: vi.fn<() => Promise>(), + stat: vi.fn(), + read: vi.fn() +})) + +vi.mock('./session-file-resolver', () => ({ + resolveSessionFilePath: mocks.resolve +})) +vi.mock('./transcript-reader', () => ({ + readNativeChatTranscript: mocks.read +})) +vi.mock('node:fs/promises', async (importOriginal) => ({ + ...(await importOriginal()), + stat: mocks.stat +})) + +import { + clearNativeChatTranscriptCache, + readNativeChatTranscriptCached +} from './transcript-read-cache' +import { WSL_TRANSCRIPT_FS_EXACT_TIMEOUT_MS } from './wsl-transcript-fs-gate' + +const SLOW_MESSAGE = + 'WSL transcript files are temporarily unavailable because filesystem access is taking too long. Try again shortly or restart Orca if the issue continues.' + +beforeEach(() => { + mocks.resolve.mockReset() + mocks.stat.mockReset() + mocks.read.mockReset() + mocks.resolve.mockResolvedValue(UNC_PATH) + clearNativeChatTranscriptCache() +}) + +describe('cached native chat transcript read with a stalled post-resolution stat', () => { + it('reports a retryable error without notFound and leaves nothing cached', async () => { + vi.useFakeTimers() + try { + mocks.stat.mockReturnValueOnce(new Promise(() => {})) + const stalled = readNativeChatTranscriptCached('claude', 'session-id') + await vi.advanceTimersByTimeAsync(WSL_TRANSCRIPT_FS_EXACT_TIMEOUT_MS + 1) + + const result = await stalled + expect(result).toEqual({ error: SLOW_MESSAGE }) + expect(result).not.toHaveProperty('notFound') + } finally { + vi.useRealTimers() + } + }) + + it('re-attempts the stat on the next call instead of serving a poisoned entry', async () => { + // The stalled task above still holds its permit, so this fast-fails on the + // stuck route rather than waiting out a second deadline — but it must still + // reach `stat` again once a different distro answers. + const healthyPath = '\\\\wsl.localhost\\Debian\\home\\ada\\.claude\\projects\\p\\session.jsonl' + mocks.resolve.mockResolvedValue(healthyPath) + mocks.stat.mockResolvedValue({ mtimeMs: 5, size: 0 }) + mocks.read.mockResolvedValue({ messages: [] }) + + await readNativeChatTranscriptCached('claude', 'session-id') + + expect(mocks.stat).toHaveBeenCalledWith(healthyPath) + }) + + it('serves the cached parse rather than a retry banner once the distro stalls', async () => { + const path = '\\\\wsl.localhost\\Fedora\\home\\ada\\.claude\\projects\\p\\session.jsonl' + mocks.resolve.mockResolvedValue(path) + mocks.stat.mockResolvedValue({ mtimeMs: 9, size: 4 }) + mocks.read.mockResolvedValue({ messages: [{ role: 'user', text: 'hi' }] }) + const warm = await readNativeChatTranscriptCached('claude', 'session-id') + + vi.useFakeTimers() + try { + mocks.stat.mockReturnValueOnce(new Promise(() => {})) + const stalled = readNativeChatTranscriptCached('claude', 'session-id') + await vi.advanceTimersByTimeAsync(WSL_TRANSCRIPT_FS_EXACT_TIMEOUT_MS + 1) + + // The stat is what the stall refuses; the parse it would have validated is + // complete and unchanged, so the chat stays rendered. + await expect(stalled).resolves.toBe(warm) + expect(mocks.read).toHaveBeenCalledOnce() + } finally { + vi.useRealTimers() + } + }) +}) diff --git a/src/main/native-chat/transcript-read-cache.ts b/src/main/native-chat/transcript-read-cache.ts index b4967cb33f4..ee0c8ddcafd 100644 --- a/src/main/native-chat/transcript-read-cache.ts +++ b/src/main/native-chat/transcript-read-cache.ts @@ -115,15 +115,22 @@ export async function readNativeChatTranscriptCached( } const key = cacheKey(agent, filePath) + const cached = cache.get(key) let mtimeMs: number let bytes: number try { ;({ mtimeMs, bytes } = await fileStat(filePath)) } catch (err) { - // Nothing cached and no `notFound`: the next call re-stats a woken distro. + // Why: the refusal says the distro stalled, not that this parse went stale — + // a complete cached transcript beats a retry banner. With nothing cached the + // error stands, and no `notFound`, so the next call re-stats a woken distro. + if (cached) { + // Bump recency so a session read through a stall survives eviction. + setCached(key, cached) + return cached.result + } return { error: wslTranscriptFsRefusal(err).message } } - const cached = cache.get(key) if (cached && Number.isFinite(mtimeMs) && cached.mtimeMs === mtimeMs) { // Bump recency so a frequently-read session survives eviction. setCached(key, cached) diff --git a/src/main/native-chat/wsl-transcript-fs-access.test.ts b/src/main/native-chat/wsl-transcript-fs-access.test.ts index 8b402a3bb98..b2bb893be4b 100644 --- a/src/main/native-chat/wsl-transcript-fs-access.test.ts +++ b/src/main/native-chat/wsl-transcript-fs-access.test.ts @@ -180,6 +180,30 @@ describe('transcript filesystem accessor on WSL UNC', () => { await new Promise((resolve) => setImmediate(resolve)) }) + it('drains handle closes one at a time so teardown cannot flood the thread pool', async () => { + let releaseFirst: (() => void) | undefined + const first = fakeHandle() + first.close.mockReturnValue( + new Promise((resolve) => { + releaseFirst = resolve + }) + ) + const second = fakeHandle() + + await closeTranscriptHandle(first as never, UNC_PATH) + await closeTranscriptHandle(second as never, UNC_PATH) + await new Promise((resolve) => setImmediate(resolve)) + + // A blocked uv_fs_close holds a libuv thread the gate cannot see, so the + // second one must wait rather than occupy a thread of its own. + expect(first.close).toHaveBeenCalledTimes(1) + expect(second.close).not.toHaveBeenCalled() + + releaseFirst?.() + await new Promise((resolve) => setImmediate(resolve)) + expect(second.close).toHaveBeenCalledTimes(1) + }) + it.each([ ['stat', wslGatedStat, mocks.stat], ['lstat', wslGatedLstat, mocks.lstat] diff --git a/src/main/native-chat/wsl-transcript-fs-access.ts b/src/main/native-chat/wsl-transcript-fs-access.ts index d9ac0170c60..b9b2efdf071 100644 --- a/src/main/native-chat/wsl-transcript-fs-access.ts +++ b/src/main/native-chat/wsl-transcript-fs-access.ts @@ -5,6 +5,10 @@ import { StringDecoder } from 'node:string_decoder' import { isWslUncPath } from '../../shared/wsl-paths' import { runWslTranscriptFsTask, type WslTranscriptFsTaskPriority } from './wsl-transcript-fs-gate' +/** Never nest a gated call inside another — that deadlocks the scan slot. */ + +// Why: one deadline per chunk instead of one for the whole file, so a large +// healthy-but-slow transcript is not false-failed by a whole-file timeout. const WSL_TRANSCRIPT_READ_CHUNK_BYTES = 1024 * 1024 type Operation = Parameters[0]['operation'] @@ -56,6 +60,7 @@ export function wslGatedReadFile( return runPathOperation('readfile', path, priority, signal, () => readFile(path, encoding)) } +// dedupe:false — two joiners would share one FileHandle and both close it. export function wslGatedOpen( path: string, priority: WslTranscriptFsTaskPriority, @@ -63,10 +68,18 @@ export function wslGatedOpen( ): Promise { return runPathOperation('open', path, priority, signal, () => open(path, 'r'), { dedupe: false, + // An unabortable open can still succeed after its caller timed out or + // cancelled; without this the descriptor leaks for the process lifetime. onAbandonedResult: (handle) => void closeTranscriptHandle(handle, path) }) } +/** + * `path` is only the gate's route key and UNC guard — the handle carries no + * path and is never re-opened. dedupe:false because `read` fills the CALLER's + * buffer: a joiner would receive the first caller's buffer while its own + * (often `Buffer.allocUnsafe`) stays uninitialized. + */ export function wslGatedRead( handle: FileHandle, path: string, @@ -87,14 +100,54 @@ export function wslGatedRead( ) } +// A blocked `uv_fs_close` holds a libuv threadpool thread just like a blocked +// read does, but it is invisible to the gate — so UNC teardown drains one handle +// at a time. Unbounded fire-and-forget closes are what would exhaust the pool the +// gate's two permits are sized against, re-creating the process-wide stall the +// gate exists to prevent; serializing them costs at most one extra busy thread. +// The queue cannot grow without bound: opens on a stuck route already fast-fail. +const MAX_CONCURRENT_UNC_CLOSES = 1 +const queuedCloses: FileHandle[] = [] +let activeCloses = 0 + +function drainQueuedCloses(): void { + while (activeCloses < MAX_CONCURRENT_UNC_CLOSES) { + const handle = queuedCloses.shift() + if (!handle) { + return + } + activeCloses += 1 + void handle + .close() + .catch(() => {}) + .finally(() => { + activeCloses -= 1 + drainQueuedCloses() + }) + } +} + +/** + * Never gated. Off UNC this is the prior contract verbatim — the caller awaits + * fd teardown and a close failure surfaces. On UNC it is fire-and-forget: + * closing a handle on a stalled mount can itself block, and a gated close would + * burn a permit and a waiter deadline purely for teardown. One leaked fd until + * the OS unblocks beats a second blocked waiter. + */ export function closeTranscriptHandle(handle: FileHandle, path: string): Promise { if (!isWslUncPath(path)) { return handle.close() } - void handle.close().catch(() => {}) + queuedCloses.push(handle) + drainQueuedCloses() return Promise.resolve() } +/** + * Open, read one slice, close — for callers that want bytes at an offset and + * never touch the handle. Each of the two syscalls is admitted separately, so a + * stalled mount fails at whichever one it blocks on. + */ export async function readTranscriptSlice( path: string, position: number, @@ -123,7 +176,11 @@ export async function readTranscriptSlice( export type TranscriptReadStreamOptions = { start?: number + /** Inclusive, matching `createReadStream`. */ end?: number + /** Set it to get decoded string chunks on both branches; leave it unset and + * the UNC branch yields `Buffer`s the consumer must decode incrementally + * itself, since a chunk boundary can split a multibyte codepoint. */ encoding?: BufferEncoding } @@ -134,6 +191,8 @@ async function* gatedChunks( signal?: AbortSignal ): AsyncGenerator { const handle = await wslGatedOpen(path, priority, signal) + // Why: chunk boundaries fall mid-codepoint, so decoding each slice + // independently would emit U+FFFD on both sides of any straddling character. const decoder = options.encoding ? new StringDecoder(options.encoding) : null try { let position = options.start ?? 0 @@ -170,15 +229,24 @@ async function* gatedChunks( yield decoded } } + // Trailing bytes of an incomplete sequence at EOF, replacement-char'd once. const trailing = decoder?.end() if (trailing) { yield trailing } } finally { + // Runs on `.destroy()` too (Readable.from calls the generator's return()), + // so an aborted head-read cannot leak the gated handle. await closeTranscriptHandle(handle, path) } } +/** + * A read stream whose UNC branch admits and deadlines each 1 MiB chunk + * separately. A gate refusal mid-stream surfaces as an `'error'` event carrying + * the `WslTranscriptFsError`, which every existing consumer funnels into its + * `catch`. + */ export function openTranscriptReadStream( path: string, options: TranscriptReadStreamOptions, diff --git a/src/main/native-chat/wsl-transcript-fs-gate.test.ts b/src/main/native-chat/wsl-transcript-fs-gate.test.ts index e4a86dc8d50..d4d87a4feeb 100644 --- a/src/main/native-chat/wsl-transcript-fs-gate.test.ts +++ b/src/main/native-chat/wsl-transcript-fs-gate.test.ts @@ -139,6 +139,29 @@ describe('WSL transcript filesystem task scheduling', () => { expect(duplicateTask).not.toHaveBeenCalled() }) + it('never joins an exact probe onto the same file queued as scan work', async () => { + const transcript = '\\\\wsl.localhost\\Ubuntu\\home\\ada\\live.jsonl' + const statTask = (priority: 'exact' | 'scan', task: () => Promise): Promise => + runWslTranscriptFsTask({ operation: 'stat', path: transcript, priority }, task) + // Hold the single scan slot on another distro so the same-file scan stat + // cannot run — a joiner would inherit exactly that queued position. + const blocked = deferred() + const holderTask = vi.fn(() => blocked.promise) + const holder = run('\\\\wsl.localhost\\Debian\\tree', 'scan', holderTask) + await vi.waitFor(() => expect(holderTask).toHaveBeenCalledOnce()) + + const scanTask = vi.fn(async () => 'scan') + const scanned = statTask('scan', scanTask) + const exactTask = vi.fn(async () => 'exact') + + await expect(statTask('exact', exactTask)).resolves.toBe('exact') + expect(exactTask).toHaveBeenCalledOnce() + expect(scanTask).not.toHaveBeenCalled() + + blocked.resolve('holder') + await expect(Promise.all([holder, scanned])).resolves.toEqual(['holder', 'scan']) + }) + it('keeps shared work needed by a remaining waiter', async () => { const pending = deferred() const task = vi.fn(() => pending.promise) diff --git a/src/main/native-chat/wsl-transcript-fs-gate.ts b/src/main/native-chat/wsl-transcript-fs-gate.ts index 90627ae7840..e214e1ddaa0 100644 --- a/src/main/native-chat/wsl-transcript-fs-gate.ts +++ b/src/main/native-chat/wsl-transcript-fs-gate.ts @@ -162,11 +162,9 @@ function stuckBlocker( // Caller-abort pre-checks live in runWslTranscriptFsTask; nothing here yields // before the waiter is attached, so no aborted-signal recheck is needed. -function attachWaiter( - task: ScheduledTask, - priority: WslTranscriptFsTaskPriority, - signal?: AbortSignal -): Promise { +// The task's own priority is every waiter's deadline: the dedupe key carries +// priority, so a joiner can only ever join a task scheduled at its own. +function attachWaiter(task: ScheduledTask, signal?: AbortSignal): Promise { if (task.waiters.size >= WSL_TRANSCRIPT_FS_MAX_WAITERS_PER_TASK) { return Promise.reject(capacityError()) } @@ -193,7 +191,7 @@ function attachWaiter( } reject(error) abandonTaskIfUnused(task as UnknownScheduledTask, error) - }, timeoutMs(priority)) + }, timeoutMs(task.priority)) waiter.timeout.unref?.() }) } @@ -316,12 +314,15 @@ export function runWslTranscriptFsTask( return Promise.reject(abortReason(options.signal)) } // Why: route spelling and Linux path case can change provider behavior, so - // only byte-identical filesystem requests are safe to share. A counter key - // never collides, so an un-deduped task simply never finds a join target. + // only byte-identical filesystem requests are safe to share. Priority is part + // of the key because it selects the task's lane, scan slot and deadline: an + // exact probe that joined a queued scan task would inherit scan scheduling and + // starve behind the very traffic the two priorities exist to bypass. A counter + // key never collides, so an un-deduped task simply never finds a join target. const key = options.dedupe === false ? `undeduped:${++undedupedTaskId}` - : JSON.stringify([options.operation, options.path]) + : JSON.stringify([options.operation, options.path, options.priority]) const existing = inFlightTasks.get(key) as ScheduledTask | undefined if (existing) { // A stuck target — or a queued target that stuck I/O keeps from ever @@ -333,7 +334,7 @@ export function runWslTranscriptFsTask( ) { return Promise.reject(unavailableError()) } - return attachWaiter(existing, options.priority, options.signal) + return attachWaiter(existing, options.signal) } const route = routeKey(options.path) // Fail fast when the permit, route, or scan slot this task needs is held by @@ -358,7 +359,7 @@ export function runWslTranscriptFsTask( stuck: false } inFlightTasks.set(key, scheduled as UnknownScheduledTask) - const result = attachWaiter(scheduled, options.priority, options.signal) + const result = attachWaiter(scheduled, options.signal) queuedTasks.push(scheduled as UnknownScheduledTask) queueMicrotask(pumpTasks) return result