From 558bfa9537f69d3a6b228f8e18b189d3402e2fbd Mon Sep 17 00:00:00 2001 From: Jinjing <6427696+AmethystLiang@users.noreply.github.com> Date: Wed, 12 Aug 2026 18:45:42 -0700 Subject: [PATCH] fix(ai-vault): gate post-resolution WSL transcript I/O (STA-4049) PR #14090 admitted only path *resolution* through the WSL transcript filesystem gate. Every byte read afterwards from the resulting \\wsl.localhost\... UNC path ran raw, so a distro that answers the first access() and then stalls hung Native Chat at "loading" and AI Vault at "scanning" with no timeout and no error. Route that I/O through a new wsl-transcript-fs-access accessor, which is a verbatim node:fs passthrough off UNC and an admitted, deadlined task on it. open/positional-read opt out of coalescing (dedupe: false): joiners would share one FileHandle or one caller's buffer. Refusals now surface as the existing retryable message rather than notFound, per-root scan failures are contained to an AiVaultScanIssue, and the memoized Codex/Kimi indexes evict on refusal so a stall cannot pin "no titles"/"no cwd" until the index changes. --- .../session-scanner-antigravity-parser.ts | 4 +- .../ai-vault/session-scanner-codex-parser.ts | 4 +- .../session-scanner-codex-title-index.ts | 27 +- .../session-scanner-copilot-parser.ts | 4 +- .../ai-vault/session-scanner-cursor-parser.ts | 4 +- .../ai-vault/session-scanner-devin-parser.ts | 8 +- ...session-scanner-discovery-wsl-gate.test.ts | 115 +++++++- .../ai-vault/session-scanner-discovery.ts | 28 +- .../ai-vault/session-scanner-droid-parser.ts | 4 +- .../session-scanner-fs-import-guard.test.ts | 81 ++++++ .../session-scanner-gemini-parsers.ts | 11 +- .../ai-vault/session-scanner-graph-parsers.ts | 9 +- .../ai-vault/session-scanner-grok-parser.ts | 11 +- .../ai-vault/session-scanner-hermes-parser.ts | 8 +- ...sion-scanner-index-cache-wsl-stall.test.ts | 124 +++++++++ .../ai-vault/session-scanner-kimi-parser.ts | 9 +- .../ai-vault/session-scanner-kimi-paths.ts | 34 ++- ...ession-scanner-omp-subagent-transcripts.ts | 4 +- .../session-scanner-opencode-parser.ts | 8 +- ...scanner-opencode-sources-wsl-stall.test.ts | 73 ++++++ .../session-scanner-opencode-sources.ts | 26 +- .../ai-vault/session-scanner-parse-cache.ts | 17 +- .../session-scanner-parse-wsl-stall.test.ts | 149 +++++++++++ .../session-scanner-primary-parsers.ts | 4 +- .../session-scanner-scope-discovery.ts | 73 ++++-- .../session-scanner-subagent-transcripts.ts | 4 +- src/main/ai-vault/session-scanner-values.ts | 4 +- src/main/ai-vault/session-scanner.ts | 4 +- ...ession-title-file-reader-wsl-stall.test.ts | 83 ++++++ .../ai-vault/session-title-file-reader.ts | 4 +- .../native-chat/transcript-file-version.ts | 4 +- .../transcript-incremental-reader.ts | 11 +- .../transcript-read-cache-wsl-stall.test.ts | 57 ++++ src/main/native-chat/transcript-read-cache.ts | 22 +- .../transcript-reader-wsl-stall.test.ts | 85 ++++++ src/main/native-chat/transcript-reader.ts | 4 +- .../transcript-tail-reader-wsl-gate.test.ts | 84 +++++- .../native-chat/transcript-tail-reader.ts | 48 +++- .../native-chat/transcript-watch-engine.ts | 32 ++- .../transcript-watch-wsl-stall.test.ts | 143 ++++++++++ src/main/native-chat/transcript-watch.ts | 18 +- .../wsl-transcript-fs-access.test.ts | 245 ++++++++++++++++++ .../native-chat/wsl-transcript-fs-access.ts | 198 ++++++++++++++ .../wsl-transcript-fs-gate.test.ts | 75 ++++++ .../native-chat/wsl-transcript-fs-gate.ts | 14 +- 45 files changed, 1813 insertions(+), 165 deletions(-) create mode 100644 src/main/ai-vault/session-scanner-fs-import-guard.test.ts create mode 100644 src/main/ai-vault/session-scanner-index-cache-wsl-stall.test.ts create mode 100644 src/main/ai-vault/session-scanner-opencode-sources-wsl-stall.test.ts create mode 100644 src/main/ai-vault/session-scanner-parse-wsl-stall.test.ts create mode 100644 src/main/ai-vault/session-title-file-reader-wsl-stall.test.ts create mode 100644 src/main/native-chat/transcript-read-cache-wsl-stall.test.ts create mode 100644 src/main/native-chat/transcript-reader-wsl-stall.test.ts create mode 100644 src/main/native-chat/transcript-watch-wsl-stall.test.ts create mode 100644 src/main/native-chat/wsl-transcript-fs-access.test.ts create mode 100644 src/main/native-chat/wsl-transcript-fs-access.ts diff --git a/src/main/ai-vault/session-scanner-antigravity-parser.ts b/src/main/ai-vault/session-scanner-antigravity-parser.ts index 818167fbd0c..3fede1d5521 100644 --- a/src/main/ai-vault/session-scanner-antigravity-parser.ts +++ b/src/main/ai-vault/session-scanner-antigravity-parser.ts @@ -1,5 +1,5 @@ import { remoteSessionContentLines } from './remote-session-content-lines' -import { createReadStream } from 'node:fs' +import { openTranscriptReadStream } from '../native-chat/wsl-transcript-fs-access' import { createInterface } from 'node:readline' import type { AiVaultSession } from '../../shared/ai-vault-types' import type { ExecutionHostId } from '../../shared/execution-host' @@ -27,7 +27,7 @@ export async function parseAntigravitySessionFile( platform: NodeJS.Platform = process.platform ): Promise { const lines = createInterface({ - input: createReadStream(file.path, { encoding: 'utf-8' }), + input: openTranscriptReadStream(file.path, { encoding: 'utf-8' }, 'scan'), crlfDelay: Infinity }) return parseAntigravitySessionLines({ file, lines, platform }) diff --git a/src/main/ai-vault/session-scanner-codex-parser.ts b/src/main/ai-vault/session-scanner-codex-parser.ts index 0f73fa54126..6887b7ed5d8 100644 --- a/src/main/ai-vault/session-scanner-codex-parser.ts +++ b/src/main/ai-vault/session-scanner-codex-parser.ts @@ -1,4 +1,4 @@ -import { createReadStream } from 'node:fs' +import { openTranscriptReadStream } from '../native-chat/wsl-transcript-fs-access' import { createInterface } from 'node:readline' import type { AiVaultSession } from '../../shared/ai-vault-types' import { readCodexSessionIndexTitle } from './session-scanner-codex-title-index' @@ -42,7 +42,7 @@ export async function parseCodexSessionFile( executionHostId?: ExecutionHostId ): Promise { const lines = createInterface({ - input: createReadStream(file.path, { encoding: 'utf-8' }), + input: openTranscriptReadStream(file.path, { encoding: 'utf-8' }, 'scan'), crlfDelay: Infinity }) diff --git a/src/main/ai-vault/session-scanner-codex-title-index.ts b/src/main/ai-vault/session-scanner-codex-title-index.ts index 8ef2d5645cb..635a215562a 100644 --- a/src/main/ai-vault/session-scanner-codex-title-index.ts +++ b/src/main/ai-vault/session-scanner-codex-title-index.ts @@ -1,7 +1,7 @@ -import { createReadStream } from 'node:fs' -import { stat } from 'node:fs/promises' import { basename, dirname, join } from 'node:path' import { createInterface } from 'node:readline' +import { openTranscriptReadStream, wslGatedStat } from '../native-chat/wsl-transcript-fs-access' +import { WslTranscriptFsError } from '../native-chat/wsl-transcript-fs-gate' import { extractString, normalizeTitleText, parseJsonObject } from './session-scanner-values' // Codex names threads lazily in /session_index.jsonl; transcripts @@ -77,7 +77,7 @@ async function readCodexSessionIndexTitles(codexHome: string): Promise ({ - signature, - titles - })) + const pending = readCodexSessionIndexTitlesFromDisk(indexPath).then(({ titles, refused }) => { + // Why: a gate refusal is a stalled distro, not a title-less index — caching + // it would pin "no titles" until the index's size or mtime changes. + if (refused && codexSessionIndexTitleCache.get(codexHome) === pending) { + codexSessionIndexTitleCache.delete(codexHome) + } + return { signature, titles } + }) storeCodexSessionIndexTitleCacheEntry(codexHome, pending) return (await pending).titles } @@ -133,11 +137,11 @@ function storeCodexSessionIndexTitleCacheEntry( async function readCodexSessionIndexTitlesFromDisk( indexPath: string -): Promise> { +): Promise<{ titles: Map; refused: boolean }> { const titleBySessionId = new Map() try { const lines = createInterface({ - input: createReadStream(indexPath, { encoding: 'utf-8' }), + input: openTranscriptReadStream(indexPath, { encoding: 'utf-8' }, 'scan'), crlfDelay: Infinity }) for await (const line of lines) { @@ -151,8 +155,9 @@ async function readCodexSessionIndexTitlesFromDisk( titleBySessionId.set(sessionId, title) } } - } catch { + } catch (error) { // Codex creates the index opportunistically; older homes may only have raw transcripts. + return { titles: titleBySessionId, refused: error instanceof WslTranscriptFsError } } - return titleBySessionId + return { titles: titleBySessionId, refused: false } } diff --git a/src/main/ai-vault/session-scanner-copilot-parser.ts b/src/main/ai-vault/session-scanner-copilot-parser.ts index 4d0140fc3df..7e5450393e0 100644 --- a/src/main/ai-vault/session-scanner-copilot-parser.ts +++ b/src/main/ai-vault/session-scanner-copilot-parser.ts @@ -1,5 +1,5 @@ import { remoteSessionContentLines } from './remote-session-content-lines' -import { createReadStream } from 'node:fs' +import { openTranscriptReadStream } from '../native-chat/wsl-transcript-fs-access' import { createInterface } from 'node:readline' import type { AiVaultSession } from '../../shared/ai-vault-types' import type { ExecutionHostId } from '../../shared/execution-host' @@ -35,7 +35,7 @@ export async function parseCopilotSessionFile( platform: NodeJS.Platform = process.platform ): Promise { const lines = createInterface({ - input: createReadStream(file.path, { encoding: 'utf-8' }), + input: openTranscriptReadStream(file.path, { encoding: 'utf-8' }, 'scan'), crlfDelay: Infinity }) return parseCopilotSessionLines({ file, lines, platform }) diff --git a/src/main/ai-vault/session-scanner-cursor-parser.ts b/src/main/ai-vault/session-scanner-cursor-parser.ts index b18402584e5..0f44aa923fa 100644 --- a/src/main/ai-vault/session-scanner-cursor-parser.ts +++ b/src/main/ai-vault/session-scanner-cursor-parser.ts @@ -1,5 +1,5 @@ import { remoteSessionContentLines } from './remote-session-content-lines' -import { createReadStream } from 'node:fs' +import { openTranscriptReadStream } from '../native-chat/wsl-transcript-fs-access' import { createInterface } from 'node:readline' import type { AiVaultSession } from '../../shared/ai-vault-types' import type { ExecutionHostId } from '../../shared/execution-host' @@ -33,7 +33,7 @@ export async function parseCursorSessionFile( platform: NodeJS.Platform = process.platform ): Promise { const lines = createInterface({ - input: createReadStream(file.path, { encoding: 'utf-8' }), + input: openTranscriptReadStream(file.path, { encoding: 'utf-8' }, 'scan'), crlfDelay: Infinity }) return parseCursorSessionLines({ file, lines, platform }) diff --git a/src/main/ai-vault/session-scanner-devin-parser.ts b/src/main/ai-vault/session-scanner-devin-parser.ts index 4ad6fa45f78..a43603bca47 100644 --- a/src/main/ai-vault/session-scanner-devin-parser.ts +++ b/src/main/ai-vault/session-scanner-devin-parser.ts @@ -1,4 +1,4 @@ -import { readFile } from 'node:fs/promises' +import { wslGatedReadFile } from '../native-chat/wsl-transcript-fs-access' import type { AiVaultSession } from '../../shared/ai-vault-types' import type { ExecutionHostId } from '../../shared/execution-host' import type { FileWithMtime } from './session-scanner-types' @@ -27,7 +27,11 @@ export async function parseDevinSessionFile( file: FileWithMtime, platform: NodeJS.Platform = process.platform ): Promise { - return parseDevinSessionContent(file, await readFile(file.path, 'utf-8'), platform) + return parseDevinSessionContent( + file, + await wslGatedReadFile(file.path, 'utf-8', 'scan'), + platform + ) } export function parseDevinSessionContent( diff --git a/src/main/ai-vault/session-scanner-discovery-wsl-gate.test.ts b/src/main/ai-vault/session-scanner-discovery-wsl-gate.test.ts index ce24afccc2c..962a2e20fa9 100644 --- a/src/main/ai-vault/session-scanner-discovery-wsl-gate.test.ts +++ b/src/main/ai-vault/session-scanner-discovery-wsl-gate.test.ts @@ -1,6 +1,54 @@ -import { describe, expect, it } from 'vitest' -import { WslTranscriptFsError } from '../native-chat/wsl-transcript-fs-gate' -import { walkSessionFiles } from './session-scanner-discovery' +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import type { Dirent } from 'node:fs' +import type * as NodeFsPromisesModule from 'node:fs/promises' +import type { AiVaultScanIssue } from '../../shared/ai-vault-types' + +const fsMocks = vi.hoisted(() => ({ readdir: vi.fn(), stat: vi.fn() })) + +vi.mock('node:fs/promises', async (importOriginal) => ({ + ...(await importOriginal()), + readdir: fsMocks.readdir, + stat: fsMocks.stat +})) + +import { + WSL_TRANSCRIPT_FS_SCAN_TIMEOUT_MS, + WslTranscriptFsError +} from '../native-chat/wsl-transcript-fs-gate' +import { discoverFiles, walkSessionFiles } from './session-scanner-discovery' + +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.' + +function dirent(name: string): Dirent { + return { name, isDirectory: () => false, isFile: () => true } as Dirent +} + +let releaseStall: (() => void) | undefined + +function stalls(): Promise { + return new Promise((resolve) => { + releaseStall = () => resolve(undefined as T) + }) +} + +beforeEach(() => { + fsMocks.readdir.mockReset() + fsMocks.stat.mockReset() + fsMocks.readdir.mockResolvedValue([]) + releaseStall = undefined +}) + +// A stalled task holds its gate permit until the call settles; releasing it +// keeps the next case from fast-failing on an already-exhausted gate. +afterEach(async () => { + if (!vi.isFakeTimers()) { + return + } + releaseStall?.() + await vi.advanceTimersByTimeAsync(0) + vi.useRealTimers() +}) describe('walkSessionFiles WSL gate refusals', () => { it('rethrows gate refusals instead of reporting an empty tree', async () => { @@ -26,3 +74,64 @@ describe('walkSessionFiles WSL gate refusals', () => { ).resolves.toEqual([]) }) }) + +describe('discoverFiles containment for a stalled WSL root', () => { + const UNC_ROOT = '\\\\wsl.localhost\\Ubuntu\\home\\ada\\.codex\\sessions' + + it('contains a stalled readdir to one root so a Promise.all fan-out still resolves', async () => { + vi.useFakeTimers() + { + fsMocks.readdir.mockImplementation((dir: string) => + dir === UNC_ROOT ? stalls() : Promise.resolve([]) + ) + const issues: AiVaultScanIssue[] = [] + const discovery = discoverFiles({ + rootDir: UNC_ROOT, + limit: 10, + agent: 'codex', + issues, + extensions: ['.jsonl'] + }) + const healthy = discoverFiles({ + rootDir: '/home/ada/.codex/sessions', + limit: 10, + agent: 'codex', + issues: [], + extensions: ['.jsonl'] + }) + + await vi.advanceTimersByTimeAsync(WSL_TRANSCRIPT_FS_SCAN_TIMEOUT_MS + 1) + + await expect(Promise.all([discovery, healthy])).resolves.toEqual([ + { agent: 'codex', rootDir: UNC_ROOT, files: [] }, + { agent: 'codex', rootDir: '/home/ada/.codex/sessions', files: [] } + ]) + expect(issues).toEqual([{ agent: 'codex', path: UNC_ROOT, message: SLOW_MESSAGE }]) + } + }) + + it('reports every stalled file without spending one deadline per file', async () => { + vi.useFakeTimers() + { + const names = ['a', 'b', 'c', 'd', 'e'].map((name) => dirent(`${name}.jsonl`)) + fsMocks.readdir.mockResolvedValue(names) + fsMocks.stat.mockImplementation(stalls) + const issues: AiVaultScanIssue[] = [] + const discovery = discoverFiles({ + rootDir: UNC_ROOT, + limit: 10, + agent: 'codex', + issues, + extensions: ['.jsonl'] + }) + + // Two deadlines total, not five: once the first stalled stat is marked + // stuck, every later file fast-fails on the route instead of queueing. + await vi.advanceTimersByTimeAsync(WSL_TRANSCRIPT_FS_SCAN_TIMEOUT_MS * 2 + 2) + + await expect(discovery).resolves.toEqual({ agent: 'codex', rootDir: UNC_ROOT, files: [] }) + expect(issues).toHaveLength(names.length) + expect(issues.every((issue) => issue.path.startsWith(UNC_ROOT))).toBe(true) + } + }) +}) diff --git a/src/main/ai-vault/session-scanner-discovery.ts b/src/main/ai-vault/session-scanner-discovery.ts index 7f72082a266..7a6e824e661 100644 --- a/src/main/ai-vault/session-scanner-discovery.ts +++ b/src/main/ai-vault/session-scanner-discovery.ts @@ -1,7 +1,7 @@ import type { Dirent } from 'node:fs' -import { readdir, stat } from 'node:fs/promises' 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 type { FileWithMtime, SessionFileDiscovery } from './session-scanner-types' import { errorMessage } from './session-scanner-values' @@ -15,15 +15,27 @@ export async function discoverFiles(args: { filePredicate?: (path: string) => boolean directoryPredicate?: (name: string, depth: number) => boolean }): Promise { - const paths = await walkSessionFiles(args.rootDir, args.agent, args.issues, { - extensions: new Set(args.extensions), - filePredicate: args.filePredicate, - directoryPredicate: args.directoryPredicate - }) + let paths: string[] + try { + paths = await walkSessionFiles(args.rootDir, args.agent, args.issues, { + extensions: new Set(args.extensions), + filePredicate: args.filePredicate, + directoryPredicate: args.directoryPredicate + }) + } catch (err) { + // Why: discoverAiVaultSessionSources fans out with Promise.all, so one + // stalled distro would otherwise reject the whole vault scan — including + // every healthy local agent. Contain it to this root. + if (!(err instanceof WslTranscriptFsError)) { + throw err + } + args.issues.push({ agent: args.agent, path: args.rootDir, message: err.message }) + return { agent: args.agent, rootDir: args.rootDir, files: [] } + } const files: FileWithMtime[] = [] for (const path of paths) { try { - const fileStat = await stat(path) + const fileStat = await wslGatedStat(path, 'scan') files.push({ path, mtimeMs: fileStat.mtimeMs, @@ -64,7 +76,7 @@ export async function walkSessionFiles( try { entries = options.readDirectory ? await options.readDirectory(dirPath) - : await readdir(dirPath, { withFileTypes: true }) + : await wslGatedReaddir(dirPath, 'scan', options.signal) } catch (error) { options.signal?.throwIfAborted() // Why: a gate refusal means the scan could not run, not that the tree is diff --git a/src/main/ai-vault/session-scanner-droid-parser.ts b/src/main/ai-vault/session-scanner-droid-parser.ts index 557920e7f40..f06051823d3 100644 --- a/src/main/ai-vault/session-scanner-droid-parser.ts +++ b/src/main/ai-vault/session-scanner-droid-parser.ts @@ -1,5 +1,5 @@ import { remoteSessionContentLines } from './remote-session-content-lines' -import { createReadStream } from 'node:fs' +import { openTranscriptReadStream } from '../native-chat/wsl-transcript-fs-access' import { createInterface } from 'node:readline' import type { AiVaultSession } from '../../shared/ai-vault-types' import type { ExecutionHostId } from '../../shared/execution-host' @@ -35,7 +35,7 @@ export async function parseDroidSessionFile( platform: NodeJS.Platform = process.platform ): Promise { const lines = createInterface({ - input: createReadStream(file.path, { encoding: 'utf-8' }), + input: openTranscriptReadStream(file.path, { encoding: 'utf-8' }, 'scan'), crlfDelay: Infinity }) return parseDroidSessionLines({ file, lines, platform }) diff --git a/src/main/ai-vault/session-scanner-fs-import-guard.test.ts b/src/main/ai-vault/session-scanner-fs-import-guard.test.ts new file mode 100644 index 00000000000..472ca932eac --- /dev/null +++ b/src/main/ai-vault/session-scanner-fs-import-guard.test.ts @@ -0,0 +1,81 @@ +import { readFile, readdir } from 'node:fs/promises' +import { join } from 'node:path' +import { describe, expect, it } from 'vitest' + +const AI_VAULT_DIR = join(import.meta.dirname, '.') +const NATIVE_CHAT_DIR = join(import.meta.dirname, '..', 'native-chat') + +// Transcript I/O in these modules must go through the WSL admission gate, so a +// stalled 9P mount cannot hang a scan or a Native Chat load (STA-4049). +const NATIVE_CHAT_MODULES = [ + 'transcript-read-cache.ts', + 'transcript-reader.ts', + 'transcript-tail-reader.ts', + 'transcript-file-version.ts', + 'transcript-incremental-reader.ts', + 'transcript-watch-engine.ts' +] + +// Adding to this list must be a reviewed act: each entry names why the file +// touches a path that can never be a WSL UNC transcript. +const ALLOWLIST = new Set([ + // `existsSync` on bundled module paths inside the app dir. + 'session-scanner-service-entry-path.ts', + 'session-scanner-service-spawn.ts', + 'session-scanner-worker-spawn.ts', + 'session-scanner-opencode-sqlite-worker-spawn.ts', + // On-demand IPC readers, gated in the STA-4049 follow-up. + 'session-scanner-claude-subagents.ts', + 'session-scanner-omp-subagent-listing.ts', + // Test-only fixture builder. + 'session-scanner-test-fixtures.ts' +]) + +const FS_IMPORT = /import\s+([\s\S]*?)\s+from\s+['"]node:fs(?:\/promises)?['"]/g + +function valueImportsOfNodeFs(source: string): string[] { + return [...source.matchAll(FS_IMPORT)] + .map((match) => match[1]!.trim()) + .filter((clause) => !clause.startsWith('type')) +} + +async function guardedFiles(): Promise<{ name: string; path: string }[]> { + const aiVault = (await readdir(AI_VAULT_DIR)) + .filter((name) => name.startsWith('session-scanner') || name === 'session-title-file-reader.ts') + .filter((name) => name.endsWith('.ts') && !name.endsWith('.test.ts')) + .filter((name) => !ALLOWLIST.has(name)) + .map((name) => ({ name, path: join(AI_VAULT_DIR, name) })) + return [ + ...aiVault, + ...NATIVE_CHAT_MODULES.map((name) => ({ name, path: join(NATIVE_CHAT_DIR, name) })) + ] +} + +describe('WSL transcript gate import guard', () => { + it('leaves no raw node:fs value import in a gated transcript module', async () => { + const files = await guardedFiles() + const offenders: string[] = [] + for (const file of files) { + for (const clause of valueImportsOfNodeFs(await readFile(file.path, 'utf-8'))) { + offenders.push(`${file.name}: import ${clause} from 'node:fs…'`) + } + } + expect(offenders).toEqual([]) + }) + + it('actually covers the modules it claims to', async () => { + const names = (await guardedFiles()).map((file) => file.name) + expect(names).toEqual(expect.arrayContaining(NATIVE_CHAT_MODULES)) + // The V12–V17 sites the previous `*parser*` wording silently skipped. + expect(names).toEqual( + expect.arrayContaining([ + 'session-title-file-reader.ts', + 'session-scanner.ts', + 'session-scanner-values.ts', + 'session-scanner-codex-title-index.ts', + 'session-scanner-kimi-paths.ts', + 'session-scanner-opencode-sources.ts' + ]) + ) + }) +}) diff --git a/src/main/ai-vault/session-scanner-gemini-parsers.ts b/src/main/ai-vault/session-scanner-gemini-parsers.ts index c3cda8856bb..d8d8a485d8f 100644 --- a/src/main/ai-vault/session-scanner-gemini-parsers.ts +++ b/src/main/ai-vault/session-scanner-gemini-parsers.ts @@ -1,6 +1,5 @@ import { remoteSessionContentLines } from './remote-session-content-lines' -import { createReadStream } from 'node:fs' -import { readFile } from 'node:fs/promises' +import { openTranscriptReadStream, wslGatedReadFile } from '../native-chat/wsl-transcript-fs-access' import { createInterface } from 'node:readline' import type { AiVaultSession } from '../../shared/ai-vault-types' import type { @@ -34,7 +33,11 @@ export async function parseGeminiSessionFile( return parseGeminiJsonlSessionFile(file, platform) } - return parseGeminiJsonSessionContent(file, await readFile(file.path, 'utf-8'), platform) + return parseGeminiJsonSessionContent( + file, + await wslGatedReadFile(file.path, 'utf-8', 'scan'), + platform + ) } export async function parseGeminiSessionContent( @@ -83,7 +86,7 @@ export async function parseGeminiJsonlSessionFile( platform: NodeJS.Platform ): Promise { const lines = createInterface({ - input: createReadStream(file.path, { encoding: 'utf-8' }), + input: openTranscriptReadStream(file.path, { encoding: 'utf-8' }, 'scan'), crlfDelay: Infinity }) return parseGeminiJsonlSessionLines({ file, lines, platform }) diff --git a/src/main/ai-vault/session-scanner-graph-parsers.ts b/src/main/ai-vault/session-scanner-graph-parsers.ts index 250ebaf9ac9..4d53f366bc7 100644 --- a/src/main/ai-vault/session-scanner-graph-parsers.ts +++ b/src/main/ai-vault/session-scanner-graph-parsers.ts @@ -1,6 +1,5 @@ import { remoteSessionContentLines } from './remote-session-content-lines' -import { createReadStream } from 'node:fs' -import { readFile } from 'node:fs/promises' +import { openTranscriptReadStream, wslGatedReadFile } from '../native-chat/wsl-transcript-fs-access' import { basename, dirname, join } from 'node:path' import { createInterface } from 'node:readline' import type { AiVaultSession } from '../../shared/ai-vault-types' @@ -41,7 +40,9 @@ export async function parseRovoSessionFile( file: FileWithMtime, platform: NodeJS.Platform = process.platform ): Promise { - const metadata = asRecord(JSON.parse(await readFile(file.path, 'utf-8')) as unknown) + const metadata = asRecord( + JSON.parse(await wslGatedReadFile(file.path, 'utf-8', 'scan')) as unknown + ) if (!metadata) { return null } @@ -176,7 +177,7 @@ export async function parseMessageGraphSessionFile( platform: NodeJS.Platform = process.platform ): Promise { const lines = createInterface({ - input: createReadStream(file.path, { encoding: 'utf-8' }), + input: openTranscriptReadStream(file.path, { encoding: 'utf-8' }, 'scan'), crlfDelay: Infinity }) return parseMessageGraphSessionLines({ agent, file, lines, platform }) diff --git a/src/main/ai-vault/session-scanner-grok-parser.ts b/src/main/ai-vault/session-scanner-grok-parser.ts index 46ed29e8f1f..05a1bdfe6da 100644 --- a/src/main/ai-vault/session-scanner-grok-parser.ts +++ b/src/main/ai-vault/session-scanner-grok-parser.ts @@ -1,5 +1,4 @@ -import { createReadStream } from 'node:fs' -import { readFile } from 'node:fs/promises' +import { openTranscriptReadStream, wslGatedReadFile } from '../native-chat/wsl-transcript-fs-access' import { dirname, join } from 'node:path' import { createInterface } from 'node:readline' import type { AiVaultSession } from '../../shared/ai-vault-types' @@ -36,7 +35,7 @@ export async function parseGrokSessionFile( file: FileWithMtime, platform: NodeJS.Platform = process.platform ): Promise { - const record = asRecord(JSON.parse(await readFile(file.path, 'utf-8')) as unknown) + const record = asRecord(JSON.parse(await wslGatedReadFile(file.path, 'utf-8', 'scan')) as unknown) if (!record) { return null } @@ -64,7 +63,11 @@ async function consumeGrokChatHistory( ): Promise { try { const lines = createInterface({ - input: createReadStream(join(sessionDir, 'chat_history.jsonl'), { encoding: 'utf-8' }), + input: openTranscriptReadStream( + join(sessionDir, 'chat_history.jsonl'), + { encoding: 'utf-8' }, + 'scan' + ), crlfDelay: Infinity }) diff --git a/src/main/ai-vault/session-scanner-hermes-parser.ts b/src/main/ai-vault/session-scanner-hermes-parser.ts index be844cc6627..017fd9c8fdc 100644 --- a/src/main/ai-vault/session-scanner-hermes-parser.ts +++ b/src/main/ai-vault/session-scanner-hermes-parser.ts @@ -1,4 +1,4 @@ -import { readFile } from 'node:fs/promises' +import { wslGatedReadFile } from '../native-chat/wsl-transcript-fs-access' import type { AiVaultSession } from '../../shared/ai-vault-types' import type { ExecutionHostId } from '../../shared/execution-host' import type { FileWithMtime } from './session-scanner-types' @@ -26,7 +26,11 @@ export async function parseHermesSessionFile( file: FileWithMtime, platform: NodeJS.Platform = process.platform ): Promise { - return parseHermesSessionContent(file, await readFile(file.path, 'utf-8'), platform) + return parseHermesSessionContent( + file, + await wslGatedReadFile(file.path, 'utf-8', 'scan'), + platform + ) } export async function parseHermesSessionContent( diff --git a/src/main/ai-vault/session-scanner-index-cache-wsl-stall.test.ts b/src/main/ai-vault/session-scanner-index-cache-wsl-stall.test.ts new file mode 100644 index 00000000000..093a5a83628 --- /dev/null +++ b/src/main/ai-vault/session-scanner-index-cache-wsl-stall.test.ts @@ -0,0 +1,124 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import type * as NodeFsPromisesModule from 'node:fs/promises' + +const CODEX_HOME = '\\\\wsl.localhost\\Ubuntu\\home\\ada\\.codex' +const CODEX_SESSION_FILE = `${CODEX_HOME}\\sessions\\2026\\01\\01\\rollout-1.jsonl` +const SESSION_ID = 'aaaaaaaa-bbbb-4ccc-8ddd-eeeeeeeeeeee' +const KIMI_INDEX = '\\\\wsl.localhost\\Ubuntu\\home\\ada\\.kimi\\session_index.jsonl' + +const mocks = vi.hoisted(() => ({ stat: vi.fn(), open: vi.fn(), readFile: vi.fn() })) + +vi.mock('node:fs/promises', async (importOriginal) => ({ + ...(await importOriginal()), + stat: mocks.stat, + open: mocks.open, + readFile: mocks.readFile +})) + +import { + readCodexSessionIndexTitle, + resetCodexSessionIndexTitleCacheForTests, + _hasCodexSessionIndexTitleCacheEntryForTest +} from './session-scanner-codex-title-index' +import { + clearKimiSessionIndexCache, + hasKimiSessionIndexCacheEntryForTests, + readKimiWorkDirBySessionId +} from './session-scanner-kimi-paths' +import { readJsonObjectIfExists } from './session-scanner-values' +import { WSL_TRANSCRIPT_FS_SCAN_TIMEOUT_MS } from '../native-chat/wsl-transcript-fs-gate' + +// Identity must not change between the refused and the recovered read, or the +// cache would miss for that reason instead of because the entry was evicted. +const INDEX_STATS = { size: 128, mtimeMs: 7, ctimeMs: 7 } + +let releaseStall: (() => void) | undefined + +function stalls(): Promise { + return new Promise((resolve) => { + releaseStall = () => resolve(undefined as never) + }) +} + +function servingHandle(body: string) { + const bytes = Buffer.from(body) + return { + read: vi.fn(async (buffer: Buffer, offset: number, length: number, position: number) => { + const slice = bytes.subarray(position, Math.min(position + length, bytes.length)) + slice.copy(buffer, offset) + return { bytesRead: slice.length, buffer } + }), + close: vi.fn(async () => {}) + } +} + +async function releaseAndSettle(): Promise { + releaseStall?.() + releaseStall = undefined + await vi.advanceTimersByTimeAsync(0) +} + +beforeEach(() => { + resetCodexSessionIndexTitleCacheForTests() + clearKimiSessionIndexCache() + mocks.stat.mockReset() + mocks.open.mockReset() + mocks.readFile.mockReset() + mocks.stat.mockResolvedValue(INDEX_STATS) + releaseStall = undefined + vi.useFakeTimers() +}) + +afterEach(async () => { + await releaseAndSettle() + vi.useRealTimers() +}) + +describe('memoized WSL session indexes under a stalled mount', () => { + it('evicts the Codex title index instead of pinning "no titles"', async () => { + mocks.open.mockResolvedValue({ read: vi.fn(stalls), close: vi.fn(async () => {}) }) + const refused = readCodexSessionIndexTitle(CODEX_SESSION_FILE, CODEX_HOME, SESSION_ID) + await vi.advanceTimersByTimeAsync(WSL_TRANSCRIPT_FS_SCAN_TIMEOUT_MS + 1) + + expect(await refused).toBeNull() + expect(_hasCodexSessionIndexTitleCacheEntryForTest(CODEX_HOME)).toBe(false) + + await releaseAndSettle() + mocks.open.mockResolvedValue( + servingHandle(`${JSON.stringify({ id: SESSION_ID, thread_name: 'Ship the gate' })}\n`) + ) + + // Same size/mtime as the refused read: only the eviction can produce a hit. + expect(await readCodexSessionIndexTitle(CODEX_SESSION_FILE, CODEX_HOME, SESSION_ID)).toBe( + 'Ship the gate' + ) + }) + + it('evicts the Kimi work-dir index instead of pinning "no cwd"', async () => { + mocks.open.mockResolvedValue({ read: vi.fn(stalls), close: vi.fn(async () => {}) }) + const refused = readKimiWorkDirBySessionId(KIMI_INDEX) + await vi.advanceTimersByTimeAsync(WSL_TRANSCRIPT_FS_SCAN_TIMEOUT_MS + 1) + + expect(await refused).toEqual(new Map()) + expect(hasKimiSessionIndexCacheEntryForTests(KIMI_INDEX)).toBe(false) + + await releaseAndSettle() + mocks.open.mockResolvedValue( + servingHandle(`${JSON.stringify({ sessionId: SESSION_ID, workDir: '/repo/app' })}\n`) + ) + + expect(await readKimiWorkDirBySessionId(KIMI_INDEX)).toEqual( + new Map([[SESSION_ID, '/repo/app']]) + ) + }) + + it('leaves optional JSON enrichment absent rather than failing the session', async () => { + mocks.readFile.mockImplementation(stalls) + const pending = readJsonObjectIfExists(`${CODEX_HOME}\\history.json`) + await vi.advanceTimersByTimeAsync(WSL_TRANSCRIPT_FS_SCAN_TIMEOUT_MS + 1) + + // `null`, not a throw: `parseSessionCandidate` would otherwise drop the + // whole session instead of listing it un-enriched. + expect(await pending).toBeNull() + }) +}) diff --git a/src/main/ai-vault/session-scanner-kimi-parser.ts b/src/main/ai-vault/session-scanner-kimi-parser.ts index 61341098e0d..018a86bfbb6 100644 --- a/src/main/ai-vault/session-scanner-kimi-parser.ts +++ b/src/main/ai-vault/session-scanner-kimi-parser.ts @@ -1,5 +1,4 @@ -import { createReadStream } from 'node:fs' -import { readFile } from 'node:fs/promises' +import { openTranscriptReadStream, wslGatedReadFile } from '../native-chat/wsl-transcript-fs-access' import { createInterface } from 'node:readline' import type { AiVaultSession } from '../../shared/ai-vault-types' import { @@ -36,7 +35,9 @@ export async function parseKimiSessionFile( ): Promise { let stateRecord: Record | null try { - stateRecord = asRecord(JSON.parse(await readFile(file.path, 'utf-8')) as unknown) + stateRecord = asRecord( + JSON.parse(await wslGatedReadFile(file.path, 'utf-8', 'scan')) as unknown + ) } catch { return null } @@ -85,7 +86,7 @@ async function consumeKimiWireTranscript( try { const lines = createInterface({ - input: createReadStream(wirePath, { encoding: 'utf-8' }), + input: openTranscriptReadStream(wirePath, { encoding: 'utf-8' }, 'scan'), crlfDelay: Infinity }) for await (const line of lines) { diff --git a/src/main/ai-vault/session-scanner-kimi-paths.ts b/src/main/ai-vault/session-scanner-kimi-paths.ts index 7ed457f6c07..ef19a8b45a3 100644 --- a/src/main/ai-vault/session-scanner-kimi-paths.ts +++ b/src/main/ai-vault/session-scanner-kimi-paths.ts @@ -1,8 +1,8 @@ -import { stat } from 'node:fs/promises' -import { createReadStream } from 'node:fs' import { homedir } from 'node:os' import { basename, dirname, join } from 'node:path' import { createInterface } from 'node:readline' +import { openTranscriptReadStream, wslGatedStat } from '../native-chat/wsl-transcript-fs-access' +import { WslTranscriptFsError } from '../native-chat/wsl-transcript-fs-gate' import { asRecord, extractString } from './session-scanner-values' import { KimiSessionIndexCache, @@ -78,16 +78,17 @@ export function hasKimiSessionIndexCacheEntryForTests(indexPath: string): boolea export async function readKimiWorkDirBySessionId(indexPath: string): Promise> { const generation = workDirCacheByIndexPath.beginRead() - let identity: Awaited> + let identity: Awaited> try { - identity = await stat(indexPath) + identity = await wslGatedStat(indexPath, 'scan') } catch { // Missing index (e.g. user deleted it): sessions still list, just without cwd. workDirCacheByIndexPath.delete(indexPath, generation) return new Map() } - return workDirCacheByIndexPath.get( + let refused = false + const workDirs = await workDirCacheByIndexPath.get( indexPath, { changeTimeMs: identity.ctimeMs, @@ -95,18 +96,30 @@ export async function readKimiWorkDirBySessionId(indexPath: string): Promise parseKimiSessionIndex(indexPath) + () => + parseKimiSessionIndex(indexPath).then((parsed) => { + refused = parsed.refused + return parsed.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(indexPath: string): Promise> { +async function parseKimiSessionIndex( + indexPath: string +): Promise<{ map: Map; refused: boolean }> { const map = new Map() // Why: never reject. This promise is memoized and shared by every session // under one Kimi home; a mid-read failure (file deleted after stat, EACCES) // must degrade to whatever was parsed so the other sessions still list. try { const lines = createInterface({ - input: createReadStream(indexPath, { encoding: 'utf-8' }), + input: openTranscriptReadStream(indexPath, { encoding: 'utf-8' }, 'scan'), crlfDelay: Infinity }) for await (const line of lines) { @@ -126,8 +139,9 @@ async function parseKimiSessionIndex(indexPath: string): Promise { let entries try { - entries = await readdir(ompArtifactDirFor(transcriptFilePath), { withFileTypes: true }) + entries = await wslGatedReaddir(ompArtifactDirFor(transcriptFilePath), 'scan') } catch { return 0 } diff --git a/src/main/ai-vault/session-scanner-opencode-parser.ts b/src/main/ai-vault/session-scanner-opencode-parser.ts index 4231f7d9de4..7b3b2ed3301 100644 --- a/src/main/ai-vault/session-scanner-opencode-parser.ts +++ b/src/main/ai-vault/session-scanner-opencode-parser.ts @@ -1,4 +1,4 @@ -import { readFile, readdir } from 'node:fs/promises' +import { wslGatedReaddir, wslGatedReadFile } from '../native-chat/wsl-transcript-fs-access' import { join } from 'node:path' import type { AiVaultSession } from '../../shared/ai-vault-types' import type { FileWithMtime, SessionAccumulator } from './session-scanner-types' @@ -26,7 +26,7 @@ export async function parseOpenCodeSessionFile( file: FileWithMtime, platform: NodeJS.Platform = process.platform ): Promise { - const record = asRecord(JSON.parse(await readFile(file.path, 'utf-8')) as unknown) + const record = asRecord(JSON.parse(await wslGatedReadFile(file.path, 'utf-8', 'scan')) as unknown) if (!record) { return null } @@ -47,7 +47,7 @@ export async function parseOpenCodeSessionFile( async function readOpenCodeMessagesInOrder(messageDir: string): Promise[]> { let entries try { - entries = await readdir(messageDir, { withFileTypes: true }) + entries = await wslGatedReaddir(messageDir, 'scan') } catch { return [] } @@ -59,7 +59,7 @@ async function readOpenCodeMessagesInOrder(messageDir: string): Promise | null = null try { message = asRecord( - JSON.parse(await readFile(join(messageDir, entry.name), 'utf-8')) as unknown + JSON.parse(await wslGatedReadFile(join(messageDir, entry.name), 'utf-8', 'scan')) as unknown ) } catch { // A live OpenCode process can leave a half-written file; skip it rather diff --git a/src/main/ai-vault/session-scanner-opencode-sources-wsl-stall.test.ts b/src/main/ai-vault/session-scanner-opencode-sources-wsl-stall.test.ts new file mode 100644 index 00000000000..b94379b64dc --- /dev/null +++ b/src/main/ai-vault/session-scanner-opencode-sources-wsl-stall.test.ts @@ -0,0 +1,73 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import type { Dirent } from 'node:fs' +import type * as NodeFsPromisesModule from 'node:fs/promises' +import type { AiVaultScanIssue } from '../../shared/ai-vault-types' + +const WSL_HOME = '\\\\wsl.localhost\\Ubuntu\\home\\ada' +const WSL_DATA_DIR = `${WSL_HOME}/.local/share/opencode` + +const mocks = vi.hoisted(() => ({ readdir: vi.fn(), stat: vi.fn() })) + +vi.mock('node:fs/promises', async (importOriginal) => ({ + ...(await importOriginal()), + readdir: mocks.readdir, + stat: mocks.stat +})) + +// The SQLite leg spawns a real worker thread, which fake timers cannot drive. +vi.mock('./session-scanner-opencode-sqlite-worker-spawn', () => ({ + listOpenCodeSqliteSessionsViaWorker: async () => [] +})) + +import { opencodeDiscoveries } from './session-scanner-opencode-sources' +import { WSL_TRANSCRIPT_FS_SCAN_TIMEOUT_MS } from '../native-chat/wsl-transcript-fs-gate' + +let releaseStall: (() => void) | undefined + +function stalls(): Promise { + return new Promise((resolve) => { + releaseStall = () => resolve(undefined as never) + }) +} + +beforeEach(() => { + mocks.readdir.mockReset() + mocks.stat.mockReset() + releaseStall = undefined + vi.useFakeTimers() +}) + +afterEach(async () => { + releaseStall?.() + releaseStall = undefined + await vi.advanceTimersByTimeAsync(0) + vi.useRealTimers() +}) + +describe('OpenCode source discovery with a stalled WSL data directory', () => { + it('reports the stalled home as an issue and still resolves every discovery', async () => { + mocks.readdir.mockImplementation((dir: string) => + dir === WSL_DATA_DIR ? stalls() : Promise.resolve([] as Dirent[]) + ) + const issues: AiVaultScanIssue[] = [] + const discoveries = Promise.all( + opencodeDiscoveries( + { opencodeStorageDir: '/home/ada/.local/share/opencode/storage' }, + [WSL_HOME], + 10, + issues + ) + ) + // Two deadlines: the data-dir probe, then the storage scan that queued + // behind it before the route was flagged stuck. + await vi.advanceTimersByTimeAsync(WSL_TRANSCRIPT_FS_SCAN_TIMEOUT_MS * 2 + 2) + + // Zero databases for that home is the degraded answer; without the issue it + // would be indistinguishable from "OpenCode was never installed there". + const resolved = await discoveries + expect(resolved).toHaveLength(2) + expect(resolved.every((discovery) => discovery.files.length === 0)).toBe(true) + expect(issues.some((issue) => issue.path === WSL_DATA_DIR)).toBe(true) + expect(issues.every((issue) => issue.agent === 'opencode')).toBe(true) + }) +}) diff --git a/src/main/ai-vault/session-scanner-opencode-sources.ts b/src/main/ai-vault/session-scanner-opencode-sources.ts index 6d84965f8c0..1e55e5dffde 100644 --- a/src/main/ai-vault/session-scanner-opencode-sources.ts +++ b/src/main/ai-vault/session-scanner-opencode-sources.ts @@ -1,6 +1,7 @@ -import { readdir } from 'node:fs/promises' import { dirname, join } from 'node:path' import type { AiVaultScanIssue } from '../../shared/ai-vault-types' +import { wslGatedReaddir } from '../native-chat/wsl-transcript-fs-access' +import { WslTranscriptFsError } from '../native-chat/wsl-transcript-fs-gate' import { resolveOpenCodeStorageDirectory } from '../opencode/opencode-data-directory' import { listOpenCodeDatabases } from '../opencode-usage/scanner' import { discoverOpenCodeSessions } from './session-scanner-opencode-sqlite-discovery' @@ -16,7 +17,7 @@ export function opencodeDiscoveries( return storageDirs.map(async (storageDir, index) => discoverOpenCodeSessions({ storageDir, - dbPaths: await opencodeDbPathsForSource(options, wslHomeDirs, storageDir, index), + dbPaths: await opencodeDbPathsForSource(options, wslHomeDirs, storageDir, index, issues), limitPerAgent: limit, issues }) @@ -37,32 +38,41 @@ async function opencodeDbPathsForSource( options: AiVaultScanOptions, wslHomeDirs: readonly string[], storageDir: string, - sourceIndex: number + sourceIndex: number, + issues: AiVaultScanIssue[] ): Promise { if (options.opencodeDbPaths) { return sourceIndex === 0 ? options.opencodeDbPaths : [] } // Why: custom OpenCode storage roots still keep SQLite DBs in the parent data dir. if (sourceIndex === 0 && options.opencodeStorageDir) { - return listOpenCodeDatabasesInDirectory(dirname(storageDir)) + return listOpenCodeDatabasesInDirectory(dirname(storageDir), issues) } if (sourceIndex === 0) { return listOpenCodeDatabases() } const wslHomeDir = wslHomeDirs[sourceIndex - 1] return wslHomeDir - ? listOpenCodeDatabasesInDirectory(join(wslHomeDir, '.local', 'share', 'opencode')) + ? listOpenCodeDatabasesInDirectory(join(wslHomeDir, '.local', 'share', 'opencode'), issues) : [] } -async function listOpenCodeDatabasesInDirectory(dataDir: string): Promise { +async function listOpenCodeDatabasesInDirectory( + dataDir: string, + issues: AiVaultScanIssue[] +): Promise { try { - const entries = await readdir(dataDir, { withFileTypes: true }) + const entries = await wslGatedReaddir(dataDir, 'scan') return entries .filter((entry) => entry.isFile() && /^opencode(?:-[A-Za-z0-9_.-]+)?\.db$/.test(entry.name)) .map((entry) => join(dataDir, entry.name)) .sort() - } catch { + } catch (error) { + // A stalled WSL data dir still degrades to "no databases", but the gap has + // to be reportable — an empty list otherwise reads as "OpenCode not used". + if (error instanceof WslTranscriptFsError) { + issues.push({ agent: 'opencode', path: dataDir, message: error.message }) + } return [] } } diff --git a/src/main/ai-vault/session-scanner-parse-cache.ts b/src/main/ai-vault/session-scanner-parse-cache.ts index 727d70ac319..563055ffa39 100644 --- a/src/main/ai-vault/session-scanner-parse-cache.ts +++ b/src/main/ai-vault/session-scanner-parse-cache.ts @@ -1,5 +1,7 @@ -import { createReadStream } from 'node:fs' -import { open } from 'node:fs/promises' +import { + openTranscriptReadStream, + readTranscriptSlice +} from '../native-chat/wsl-transcript-fs-access' import type { AiVaultSession } from '../../shared/ai-vault-types' import { createAntigravitySessionResumeState } from './session-scanner-antigravity-parser' import { parseAgentSessionFile } from './session-scanner-agent-parser' @@ -305,13 +307,8 @@ async function parseResumableCandidate(args: { // agent transcripts are append-only so that trade is accepted (worst case is // a stale vault row until the file is next truncated or the app restarts). async function endsWithNewlineAt(path: string, offset: number): Promise { - const handle = await open(path, 'r') - try { - const { bytesRead, buffer } = await handle.read(Buffer.alloc(1), 0, 1, offset - 1) - return bytesRead === 1 && buffer[0] === NEWLINE_BYTE - } finally { - await handle.close() - } + const slice = await readTranscriptSlice(path, offset - 1, 1, 'scan') + return slice.length === 1 && slice[0] === NEWLINE_BYTE } type JsonlReadResult = { @@ -336,7 +333,7 @@ async function consumeCompleteJsonlLines(args: { let remainderParts: Buffer[] = [] let remainderLength = 0 - const stream = createReadStream(args.path, { start: args.start }) + const stream = openTranscriptReadStream(args.path, { start: args.start }, 'scan') for await (const chunk of stream as AsyncIterable) { bytesRead += chunk.length // Why check the chunk alone: the pieces held over are all mid-line, so none diff --git a/src/main/ai-vault/session-scanner-parse-wsl-stall.test.ts b/src/main/ai-vault/session-scanner-parse-wsl-stall.test.ts new file mode 100644 index 00000000000..a79063194a8 --- /dev/null +++ b/src/main/ai-vault/session-scanner-parse-wsl-stall.test.ts @@ -0,0 +1,149 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import type * as NodeFsPromisesModule from 'node:fs/promises' +import type { SessionFileCandidate } from './session-scanner-types' + +const STALLED_PATH = '\\\\wsl.localhost\\Ubuntu\\home\\ada\\.claude\\projects\\p\\a.jsonl' +const SIBLING_PATH = '\\\\wsl.localhost\\Debian\\home\\ada\\.claude\\projects\\p\\b.jsonl' + +const mocks = vi.hoisted(() => ({ open: vi.fn() })) + +vi.mock('node:fs/promises', async (importOriginal) => ({ + ...(await importOriginal()), + open: mocks.open +})) + +import { + parseAgentSessionFileCached, + resetSessionParseCacheForTests +} from './session-scanner-parse-cache' +import { + WSL_TRANSCRIPT_FS_SCAN_TIMEOUT_MS, + WslTranscriptFsError +} from '../native-chat/wsl-transcript-fs-gate' + +type ReadResult = { bytesRead: number; buffer: Buffer } + +// A stalled task keeps the gate's single scan slot until the call settles, so +// every case releases its stall before the next one runs. +let releaseStall: (() => void) | undefined + +function stalls(): Promise { + return new Promise((resolve) => { + releaseStall = () => resolve({ bytesRead: 0, buffer: Buffer.alloc(0) }) + }) +} + +function servingHandle(body: Buffer) { + return { + read: vi.fn(async (buffer: Buffer, offset: number, length: number, position: number) => { + const slice = body.subarray(position, Math.min(position + length, body.length)) + slice.copy(buffer, offset) + return { bytesRead: slice.length, buffer } + }), + close: vi.fn(async () => {}) + } +} + +function record(index: number, text: string): string { + return JSON.stringify({ + type: 'user', + sessionId: 'aaaaaaaa-bbbb-4ccc-8ddd-eeeeeeeeeeee', + timestamp: new Date(1740000000000 + index * 60_000).toISOString(), + cwd: '/repo/app', + gitBranch: 'main', + message: { role: 'user', content: text } + }) +} + +function body(count: number): Buffer { + return Buffer.from( + Array.from({ length: count }, (_, index) => `${record(index, `line ${index}`)}\n`).join('') + ) +} + +function candidate(path: string, bytes: Buffer, mtimeMs: number): SessionFileCandidate { + return { + agent: 'claude', + file: { + path, + mtimeMs, + modifiedAt: new Date(mtimeMs).toISOString(), + sizeBytes: bytes.length + }, + codexHome: null + } +} + +async function releaseAndSettle(): Promise { + releaseStall?.() + releaseStall = undefined + await vi.advanceTimersByTimeAsync(0) +} + +beforeEach(() => { + resetSessionParseCacheForTests() + mocks.open.mockReset() + releaseStall = undefined + vi.useFakeTimers() +}) + +afterEach(async () => { + await releaseAndSettle() + vi.useRealTimers() +}) + +describe('AI Vault session parse with a stalled WSL transcript body read', () => { + it('refuses the stalled candidate and still parses a sibling on a healthy distro', async () => { + mocks.open.mockResolvedValue({ read: vi.fn(stalls), close: vi.fn(async () => {}) }) + // Rejecting (not resolving null) is what lets `parseSessionCandidate` + // report it as an issue instead of silently dropping the session. The + // assertion is attached before the clock moves so the rejection is never + // momentarily unhandled. + const refusal = expect( + parseAgentSessionFileCached(candidate(STALLED_PATH, body(2), 1), 'linux') + ).rejects.toBeInstanceOf(WslTranscriptFsError) + await vi.advanceTimersByTimeAsync(WSL_TRANSCRIPT_FS_SCAN_TIMEOUT_MS + 1) + await refusal + + await releaseAndSettle() + const siblingBytes = body(2) + mocks.open.mockResolvedValue(servingHandle(siblingBytes)) + + const sibling = await parseAgentSessionFileCached( + candidate(SIBLING_PATH, siblingBytes, 1), + 'linux' + ) + expect(sibling?.messageCount).toBe(2) + }) + + it('caches no resume point when the newline probe is refused mid-append', async () => { + const first = body(2) + mocks.open.mockResolvedValue(servingHandle(first)) + const seeded = await parseAgentSessionFileCached(candidate(STALLED_PATH, first, 1), 'linux') + expect(seeded?.messageCount).toBe(2) + + const grown = body(3) + const serving = servingHandle(grown) + mocks.open.mockResolvedValue({ + // Only the 1-byte resume probe stalls; the body read stays healthy. + read: vi.fn((buffer: Buffer, offset: number, length: number, position: number) => + length === 1 ? stalls() : serving.read(buffer, offset, length, position) + ), + close: vi.fn(async () => {}) + }) + const refused = expect( + parseAgentSessionFileCached(candidate(STALLED_PATH, grown, 2), 'linux') + ).rejects.toBeInstanceOf(WslTranscriptFsError) + await vi.advanceTimersByTimeAsync(WSL_TRANSCRIPT_FS_SCAN_TIMEOUT_MS + 1) + await refused + + await releaseAndSettle() + mocks.open.mockResolvedValue(servingHandle(grown)) + const recovered = await parseAgentSessionFileCached(candidate(STALLED_PATH, grown, 2), 'linux') + + resetSessionParseCacheForTests() + const cold = await parseAgentSessionFileCached(candidate(STALLED_PATH, grown, 2), 'linux') + expect(recovered).toEqual(cold) + expect(recovered?.messageCount).toBe(3) + }) +}) diff --git a/src/main/ai-vault/session-scanner-primary-parsers.ts b/src/main/ai-vault/session-scanner-primary-parsers.ts index 92db2241320..f554268c7e5 100644 --- a/src/main/ai-vault/session-scanner-primary-parsers.ts +++ b/src/main/ai-vault/session-scanner-primary-parsers.ts @@ -1,5 +1,5 @@ import { remoteSessionContentLines } from './remote-session-content-lines' -import { createReadStream } from 'node:fs' +import { openTranscriptReadStream } from '../native-chat/wsl-transcript-fs-access' import { createInterface } from 'node:readline' import type { AiVaultSession } from '../../shared/ai-vault-types' import { LOCAL_EXECUTION_HOST_ID, type ExecutionHostId } from '../../shared/execution-host' @@ -210,7 +210,7 @@ export async function parseClaudeSessionFile( platform: NodeJS.Platform = process.platform ): Promise { const lines = createInterface({ - input: createReadStream(file.path, { encoding: 'utf-8' }), + input: openTranscriptReadStream(file.path, { encoding: 'utf-8' }, 'scan'), crlfDelay: Infinity }) return parseClaudeSessionLines({ file, lines, platform }) diff --git a/src/main/ai-vault/session-scanner-scope-discovery.ts b/src/main/ai-vault/session-scanner-scope-discovery.ts index b0a4792ffd6..3ae8a75abfa 100644 --- a/src/main/ai-vault/session-scanner-scope-discovery.ts +++ b/src/main/ai-vault/session-scanner-scope-discovery.ts @@ -1,7 +1,11 @@ -import { createReadStream } from 'node:fs' -import { readdir, stat } from 'node:fs/promises' import { createInterface } from 'node:readline' import { extname, join } from 'node:path' +import { + openTranscriptReadStream, + wslGatedReaddir, + wslGatedStat +} from '../native-chat/wsl-transcript-fs-access' +import { WslTranscriptFsError } from '../native-chat/wsl-transcript-fs-gate' import { isPathInsideOrEqual, normalizeRuntimePathSeparators @@ -26,7 +30,18 @@ export function resetProjectDirCwdCacheForTests(): void { projectDirCwdCache.clear() } -async function cachedProjectDirCwd(projectDir: string): Promise { +// A gate refusal is a stalled WSL root, not an empty one: keep the existing +// 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 }) + } +} + +async function cachedProjectDirCwd( + projectDir: string, + issues: AiVaultScanIssue[] +): Promise { const cached = projectDirCwdCache.get(projectDir) if (cached !== undefined) { // Refresh recency so hot in-scope dirs outlive one-off ones at the cap. @@ -34,7 +49,7 @@ async function cachedProjectDirCwd(projectDir: string): Promise { projectDirCwdCache.set(projectDir, cached) return cached } - const cwd = await readProjectDirCwd(projectDir) + const cwd = await readProjectDirCwd(projectDir, issues) if (cwd) { if (projectDirCwdCache.size >= PROJECT_DIR_CWD_CACHE_MAX) { const oldest = projectDirCwdCache.keys().next() @@ -70,8 +85,8 @@ export async function discoverInScopeClaudeFiles(args: { const scopeProjectPrefixes = claudeProjectScopePrefixes(args.scopePaths) const collected = new Map() for (const rootDir of args.rootDirs) { - for (const projectDir of await listProjectDirs(rootDir, scopeProjectPrefixes)) { - const cwd = await cachedProjectDirCwd(projectDir) + for (const projectDir of await listProjectDirs(rootDir, scopeProjectPrefixes, args.issues)) { + const cwd = await cachedProjectDirCwd(projectDir, args.issues) if (!cwd || !args.scopePaths.some((scopePath) => isCwdInsideScopePath(scopePath, cwd))) { continue } @@ -148,12 +163,14 @@ function isCwdInsideScopePath(scopePath: string, cwd: string): boolean { async function listProjectDirs( rootDir: string, - scopeProjectPrefixes: ReadonlySet + scopeProjectPrefixes: ReadonlySet, + issues: AiVaultScanIssue[] ): Promise { let entries try { - entries = await readdir(rootDir, { withFileTypes: true }) - } catch { + entries = await wslGatedReaddir(rootDir, 'scan') + } catch (err) { + recordRefusal(issues, rootDir, err) return [] } return entries @@ -163,10 +180,13 @@ async function listProjectDirs( .map((entry) => join(rootDir, entry.name)) } -async function readProjectDirCwd(projectDir: string): Promise { - const files = await newestClaudeFilesInDir(projectDir) +async function readProjectDirCwd( + projectDir: string, + issues: AiVaultScanIssue[] +): Promise { + const files = await newestClaudeFilesInDir(projectDir, issues) for (const file of files.slice(0, REPRESENTATIVE_FILE_LIMIT)) { - const cwd = await readFirstCwd(file) + const cwd = await readFirstCwd(file, issues) if (cwd) { return cwd } @@ -174,11 +194,15 @@ async function readProjectDirCwd(projectDir: string): Promise { return null } -async function newestClaudeFilesInDir(projectDir: string): Promise { +async function newestClaudeFilesInDir( + projectDir: string, + issues: AiVaultScanIssue[] +): Promise { let entries try { - entries = await readdir(projectDir, { withFileTypes: true }) - } catch { + entries = await wslGatedReaddir(projectDir, 'scan') + } catch (err) { + recordRefusal(issues, projectDir, err) return [] } const newest: { path: string; mtimeMs: number }[] = [] @@ -190,18 +214,19 @@ async function newestClaudeFilesInDir(projectDir: string): Promise { try { addBoundedPath(newest, REPRESENTATIVE_FILE_LIMIT, { path, - mtimeMs: (await stat(path)).mtimeMs + mtimeMs: (await wslGatedStat(path, 'scan')).mtimeMs }) - } catch { + } catch (err) { // Best effort: unreadable candidates are ignored here and reported during // full collection if the project directory proves in-scope. + recordRefusal(issues, path, err) } } return newest.sort((left, right) => right.mtimeMs - left.mtimeMs).map((value) => value.path) } -async function readFirstCwd(filePath: string): Promise { - const input = createReadStream(filePath, { encoding: 'utf-8' }) +async function readFirstCwd(filePath: string, issues: AiVaultScanIssue[]): Promise { + const input = openTranscriptReadStream(filePath, { encoding: 'utf-8' }, 'scan') const lines = createInterface({ input, crlfDelay: Infinity }) let read = 0 try { @@ -214,7 +239,8 @@ async function readFirstCwd(filePath: string): Promise { return cwd } } - } catch { + } catch (err) { + recordRefusal(issues, filePath, err) return null } finally { // readline.close() leaves the underlying stream open; destroy it so the early @@ -234,8 +260,9 @@ async function collectClaudeFiles(args: { }): Promise { let entries try { - entries = await readdir(args.projectDir, { withFileTypes: true }) - } catch { + entries = await wslGatedReaddir(args.projectDir, 'scan') + } catch (err) { + recordRefusal(args.issues, args.projectDir, err) return } for (const entry of entries) { @@ -247,7 +274,7 @@ async function collectClaudeFiles(args: { continue } try { - const fileStat = await stat(path) + const fileStat = await wslGatedStat(path, 'scan') addBoundedFile(args.collected, args.limit, { path, mtimeMs: fileStat.mtimeMs, diff --git a/src/main/ai-vault/session-scanner-subagent-transcripts.ts b/src/main/ai-vault/session-scanner-subagent-transcripts.ts index d7df4aeedac..8d477f450af 100644 --- a/src/main/ai-vault/session-scanner-subagent-transcripts.ts +++ b/src/main/ai-vault/session-scanner-subagent-transcripts.ts @@ -1,4 +1,4 @@ -import { readdir } from 'node:fs/promises' +import { wslGatedReaddir } from '../native-chat/wsl-transcript-fs-access' import { basename, dirname, extname, join } from 'node:path' // Exported so discovery can prune these subtrees using the same literal that @@ -38,7 +38,7 @@ export function subagentTranscriptsDirFor(transcriptFilePath: string): string { export async function countSubagentTranscripts(transcriptFilePath: string): Promise { let entries try { - entries = await readdir(subagentTranscriptsDirFor(transcriptFilePath), { withFileTypes: true }) + entries = await wslGatedReaddir(subagentTranscriptsDirFor(transcriptFilePath), 'scan') } catch { return 0 } diff --git a/src/main/ai-vault/session-scanner-values.ts b/src/main/ai-vault/session-scanner-values.ts index b058876fda3..073a58ff359 100644 --- a/src/main/ai-vault/session-scanner-values.ts +++ b/src/main/ai-vault/session-scanner-values.ts @@ -1,6 +1,6 @@ import { homedir } from 'node:os' import { basename, dirname, isAbsolute, join } from 'node:path' -import { readFile } from 'node:fs/promises' +import { wslGatedReadFile } from '../native-chat/wsl-transcript-fs-access' import { asRecord } from './session-scanner-record-value' export { asRecord } @@ -76,7 +76,7 @@ export async function readJsonObjectIfExists( filePath: string ): Promise | null> { try { - return asRecord(JSON.parse(await readFile(filePath, 'utf-8')) as unknown) + return asRecord(JSON.parse(await wslGatedReadFile(filePath, 'utf-8', 'scan')) as unknown) } catch { return null } diff --git a/src/main/ai-vault/session-scanner.ts b/src/main/ai-vault/session-scanner.ts index 3a98b7018ac..68f4af09ddf 100644 --- a/src/main/ai-vault/session-scanner.ts +++ b/src/main/ai-vault/session-scanner.ts @@ -1,4 +1,4 @@ -import { readFile } from 'node:fs/promises' +import { wslGatedReadFile } from '../native-chat/wsl-transcript-fs-access' import type { AiVaultListResult, AiVaultScanIssue, @@ -307,7 +307,7 @@ async function parseSessionCandidate( async function readOptionalTextFile(path: string): Promise { try { - return await readFile(path, 'utf-8') + return await wslGatedReadFile(path, 'utf-8', 'scan') } catch { return null } diff --git a/src/main/ai-vault/session-title-file-reader-wsl-stall.test.ts b/src/main/ai-vault/session-title-file-reader-wsl-stall.test.ts new file mode 100644 index 00000000000..337c2064d8a --- /dev/null +++ b/src/main/ai-vault/session-title-file-reader-wsl-stall.test.ts @@ -0,0 +1,83 @@ +import { afterEach, 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\\stalled.jsonl' +const STALLED_SESSION_ID = 'aaaaaaaa-bbbb-4ccc-8ddd-eeeeeeeeeeee' +const LOCAL_SESSION_ID = 'bbbbbbbb-cccc-4ddd-8eee-ffffffffffff' + +const mocks = vi.hoisted(() => ({ lstat: vi.fn() })) + +vi.mock('node:fs/promises', async (importOriginal) => { + const original = await importOriginal() + return { + ...original, + // Only the WSL request is faked; the local sibling reads a real temp file. + lstat: (path: string, ...rest: unknown[]) => + path.startsWith('\\\\') + ? mocks.lstat(path) + : (original.lstat as (...args: unknown[]) => unknown)(path, ...rest) + } +}) + +import { mkdtemp, rm, writeFile } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { readAiVaultSessionTitlesFromFiles } from './session-title-file-reader' +import { resetSessionParseCacheForTests } from './session-scanner-parse-cache' +import { WSL_TRANSCRIPT_FS_SCAN_TIMEOUT_MS } from '../native-chat/wsl-transcript-fs-gate' + +let releaseStall: (() => void) | undefined +let tempRoot: string | undefined + +function stalls(): Promise { + return new Promise((resolve) => { + releaseStall = () => resolve(undefined as never) + }) +} + +function userRecord(sessionId: string, text: string): string { + return JSON.stringify({ + type: 'user', + sessionId, + timestamp: new Date(1740000000000).toISOString(), + cwd: '/repo/app', + gitBranch: 'main', + message: { role: 'user', content: text } + }) +} + +beforeEach(async () => { + resetSessionParseCacheForTests() + mocks.lstat.mockReset() + releaseStall = undefined + tempRoot = await mkdtemp(join(tmpdir(), 'orca-title-stall-')) +}) + +afterEach(async () => { + releaseStall?.() + releaseStall = undefined + await vi.advanceTimersByTimeAsync(0) + vi.useRealTimers() + if (tempRoot) { + await rm(tempRoot, { recursive: true, force: true }) + tempRoot = undefined + } +}) + +describe('AI Vault session titles with a stalled WSL transcript lstat', () => { + it('drops only the stalled request and still returns the rest of the batch', async () => { + const localPath = join(tempRoot!, `${LOCAL_SESSION_ID}.jsonl`) + await writeFile(localPath, `${userRecord(LOCAL_SESSION_ID, 'hello world')}\n`) + mocks.lstat.mockImplementation(stalls) + vi.useFakeTimers() + + const pending = readAiVaultSessionTitlesFromFiles([ + { agent: 'claude', sessionId: STALLED_SESSION_ID, transcriptPath: UNC_PATH }, + { agent: 'claude', sessionId: LOCAL_SESSION_ID, transcriptPath: localPath } + ]) + await vi.advanceTimersByTimeAsync(WSL_TRANSCRIPT_FS_SCAN_TIMEOUT_MS + 1) + + const { titles } = await pending + expect(titles).toEqual([{ agent: 'claude', sessionId: LOCAL_SESSION_ID, title: 'hello world' }]) + }) +}) diff --git a/src/main/ai-vault/session-title-file-reader.ts b/src/main/ai-vault/session-title-file-reader.ts index 799d2b95da9..dd37ebed18d 100644 --- a/src/main/ai-vault/session-title-file-reader.ts +++ b/src/main/ai-vault/session-title-file-reader.ts @@ -1,4 +1,4 @@ -import { lstat } from 'node:fs/promises' +import { wslGatedLstat } from '../native-chat/wsl-transcript-fs-access' import { AI_VAULT_SESSION_TITLE_REQUEST_MAX_COUNT, type AiVaultSessionTitle, @@ -27,7 +27,7 @@ async function readOneTitle( return cache?.get(request) ?? null } try { - const stats = await lstat(transcriptPath) + const stats = await wslGatedLstat(transcriptPath, 'scan') if (!stats.isFile() || signal?.aborted) { return null } diff --git a/src/main/native-chat/transcript-file-version.ts b/src/main/native-chat/transcript-file-version.ts index 69634c0a957..ff6001baad9 100644 --- a/src/main/native-chat/transcript-file-version.ts +++ b/src/main/native-chat/transcript-file-version.ts @@ -1,4 +1,4 @@ -import { stat } from 'node:fs/promises' +import { wslGatedStat } from './wsl-transcript-fs-access' export type TranscriptFileVersion = { identity: string @@ -8,7 +8,7 @@ export type TranscriptFileVersion = { } export async function readTranscriptFileVersion(filePath: string): Promise { - const value = await stat(filePath) + const value = await wslGatedStat(filePath, 'exact') return { identity: `${value.dev}:${value.ino}`, size: value.size, diff --git a/src/main/native-chat/transcript-incremental-reader.ts b/src/main/native-chat/transcript-incremental-reader.ts index 8ad8cc335d1..75444acc9b6 100644 --- a/src/main/native-chat/transcript-incremental-reader.ts +++ b/src/main/native-chat/transcript-incremental-reader.ts @@ -1,10 +1,10 @@ -import { open, stat } from 'node:fs/promises' import type { NativeChatMessage, NativeChatTurnLifecycle } from '../../shared/native-chat-types' import { transcriptFallbackId } from './transcript-fallback-id' import { MAX_NATIVE_CHAT_TRANSCRIPT_RECORD_BYTES, type NativeChatLineDecoder } from './transcript-tail-reader' +import { openTranscriptReadStream, wslGatedStat } from './wsl-transcript-fs-access' const APPEND_BATCH_MESSAGE_LIMIT = 40 @@ -32,14 +32,13 @@ export async function readIncrementalTranscriptMessages( decodeLifecycle?: (line: string, fallbackId: string) => NativeChatTurnLifecycle | null, onLifecycle?: (lifecycle: NativeChatTurnLifecycle) => void ): Promise { - const end = (await stat(filePath)).size + const end = (await wslGatedStat(filePath, 'exact')).size if (end <= state.offset) { return [] } const messages: NativeChatMessage[] = [] - const handle = await open(filePath, 'r') + const stream = openTranscriptReadStream(filePath, { start: state.offset, end: end - 1 }, 'exact') try { - const stream = handle.createReadStream({ start: state.offset, end: end - 1, autoClose: false }) let absoluteOffset = state.offset for await (const rawChunk of stream) { const chunk = Buffer.isBuffer(rawChunk) ? rawChunk : Buffer.from(rawChunk) @@ -62,7 +61,9 @@ export async function readIncrementalTranscriptMessages( } return messages } finally { - await handle.close() + // Early exits (throw/oversized-record bail) must not leak the fd or, on + // UNC, the gated handle the generator's finally closes. + stream.destroy() } function retainPart(part: Buffer): void { 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..b4696fe0dd1 --- /dev/null +++ b/src/main/native-chat/transcript-read-cache-wsl-stall.test.ts @@ -0,0 +1,57 @@ +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() +})) + +vi.mock('./session-file-resolver', () => ({ resolveSessionFilePath: mocks.resolve })) +vi.mock('node:fs/promises', async (importOriginal) => ({ + ...(await importOriginal()), + stat: mocks.stat +})) + +import { 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.resolve.mockResolvedValue(UNC_PATH) +}) + +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 }) + + await readNativeChatTranscriptCached('claude', 'session-id') + + expect(mocks.stat).toHaveBeenCalledWith(healthyPath) + }) +}) diff --git a/src/main/native-chat/transcript-read-cache.ts b/src/main/native-chat/transcript-read-cache.ts index d2a4b4d3de2..fdc9041032d 100644 --- a/src/main/native-chat/transcript-read-cache.ts +++ b/src/main/native-chat/transcript-read-cache.ts @@ -1,8 +1,8 @@ -import { stat } from 'node:fs/promises' import type { AgentType } from '../../shared/native-chat-types' import { resolveSessionFilePath } from './session-file-resolver' import { readNativeChatTranscript, type ReadTranscriptResult } from './transcript-reader' -import { wslTranscriptFsRefusal } from './wsl-transcript-fs-gate' +import { wslGatedStat } from './wsl-transcript-fs-access' +import { WslTranscriptFsError, wslTranscriptFsRefusal } from './wsl-transcript-fs-gate' // Why: both the desktop IPC handler and the runtime RPC handler read the same // host-filesystem transcript, so a single process-global cache keyed by the @@ -72,9 +72,14 @@ function cacheKey(agent: AgentType, filePath: string): string { async function fileStat(filePath: string): Promise<{ mtimeMs: number; bytes: number }> { try { - const stats = await stat(filePath) + const stats = await wslGatedStat(filePath, 'exact') return { mtimeMs: stats.mtimeMs, bytes: stats.size } - } catch { + } catch (err) { + // Why: swallowing a gate refusal into an unknown mtime falls through to a + // full read that burns a second deadline. Let the caller report it instead. + if (err instanceof WslTranscriptFsError) { + throw err + } return { mtimeMs: Number.NaN, bytes: 0 } } } @@ -105,7 +110,14 @@ export async function readNativeChatTranscriptCached( } const key = cacheKey(agent, filePath) - const { mtimeMs, bytes } = await fileStat(filePath) + 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. + 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. diff --git a/src/main/native-chat/transcript-reader-wsl-stall.test.ts b/src/main/native-chat/transcript-reader-wsl-stall.test.ts new file mode 100644 index 00000000000..47d50ab74bc --- /dev/null +++ b/src/main/native-chat/transcript-reader-wsl-stall.test.ts @@ -0,0 +1,85 @@ +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(() => ({ open: vi.fn() })) + +vi.mock('node:fs/promises', async (importOriginal) => ({ + ...(await importOriginal()), + open: mocks.open +})) + +import { readNativeChatTranscript } from './transcript-reader' +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.' + +const LINE = JSON.stringify({ + type: 'user', + uuid: 'u1', + timestamp: '2026-01-01T00:00:00.000Z', + message: { role: 'user', content: 'hi' } +}) + +function handleServing(body: Buffer, delayMs = 0) { + return { + read: vi.fn( + (buffer: Buffer, offset: number, length: number, position: number) => + new Promise((resolve) => { + const slice = body.subarray(position, Math.min(position + length, body.length)) + slice.copy(buffer, offset) + const settle = () => resolve({ bytesRead: slice.length, buffer }) + if (delayMs > 0) { + setTimeout(settle, delayMs) + } else { + settle() + } + }) + ), + close: vi.fn(async () => {}) + } +} + +beforeEach(() => { + mocks.open.mockReset() +}) + +describe('native chat transcript read with a stalled post-resolution body read', () => { + it('surfaces a retryable error without notFound when the first chunk never settles', async () => { + vi.useFakeTimers() + try { + mocks.open.mockResolvedValue({ + read: vi.fn(() => new Promise(() => {})), + close: vi.fn(async () => {}) + }) + const reading = readNativeChatTranscript('claude', 'session-id', { filePath: UNC_PATH }) + await vi.advanceTimersByTimeAsync(WSL_TRANSCRIPT_FS_EXACT_TIMEOUT_MS + 1) + + const result = await reading + expect(result).toEqual({ error: SLOW_MESSAGE }) + expect(result).not.toHaveProperty('notFound') + } finally { + vi.useRealTimers() + } + }) + + it('succeeds when every chunk is slow but under the deadline, past the whole-file budget', async () => { + vi.useFakeTimers() + const healthyPath = '\\\\wsl.localhost\\Debian\\home\\ada\\.claude\\projects\\p\\session.jsonl' + try { + const chunkDelayMs = WSL_TRANSCRIPT_FS_EXACT_TIMEOUT_MS / 2 + mocks.open.mockResolvedValue(handleServing(Buffer.from(`${LINE}\n`), chunkDelayMs)) + const reading = readNativeChatTranscript('claude', 'session-id', { filePath: healthyPath }) + + // Two chunk admissions (body + EOF probe) span more than one exact deadline. + await vi.advanceTimersByTimeAsync(chunkDelayMs * 4) + + const result = await reading + expect(result).toMatchObject({ messages: [expect.objectContaining({ id: 'u1' })] }) + } finally { + vi.useRealTimers() + } + }) +}) diff --git a/src/main/native-chat/transcript-reader.ts b/src/main/native-chat/transcript-reader.ts index a8e14e0e180..a4d3acb7465 100644 --- a/src/main/native-chat/transcript-reader.ts +++ b/src/main/native-chat/transcript-reader.ts @@ -1,4 +1,3 @@ -import { createReadStream } from 'node:fs' import type { AgentType, NativeChatMessage, @@ -7,6 +6,7 @@ import type { import { resolveNativeChatTranscriptAgent } from '../../shared/native-chat-agent-support' import { errorMessage } from '../ai-vault/session-scanner-values' import { resolveSessionFilePath, type ResolveSessionFileOptions } from './session-file-resolver' +import { openTranscriptReadStream } from './wsl-transcript-fs-access' import { wslTranscriptFsRefusal } from './wsl-transcript-fs-gate' import { decodeClaudeTranscriptLine, @@ -82,7 +82,7 @@ async function readTranscript( filePath: string, decode: (line: string, fallbackId: string) => NativeChatMessage | null ): Promise { - const stream = createReadStream(filePath, { encoding: 'utf-8' }) + const stream = openTranscriptReadStream(filePath, { encoding: 'utf-8' }, 'exact') const { messages } = await decodeTranscriptStream(stream, filePath, 0, decode, true) return messages } diff --git a/src/main/native-chat/transcript-tail-reader-wsl-gate.test.ts b/src/main/native-chat/transcript-tail-reader-wsl-gate.test.ts index ff5194668cd..da040de2794 100644 --- a/src/main/native-chat/transcript-tail-reader-wsl-gate.test.ts +++ b/src/main/native-chat/transcript-tail-reader-wsl-gate.test.ts @@ -1,15 +1,27 @@ -import { beforeEach, describe, expect, it, vi } from 'vitest' +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import type * as NodeFsPromisesModule from 'node:fs/promises' const mocks = vi.hoisted(() => ({ - resolve: vi.fn<() => Promise>() + resolve: vi.fn<() => Promise>(), + stat: vi.fn(), + open: vi.fn() })) vi.mock('./session-file-resolver', () => ({ resolveSessionFilePath: mocks.resolve })) +vi.mock('node:fs/promises', async (importOriginal) => ({ + ...(await importOriginal()), + stat: mocks.stat, + open: mocks.open +})) + import { readNativeChatTranscriptTail } from './transcript-tail-reader' -import { WslTranscriptFsError } from './wsl-transcript-fs-gate' +import { WSL_TRANSCRIPT_FS_EXACT_TIMEOUT_MS, WslTranscriptFsError } 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.' describe('native chat transcript tail under WSL gate refusals', () => { beforeEach(() => { @@ -32,3 +44,69 @@ describe('native chat transcript tail under WSL gate refusals', () => { ).rejects.toThrow('resolver crashed') }) }) + +describe('native chat transcript tail with stalled post-resolution UNC I/O', () => { + const UNC_PATH = '\\\\wsl.localhost\\Ubuntu\\home\\ada\\.codex\\sessions\\a.jsonl' + // A stalled task holds its gate permit until the underlying call settles, so + // each case releases its stall before the next one runs. + let releaseStall: (() => void) | undefined + + function stalls(): Promise { + return new Promise((resolve) => { + releaseStall = () => resolve(undefined as T) + }) + } + + async function tailAfterStall(): Promise { + const pending = readNativeChatTranscriptTail({ + agent: 'codex', + sessionId: 'session-id', + filePath: UNC_PATH, + limit: 10 + }) + await vi.advanceTimersByTimeAsync(WSL_TRANSCRIPT_FS_EXACT_TIMEOUT_MS + 1) + return pending + } + + beforeEach(() => { + mocks.stat.mockReset() + mocks.open.mockReset() + releaseStall = undefined + vi.useFakeTimers() + }) + + afterEach(async () => { + releaseStall?.() + await vi.advanceTimersByTimeAsync(0) + vi.useRealTimers() + }) + + it('reports a stalled stat as retryable without notFound', async () => { + mocks.stat.mockImplementation(stalls) + const result = await tailAfterStall() + expect(result).toEqual({ error: SLOW_MESSAGE }) + expect(result).not.toHaveProperty('notFound') + }) + + it('reports a stalled open as retryable without notFound', async () => { + mocks.stat.mockResolvedValue({ size: 4096 }) + mocks.open.mockImplementation(stalls) + const result = await tailAfterStall() + expect(result).toEqual({ error: SLOW_MESSAGE }) + expect(result).not.toHaveProperty('notFound') + }) + + it('closes the handle without a second admission when a positional read stalls', async () => { + const close = vi.fn(async () => {}) + mocks.stat.mockResolvedValue({ size: 4096 }) + mocks.open.mockResolvedValue({ read: vi.fn(stalls), close }) + + const result = await tailAfterStall() + + expect(result).toEqual({ error: SLOW_MESSAGE }) + expect(result).not.toHaveProperty('notFound') + // Fire-and-forget close: never gated, never awaited, exactly once. + await vi.advanceTimersByTimeAsync(0) + expect(close).toHaveBeenCalledTimes(1) + }) +}) diff --git a/src/main/native-chat/transcript-tail-reader.ts b/src/main/native-chat/transcript-tail-reader.ts index 1264c2e3d98..c1883c786b0 100644 --- a/src/main/native-chat/transcript-tail-reader.ts +++ b/src/main/native-chat/transcript-tail-reader.ts @@ -1,4 +1,4 @@ -import { open, stat } from 'node:fs/promises' +import type { FileHandle } from 'node:fs/promises' import type { AgentType, NativeChatMessage, @@ -17,6 +17,12 @@ import { nativeChatTurnLifecycleDecoderForAgent, type NativeChatTurnLifecycleDecoder } from './transcript-turn-lifecycle' +import { + closeTranscriptHandle, + wslGatedOpen, + wslGatedRead, + wslGatedStat +} from './wsl-transcript-fs-access' import { wslTranscriptFsRefusal } from './wsl-transcript-fs-gate' export const MAX_NATIVE_CHAT_TRANSCRIPT_RECORD_BYTES = 2 * 1024 * 1024 @@ -59,12 +65,15 @@ export async function readNativeChatTranscriptTailFile( oversizedRecordCount?: number }> { signal?.throwIfAborted() - const end = Math.min((await stat(filePath)).size, endOffset ?? Number.MAX_SAFE_INTEGER) + const end = Math.min( + (await wslGatedStat(filePath, 'exact')).size, + endOffset ?? Number.MAX_SAFE_INTEGER + ) signal?.throwIfAborted() if (end === 0) { return { messages: [], consumedTo: 0, hasMore: false, beforeOffset: 0 } } - const handle = await open(filePath, 'r') + const handle = await wslGatedOpen(filePath, 'exact', signal) const lineParts: Buffer[] = [] let lineBytes = 0 let lineOversized = false @@ -76,13 +85,13 @@ export async function readNativeChatTranscriptTailFile( signal?.throwIfAborted() const consumedTo = includeTrailingLine ? end - : await findLastCompleteLineEnd(handle, end, signal) + : await findLastCompleteLineEnd(handle, filePath, end, signal) if (consumedTo === 0) { return { messages: [], consumedTo: 0, hasMore: false, beforeOffset: 0 } } const newestFirst: { message: NativeChatMessage; offset: number }[] = [] const finalByte = Buffer.allocUnsafe(1) - await handle.read(finalByte, 0, 1, consumedTo - 1) + await wslGatedRead(handle, filePath, finalByte, 0, 1, consumedTo - 1, 'exact', signal) signal?.throwIfAborted() ignoreNextMalformedRecord = finalByte[0] !== 0x0a let cursor = consumedTo - (finalByte[0] === 0x0a ? 1 : 0) @@ -90,7 +99,16 @@ export async function readNativeChatTranscriptTailFile( signal?.throwIfAborted() const start = Math.max(0, cursor - TAIL_CHUNK_BYTES) const buffer = Buffer.allocUnsafe(cursor - start) - const { bytesRead } = await handle.read(buffer, 0, buffer.length, start) + const { bytesRead } = await wslGatedRead( + handle, + filePath, + buffer, + 0, + buffer.length, + start, + 'exact', + signal + ) signal?.throwIfAborted() let segmentEnd = bytesRead for (let index = bytesRead - 1; index >= 0 && newestFirst.length <= limit; index--) { @@ -126,7 +144,7 @@ export async function readNativeChatTranscriptTailFile( ...(oversizedRecordCount > 0 ? { oversizedRecordCount } : {}) } } finally { - await handle.close() + closeTranscriptHandle(handle) } function retainPart(part: Buffer): void { @@ -184,13 +202,14 @@ export async function readNativeChatTranscriptTailFile( } async function findLastCompleteLineEnd( - handle: Awaited>, + handle: FileHandle, + filePath: string, end: number, signal?: AbortSignal ): Promise { signal?.throwIfAborted() const lastByte = Buffer.allocUnsafe(1) - await handle.read(lastByte, 0, 1, end - 1) + await wslGatedRead(handle, filePath, lastByte, 0, 1, end - 1, 'exact', signal) signal?.throwIfAborted() if (lastByte[0] === 0x0a) { return end @@ -200,7 +219,16 @@ async function findLastCompleteLineEnd( signal?.throwIfAborted() const start = Math.max(0, cursor - TAIL_CHUNK_BYTES) const buffer = Buffer.allocUnsafe(cursor - start) - const { bytesRead } = await handle.read(buffer, 0, buffer.length, start) + const { bytesRead } = await wslGatedRead( + handle, + filePath, + buffer, + 0, + buffer.length, + start, + 'exact', + signal + ) signal?.throwIfAborted() const newline = buffer.subarray(0, bytesRead).lastIndexOf(0x0a) if (newline !== -1) { diff --git a/src/main/native-chat/transcript-watch-engine.ts b/src/main/native-chat/transcript-watch-engine.ts index ff1df71ee0f..e2b3c610e87 100644 --- a/src/main/native-chat/transcript-watch-engine.ts +++ b/src/main/native-chat/transcript-watch-engine.ts @@ -1,4 +1,3 @@ -import { open, stat } from 'node:fs/promises' import type { NativeChatMessage, NativeChatTurnLifecycle } from '../../shared/native-chat-types' import { readTranscriptFileVersion, @@ -18,6 +17,8 @@ import type { SubscribeNativeChatTranscriptArgs } from './transcript-watch-contract' import { createTranscriptWatchScheduler } from './transcript-watch-scheduler' +import { readTranscriptSlice, wslGatedStat } from './wsl-transcript-fs-access' +import { WslTranscriptFsError } from './wsl-transcript-fs-gate' const ROTATION_RETRY_MS = 25 const MAX_ROTATION_RETRY_MS = 2_000 @@ -32,14 +33,8 @@ async function boundaryFingerprint(filePath: string, offset: number): Promise { try { - await stat(filePath) - } catch { + await wslGatedStat(filePath, 'exact') + } catch (error) { + // Why: "not flushed yet" degrades to resolve-polling, but a stalled distro + // must reach the caller so it can surface a retryable message instead of + // stranding the client at `loading`. + if (error instanceof WslTranscriptFsError) { + throw error + } return null } const { onAppend, onInitialSnapshot, onReplace, initialLimit } = args @@ -253,7 +254,7 @@ export async function installTranscriptWatcher( pendingReadRequested = false try { await drainOnce() - } catch { + } catch (error) { // Why: unlink/recreate can detach fs.watch from the pathname. Keep one // capped-backoff retry alive until a successor appears or we unsubscribe. // A still-pending initial drain also surfaces one error snapshot so a @@ -261,7 +262,12 @@ export async function installTranscriptWatcher( // throwing; initialDrain stays true so a recovered read can still win. if (!closed && initialDrain && onInitialSnapshot && !initialErrorEmitted) { initialErrorEmitted = true - onInitialSnapshot([], false, 0, 'Transcript unavailable') + onInitialSnapshot( + [], + false, + 0, + error instanceof WslTranscriptFsError ? error.message : 'Transcript unavailable' + ) } scheduleRotationRetry() break diff --git a/src/main/native-chat/transcript-watch-wsl-stall.test.ts b/src/main/native-chat/transcript-watch-wsl-stall.test.ts new file mode 100644 index 00000000000..063ecb25139 --- /dev/null +++ b/src/main/native-chat/transcript-watch-wsl-stall.test.ts @@ -0,0 +1,143 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import type * as NodeFsPromisesModule from 'node:fs/promises' +import type { NativeChatMessage } from '../../shared/native-chat-types' + +const UNC_PATH = '\\\\wsl.localhost\\Ubuntu\\home\\ada\\.claude\\projects\\p\\session.jsonl' + +const mocks = vi.hoisted(() => ({ + resolve: vi.fn<() => Promise>(), + stat: vi.fn(), + open: vi.fn() +})) + +vi.mock('./session-file-resolver', () => ({ resolveSessionFilePath: mocks.resolve })) +vi.mock('./transcript-native-watcher', () => ({ + createTranscriptNativeWatcher: () => ({ + bind: () => true, + needsRebind: () => false, + invalidate: () => {}, + dispose: () => {} + }) +})) +vi.mock('node:fs/promises', async (importOriginal) => ({ + ...(await importOriginal()), + stat: mocks.stat, + open: mocks.open +})) + +import { subscribeNativeChatTranscript } from './transcript-watch' +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.' + +type Snapshot = [NativeChatMessage[], boolean, number, string | undefined] + +let releaseStall: (() => void) | undefined +let unhandled: unknown[] = [] + +function stalls(): Promise { + return new Promise((resolve) => { + releaseStall = () => resolve(undefined as T) + }) +} + +function subscribeCollecting(snapshots: Snapshot[]) { + return subscribeNativeChatTranscript({ + agent: 'claude', + sessionId: 'session-id', + onAppend: () => {}, + onInitialSnapshot: (messages, hasMore, beforeOffset, error) => { + snapshots.push([messages, hasMore, beforeOffset, error]) + }, + initialLimit: 10, + resolvePollIntervalMs: 50, + reconciliationIntervalMs: 1_000 + }) +} + +function trackUnhandled(reason: unknown): void { + unhandled.push(reason) +} + +beforeEach(() => { + mocks.resolve.mockReset() + mocks.stat.mockReset() + mocks.open.mockReset() + releaseStall = undefined + unhandled = [] + process.on('unhandledRejection', trackUnhandled) + vi.useFakeTimers() +}) + +afterEach(async () => { + releaseStall?.() + await vi.advanceTimersByTimeAsync(0) + vi.useRealTimers() + process.off('unhandledRejection', trackUnhandled) +}) + +const EMPTY_STATS = { size: 0, mtimeMs: 1, ctimeMs: 1, ino: 1, dev: 1, mtime: new Date(0) } + +describe('native chat transcript subscription with a stalled install stat', () => { + it('keeps watching and emits exactly one retryable snapshot, then the real one', async () => { + mocks.resolve.mockResolvedValue(UNC_PATH) + mocks.stat.mockImplementationOnce(stalls).mockResolvedValue(EMPTY_STATS) + const snapshots: Snapshot[] = [] + + // Not awaited yet: the setup install is itself blocked on the stalled stat. + const subscribing = subscribeCollecting(snapshots) + await vi.advanceTimersByTimeAsync(WSL_TRANSCRIPT_FS_EXACT_TIMEOUT_MS + 1) + const subscription = await subscribing + expect(subscription.watching).toBe(true) + + await vi.advanceTimersByTimeAsync(WSL_TRANSCRIPT_FS_EXACT_TIMEOUT_MS + 1) + expect(snapshots).toEqual([[[], false, 0, SLOW_MESSAGE]]) + + // Later poll ticks fast-fail on the still-stuck route without re-emitting. + await vi.advanceTimersByTimeAsync(500) + expect(snapshots).toHaveLength(1) + + // The distro wakes: the same subscription still delivers a real snapshot. + releaseStall?.() + await vi.advanceTimersByTimeAsync(500) + + expect(snapshots.length).toBeGreaterThan(1) + expect(snapshots.at(-1)?.[3]).toBeUndefined() + subscription.unsubscribe() + expect(unhandled).toEqual([]) + }) + + it('keeps the poll loop alive on a non-gate resolve failure without emitting a frame', async () => { + // First resolve misses (transcript not flushed) so setup falls through to + // the poll loop; every later tick throws a non-gate error. + mocks.resolve + .mockResolvedValueOnce(null) + .mockRejectedValue(Object.assign(new Error('EACCES'), { code: 'EACCES' })) + const snapshots: Snapshot[] = [] + + const subscription = await subscribeCollecting(snapshots) + const attemptsAfterSetup = mocks.resolve.mock.calls.length + await vi.advanceTimersByTimeAsync(500) + + expect(snapshots).toEqual([]) + // Still retrying: the rethrow-free branch must not kill `void runAttempt()`. + expect(mocks.resolve.mock.calls.length).toBeGreaterThan(attemptsAfterSetup) + expect(unhandled).toEqual([]) + subscription.unsubscribe() + }) + + it('tears down cleanly while an admission is still stalled', async () => { + mocks.resolve.mockResolvedValue(UNC_PATH) + mocks.stat.mockImplementation(stalls) + const snapshots: Snapshot[] = [] + + const subscribing = subscribeCollecting(snapshots) + await vi.advanceTimersByTimeAsync(WSL_TRANSCRIPT_FS_EXACT_TIMEOUT_MS + 1) + const subscription = await subscribing + + expect(() => subscription.unsubscribe()).not.toThrow() + await vi.advanceTimersByTimeAsync(WSL_TRANSCRIPT_FS_EXACT_TIMEOUT_MS + 1) + expect(unhandled).toEqual([]) + }) +}) diff --git a/src/main/native-chat/transcript-watch.ts b/src/main/native-chat/transcript-watch.ts index 2370d7b6410..774d8d7aa2c 100644 --- a/src/main/native-chat/transcript-watch.ts +++ b/src/main/native-chat/transcript-watch.ts @@ -11,7 +11,7 @@ import type { SubscribeNativeChatTranscriptArgs } from './transcript-watch-contract' import { nativeChatLineDecoderForAgent } from './transcript-tail-reader' -import { wslTranscriptFsRefusal } from './wsl-transcript-fs-gate' +import { WslTranscriptFsError, wslTranscriptFsRefusal } from './wsl-transcript-fs-gate' export { readNativeChatTranscriptTail } from './transcript-tail-reader' export { getActiveNativeChatWatcherCount } from './transcript-watch-engine' @@ -78,6 +78,9 @@ function subscribeViaResolvePoll( // the exact-path install doesn't wait on the slower id-glob (#10326). let hostReadableExactPath: string | null = null let lastWslTranslateAt = 0 + // Latches only once a frame was actually emitted, so a subscriber without the + // callback can't suppress it for a later one. + let gateErrorEmitted = false const resolveController = new AbortController() function scheduleAttempt(): void { @@ -141,9 +144,18 @@ function subscribeViaResolvePoll( lastFallbackResolveAt = Date.now() result = await attemptInstall(args, decode, resolveController.signal) } - } catch { + } catch (error) { // Why: a transient resolve failure (EACCES/EIO during the glob) must not - // kill the poll loop with an unhandled rejection — retry like a miss. + // kill the poll loop with an unhandled rejection — retry like a miss. A + // stalled WSL distro would otherwise poll silently forever, leaving the + // client at 'loading'; emit its retryable message once and keep polling, + // so a later tick's real snapshot still replaces it. Narrowed with a bare + // instanceof, never wslTranscriptFsRefusal — that helper rethrows, and + // runAttempt is invoked as `void runAttempt()`. + if (error instanceof WslTranscriptFsError && !gateErrorEmitted && args.onInitialSnapshot) { + gateErrorEmitted = true + args.onInitialSnapshot([], false, 0, error.message) + } result = null } if (closed) { diff --git a/src/main/native-chat/wsl-transcript-fs-access.test.ts b/src/main/native-chat/wsl-transcript-fs-access.test.ts new file mode 100644 index 00000000000..0e6615de0fe --- /dev/null +++ b/src/main/native-chat/wsl-transcript-fs-access.test.ts @@ -0,0 +1,245 @@ +import { beforeEach, describe, expect, it, vi } from 'vitest' +import type * as NodeFsModule from 'node:fs' +import type * as NodeFsPromisesModule from 'node:fs/promises' +import type * as GateModule from './wsl-transcript-fs-gate' + +const UNC_PATH = '\\\\wsl.localhost\\Ubuntu\\home\\ada\\.codex\\sessions\\a.jsonl' +const LEGACY_UNC_PATH = '\\\\wsl$\\Ubuntu\\home\\ada\\.codex\\sessions\\a.jsonl' +const WINDOWS_PATH = 'C:\\Users\\ada\\.codex\\sessions\\a.jsonl' +const POSIX_PATH = '/home/ada/.codex/sessions/a.jsonl' + +const mocks = vi.hoisted(() => ({ + stat: vi.fn(), + lstat: vi.fn(), + readdir: vi.fn(), + readFile: vi.fn(), + open: vi.fn(), + createReadStream: vi.fn(), + runTask: vi.fn() +})) + +vi.mock('node:fs', async (importOriginal) => ({ + ...(await importOriginal()), + createReadStream: mocks.createReadStream +})) + +vi.mock('node:fs/promises', async (importOriginal) => ({ + ...(await importOriginal()), + stat: mocks.stat, + lstat: mocks.lstat, + readdir: mocks.readdir, + readFile: mocks.readFile, + open: mocks.open +})) + +vi.mock('./wsl-transcript-fs-gate', async (importOriginal) => { + const original = await importOriginal() + mocks.runTask.mockImplementation(original.runWslTranscriptFsTask) + return { ...original, runWslTranscriptFsTask: mocks.runTask } +}) + +import { + closeTranscriptHandle, + openTranscriptReadStream, + readTranscriptSlice, + wslGatedLstat, + wslGatedOpen, + wslGatedRead, + wslGatedReaddir, + wslGatedReadFile, + wslGatedStat +} from './wsl-transcript-fs-access' +import { WSL_TRANSCRIPT_FS_EXACT_TIMEOUT_MS, WslTranscriptFsError } from './wsl-transcript-fs-gate' + +function fakeHandle() { + return { read: vi.fn(), close: vi.fn(async () => {}) } +} + +beforeEach(() => { + // runTask keeps the real gate implementation installed by the mock factory; + // only its call log is cleared. + mocks.runTask.mockClear() + for (const [name, mock] of Object.entries(mocks)) { + if (name !== 'runTask') { + mock.mockReset() + } + } +}) + +describe('transcript filesystem accessor off WSL UNC', () => { + it.each([ + ['a posix path', POSIX_PATH], + ['a Windows drive path', WINDOWS_PATH] + ])('never enters the gate for %s', async (_label, path) => { + mocks.stat.mockResolvedValue({ size: 1 }) + mocks.lstat.mockResolvedValue({ size: 1 }) + mocks.readdir.mockResolvedValue([]) + mocks.readFile.mockResolvedValue('body') + const handle = fakeHandle() + handle.read.mockResolvedValue({ bytesRead: 0, buffer: Buffer.alloc(0) }) + mocks.open.mockResolvedValue(handle) + mocks.createReadStream.mockReturnValue('raw-stream') + + await wslGatedStat(path, 'exact') + await wslGatedLstat(path, 'scan') + await wslGatedReaddir(path, 'scan') + await wslGatedReadFile(path, 'utf-8', 'scan') + const opened = await wslGatedOpen(path, 'exact') + await wslGatedRead(opened, path, Buffer.alloc(1), 0, 1, 0, 'exact') + const stream = openTranscriptReadStream(path, { encoding: 'utf-8' }, 'scan') + + expect(mocks.runTask).not.toHaveBeenCalled() + expect(mocks.stat).toHaveBeenCalledWith(path) + expect(mocks.readdir).toHaveBeenCalledWith(path, { withFileTypes: true }) + expect(mocks.readFile).toHaveBeenCalledWith(path, 'utf-8') + expect(mocks.open).toHaveBeenCalledWith(path, 'r') + // Off UNC the raw stream is handed back verbatim, encoding included. + expect(stream).toBe('raw-stream') + expect(mocks.createReadStream).toHaveBeenCalledWith(path, { encoding: 'utf-8' }) + }) +}) + +describe('transcript filesystem accessor on WSL UNC', () => { + it.each([ + ['wsl.localhost', UNC_PATH], + ['wsl$', LEGACY_UNC_PATH] + ])('routes %s paths through the gate', async (_label, path) => { + mocks.stat.mockResolvedValue({ size: 7 }) + await expect(wslGatedStat(path, 'exact')).resolves.toEqual({ size: 7 }) + expect(mocks.runTask).toHaveBeenCalledWith( + expect.objectContaining({ operation: 'stat', path, priority: 'exact' }), + expect.any(Function) + ) + expect(mocks.createReadStream).not.toHaveBeenCalled() + }) + + it('opts positional reads and opens out of coalescing', async () => { + const handle = fakeHandle() + handle.read.mockResolvedValue({ bytesRead: 1, buffer: Buffer.alloc(1) }) + mocks.open.mockResolvedValue(handle) + + const opened = await wslGatedOpen(UNC_PATH, 'exact') + await wslGatedRead(opened, UNC_PATH, Buffer.alloc(1), 0, 1, 0, 'exact') + + for (const call of mocks.runTask.mock.calls) { + expect(call[0]).toMatchObject({ dedupe: false }) + } + }) + + it('reads a slice and closes the handle even when the read rejects', async () => { + const handle = fakeHandle() + handle.read.mockRejectedValue(new Error('EIO')) + mocks.open.mockResolvedValue(handle) + + await expect(readTranscriptSlice(UNC_PATH, 4, 8, 'scan')).rejects.toThrow('EIO') + expect(handle.close).toHaveBeenCalledTimes(1) + }) + + it('yields Buffer chunks and closes the handle when the consumer destroys the stream', async () => { + const handle = fakeHandle() + handle.read.mockImplementation( + async (buffer: Buffer, offset: number, length: number, position: number) => { + const body = Buffer.from('{"a":1}\n{"b":2}\n') + const slice = body.subarray(position, Math.min(position + length, body.length)) + slice.copy(buffer, offset) + return { bytesRead: slice.length, buffer } + } + ) + mocks.open.mockResolvedValue(handle) + + const stream = openTranscriptReadStream(UNC_PATH, {}, 'exact') + const chunks: unknown[] = [] + for await (const chunk of stream) { + chunks.push(chunk) + break + } + stream.destroy() + await new Promise((resolve) => setImmediate(resolve)) + + expect(chunks).toHaveLength(1) + expect(Buffer.isBuffer(chunks[0])).toBe(true) + expect((chunks[0] as Buffer).toString('utf-8')).toBe('{"a":1}\n{"b":2}\n') + expect(handle.close).toHaveBeenCalledTimes(1) + }) + + it('swallows close failures so teardown never rejects', async () => { + const handle = fakeHandle() + handle.close.mockRejectedValue(new Error('stalled close')) + expect(() => closeTranscriptHandle(handle as never)).not.toThrow() + await new Promise((resolve) => setImmediate(resolve)) + }) +}) + +describe('per-chunk admission', () => { + it('streams a long healthy-but-slow file past the whole-file deadline', async () => { + vi.useFakeTimers() + try { + const chunkDelayMs = WSL_TRANSCRIPT_FS_EXACT_TIMEOUT_MS / 4 + const totalChunks = 6 + let served = 0 + const handle = fakeHandle() + handle.read.mockImplementation( + (buffer: Buffer) => + new Promise((resolve) => { + setTimeout(() => { + if (served++ >= totalChunks) { + resolve({ bytesRead: 0, buffer }) + return + } + buffer.write('x') + resolve({ bytesRead: 1, buffer }) + }, chunkDelayMs) + }) + ) + mocks.open.mockResolvedValue(handle) + + const stream = openTranscriptReadStream(UNC_PATH, {}, 'exact') + const collected: Promise = (async () => { + const out: Buffer[] = [] + for await (const chunk of stream) { + out.push(chunk as Buffer) + } + return out + })() + + // Elapsed time far exceeds one exact deadline; each chunk stays under it. + await vi.advanceTimersByTimeAsync(chunkDelayMs * (totalChunks + 2)) + await expect(collected).resolves.toHaveLength(totalChunks) + } finally { + vi.useRealTimers() + } + }) + + it('surfaces a gate refusal mid-stream as an error event', async () => { + vi.useFakeTimers() + try { + let served = 0 + const handle = fakeHandle() + handle.read.mockImplementation((buffer: Buffer) => { + if (served++ === 0) { + buffer.write('x') + return Promise.resolve({ bytesRead: 1, buffer }) + } + return new Promise(() => {}) + }) + mocks.open.mockResolvedValue(handle) + + const stream = openTranscriptReadStream(UNC_PATH, {}, 'exact') + const drained = (async () => { + for await (const chunk of stream) { + void chunk + } + })() + const settled = drained.then( + () => null, + (error: unknown) => error + ) + await vi.advanceTimersByTimeAsync(WSL_TRANSCRIPT_FS_EXACT_TIMEOUT_MS + 1) + + await expect(settled).resolves.toBeInstanceOf(WslTranscriptFsError) + expect(handle.close).toHaveBeenCalledTimes(1) + } finally { + vi.useRealTimers() + } + }) +}) diff --git a/src/main/native-chat/wsl-transcript-fs-access.ts b/src/main/native-chat/wsl-transcript-fs-access.ts new file mode 100644 index 00000000000..12ae1f78ee1 --- /dev/null +++ b/src/main/native-chat/wsl-transcript-fs-access.ts @@ -0,0 +1,198 @@ +import { createReadStream, type Dirent, type Stats } from 'node:fs' +import { lstat, open, readdir, readFile, stat, type FileHandle } from 'node:fs/promises' +import { Readable } from 'node:stream' +import { isWslUncPath } from '../../shared/wsl-paths' +import { runWslTranscriptFsTask, type WslTranscriptFsTaskPriority } from './wsl-transcript-fs-gate' + +/** + * Transcript-layer filesystem primitives that route WSL UNC paths through the + * admission gate and everything else straight to `node:fs`. The `isWslUncPath` + * guard is the single switch: macOS, Linux, Windows drive paths, SSH remotes and + * folder workspaces execute the exact call they execute today. + * + * Invariant future call sites must preserve: every gated unit here is a single + * syscall, and `openTranscriptReadStream` composes them OUTSIDE the gate. A + * gated call awaited from inside another gated task's operation would deadlock + * on the single scan slot or the route+priority lane. + */ + +// 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 + +export function wslGatedStat(path: string, priority: WslTranscriptFsTaskPriority): Promise { + if (!isWslUncPath(path)) { + return stat(path) + } + return runWslTranscriptFsTask({ operation: 'stat', path, priority }, () => stat(path)) +} + +export function wslGatedLstat(path: string, priority: WslTranscriptFsTaskPriority): Promise { + if (!isWslUncPath(path)) { + return lstat(path) + } + return runWslTranscriptFsTask({ operation: 'lstat', path, priority }, () => lstat(path)) +} + +export function wslGatedReaddir( + path: string, + priority: WslTranscriptFsTaskPriority, + signal?: AbortSignal +): Promise { + if (!isWslUncPath(path)) { + return readdir(path, { withFileTypes: true }) + } + return runWslTranscriptFsTask({ operation: 'readdir', path, priority, signal }, () => + readdir(path, { withFileTypes: true }) + ) +} + +export function wslGatedReadFile( + path: string, + encoding: BufferEncoding, + priority: WslTranscriptFsTaskPriority, + signal?: AbortSignal +): Promise { + if (!isWslUncPath(path)) { + return readFile(path, encoding) + } + return runWslTranscriptFsTask({ operation: '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, + signal?: AbortSignal +): Promise { + if (!isWslUncPath(path)) { + return open(path, 'r') + } + return runWslTranscriptFsTask({ operation: 'open', path, priority, signal, dedupe: false }, () => + open(path, 'r') + ) +} + +/** + * `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, + buffer: Buffer, + offset: number, + length: number, + position: number, + priority: WslTranscriptFsTaskPriority, + signal?: AbortSignal +): Promise<{ bytesRead: number; buffer: Buffer }> { + if (!isWslUncPath(path)) { + return handle.read(buffer, offset, length, position) + } + return runWslTranscriptFsTask({ operation: 'read', path, priority, signal, dedupe: false }, () => + handle.read(buffer, offset, length, position) + ) +} + +/** + * Fire-and-forget, never gated: 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): void { + void handle.close().catch(() => {}) +} + +/** + * 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, + length: number, + priority: WslTranscriptFsTaskPriority, + signal?: AbortSignal +): Promise { + const handle = await wslGatedOpen(path, priority, signal) + try { + const buffer = Buffer.allocUnsafe(length) + const read = await wslGatedRead(handle, path, buffer, 0, length, position, priority, signal) + return buffer.subarray(0, read.bytesRead) + } finally { + closeTranscriptHandle(handle) + } +} + +export type TranscriptReadStreamOptions = { + start?: number + /** Inclusive, matching `createReadStream`. */ + end?: number + /** Honoured off UNC only; the UNC branch always yields `Buffer` chunks. */ + encoding?: BufferEncoding +} + +async function* gatedChunks( + path: string, + options: TranscriptReadStreamOptions, + priority: WslTranscriptFsTaskPriority, + signal?: AbortSignal +): AsyncGenerator { + const handle = await wslGatedOpen(path, priority, signal) + try { + let position = options.start ?? 0 + for (;;) { + const length = + options.end === undefined + ? WSL_TRANSCRIPT_READ_CHUNK_BYTES + : Math.min(WSL_TRANSCRIPT_READ_CHUNK_BYTES, options.end - position + 1) + if (length <= 0) { + return + } + const buffer = Buffer.allocUnsafe(length) + const { bytesRead } = await wslGatedRead( + handle, + path, + buffer, + 0, + length, + position, + priority, + signal + ) + if (bytesRead <= 0) { + return + } + position += bytesRead + yield buffer.subarray(0, bytesRead) + } + } finally { + // Runs on `.destroy()` too (Readable.from calls the generator's return()), + // so an aborted head-read cannot leak the gated handle. + closeTranscriptHandle(handle) + } +} + +/** + * 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, + priority: WslTranscriptFsTaskPriority, + signal?: AbortSignal +): Readable { + if (!isWslUncPath(path)) { + return createReadStream(path, options) + } + return Readable.from(gatedChunks(path, options, priority, signal)) +} 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 1f82ab25277..e4a86dc8d50 100644 --- a/src/main/native-chat/wsl-transcript-fs-gate.test.ts +++ b/src/main/native-chat/wsl-transcript-fs-gate.test.ts @@ -632,3 +632,78 @@ describe('WSL transcript filesystem task scheduling', () => { } }) }) + +describe('WSL transcript filesystem task coalescing opt-out', () => { + const READ_PATH = '\\\\wsl.localhost\\Alpine\\home\\ada\\transcript.jsonl' + + function gatedRead( + buffer: Buffer, + task: (signal: AbortSignal) => Promise + ): Promise { + void buffer + return runWslTranscriptFsTask( + { operation: 'read', path: READ_PATH, priority: 'exact', dedupe: false }, + task + ) + } + + it('fills each caller its own buffer for identical positional reads', async () => { + const first = Buffer.alloc(4, 0xa1) + const second = Buffer.alloc(4, 0xb2) + const bodies = [Buffer.from('AAAA'), Buffer.from('BBBB')] + + const reads = Promise.all([ + gatedRead(first, async () => { + bodies.shift()!.copy(first) + return first + }), + gatedRead(second, async () => { + bodies.shift()!.copy(second) + return second + }) + ]) + + await expect(reads).resolves.toEqual([first, second]) + expect(first.toString()).toBe('AAAA') + expect(second.toString()).toBe('BBBB') + // Neither caller kept its sentinel nor received the other's bytes. + expect(first.equals(second)).toBe(false) + }) + + it('gives each open call a distinct handle', async () => { + const handles = [ + { id: 1, closed: false }, + { id: 2, closed: false } + ] + let served = 0 + const opened = await Promise.all( + handles.map(() => + runWslTranscriptFsTask( + { operation: 'open', path: READ_PATH, priority: 'exact', dedupe: false }, + async () => handles[served++] + ) + ) + ) + + expect(opened[0]).not.toBe(opened[1]) + opened[0].closed = true + expect(opened[1].closed).toBe(false) + }) + + it('still coalesces identical work when dedupe is omitted', async () => { + const stalled = deferred() + const task = vi.fn(() => stalled.promise) + const path = '\\\\wsl.localhost\\Alpine\\home\\ada' + const first = run(path, 'scan', task) + await vi.waitFor(() => expect(task).toHaveBeenCalledOnce()) + const joiner = run( + path, + 'scan', + vi.fn(async () => 'never') + ) + + stalled.resolve('shared') + await expect(Promise.all([first, joiner])).resolves.toEqual(['shared', 'shared']) + expect(task).toHaveBeenCalledOnce() + }) +}) diff --git a/src/main/native-chat/wsl-transcript-fs-gate.ts b/src/main/native-chat/wsl-transcript-fs-gate.ts index 52b4b22f81b..eee1b2d733d 100644 --- a/src/main/native-chat/wsl-transcript-fs-gate.ts +++ b/src/main/native-chat/wsl-transcript-fs-gate.ts @@ -59,6 +59,7 @@ type ScheduledTask = { type UnknownScheduledTask = ScheduledTask let activeScanCount = 0 +let undedupedTaskId = 0 const activeLaneKeys = new Set() const queuedTasks: UnknownScheduledTask[] = [] const inFlightTasks = new Map() @@ -284,10 +285,13 @@ function pumpTasks(): void { /** Bound 9P work without letting scans delay exact transcript probes. */ export function runWslTranscriptFsTask( options: { - operation: 'access' | 'readdir' + operation: 'access' | 'readdir' | 'stat' | 'lstat' | 'open' | 'read' | 'readfile' path: string priority: WslTranscriptFsTaskPriority signal?: AbortSignal + /** Opt out of coalescing when the result is not the whole answer (`open`, + * positional `read`): a joiner would share the handle or the buffer. */ + dedupe?: boolean }, task: (signal: AbortSignal) => Promise ): Promise { @@ -295,8 +299,12 @@ 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. - const key = JSON.stringify([options.operation, options.path]) + // 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. + const key = + options.dedupe === false + ? `undeduped:${++undedupedTaskId}` + : JSON.stringify([options.operation, options.path]) const existing = inFlightTasks.get(key) as ScheduledTask | undefined if (existing) { // A stuck target — or a queued target that stuck I/O keeps from ever