diff --git a/src/main/ai-vault/session-scanner-antigravity-parser.ts b/src/main/ai-vault/session-scanner-antigravity-parser.ts index 3fede1d5521..1896138749c 100644 --- a/src/main/ai-vault/session-scanner-antigravity-parser.ts +++ b/src/main/ai-vault/session-scanner-antigravity-parser.ts @@ -26,11 +26,16 @@ export async function parseAntigravitySessionFile( file: FileWithMtime, platform: NodeJS.Platform = process.platform ): Promise { - const lines = createInterface({ - input: openTranscriptReadStream(file.path, { encoding: 'utf-8' }, 'scan'), - crlfDelay: Infinity - }) - return parseAntigravitySessionLines({ file, lines, platform }) + const input = openTranscriptReadStream(file.path, { encoding: 'utf-8' }, 'scan') + const lines = createInterface({ input, crlfDelay: Infinity }) + try { + return await parseAntigravitySessionLines({ file, lines, platform }) + } finally { + // readline.close() leaves the underlying stream open; destroy it so a + // mid-parse throw cannot leak the gated transcript handle. + lines.close() + input.destroy() + } } export async function parseAntigravitySessionContent( diff --git a/src/main/ai-vault/session-scanner-droid-parser.ts b/src/main/ai-vault/session-scanner-droid-parser.ts index f06051823d3..c896b47a3e8 100644 --- a/src/main/ai-vault/session-scanner-droid-parser.ts +++ b/src/main/ai-vault/session-scanner-droid-parser.ts @@ -34,11 +34,16 @@ export async function parseDroidSessionFile( file: FileWithMtime, platform: NodeJS.Platform = process.platform ): Promise { - const lines = createInterface({ - input: openTranscriptReadStream(file.path, { encoding: 'utf-8' }, 'scan'), - crlfDelay: Infinity - }) - return parseDroidSessionLines({ file, lines, platform }) + const input = openTranscriptReadStream(file.path, { encoding: 'utf-8' }, 'scan') + const lines = createInterface({ input, crlfDelay: Infinity }) + try { + return await parseDroidSessionLines({ file, lines, platform }) + } finally { + // readline.close() leaves the underlying stream open; destroy it so a + // mid-parse throw cannot leak the gated transcript handle. + lines.close() + input.destroy() + } } export async function parseDroidSessionContent( diff --git a/src/main/ai-vault/session-scanner-graph-parsers.ts b/src/main/ai-vault/session-scanner-graph-parsers.ts index 4d53f366bc7..d99893acaa2 100644 --- a/src/main/ai-vault/session-scanner-graph-parsers.ts +++ b/src/main/ai-vault/session-scanner-graph-parsers.ts @@ -176,11 +176,16 @@ export async function parseMessageGraphSessionFile( file: FileWithMtime, platform: NodeJS.Platform = process.platform ): Promise { - const lines = createInterface({ - input: openTranscriptReadStream(file.path, { encoding: 'utf-8' }, 'scan'), - crlfDelay: Infinity - }) - return parseMessageGraphSessionLines({ agent, file, lines, platform }) + const input = openTranscriptReadStream(file.path, { encoding: 'utf-8' }, 'scan') + const lines = createInterface({ input, crlfDelay: Infinity }) + try { + return await parseMessageGraphSessionLines({ agent, file, lines, platform }) + } finally { + // readline.close() leaves the underlying stream open; destroy it so a + // mid-parse throw cannot leak the gated transcript handle. + lines.close() + input.destroy() + } } export async function parseMessageGraphSessionContent( diff --git a/src/main/ai-vault/session-scanner-grok-parser.ts b/src/main/ai-vault/session-scanner-grok-parser.ts index c4a239ff91c..94e41e4b052 100644 --- a/src/main/ai-vault/session-scanner-grok-parser.ts +++ b/src/main/ai-vault/session-scanner-grok-parser.ts @@ -62,16 +62,13 @@ async function consumeGrokChatHistory( accumulator: SessionAccumulator, sessionDir: string ): Promise { + const input = openTranscriptReadStream( + join(sessionDir, 'chat_history.jsonl'), + { encoding: 'utf-8' }, + 'scan' + ) + const lines = createInterface({ input, crlfDelay: Infinity }) try { - const lines = createInterface({ - input: openTranscriptReadStream( - join(sessionDir, 'chat_history.jsonl'), - { encoding: 'utf-8' }, - 'scan' - ), - crlfDelay: Infinity - }) - for await (const line of lines) { const record = parseJsonObject(line) if (!record) { @@ -122,6 +119,11 @@ async function consumeGrokChatHistory( if (error instanceof WslTranscriptFsError) { throw error } + } finally { + // readline.close() leaves the underlying stream open; destroy it so a + // mid-read failure cannot leak the gated transcript handle. + lines.close() + input.destroy() } } diff --git a/src/main/ai-vault/session-scanner-kimi-parser.ts b/src/main/ai-vault/session-scanner-kimi-parser.ts index d2d43036b95..ba6deaa8fb8 100644 --- a/src/main/ai-vault/session-scanner-kimi-parser.ts +++ b/src/main/ai-vault/session-scanner-kimi-parser.ts @@ -91,11 +91,9 @@ async function consumeKimiWireTranscript( } } + const input = openTranscriptReadStream(wirePath, { encoding: 'utf-8' }, 'scan') + const lines = createInterface({ input, crlfDelay: Infinity }) try { - const lines = createInterface({ - input: openTranscriptReadStream(wirePath, { encoding: 'utf-8' }, 'scan'), - crlfDelay: Infinity - }) for await (const line of lines) { const record = parseJsonObject(line) if (!record) { @@ -126,6 +124,11 @@ async function consumeKimiWireTranscript( if (error instanceof WslTranscriptFsError) { throw error } + } finally { + // readline.close() leaves the underlying stream open; destroy it so a + // mid-read failure cannot leak the gated transcript handle. + lines.close() + input.destroy() } flushAssistant() } diff --git a/src/main/ai-vault/session-scanner-opencode-sources.ts b/src/main/ai-vault/session-scanner-opencode-sources.ts index 3c5bc6dc476..365e6334512 100644 --- a/src/main/ai-vault/session-scanner-opencode-sources.ts +++ b/src/main/ai-vault/session-scanner-opencode-sources.ts @@ -4,6 +4,7 @@ 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 { recordSessionScanIssue } from './session-scan-issues' import { discoverOpenCodeSessions } from './session-scanner-opencode-sqlite-discovery' import type { AiVaultScanOptions, SessionFileDiscovery } from './session-scanner-types' @@ -50,7 +51,7 @@ async function opencodeDbPathsForSource( } if (sourceIndex === 0) { return listOpenCodeDatabases((path, error) => { - issues.push({ agent: 'opencode', path, message: error.message }) + recordSessionScanIssue(issues, { agent: 'opencode', path, message: error.message }) }) } const wslHomeDir = wslHomeDirs[sourceIndex - 1] @@ -73,7 +74,11 @@ async function listOpenCodeDatabasesInDirectory( // 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 }) + recordSessionScanIssue(issues, { + agent: 'opencode', + path: dataDir, + message: error.message + }) } return [] } 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 index a79063194a8..9eea5176c66 100644 --- a/src/main/ai-vault/session-scanner-parse-wsl-stall.test.ts +++ b/src/main/ai-vault/session-scanner-parse-wsl-stall.test.ts @@ -5,11 +5,14 @@ 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() })) +const mocks = vi.hoisted(() => ({ open: vi.fn(), readdir: vi.fn() })) +// readdir too: every claude parse counts sibling subagent transcripts, and an +// unmocked one would reach the host UNC path and stall under fake timers. vi.mock('node:fs/promises', async (importOriginal) => ({ ...(await importOriginal()), - open: mocks.open + open: mocks.open, + readdir: mocks.readdir })) import { @@ -83,6 +86,8 @@ async function releaseAndSettle(): Promise { beforeEach(() => { resetSessionParseCacheForTests() mocks.open.mockReset() + mocks.readdir.mockReset() + mocks.readdir.mockResolvedValue([]) releaseStall = undefined vi.useFakeTimers() }) diff --git a/src/main/ai-vault/session-scanner-parser-stream-cleanup.test.ts b/src/main/ai-vault/session-scanner-parser-stream-cleanup.test.ts new file mode 100644 index 00000000000..7470b3430d2 --- /dev/null +++ b/src/main/ai-vault/session-scanner-parser-stream-cleanup.test.ts @@ -0,0 +1,127 @@ +import { beforeEach, describe, expect, it, vi } from 'vitest' +import { Readable } from 'node:stream' +import type * as NodeReadlineModule from 'node:readline' +import type * as WslFsAccessModule from '../native-chat/wsl-transcript-fs-access' +import type { FileWithMtime } from './session-scanner-types' + +// readline.close() leaves its input open, so a parser that stops consuming +// mid-file (a throw, a break) strands the gated transcript handle until the +// gate's deadline and delays every later scan. These pin the finally blocks. +const mocks = vi.hoisted(() => ({ + openStream: vi.fn(), + readFile: vi.fn(), + stat: vi.fn(), + createInterface: vi.fn() +})) + +vi.mock('../native-chat/wsl-transcript-fs-access', async (importOriginal) => ({ + ...(await importOriginal()), + openTranscriptReadStream: mocks.openStream, + wslGatedReadFile: mocks.readFile, + wslGatedStat: mocks.stat +})) + +vi.mock('node:readline', async (importOriginal) => ({ + ...(await importOriginal()), + createInterface: mocks.createInterface +})) + +import { parseAntigravitySessionFile } from './session-scanner-antigravity-parser' +import { parseDroidSessionFile } from './session-scanner-droid-parser' +import { parseMessageGraphSessionFile } from './session-scanner-graph-parsers' +import { parseGrokSessionFile } from './session-scanner-grok-parser' +import { parseKimiSessionFile } from './session-scanner-kimi-parser' +import { clearKimiSessionIndexCache } from './session-scanner-kimi-paths' + +const PARSE_FAILURE = 'parser failed mid-transcript' + +const opened: { path: string; stream: Readable }[] = [] +const interfaces: { close: ReturnType }[] = [] + +function file(path: string): FileWithMtime { + return { path, mtimeMs: 1, modifiedAt: '2026-06-01T10:05:00.000Z', sizeBytes: 128 } +} + +function lastOpened(): { path: string; stream: Readable } { + const entry = opened.at(-1) + if (!entry) { + throw new Error('no transcript stream was opened') + } + return entry +} + +function expectStreamTornDown(): void { + expect(lastOpened().stream.destroyed).toBe(true) + expect(interfaces.at(-1)?.close).toHaveBeenCalled() +} + +beforeEach(() => { + opened.length = 0 + interfaces.length = 0 + clearKimiSessionIndexCache() + mocks.openStream.mockReset() + mocks.readFile.mockReset() + mocks.stat.mockReset() + mocks.createInterface.mockReset() + + mocks.openStream.mockImplementation((path: string) => { + const stream = new Readable({ + read() { + this.push(null) + } + }) + opened.push({ path, stream }) + return stream + }) + // One line, then a consumer-side throw: the parser must still tear the + // stream down on its way out. + mocks.createInterface.mockImplementation(() => { + const lines = { + close: vi.fn(), + [Symbol.asyncIterator]: async function* () { + yield '{}' + throw new Error(PARSE_FAILURE) + } + } + interfaces.push(lines) + return lines + }) +}) + +describe('session parsers that stop consuming a gated transcript early', () => { + it.each([ + ['antigravity', () => parseAntigravitySessionFile(file('/w/conversation.jsonl'), 'linux')], + ['droid', () => parseDroidSessionFile(file('/w/session.jsonl'), 'linux')], + ['message graph', () => parseMessageGraphSessionFile('pi', file('/w/session.jsonl'), 'linux')] + ])('destroys the stream when the %s parse throws', async (_agent, parse) => { + await expect(parse()).rejects.toThrow(PARSE_FAILURE) + + expectStreamTornDown() + }) + + it('destroys the chat_history stream when the Grok parse swallows the failure', async () => { + mocks.readFile.mockResolvedValue(JSON.stringify({ info: { id: 'ses-1' } })) + + // Grok degrades to a summary-only session on a non-gate failure, so the + // teardown has no rejection to ride out on. + await expect( + parseGrokSessionFile(file('/w/.grok/sessions/ses-1/session.json')) + ).resolves.toBeTruthy() + + expect(lastOpened().path).toContain('chat_history.jsonl') + expectStreamTornDown() + }) + + it('destroys the wire stream when the Kimi parse swallows the failure', async () => { + mocks.readFile.mockResolvedValue(JSON.stringify({ title: 'Kimi session' })) + // No session_index.jsonl, so only the wire transcript opens a stream. + mocks.stat.mockRejectedValue(Object.assign(new Error('ENOENT'), { code: 'ENOENT' })) + + await expect( + parseKimiSessionFile(file('/w/.kimi-code/sessions/wd_app/session_abc/state.json')) + ).resolves.toBeTruthy() + + expect(lastOpened().path).toContain('wire.jsonl') + expectStreamTornDown() + }) +}) diff --git a/src/main/native-chat/wsl-transcript-fs-access.test.ts b/src/main/native-chat/wsl-transcript-fs-access.test.ts index b2bb893be4b..6f978453239 100644 --- a/src/main/native-chat/wsl-transcript-fs-access.test.ts +++ b/src/main/native-chat/wsl-transcript-fs-access.test.ts @@ -4,6 +4,7 @@ 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 OTHER_DISTRO_UNC_PATH = '\\\\wsl.localhost\\Debian\\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' @@ -47,15 +48,24 @@ import { wslGatedRead, wslGatedReaddir, wslGatedReadFile, - wslGatedStat + wslGatedStat, + WSL_TRANSCRIPT_READ_CHUNK_BYTES } from './wsl-transcript-fs-access' -import { WSL_TRANSCRIPT_FS_EXACT_TIMEOUT_MS, WslTranscriptFsError } from './wsl-transcript-fs-gate' +import { + resetWslTranscriptFsGateForTests, + WSL_TRANSCRIPT_FS_EXACT_TIMEOUT_MS, + WslTranscriptFsError +} from './wsl-transcript-fs-gate' function fakeHandle() { return { read: vi.fn(), close: vi.fn(async () => {}) } } beforeEach(() => { + // The gate mock delegates to the real implementation, so its module state + // survives between cases — a case that leaves a task stalled would otherwise + // mark that route stuck and fast-fail every later case on it. + resetWslTranscriptFsGateForTests() // runTask keeps the real gate implementation installed by the mock factory; // only its call log is cleared. mocks.runTask.mockClear() @@ -96,7 +106,20 @@ describe('transcript filesystem accessor off WSL UNC', () => { // Off UNC the raw stream is handed back verbatim, encoding included. expect(stream).toBe('raw-stream') expect(mocks.createReadStream).toHaveBeenCalledWith(path, { - encoding: 'utf-8' + encoding: 'utf-8', + signal: undefined + }) + }) + + it('forwards the caller signal so the local stream honours cancellation', () => { + mocks.createReadStream.mockReturnValue('raw-stream') + const controller = new AbortController() + + openTranscriptReadStream(POSIX_PATH, { start: 4 }, 'exact', controller.signal) + + expect(mocks.createReadStream).toHaveBeenCalledWith(POSIX_PATH, { + start: 4, + signal: controller.signal }) }) }) @@ -204,6 +227,31 @@ describe('transcript filesystem accessor on WSL UNC', () => { expect(second.close).toHaveBeenCalledTimes(1) }) + it('keeps a blocked close on one distro from stranding teardown on another', async () => { + let releaseStuck: (() => void) | undefined + const stuck = fakeHandle() + stuck.close.mockReturnValue( + new Promise((resolve) => { + releaseStuck = resolve + }) + ) + const healthy = fakeHandle() + + try { + await closeTranscriptHandle(stuck as never, UNC_PATH) + await closeTranscriptHandle(healthy as never, OTHER_DISTRO_UNC_PATH) + await new Promise((resolve) => setImmediate(resolve)) + + // A close that never settles on a stalled mount would hold a shared lane + // for the process lifetime, leaking every later descriptor with it. + expect(stuck.close).toHaveBeenCalledTimes(1) + expect(healthy.close).toHaveBeenCalledTimes(1) + } finally { + releaseStuck?.() + await new Promise((resolve) => setImmediate(resolve)) + } + }) + it.each([ ['stat', wslGatedStat, mocks.stat], ['lstat', wslGatedLstat, mocks.lstat] @@ -287,11 +335,11 @@ describe('transcript handle close off WSL UNC', () => { describe('per-chunk admission', () => { it('carries a codepoint straddling the 1 MiB chunk boundary across chunks', async () => { - const chunkBytes = 1024 * 1024 const emoji = Buffer.from('😀', 'utf8') - // Two of the emoji's four bytes land in chunk 1, two in chunk 2. + // Two of the emoji's four bytes land in chunk 1, two in chunk 2 — derived + // from the production constant so the fixture cannot drift off the boundary. const body = Buffer.concat([ - Buffer.alloc(chunkBytes - 2, 0x61), + Buffer.alloc(WSL_TRANSCRIPT_READ_CHUNK_BYTES - 2, 0x61), emoji, Buffer.from('tail\n', 'utf8') ]) diff --git a/src/main/native-chat/wsl-transcript-fs-access.ts b/src/main/native-chat/wsl-transcript-fs-access.ts index b9b2efdf071..b50c1b1f5f6 100644 --- a/src/main/native-chat/wsl-transcript-fs-access.ts +++ b/src/main/native-chat/wsl-transcript-fs-access.ts @@ -4,12 +4,13 @@ import { Readable } from 'node:stream' import { StringDecoder } from 'node:string_decoder' import { isWslUncPath } from '../../shared/wsl-paths' import { runWslTranscriptFsTask, type WslTranscriptFsTaskPriority } from './wsl-transcript-fs-gate' +import { wslTranscriptFsRouteKey } from './wsl-transcript-fs-route' /** Never nest a gated call inside another — that deadlocks the scan slot. */ // Why: one deadline per chunk instead of one for the whole file, so a large // healthy-but-slow transcript is not false-failed by a whole-file timeout. -const WSL_TRANSCRIPT_READ_CHUNK_BYTES = 1024 * 1024 +export const WSL_TRANSCRIPT_READ_CHUNK_BYTES = 1024 * 1024 type Operation = Parameters[0]['operation'] function runPathOperation( @@ -106,23 +107,33 @@ export function wslGatedRead( // gate's two permits are sized against, re-creating the process-wide stall the // gate exists to prevent; serializing them costs at most one extra busy thread. // The queue cannot grow without bound: opens on a stuck route already fast-fail. -const MAX_CONCURRENT_UNC_CLOSES = 1 -const queuedCloses: FileHandle[] = [] -let activeCloses = 0 +// Keyed by route like the gate's own admission, because a close that blocks on a +// stalled distro never settles — a shared queue would strand every later close, +// including handles on healthy distros, for the process lifetime. +const MAX_CONCURRENT_UNC_CLOSES_PER_ROUTE = 1 +type RouteCloseQueue = { queued: FileHandle[]; active: number } +const closeQueuesByRoute = new Map() -function drainQueuedCloses(): void { - while (activeCloses < MAX_CONCURRENT_UNC_CLOSES) { - const handle = queuedCloses.shift() +function drainQueuedCloses(route: string): void { + const lane = closeQueuesByRoute.get(route) + if (!lane) { + return + } + while (lane.active < MAX_CONCURRENT_UNC_CLOSES_PER_ROUTE) { + const handle = lane.queued.shift() if (!handle) { + if (lane.active === 0) { + closeQueuesByRoute.delete(route) + } return } - activeCloses += 1 + lane.active += 1 void handle .close() .catch(() => {}) .finally(() => { - activeCloses -= 1 - drainQueuedCloses() + lane.active -= 1 + drainQueuedCloses(route) }) } } @@ -138,8 +149,11 @@ export function closeTranscriptHandle(handle: FileHandle, path: string): Promise if (!isWslUncPath(path)) { return handle.close() } - queuedCloses.push(handle) - drainQueuedCloses() + const route = wslTranscriptFsRouteKey(path) + const lane = closeQueuesByRoute.get(route) ?? { queued: [], active: 0 } + closeQueuesByRoute.set(route, lane) + lane.queued.push(handle) + drainQueuedCloses(route) return Promise.resolve() } @@ -254,7 +268,9 @@ export function openTranscriptReadStream( signal?: AbortSignal ): Readable { if (!isWslUncPath(path)) { - return createReadStream(path, options) + // Node destroys the stream with an AbortError on abort, matching how the + // gated branch surfaces cancellation to the same consumers. + return createReadStream(path, { ...options, signal }) } 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 d4d87a4feeb..9c979002d01 100644 --- a/src/main/native-chat/wsl-transcript-fs-gate.test.ts +++ b/src/main/native-chat/wsl-transcript-fs-gate.test.ts @@ -659,11 +659,7 @@ 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 + function gatedRead(task: (signal: AbortSignal) => Promise): Promise { return runWslTranscriptFsTask( { operation: 'read', path: READ_PATH, priority: 'exact', dedupe: false }, task @@ -676,11 +672,11 @@ describe('WSL transcript filesystem task coalescing opt-out', () => { const bodies = [Buffer.from('AAAA'), Buffer.from('BBBB')] const reads = Promise.all([ - gatedRead(first, async () => { + gatedRead(async () => { bodies.shift()!.copy(first) return first }), - gatedRead(second, async () => { + gatedRead(async () => { bodies.shift()!.copy(second) return second }) diff --git a/src/main/native-chat/wsl-transcript-fs-gate.ts b/src/main/native-chat/wsl-transcript-fs-gate.ts index e214e1ddaa0..84848e13f11 100644 --- a/src/main/native-chat/wsl-transcript-fs-gate.ts +++ b/src/main/native-chat/wsl-transcript-fs-gate.ts @@ -1,4 +1,4 @@ -import { parseWslUncPath } from '../../shared/wsl-paths' +import { wslTranscriptFsRouteKey } from './wsl-transcript-fs-route' const MAX_CONCURRENT_WSL_TRANSCRIPT_FS_TASKS = 2 export const WSL_TRANSCRIPT_FS_EXACT_TIMEOUT_MS = 30_000 @@ -71,18 +71,6 @@ function abortReason(signal: AbortSignal): unknown { return signal.reason ?? new Error('WSL transcript filesystem task aborted') } -// wsl$ and wsl.localhost spellings of one distro stay distinct routes on -// purpose (provider spelling can change behavior), so a stuck spelling never -// fast-fails its twin — the twin is bounded by its own waiter deadlines. -function routeKey(path: string): string { - const normalized = path.replace(/\\/g, '/') - const match = normalized.match(/^\/\/(wsl\.localhost|wsl\$)\/([^/]+)/i) - if (match) { - return `${match[1].toLowerCase()}/${match[2].trim().toLowerCase()}` - } - return parseWslUncPath(path)?.distro.trim().toLowerCase() ?? path -} - function removeQueuedTask(task: UnknownScheduledTask): void { const index = queuedTasks.indexOf(task) if (index !== -1) { @@ -295,6 +283,27 @@ function pumpTasks(): void { } } +/** + * Test-only: drop every task and counter so a case that leaves a task stalled + * (holding a permit, marking its route stuck) cannot fast-fail the next one. + * Tasks are marked settled first, so a late resolution from the unabortable + * syscall they are still blocked on cannot decrement counters this just zeroed. + */ +export function resetWslTranscriptFsGateForTests(): void { + for (const task of [...activeTasks, ...queuedTasks]) { + task.state = 'settled' + clearTimeout(task.stuckTimer) + for (const waiter of task.waiters) { + removeWaiter(task, waiter) + } + } + activeTasks.clear() + queuedTasks.length = 0 + inFlightTasks.clear() + activeLaneKeys.clear() + activeScanCount = 0 +} + /** Bound 9P work without letting scans delay exact transcript probes. */ export function runWslTranscriptFsTask( options: { @@ -336,7 +345,7 @@ export function runWslTranscriptFsTask( } return attachWaiter(existing, options.signal) } - const route = routeKey(options.path) + const route = wslTranscriptFsRouteKey(options.path) // Fail fast when the permit, route, or scan slot this task needs is held by // stuck I/O — queueing would only burn the caller's full deadline. if (stuckBlocker(route, options.priority)) { diff --git a/src/main/native-chat/wsl-transcript-fs-route.ts b/src/main/native-chat/wsl-transcript-fs-route.ts new file mode 100644 index 00000000000..dc576c066d7 --- /dev/null +++ b/src/main/native-chat/wsl-transcript-fs-route.ts @@ -0,0 +1,19 @@ +import { parseWslUncPath } from '../../shared/wsl-paths' + +/** + * The isolation unit for WSL transcript filesystem work: one stalled distro + * must not fast-fail or strand work on any other. Shared by the gate's + * admission and by the ungated close queue so both isolate on the same key. + * + * wsl$ and wsl.localhost spellings of one distro stay distinct routes on + * purpose (provider spelling can change behavior), so a stuck spelling never + * fast-fails its twin — the twin is bounded by its own waiter deadlines. + */ +export function wslTranscriptFsRouteKey(path: string): string { + const normalized = path.replace(/\\/g, '/') + const match = normalized.match(/^\/\/(wsl\.localhost|wsl\$)\/([^/]+)/i) + if (match) { + return `${match[1].toLowerCase()}/${match[2].trim().toLowerCase()}` + } + return parseWslUncPath(path)?.distro.trim().toLowerCase() ?? path +} diff --git a/src/main/opencode-usage/scanner-wsl-gate.test.ts b/src/main/opencode-usage/scanner-wsl-gate.test.ts index 320077444aa..b7c94000635 100644 --- a/src/main/opencode-usage/scanner-wsl-gate.test.ts +++ b/src/main/opencode-usage/scanner-wsl-gate.test.ts @@ -21,7 +21,10 @@ vi.mock('node:fs/promises', async (importOriginal) => ({ })) import { listOpenCodeDatabases } from './scanner' -import { WSL_TRANSCRIPT_FS_SCAN_TIMEOUT_MS } from '../native-chat/wsl-transcript-fs-gate' +import { + WSL_TRANSCRIPT_FS_SCAN_TIMEOUT_MS, + WslTranscriptFsError +} from '../native-chat/wsl-transcript-fs-gate' // A stalled task holds the gate's single scan slot until it settles, so every // case releases its stall before the next one runs. @@ -82,16 +85,27 @@ afterEach(async () => { describe('OpenCode database discovery on a stalled WSL data directory', () => { it('gates the data-directory listing instead of hanging the scan', async () => { mocks.readdir.mockImplementation(stalls) + // Asserted because an empty list alone also matches a missing data dir; the + // AI Vault's OpenCode source turns this callback into the scan issue. + const onRefusal = vi.fn() - expect(await settlesOnlyAtTheScanDeadline(listOpenCodeDatabases())).toEqual([]) + expect(await settlesOnlyAtTheScanDeadline(listOpenCodeDatabases(onRefusal))).toEqual([]) + expect(onRefusal).toHaveBeenCalledWith(UNC_DATA_DIR, expect.any(WslTranscriptFsError)) }) it('gates an absolute UNC OPENCODE_DB probe instead of hanging the scan', async () => { process.env.OPENCODE_DB = UNC_DATABASE mocks.stat.mockImplementation(stalls) + const onRefusal = vi.fn() - expect(await settlesOnlyAtTheScanDeadline(listOpenCodeDatabases())).toEqual([]) + expect(await settlesOnlyAtTheScanDeadline(listOpenCodeDatabases(onRefusal))).toEqual([]) expect(mocks.readdir).not.toHaveBeenCalled() + // Loose on the path: `isAbsolute` is host-flavoured, so a UNC override only + // stays verbatim on Windows. The refusal reaching the caller is the contract. + expect(onRefusal).toHaveBeenCalledWith( + expect.stringContaining('opencode.db'), + expect.any(WslTranscriptFsError) + ) }) it('still lists databases when the distro answers', async () => {