diff --git a/config/build-plugins/plain-node-entry-guard.ts b/config/build-plugins/plain-node-entry-guard.ts index 45d34b944f7..441536e7edb 100644 --- a/config/build-plugins/plain-node-entry-guard.ts +++ b/config/build-plugins/plain-node-entry-guard.ts @@ -17,12 +17,13 @@ type OutputChunk = Rollup.OutputChunk // graph still resolves. // Entries executed as plain Node (ELECTRON_RUN_AS_NODE / no electron runtime): -// forked daemon, parcel-watcher and computer sidecars, and the CLI-run +// forked daemon, parcel-watcher, WSL filesystem and computer sidecars, and the CLI-run // agent-hooks entry. require("electron") throws MODULE_NOT_FOUND in all of them. const PLAIN_NODE_ENTRY_NAMES = [ 'daemon-entry', 'parcel-watcher-process-entry', 'computer-sidecar', + 'wsl-transcript-fs-process-entry', 'agent-hooks/managed-agent-hook-controls', 'codex/codex-app-server-grant-entry' ] as const diff --git a/config/electron-builder.config.cjs b/config/electron-builder.config.cjs index 577cbd97a93..f1a536bed5a 100644 --- a/config/electron-builder.config.cjs +++ b/config/electron-builder.config.cjs @@ -192,6 +192,7 @@ module.exports = { 'out/main/hermes/**', 'out/main/daemon-entry.js', 'out/main/session-scanner-service-entry.js', + 'out/main/wsl-transcript-fs-process-entry.js', 'out/main/session-scanner-opencode-sqlite-worker-entry.js', 'out/main/plugin-host-entry.js', 'out/main/computer-sidecar.js', diff --git a/config/scripts/electron-builder-config.test.mjs b/config/scripts/electron-builder-config.test.mjs index 75c3898858e..bf28401fd35 100644 --- a/config/scripts/electron-builder-config.test.mjs +++ b/config/scripts/electron-builder-config.test.mjs @@ -207,6 +207,14 @@ describe('electron-builder config', () => { ) }) + it('unpacks the replaceable WSL transcript filesystem process entry', async () => { + const entryFilename = 'wsl-transcript-fs-process-entry.js' + expect(electronBuilderConfig.asarUnpack).toContain(`out/main/${entryFilename}`) + + const viteConfig = await readFile(join(REPO_ROOT, 'electron.vite.config.ts'), 'utf8') + expect(viteConfig).toMatch(new RegExp(`'${entryFilename.replace(/\.js$/, '')}':\\s*resolve\\(`)) + }) + // Why: the scanner service is forked with ELECTRON_RUN_AS_NODE, so asar is // invisible to it and a packed worker entry fails closed — dropping every // OpenCode session in packaged builds while dev stays green. Three legs must diff --git a/electron.vite.config.ts b/electron.vite.config.ts index 7db7df7ca69..0b9ca284b95 100644 --- a/electron.vite.config.ts +++ b/electron.vite.config.ts @@ -223,6 +223,9 @@ export const electronViteConfig: UserConfig = { 'session-scanner-service-entry': resolve( 'src/main/ai-vault/session-scanner-service-entry.ts' ), + 'wsl-transcript-fs-process-entry': resolve( + 'src/main/native-chat/wsl-transcript-fs-process-entry.ts' + ), // Why: libuv spawns processes inline on the calling loop, so the port // scan's probe commands run on a worker thread instead of the UI one. 'port-scan-command-worker-entry': resolve( diff --git a/src/main/ai-vault/session-scanner-core-parser-wsl-stall.test.ts b/src/main/ai-vault/session-scanner-core-parser-wsl-stall.test.ts index 44fe71c80b2..5a07845aea0 100644 --- a/src/main/ai-vault/session-scanner-core-parser-wsl-stall.test.ts +++ b/src/main/ai-vault/session-scanner-core-parser-wsl-stall.test.ts @@ -29,6 +29,8 @@ import { resetSessionParseCacheForTests } from './session-scanner-parse-cache' import { + resetWslTranscriptFsGateForTests, + WSL_TRANSCRIPT_FS_ROUTE_QUARANTINE_BASE_MS, WSL_TRANSCRIPT_FS_SCAN_TIMEOUT_MS, WslTranscriptFsError } from '../native-chat/wsl-transcript-fs-gate' @@ -107,8 +109,19 @@ function missing(): Error { return Object.assign(new Error('ENOENT'), { code: 'ENOENT' }) } +// Complete: UNC readdir results pass through the child dispatcher's dirent +// serializer, which reads every kind flag. function dirent(name: string) { - return { name, isFile: () => true } + return { + name, + isBlockDevice: () => false, + isCharacterDevice: () => false, + isDirectory: () => false, + isFIFO: () => false, + isFile: () => true, + isSocket: () => false, + isSymbolicLink: () => false + } } function candidate(agent: SessionFileCandidate['agent'], path: string): SessionFileCandidate { @@ -132,13 +145,18 @@ async function expectRefusal(target: SessionFileCandidate): Promise { await refusal } +// A result that lands past the deadline never lifts the route quarantine, so +// recovery waits out the back-off window the same way production does. async function releaseAndSettle(): Promise { releaseStall?.() releaseStall = undefined - await vi.advanceTimersByTimeAsync(0) + await vi.advanceTimersByTimeAsync(WSL_TRANSCRIPT_FS_ROUTE_QUARANTINE_BASE_MS) } beforeEach(() => { + // blockedRoutes is persistent gate state: a prior stall must not quarantine + // this test's route. + resetWslTranscriptFsGateForTests() resetSessionParseCacheForTests() mocks.open.mockReset() mocks.readFile.mockReset() @@ -146,7 +164,8 @@ beforeEach(() => { mocks.stat.mockReset() releaseStall = undefined mocks.stat.mockRejectedValue(missing()) - vi.useFakeTimers() + // performance.now drives the route quarantine clock, so it must be faked too. + vi.useFakeTimers({ toFake: ['setTimeout', 'clearTimeout', 'Date', 'performance'] }) }) afterEach(async () => { 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 962a2e20fa9..5e0e686651c 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 @@ -12,6 +12,7 @@ vi.mock('node:fs/promises', async (importOriginal) => ({ })) import { + resetWslTranscriptFsGateForTests, WSL_TRANSCRIPT_FS_SCAN_TIMEOUT_MS, WslTranscriptFsError } from '../native-chat/wsl-transcript-fs-gate' @@ -20,8 +21,19 @@ 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.' +// Complete: UNC readdir results pass through the child dispatcher's dirent +// serializer, which reads every kind flag. function dirent(name: string): Dirent { - return { name, isDirectory: () => false, isFile: () => true } as Dirent + return { + name, + isBlockDevice: () => false, + isCharacterDevice: () => false, + isDirectory: () => false, + isFIFO: () => false, + isFile: () => true, + isSocket: () => false, + isSymbolicLink: () => false + } as Dirent } let releaseStall: (() => void) | undefined @@ -33,6 +45,7 @@ function stalls(): Promise { } beforeEach(() => { + resetWslTranscriptFsGateForTests() fsMocks.readdir.mockReset() fsMocks.stat.mockReset() fsMocks.readdir.mockResolvedValue([]) diff --git a/src/main/ai-vault/session-scanner-service-env.ts b/src/main/ai-vault/session-scanner-service-env.ts index 1bfb75344ca..e2f5309e70e 100644 --- a/src/main/ai-vault/session-scanner-service-env.ts +++ b/src/main/ai-vault/session-scanner-service-env.ts @@ -8,7 +8,8 @@ */ // What Node and libuv need to start and resolve a home, temp dir and locale. -const RUNTIME_ENV_ALLOWLIST = [ +// Exported for sibling plain-node forks (the WSL transcript fs process). +export const RUNTIME_ENV_ALLOWLIST = [ 'PATH', 'HOME', 'USERPROFILE', @@ -49,7 +50,7 @@ const AGENT_ROOT_ENV_ALLOWLIST = [ 'XDG_DATA_HOME' ] as const -function pickAllowedEnv( +export function pickAllowedEnv( keys: readonly string[], baseEnv: NodeJS.ProcessEnv, platform: NodeJS.Platform diff --git a/src/main/native-chat/host-readable-transcript-path.ts b/src/main/native-chat/host-readable-transcript-path.ts index 4d919099536..9337281cf7f 100644 --- a/src/main/native-chat/host-readable-transcript-path.ts +++ b/src/main/native-chat/host-readable-transcript-path.ts @@ -1,14 +1,10 @@ import { existsSync } from 'node:fs' -import { access } from 'node:fs/promises' import { isAbsolute } from 'node:path' import { isWslUncPath, parseWslUncPath, toWindowsWslPath } from '../../shared/wsl-paths' import { WSL_CODEX_RUNTIME_HOME_SEGMENTS } from '../pty/codex-home-wsl-env' import { getWslHomeAsync, listWslDistrosAsync } from '../wsl' -import { - runWslTranscriptFsTask, - wslTranscriptFsRefusal, - type WslTranscriptFsError -} from './wsl-transcript-fs-gate' +import { wslGatedAccess } from './wsl-transcript-fs-access' +import { WslTranscriptFsError, wslTranscriptFsRefusal } from './wsl-transcript-fs-gate' /** * True for guest-absolute Linux paths that Win32 cannot open as-is. @@ -50,15 +46,18 @@ async function pathExistsAsync(path: string, signal?: AbortSignal): Promise => { - try { - await access(path) - return true - } catch { - return false + try { + return await wslGatedAccess(path, 'exact', signal) + } catch (error) { + if (error instanceof WslTranscriptFsError) { + throw error } + // A caller abort stays authoritative — it must never read as "missing". + if (signal?.aborted) { + throw error + } + return false } - return runWslTranscriptFsTask({ operation: 'access', path, priority: 'exact', signal }, probe) } // Why: resolveSessionFilePath runs on a 500ms–5s poll loop. listWslDistrosAsync diff --git a/src/main/native-chat/transcript-read-cache-refusal-recovery.test.ts b/src/main/native-chat/transcript-read-cache-refusal-recovery.test.ts index 4e865b306fc..6fd5aa85aee 100644 --- a/src/main/native-chat/transcript-read-cache-refusal-recovery.test.ts +++ b/src/main/native-chat/transcript-read-cache-refusal-recovery.test.ts @@ -22,7 +22,9 @@ import { readNativeChatTranscriptCached } from './transcript-read-cache' import { + resetWslTranscriptFsGateForTests, WSL_TRANSCRIPT_FS_EXACT_TIMEOUT_MS, + WSL_TRANSCRIPT_FS_ROUTE_QUARANTINE_BASE_MS, WSL_TRANSCRIPT_FS_SLOW_MESSAGE } from './wsl-transcript-fs-gate' @@ -55,6 +57,8 @@ function transcriptHandle(body = BODY) { } beforeEach(() => { + // The deadline quarantines the route, and a late release never lifts it. + resetWslTranscriptFsGateForTests() clearNativeChatTranscriptCache() mocks.resolve.mockReset() mocks.stat.mockReset() @@ -63,7 +67,8 @@ beforeEach(() => { mocks.resolve.mockResolvedValue(UNC_PATH) // The file itself never changes across the refusal and the recovery. mocks.stat.mockResolvedValue({ mtimeMs: 42, size: BODY.length }) - vi.useFakeTimers() + // performance.now drives the route quarantine clock, so it must be faked too. + vi.useFakeTimers({ toFake: ['setTimeout', 'clearTimeout', 'Date', 'performance'] }) }) afterEach(async () => { @@ -82,7 +87,8 @@ describe('cached native chat transcript read after WSL gate refusals', () => { expect(await refused).toEqual({ error: WSL_TRANSCRIPT_FS_SLOW_MESSAGE }) releaseStall?.() releaseStall = undefined - await vi.advanceTimersByTimeAsync(0) + // The deadline's route back-off has to expire before the retry is admitted. + await vi.advanceTimersByTimeAsync(WSL_TRANSCRIPT_FS_ROUTE_QUARANTINE_BASE_MS) mocks.open.mockResolvedValue(transcriptHandle()) await expect(readNativeChatTranscriptCached('claude', 'session-id')).resolves.toMatchObject({ @@ -102,7 +108,8 @@ describe('cached native chat transcript read after WSL gate refusals', () => { releaseStall?.() releaseStall = undefined - await vi.advanceTimersByTimeAsync(0) + // The deadline's route back-off has to expire before the retry is admitted. + await vi.advanceTimersByTimeAsync(WSL_TRANSCRIPT_FS_ROUTE_QUARANTINE_BASE_MS) mocks.open.mockResolvedValue(transcriptHandle()) const recovered = await readNativeChatTranscriptCached('claude', 'session-id') diff --git a/src/main/native-chat/transcript-read-cache.ts b/src/main/native-chat/transcript-read-cache.ts index ee0c8ddcafd..ab5c220bfd1 100644 --- a/src/main/native-chat/transcript-read-cache.ts +++ b/src/main/native-chat/transcript-read-cache.ts @@ -3,11 +3,10 @@ import { resolveSessionFilePath } from './session-file-resolver' import { readNativeChatTranscript, type ReadTranscriptResult } from './transcript-reader' import { wslGatedStat } from './wsl-transcript-fs-access' import { - WSL_TRANSCRIPT_FS_CAPACITY_MESSAGE, - WSL_TRANSCRIPT_FS_SLOW_MESSAGE, + isWslTranscriptFsRefusalMessage, WslTranscriptFsError, wslTranscriptFsRefusal -} from './wsl-transcript-fs-gate' +} from './wsl-transcript-fs-error' // 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 @@ -150,11 +149,7 @@ export async function readNativeChatTranscriptCached( // The reader flattens a refusal into its message, so that is the only handle // this layer has on one. function isGateRefusal(result: ReadTranscriptResult): boolean { - return ( - 'error' in result && - (result.error === WSL_TRANSCRIPT_FS_SLOW_MESSAGE || - result.error === WSL_TRANSCRIPT_FS_CAPACITY_MESSAGE) - ) + return 'error' in result && isWslTranscriptFsRefusalMessage(result.error) } /** Test-only: drop the transcript parse cache between runs. */ diff --git a/src/main/native-chat/transcript-tail-boundary.ts b/src/main/native-chat/transcript-tail-boundary.ts new file mode 100644 index 00000000000..b4479ac2bd0 --- /dev/null +++ b/src/main/native-chat/transcript-tail-boundary.ts @@ -0,0 +1,65 @@ +import { wslGatedRead } from './wsl-transcript-fs-access' +import type { TranscriptFileHandle } from './wsl-transcript-fs-access' + +export const TAIL_CHUNK_BYTES = 64 * 1024 + +/** + * The byte at `position`, or null when the file shrank below it between the + * caller's stat and this read (allocUnsafe would otherwise hand back garbage). + */ +export async function readTranscriptByteAt( + handle: TranscriptFileHandle, + filePath: string, + position: number, + signal?: AbortSignal +): Promise { + const byte = Buffer.allocUnsafe(1) + const { bytesRead } = await wslGatedRead(handle, filePath, byte, 0, 1, position, 'exact', signal) + signal?.throwIfAborted() + return bytesRead === 1 ? byte[0] : null +} + +/** End offset (exclusive) of the last newline-terminated line at or before `end`. */ +export async function findLastCompleteLineEnd( + handle: TranscriptFileHandle, + filePath: string, + end: number, + signal?: AbortSignal +): Promise { + signal?.throwIfAborted() + const lastByte = await readTranscriptByteAt(handle, filePath, end - 1, signal) + if (lastByte === null) { + // File shrank between stat and probe. + return 0 + } + if (lastByte === 0x0a) { + return end + } + let cursor = end + while (cursor > 0) { + signal?.throwIfAborted() + const start = Math.max(0, cursor - TAIL_CHUNK_BYTES) + const buffer = Buffer.allocUnsafe(cursor - start) + const { bytesRead } = await wslGatedRead( + handle, + filePath, + buffer, + 0, + buffer.length, + start, + 'exact', + signal + ) + signal?.throwIfAborted() + if (bytesRead < buffer.length) { + // File shrank mid-walk: any boundary computed from stale offsets is wrong. + return 0 + } + const newline = buffer.subarray(0, bytesRead).lastIndexOf(0x0a) + if (newline !== -1) { + return start + newline + 1 + } + cursor = start + } + return 0 +} 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 da040de2794..89e06bc8814 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 @@ -18,7 +18,11 @@ vi.mock('node:fs/promises', async (importOriginal) => ({ })) import { readNativeChatTranscriptTail } from './transcript-tail-reader' -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' 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.' @@ -48,7 +52,9 @@ describe('native chat transcript tail under WSL gate refusals', () => { 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. + // each case releases its stall before the next one runs. The deadline also + // quarantines the route, and a late release never lifts that, so each case + // starts from a reset gate. let releaseStall: (() => void) | undefined function stalls(): Promise { @@ -69,6 +75,7 @@ describe('native chat transcript tail with stalled post-resolution UNC I/O', () } beforeEach(() => { + resetWslTranscriptFsGateForTests() mocks.stat.mockReset() mocks.open.mockReset() releaseStall = undefined diff --git a/src/main/native-chat/transcript-tail-reader.ts b/src/main/native-chat/transcript-tail-reader.ts index e9cbe430090..b6fa9bafdb0 100644 --- a/src/main/native-chat/transcript-tail-reader.ts +++ b/src/main/native-chat/transcript-tail-reader.ts @@ -1,4 +1,3 @@ -import type { FileHandle } from 'node:fs/promises' import type { AgentType, NativeChatMessage, @@ -17,6 +16,11 @@ import { nativeChatTurnLifecycleDecoderForAgent, type NativeChatTurnLifecycleDecoder } from './transcript-turn-lifecycle' +import { + findLastCompleteLineEnd, + readTranscriptByteAt, + TAIL_CHUNK_BYTES +} from './transcript-tail-boundary' import { closeTranscriptHandle, wslGatedOpen, @@ -26,7 +30,6 @@ import { import { wslTranscriptFsRefusal } from './wsl-transcript-fs-gate' export const MAX_NATIVE_CHAT_TRANSCRIPT_RECORD_BYTES = 2 * 1024 * 1024 -const TAIL_CHUNK_BYTES = 64 * 1024 export type NativeChatLineDecoder = (line: string, fallbackId: string) => NativeChatMessage | null @@ -90,11 +93,13 @@ export async function readNativeChatTranscriptTailFile( return { messages: [], consumedTo: 0, hasMore: false, beforeOffset: 0 } } const newestFirst: { message: NativeChatMessage; offset: number }[] = [] - const finalByte = Buffer.allocUnsafe(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) + const finalByte = await readTranscriptByteAt(handle, filePath, consumedTo - 1, signal) + if (finalByte === null) { + // File shrank between stat and probe: report empty, the next poll re-stats. + return { messages: [], consumedTo: 0, hasMore: false, beforeOffset: 0 } + } + ignoreNextMalformedRecord = finalByte !== 0x0a + let cursor = consumedTo - (finalByte === 0x0a ? 1 : 0) while (cursor > 0 && newestFirst.length <= limit) { signal?.throwIfAborted() const start = Math.max(0, cursor - TAIL_CHUNK_BYTES) @@ -110,6 +115,11 @@ export async function readNativeChatTranscriptTailFile( signal ) signal?.throwIfAborted() + // A short read means the file shrank mid-walk: stop paging back rather + // than stitch non-adjacent bytes into records. + if (bytesRead < buffer.length) { + break + } let segmentEnd = bytesRead for (let index = bytesRead - 1; index >= 0 && newestFirst.length <= limit; index--) { if (buffer[index] !== 0x0a) { @@ -201,44 +211,6 @@ export async function readNativeChatTranscriptTailFile( } } -async function findLastCompleteLineEnd( - handle: FileHandle, - filePath: string, - end: number, - signal?: AbortSignal -): Promise { - signal?.throwIfAborted() - const lastByte = Buffer.allocUnsafe(1) - await wslGatedRead(handle, filePath, lastByte, 0, 1, end - 1, 'exact', signal) - signal?.throwIfAborted() - if (lastByte[0] === 0x0a) { - return end - } - let cursor = end - while (cursor > 0) { - signal?.throwIfAborted() - const start = Math.max(0, cursor - TAIL_CHUNK_BYTES) - const buffer = Buffer.allocUnsafe(cursor - 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) { - return start + newline + 1 - } - cursor = start - } - return 0 -} - export async function readNativeChatTranscriptTail( args: ResolveSessionFileOptions & { agent: AgentType diff --git a/src/main/native-chat/transcript-watch-wsl-stall.test.ts b/src/main/native-chat/transcript-watch-wsl-stall.test.ts index 3fd066325bb..3c508151016 100644 --- a/src/main/native-chat/transcript-watch-wsl-stall.test.ts +++ b/src/main/native-chat/transcript-watch-wsl-stall.test.ts @@ -26,7 +26,10 @@ vi.mock('node:fs/promises', async (importOriginal) => ({ })) import { subscribeNativeChatTranscript } from './transcript-watch' -import { WSL_TRANSCRIPT_FS_EXACT_TIMEOUT_MS } from './wsl-transcript-fs-gate' +import { + resetWslTranscriptFsGateForTests, + 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.' @@ -61,6 +64,9 @@ function trackUnhandled(reason: unknown): void { } beforeEach(() => { + // blockedRoutes is persistent gate state: a prior stall must not quarantine + // this test's route. + resetWslTranscriptFsGateForTests() mocks.resolve.mockReset() mocks.stat.mockReset() mocks.open.mockReset() @@ -82,7 +88,13 @@ const EMPTY_STATS = { size: 0, mtimeMs: 1, ctimeMs: 1, ino: 1, dev: 1, mtime: ne 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) + // Every stat stalls until the distro "wakes": the first-strike quarantine + // runs on the real performance.now clock, so depending on wall time a retry + // probe may be re-admitted mid-test and must find the mount still hung. + const pendingStats: ((stats: typeof EMPTY_STATS) => void)[] = [] + mocks.stat.mockImplementation( + () => new Promise((resolve) => pendingStats.push(resolve as never)) + ) const snapshots: Snapshot[] = [] // Not awaited yet: the setup install is itself blocked on the stalled stat. @@ -99,7 +111,10 @@ describe('native chat transcript subscription with a stalled install stat', () = expect(snapshots).toHaveLength(1) // The distro wakes: the same subscription still delivers a real snapshot. - releaseStall?.() + mocks.stat.mockResolvedValue(EMPTY_STATS) + for (const release of pendingStats.splice(0)) { + release(EMPTY_STATS) + } await vi.advanceTimersByTimeAsync(500) expect(snapshots.length).toBeGreaterThan(1) diff --git a/src/main/native-chat/wsl-codex-session-path-scan.ts b/src/main/native-chat/wsl-codex-session-path-scan.ts index 804bcfe858e..fbffa297ee8 100644 --- a/src/main/native-chat/wsl-codex-session-path-scan.ts +++ b/src/main/native-chat/wsl-codex-session-path-scan.ts @@ -1,8 +1,7 @@ import type { Dirent } from 'node:fs' -import { readdir } from 'node:fs/promises' import { basename, extname } from 'node:path' import { walkSessionFiles } from '../ai-vault/session-scanner-discovery' -import { runWslTranscriptFsTask } from './wsl-transcript-fs-gate' +import { wslGatedReaddir } from './wsl-transcript-fs-access' type ScanWaiter = { sessionId: string @@ -23,10 +22,7 @@ type ScanGeneration = { const inFlightScans = new Map() function readDirectory(dirPath: string, signal: AbortSignal): Promise { - return runWslTranscriptFsTask( - { operation: 'readdir', path: dirPath, priority: 'scan', signal }, - () => readdir(dirPath, { withFileTypes: true }) - ) + return wslGatedReaddir(dirPath, 'scan', signal) } function sessionFileName(path: string): string { 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 6f978453239..aaaf18f2698 100644 --- a/src/main/native-chat/wsl-transcript-fs-access.test.ts +++ b/src/main/native-chat/wsl-transcript-fs-access.test.ts @@ -4,7 +4,6 @@ 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' @@ -166,7 +165,7 @@ describe('transcript filesystem accessor on WSL UNC', () => { mocks.open.mockResolvedValue(handle) await expect(readTranscriptSlice(UNC_PATH, 4, 8, 'scan')).rejects.toThrow('EIO') - expect(handle.close).toHaveBeenCalledTimes(1) + expect(handle.close).toHaveBeenCalled() }) it('yields Buffer chunks and closes the handle when the consumer destroys the stream', async () => { @@ -193,7 +192,7 @@ describe('transcript filesystem accessor on WSL UNC', () => { 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) + expect(handle.close).toHaveBeenCalled() }) it('swallows close failures so teardown never rejects', async () => { @@ -203,31 +202,7 @@ describe('transcript filesystem accessor on WSL UNC', () => { await new Promise((resolve) => setImmediate(resolve)) }) - it('drains handle closes one at a time so teardown cannot flood the thread pool', async () => { - let releaseFirst: (() => void) | undefined - const first = fakeHandle() - first.close.mockReturnValue( - new Promise((resolve) => { - releaseFirst = resolve - }) - ) - const second = fakeHandle() - - await closeTranscriptHandle(first as never, UNC_PATH) - await closeTranscriptHandle(second as never, UNC_PATH) - await new Promise((resolve) => setImmediate(resolve)) - - // A blocked uv_fs_close holds a libuv thread the gate cannot see, so the - // second one must wait rather than occupy a thread of its own. - expect(first.close).toHaveBeenCalledTimes(1) - expect(second.close).not.toHaveBeenCalled() - - releaseFirst?.() - await new Promise((resolve) => setImmediate(resolve)) - expect(second.close).toHaveBeenCalledTimes(1) - }) - - it('keeps a blocked close on one distro from stranding teardown on another', async () => { + it('does not await a UNC close, mirroring the process-handle contract', async () => { let releaseStuck: (() => void) | undefined const stuck = fakeHandle() stuck.close.mockReturnValue( @@ -235,17 +210,11 @@ describe('transcript filesystem accessor on WSL UNC', () => { 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. + // A close blocked on a stalled mount must not block the caller's teardown. + await expect(closeTranscriptHandle(stuck as never, UNC_PATH)).resolves.toBeUndefined() expect(stuck.close).toHaveBeenCalledTimes(1) - expect(healthy.close).toHaveBeenCalledTimes(1) } finally { releaseStuck?.() await new Promise((resolve) => setImmediate(resolve)) @@ -289,7 +258,7 @@ describe('transcript filesystem accessor on WSL UNC', () => { } ) - it('closes an abandoned open whose syscall lands after the caller gave up', async () => { + it('disposes a late open result after the deadline already settled the task', async () => { vi.useFakeTimers() let release: ((handle: unknown) => void) | undefined mocks.open.mockReturnValue( @@ -306,7 +275,7 @@ describe('transcript filesystem accessor on WSL UNC', () => { release?.(handle) await vi.advanceTimersByTimeAsync(0) - // Nobody received the handle, so the gate owns closing it. + // Nobody was left to own the descriptor, so the gate's disposer closed it. 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 index b50c1b1f5f6..c048fed0605 100644 --- a/src/main/native-chat/wsl-transcript-fs-access.ts +++ b/src/main/native-chat/wsl-transcript-fs-access.ts @@ -1,29 +1,58 @@ import { createReadStream, type Dirent, type Stats } from 'node:fs' -import { lstat, open, readdir, readFile, stat, type FileHandle } from 'node:fs/promises' +import { access, lstat, open, readdir, readFile, stat, type FileHandle } from 'node:fs/promises' 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' +import { + closeWslTranscriptFsProcess, + isWslTranscriptFsProcessHandle, + openWslTranscriptFsProcess, + readWslTranscriptFsProcess, + runWslTranscriptFsProcess, + type WslTranscriptFsProcessHandle +} from './wsl-transcript-fs-process-dispatch' +import type { WslTranscriptFsReusableProcessCall } from './wsl-transcript-fs-process-protocol' +import { wslTranscriptFsLaneKey } 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. export const WSL_TRANSCRIPT_READ_CHUNK_BYTES = 1024 * 1024 -type Operation = Parameters[0]['operation'] -function runPathOperation( - operation: Operation, - path: string, +export type TranscriptFileHandle = FileHandle | WslTranscriptFsProcessHandle + +// One request object drives both the gate's dedupe/route key and the child +// call, so the two can never disagree on the operation or path. +function runReusableFsOperation( + request: WslTranscriptFsReusableProcessCall, priority: WslTranscriptFsTaskPriority, signal: AbortSignal | undefined, - task: () => Promise, - options?: { dedupe?: boolean; onAbandonedResult?: (value: T) => void } + localTask: () => Promise ): Promise { - return isWslUncPath(path) - ? runWslTranscriptFsTask({ operation, path, priority, signal, ...options }, task) - : task() + return isWslUncPath(request.path) + ? runWslTranscriptFsTask( + { operation: request.operation, path: request.path, priority, signal }, + (taskSignal) => + runWslTranscriptFsProcess( + request, + taskSignal, + wslTranscriptFsLaneKey(request.path, priority) + ) + ) + : localTask() +} + +export function wslGatedAccess( + path: string, + priority: WslTranscriptFsTaskPriority, + signal?: AbortSignal +): Promise { + return runReusableFsOperation({ operation: 'access', path }, priority, signal, async () => { + await access(path) + return true + }) } export function wslGatedStat( @@ -31,7 +60,7 @@ export function wslGatedStat( priority: WslTranscriptFsTaskPriority, signal?: AbortSignal ): Promise { - return runPathOperation('stat', path, priority, signal, () => stat(path)) + return runReusableFsOperation({ operation: 'stat', path }, priority, signal, () => stat(path)) } export function wslGatedLstat( @@ -39,7 +68,7 @@ export function wslGatedLstat( priority: WslTranscriptFsTaskPriority, signal?: AbortSignal ): Promise { - return runPathOperation('lstat', path, priority, signal, () => lstat(path)) + return runReusableFsOperation({ operation: 'lstat', path }, priority, signal, () => lstat(path)) } export function wslGatedReaddir( @@ -47,7 +76,7 @@ export function wslGatedReaddir( priority: WslTranscriptFsTaskPriority, signal?: AbortSignal ): Promise { - return runPathOperation('readdir', path, priority, signal, () => + return runReusableFsOperation({ operation: 'readdir', path }, priority, signal, () => readdir(path, { withFileTypes: true }) ) } @@ -58,7 +87,9 @@ export function wslGatedReadFile( priority: WslTranscriptFsTaskPriority, signal?: AbortSignal ): Promise { - return runPathOperation('readfile', path, priority, signal, () => readFile(path, encoding)) + return runReusableFsOperation({ operation: 'readfile', path, encoding }, priority, signal, () => + readFile(path, encoding) + ) } // dedupe:false — two joiners would share one FileHandle and both close it. @@ -66,13 +97,22 @@ export function wslGatedOpen( path: string, priority: WslTranscriptFsTaskPriority, signal?: AbortSignal -): Promise { - return runPathOperation('open', path, priority, signal, () => open(path, 'r'), { - dedupe: false, - // An unabortable open can still succeed after its caller timed out or - // cancelled; without this the descriptor leaks for the process lifetime. - onAbandonedResult: (handle) => void closeTranscriptHandle(handle, path) - }) +): Promise { + if (!isWslUncPath(path)) { + return open(path, 'r') + } + return runWslTranscriptFsTask( + { + operation: 'open', + path, + priority, + signal, + dedupe: false, + onAbandonedResult: (handle) => void closeTranscriptHandle(handle, path) + }, + (taskSignal) => + openWslTranscriptFsProcess(path, taskSignal, wslTranscriptFsLaneKey(path, priority)) + ) } /** @@ -82,7 +122,7 @@ export function wslGatedOpen( * (often `Buffer.allocUnsafe`) stays uninitialized. */ export function wslGatedRead( - handle: FileHandle, + handle: TranscriptFileHandle, path: string, buffer: Buffer, offset: number, @@ -91,69 +131,37 @@ export function wslGatedRead( priority: WslTranscriptFsTaskPriority, signal?: AbortSignal ): Promise<{ bytesRead: number; buffer: Buffer }> { - return runPathOperation( - 'read', - path, - priority, - signal, - () => handle.read(buffer, offset, length, position), - { dedupe: false } + // Handle kind decides before path spelling: a process-owned handle must never + // hit the FileHandle branch even if a caller re-derives the path off-UNC. + if (!isWslUncPath(path) && !isWslTranscriptFsProcessHandle(handle)) { + return (handle as FileHandle).read(buffer, offset, length, position) + } + return runWslTranscriptFsTask( + { operation: 'read', path, priority, signal, dedupe: false }, + async (taskSignal) => { + if (!isWslTranscriptFsProcessHandle(handle)) { + // The vitest fallback: a FileHandle read fills the caller's buffer itself. + return (handle as FileHandle).read(buffer, offset, length, position) + } + const body = await readWslTranscriptFsProcess(handle, position, length, taskSignal) + buffer.set(body, offset) + return { bytesRead: body.byteLength, buffer } + } ) } -// A blocked `uv_fs_close` holds a libuv threadpool thread just like a blocked -// read does, but it is invisible to the gate — so UNC teardown drains one handle -// at a time. Unbounded fire-and-forget closes are what would exhaust the pool the -// gate's two permits are sized against, re-creating the process-wide stall the -// gate exists to prevent; serializing them costs at most one extra busy thread. -// The queue cannot grow without bound: opens on a stuck route already fast-fail. -// 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(route: string): void { - const lane = closeQueuesByRoute.get(route) - if (!lane) { - return +/** Never gated; process-owned handles retire on the client's bounded deadline. */ +export function closeTranscriptHandle(handle: TranscriptFileHandle, path: string): Promise { + if (isWslTranscriptFsProcessHandle(handle)) { + void closeWslTranscriptFsProcess(handle).catch(() => {}) + return Promise.resolve() } - while (lane.active < MAX_CONCURRENT_UNC_CLOSES_PER_ROUTE) { - const handle = lane.queued.shift() - if (!handle) { - if (lane.active === 0) { - closeQueuesByRoute.delete(route) - } - return - } - lane.active += 1 - void handle - .close() - .catch(() => {}) - .finally(() => { - lane.active -= 1 - drainQueuedCloses(route) - }) - } -} - -/** - * Never gated. Off UNC this is the prior contract verbatim — the caller awaits - * fd teardown and a close failure surfaces. On UNC it is fire-and-forget: - * closing a handle on a stalled mount can itself block, and a gated close would - * burn a permit and a waiter deadline purely for teardown. One leaked fd until - * the OS unblocks beats a second blocked waiter. - */ -export function closeTranscriptHandle(handle: FileHandle, path: string): Promise { if (!isWslUncPath(path)) { return handle.close() } - const route = wslTranscriptFsRouteKey(path) - const lane = closeQueuesByRoute.get(route) ?? { queued: [], active: 0 } - closeQueuesByRoute.set(route, lane) - lane.queued.push(handle) - drainQueuedCloses(route) + // Only the vitest fallback pairs a FileHandle with a UNC path; mirror the + // process-handle contract there: fire-and-forget, close failures swallowed. + void handle.close().catch(() => {}) return Promise.resolve() } diff --git a/src/main/native-chat/wsl-transcript-fs-error.ts b/src/main/native-chat/wsl-transcript-fs-error.ts new file mode 100644 index 00000000000..f9f38812545 --- /dev/null +++ b/src/main/native-chat/wsl-transcript-fs-error.ts @@ -0,0 +1,56 @@ +export const WSL_TRANSCRIPT_FS_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.' +export const WSL_TRANSCRIPT_FS_CAPACITY_MESSAGE = + 'WSL transcript discovery is temporarily unavailable because too many filesystem requests are already waiting. Try again shortly or restart Orca if the issue continues.' +const WSL_TRANSCRIPT_FS_PROCESS_FAILURE_PREFIX = + 'WSL transcript files are temporarily unavailable because the filesystem helper process failed' + +export type WslTranscriptFsFailureCode = 'timeout' | 'capacity' | 'unavailable' + +export class WslTranscriptFsError extends Error { + constructor( + readonly code: WslTranscriptFsFailureCode, + message: string + ) { + super(message) + this.name = 'WslTranscriptFsError' + } +} + +export function wslTranscriptFsTimeoutError(): WslTranscriptFsError { + return new WslTranscriptFsError('timeout', WSL_TRANSCRIPT_FS_SLOW_MESSAGE) +} + +export function wslTranscriptFsCapacityError(): WslTranscriptFsError { + return new WslTranscriptFsError('capacity', WSL_TRANSCRIPT_FS_CAPACITY_MESSAGE) +} + +export function wslTranscriptFsUnavailableError(): WslTranscriptFsError { + return new WslTranscriptFsError('unavailable', WSL_TRANSCRIPT_FS_SLOW_MESSAGE) +} + +/** Helper-process transport fault: nothing was consulted about the mount. */ +export function wslTranscriptFsProcessFailureError(detail: unknown): WslTranscriptFsError { + const text = detail instanceof Error ? detail.message : String(detail) + return new WslTranscriptFsError( + 'unavailable', + `${WSL_TRANSCRIPT_FS_PROCESS_FAILURE_PREFIX} (${text}). Try again shortly or restart Orca if the issue continues.` + ) +} + +/** Layers that only see a flattened message use this to spot a refusal. */ +export function isWslTranscriptFsRefusalMessage(message: string): boolean { + return ( + message === WSL_TRANSCRIPT_FS_SLOW_MESSAGE || + message === WSL_TRANSCRIPT_FS_CAPACITY_MESSAGE || + message.startsWith(WSL_TRANSCRIPT_FS_PROCESS_FAILURE_PREFIX) + ) +} + +/** Narrow a caught error to a gate refusal, rethrowing anything else. */ +export function wslTranscriptFsRefusal(error: unknown): WslTranscriptFsError { + if (error instanceof WslTranscriptFsError) { + return error + } + throw error +} 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 9c979002d01..b886168949a 100644 --- a/src/main/native-chat/wsl-transcript-fs-gate.test.ts +++ b/src/main/native-chat/wsl-transcript-fs-gate.test.ts @@ -1,9 +1,11 @@ import { afterEach, beforeEach, describe, expect, it, vi, type MockInstance } from 'vitest' import { + resetWslTranscriptFsGateForTests, runWslTranscriptFsTask, WSL_TRANSCRIPT_FS_EXACT_TIMEOUT_MS, WSL_TRANSCRIPT_FS_MAX_PENDING_TASKS, WSL_TRANSCRIPT_FS_MAX_WAITERS_PER_TASK, + WSL_TRANSCRIPT_FS_ROUTE_QUARANTINE_BASE_MS, WSL_TRANSCRIPT_FS_SCAN_TIMEOUT_MS, WslTranscriptFsError } from './wsl-transcript-fs-gate' @@ -43,6 +45,7 @@ describe('WSL transcript filesystem task scheduling', () => { let warnSpy: MockInstance beforeEach(() => { + resetWslTranscriptFsGateForTests() warnSpy = vi.spyOn(console, 'warn').mockImplementation(() => {}) }) @@ -278,7 +281,7 @@ describe('WSL transcript filesystem task scheduling', () => { } }) - it('does not let one timed-out waiter settle a later shared waiter', async () => { + it('expires every waiter when shared work reaches its enforced deadline', async () => { vi.useFakeTimers() const work = deferred() const task = vi.fn(() => work.promise) @@ -288,11 +291,11 @@ describe('WSL transcript filesystem task scheduling', () => { const firstRejected = expect(first).rejects.toMatchObject({ code: 'timeout' }) await vi.advanceTimersByTimeAsync(10_000) const second = run(path, 'exact', task) + const secondRejected = expect(second).rejects.toMatchObject({ code: 'timeout' }) await vi.advanceTimersByTimeAsync(WSL_TRANSCRIPT_FS_EXACT_TIMEOUT_MS - 10_000) - await firstRejected + await Promise.all([firstRejected, secondRejected]) work.resolve('shared') - await expect(second).resolves.toBe('shared') expect(task).toHaveBeenCalledOnce() expect(vi.getTimerCount()).toBe(0) } finally { @@ -380,7 +383,7 @@ describe('WSL transcript filesystem task scheduling', () => { } }) - it('fails new work fast after both running permits stall', async () => { + it('restores both permits after running work exceeds its deadline', async () => { vi.useFakeTimers() const ubuntu = deferred() const debian = deferred() @@ -393,10 +396,10 @@ describe('WSL transcript filesystem task scheduling', () => { await vi.advanceTimersByTimeAsync(WSL_TRANSCRIPT_FS_EXACT_TIMEOUT_MS) await Promise.all([firstRejected, secondRejected]) const laterTask = vi.fn(async () => 'later') - await expect( - run('\\\\wsl.localhost\\Fedora\\later', 'exact', laterTask) - ).rejects.toMatchObject({ code: 'unavailable', message: SLOW_MESSAGE }) - expect(laterTask).not.toHaveBeenCalled() + await expect(run('\\\\wsl.localhost\\Fedora\\later', 'exact', laterTask)).resolves.toBe( + 'later' + ) + expect(laterTask).toHaveBeenCalledOnce() expect(vi.getTimerCount()).toBe(0) } finally { ubuntu.resolve('ubuntu') @@ -406,7 +409,144 @@ describe('WSL transcript filesystem task scheduling', () => { } }) - it('fails only work needing a stuck lane fast, keeping the other permit usable', async () => { + // Why: production tasks reject promptly on abort (the isolated process is + // killed). The sole waiter's timeout must not pre-empt the deadline timer — + // that would settle the task first and skip the route quarantine entirely. + it('quarantines the route when a sole-waiter abort-responsive task hits the deadline', async () => { + vi.useFakeTimers() + try { + const path = '\\\\wsl.localhost\\Ubuntu\\solo-stall' + const stalled = runWslTranscriptFsTask( + { operation: 'open', path, priority: 'exact', dedupe: false }, + (signal) => + new Promise((_resolve, reject) => + signal.addEventListener('abort', () => reject(signal.reason), { once: true }) + ) + ) + const stalledRejected = expect(stalled).rejects.toMatchObject({ code: 'timeout' }) + await vi.advanceTimersByTimeAsync(WSL_TRANSCRIPT_FS_EXACT_TIMEOUT_MS) + await stalledRejected + const retryTask = vi.fn(async () => 'retry') + await expect(run(path, 'exact', retryTask)).rejects.toMatchObject({ code: 'unavailable' }) + expect(retryTask).not.toHaveBeenCalled() + expect(vi.getTimerCount()).toBe(0) + } finally { + vi.useRealTimers() + } + }) + + it('lifts a first-strike quarantine quickly and escalates on repeat stalls', async () => { + // performance.now drives the quarantine clock, so it must be faked too. + vi.useFakeTimers({ toFake: ['setTimeout', 'clearTimeout', 'Date', 'performance'] }) + const path = '\\\\wsl.localhost\\Ubuntu\\cold-start' + const stallOnce = (): Promise => + runWslTranscriptFsTask( + { operation: 'open', path, priority: 'exact', dedupe: false }, + (signal) => + new Promise((_resolve, reject) => + signal.addEventListener('abort', () => reject(signal.reason), { once: true }) + ) + ) + try { + const first = stallOnce() + const firstRejected = expect(first).rejects.toMatchObject({ code: 'timeout' }) + await vi.advanceTimersByTimeAsync(WSL_TRANSCRIPT_FS_EXACT_TIMEOUT_MS) + await firstRejected + await expect(run(path, 'exact', async () => 'early')).rejects.toMatchObject({ + code: 'unavailable' + }) + + // A cold-booted distro recovers here: admitted again after the base window. + await vi.advanceTimersByTimeAsync(WSL_TRANSCRIPT_FS_ROUTE_QUARANTINE_BASE_MS) + const second = stallOnce() + const secondRejected = expect(second).rejects.toMatchObject({ code: 'timeout' }) + await vi.advanceTimersByTimeAsync(WSL_TRANSCRIPT_FS_EXACT_TIMEOUT_MS) + await secondRejected + + // Second strike doubles the back-off: still blocked after one base window. + await vi.advanceTimersByTimeAsync(WSL_TRANSCRIPT_FS_ROUTE_QUARANTINE_BASE_MS) + await expect(run(path, 'exact', async () => 'still-blocked')).rejects.toMatchObject({ + code: 'unavailable' + }) + await vi.advanceTimersByTimeAsync(WSL_TRANSCRIPT_FS_ROUTE_QUARANTINE_BASE_MS) + await expect(run(path, 'exact', async () => 'recovered')).resolves.toBe('recovered') + + // The success reset the strikes: the next stall quarantines for the base + // window again instead of continuing the escalation. + const third = stallOnce() + const thirdRejected = expect(third).rejects.toMatchObject({ code: 'timeout' }) + await vi.advanceTimersByTimeAsync(WSL_TRANSCRIPT_FS_EXACT_TIMEOUT_MS) + await thirdRejected + await vi.advanceTimersByTimeAsync(WSL_TRANSCRIPT_FS_ROUTE_QUARANTINE_BASE_MS) + await expect(run(path, 'exact', async () => 'reset')).resolves.toBe('reset') + } finally { + vi.useRealTimers() + } + }) + + // Why: a helper transport fault (child died, fork failed) consulted nothing + // about the mount, so it must not erase a live quarantine or its strikes. + it('keeps the quarantine when a same-route transport fault settles', async () => { + vi.useFakeTimers() + const stalledExact = deferred() + const scanWork = deferred() + try { + const stalled = run('\\\\wsl.localhost\\Ubuntu\\quarantine-me', 'exact', () => { + return stalledExact.promise + }) + const stalledRejected = expect(stalled).rejects.toMatchObject({ code: 'timeout' }) + const scan = run('\\\\wsl.localhost\\Ubuntu\\scan-tree', 'scan', () => scanWork.promise) + const scanRejected = expect(scan).rejects.toMatchObject({ code: 'unavailable' }) + + await vi.advanceTimersByTimeAsync(WSL_TRANSCRIPT_FS_EXACT_TIMEOUT_MS) + await stalledRejected + + // The still-running scan's helper dies: a transport fault, not an answer. + scanWork.reject(new WslTranscriptFsError('unavailable', 'helper process died')) + await vi.advanceTimersByTimeAsync(0) + await scanRejected + + const retryTask = vi.fn(async () => 'retry') + await expect( + run('\\\\wsl.localhost\\Ubuntu\\retry', 'exact', retryTask) + ).rejects.toMatchObject({ code: 'unavailable' }) + expect(retryTask).not.toHaveBeenCalled() + } finally { + stalledExact.resolve('late') + await vi.advanceTimersByTimeAsync(0) + vi.useRealTimers() + } + }) + + it('lets abandoned running work finish instead of aborting its process', async () => { + const work = deferred() + const controller = new AbortController() + const sawAbort = vi.fn() + const task = vi.fn((signal: AbortSignal) => { + signal.addEventListener('abort', sawAbort, { once: true }) + return work.promise + }) + const pending = runWslTranscriptFsTask( + { + operation: 'read', + path: '\\\\wsl.localhost\\Ubuntu\\abandoned-read', + priority: 'exact', + dedupe: false, + signal: controller.signal + }, + task + ) + await vi.waitFor(() => expect(task).toHaveBeenCalledOnce()) + const reason = new Error('caller moved on') + controller.abort(reason) + await expect(pending).rejects.toBe(reason) + // A healthy in-flight process op keeps running; only the deadline kills. + expect(sawAbort).not.toHaveBeenCalled() + work.resolve('late') + await Promise.resolve() + }) + + it('quarantines an expired route while keeping healthy routes usable', async () => { vi.useFakeTimers() const stalled = deferred() try { @@ -418,7 +558,7 @@ describe('WSL transcript filesystem task scheduling', () => { const sameLaneTask = vi.fn(async () => 'same-lane') await expect( run('\\\\wsl.localhost\\Ubuntu\\lane-stuck-sibling', 'exact', sameLaneTask) - ).rejects.toMatchObject({ code: 'unavailable', message: SLOW_MESSAGE }) + ).rejects.toMatchObject({ code: 'unavailable' }) expect(sameLaneTask).not.toHaveBeenCalled() await expect( run('\\\\wsl.localhost\\Debian\\healthy-lane', 'exact', async () => 'debian') @@ -430,7 +570,7 @@ describe('WSL transcript filesystem task scheduling', () => { } }) - it('fails scans and same-route probes fast while a scan is stuck', async () => { + it('restores the scan slot while quarantining the expired route', async () => { vi.useFakeTimers() const stalled = deferred() try { @@ -442,10 +582,8 @@ describe('WSL transcript filesystem task scheduling', () => { const otherScanTask = vi.fn(async () => 'other-scan') await expect( run('\\\\wsl.localhost\\Debian\\other-tree', 'scan', otherScanTask) - ).rejects.toMatchObject({ code: 'unavailable' }) - expect(otherScanTask).not.toHaveBeenCalled() - // The whole Ubuntu mount is hung — an exact probe there would only burn - // the remaining permit on it. + ).resolves.toBe('other-scan') + expect(otherScanTask).toHaveBeenCalledOnce() await expect( run('\\\\wsl.localhost\\Ubuntu\\live-probe', 'exact', async () => 'exact') ).rejects.toMatchObject({ code: 'unavailable' }) @@ -459,7 +597,7 @@ describe('WSL transcript filesystem task scheduling', () => { } }) - it('keeps a stuck exact probe from feeding a doomed scan the second permit', async () => { + it('does not spend a replacement process on a quarantined route', async () => { vi.useFakeTimers() const stalled = deferred() try { @@ -483,8 +621,9 @@ describe('WSL transcript filesystem task scheduling', () => { } }) - it('admits new work again once stuck work finally settles', async () => { - vi.useFakeTimers() + it('ignores a late result after admitting replacement work', async () => { + // performance.now drives the quarantine clock, so it must be faked too. + vi.useFakeTimers({ toFake: ['setTimeout', 'clearTimeout', 'Date', 'performance'] }) const stalled = deferred() try { const stuck = run('\\\\wsl.localhost\\Ubuntu\\recovering', 'exact', () => stalled.promise) @@ -495,8 +634,15 @@ describe('WSL transcript filesystem task scheduling', () => { run('\\\\wsl.localhost\\Ubuntu\\recovering-next', 'exact', async () => 'blocked') ).rejects.toMatchObject({ code: 'unavailable' }) + // A value that only lands past the deadline is the stall itself, not + // proof of health: it must not cut the back-off short. stalled.resolve('late') await vi.advanceTimersByTimeAsync(0) + await expect( + run('\\\\wsl.localhost\\Ubuntu\\recovering-next', 'exact', async () => 'still-blocked') + ).rejects.toMatchObject({ code: 'unavailable' }) + + await vi.advanceTimersByTimeAsync(WSL_TRANSCRIPT_FS_ROUTE_QUARANTINE_BASE_MS) await expect( run('\\\\wsl.localhost\\Ubuntu\\recovering-next', 'exact', async () => 'recovered') ).resolves.toBe('recovered') @@ -508,8 +654,9 @@ describe('WSL transcript filesystem task scheduling', () => { } }) - it('refuses to join in-flight work already past its own deadline', async () => { - vi.useFakeTimers() + it('starts a new generation after shared work reaches its deadline', async () => { + // performance.now drives the quarantine clock, so it must be faked too. + vi.useFakeTimers({ toFake: ['setTimeout', 'clearTimeout', 'Date', 'performance'] }) const work = deferred() const task = vi.fn(() => work.promise) try { @@ -521,12 +668,13 @@ describe('WSL transcript filesystem task scheduling', () => { const secondRejected = expect(second).rejects.toMatchObject({ code: 'timeout' }) await vi.advanceTimersByTimeAsync(WSL_TRANSCRIPT_FS_EXACT_TIMEOUT_MS - 10_000) - await firstRejected + await Promise.all([firstRejected, secondRejected]) + work.resolve('late') + // The late value settles nothing, so the deadline's back-off still runs. + await vi.advanceTimersByTimeAsync(WSL_TRANSCRIPT_FS_ROUTE_QUARANTINE_BASE_MS) const joinTask = vi.fn(async () => 'join') - await expect(run(path, 'exact', joinTask)).rejects.toMatchObject({ code: 'unavailable' }) - expect(joinTask).not.toHaveBeenCalled() - await vi.advanceTimersByTimeAsync(10_000) - await secondRejected + await expect(run(path, 'exact', joinTask)).resolves.toBe('join') + expect(joinTask).toHaveBeenCalledOnce() } finally { work.resolve('late') await vi.advanceTimersByTimeAsync(0) @@ -534,7 +682,7 @@ describe('WSL transcript filesystem task scheduling', () => { } }) - it('fails joins onto queued work fast when stuck I/O keeps it from running', async () => { + it('fails queued same-route work fast when the deadline quarantines it', async () => { vi.useFakeTimers() const stalled = deferred() const queuedTask = vi.fn(async () => 'queued') @@ -542,19 +690,15 @@ describe('WSL transcript filesystem task scheduling', () => { const stuck = run('\\\\wsl.localhost\\Ubuntu\\trap-hung', 'exact', () => stalled.promise) const stuckRejected = expect(stuck).rejects.toMatchObject({ code: 'timeout' }) await vi.advanceTimersByTimeAsync(5_000) - // Queued behind the same route before the hang is detected. + // Queued behind the same route before the hang is detected. Left queued + // it would strand a full waiter deadline; a sequential multi-file scan + // would then pay one deadline per file. const queuedPath = '\\\\wsl.localhost\\Ubuntu\\trap-queued' const queued = run(queuedPath, 'exact', queuedTask) - const queuedRejected = expect(queued).rejects.toMatchObject({ code: 'timeout' }) + const queuedRejected = expect(queued).rejects.toMatchObject({ code: 'unavailable' }) await vi.advanceTimersByTimeAsync(WSL_TRANSCRIPT_FS_EXACT_TIMEOUT_MS - 5_000) await stuckRejected - // Re-requests must not keep the doomed queued task alive with fresh - // deadlines — that would defeat the fail-fast for as long as I/O hangs. - await expect(run(queuedPath, 'exact', queuedTask)).rejects.toMatchObject({ - code: 'unavailable' - }) - await vi.advanceTimersByTimeAsync(5_000) await queuedRejected expect(queuedTask).not.toHaveBeenCalled() } finally { @@ -564,7 +708,7 @@ describe('WSL transcript filesystem task scheduling', () => { } }) - it('does not feed a stuck route the other permit from the queue', async () => { + it('keeps exact work moving while an isolated scan process is stalled', async () => { vi.useFakeTimers() const scanWork = deferred() const debianWork = deferred() @@ -587,16 +731,13 @@ describe('WSL transcript filesystem task scheduling', () => { // Queued before Ubuntu's hang is detected at the 60s scan deadline. const queuedU = run('\\\\wsl.localhost\\Ubuntu\\queued-route-file', 'exact', ubuntuTask) - const queuedURejected = expect(queuedU).rejects.toMatchObject({ code: 'timeout' }) + await expect(queuedU).resolves.toBe('ubuntu') await vi.advanceTimersByTimeAsync(25_000) await scanRejected debianWork.resolve('debian') await vi.advanceTimersByTimeAsync(0) - expect(ubuntuTask).not.toHaveBeenCalled() - await vi.advanceTimersByTimeAsync(5_000) - await queuedURejected - expect(ubuntuTask).not.toHaveBeenCalled() + expect(ubuntuTask).toHaveBeenCalledOnce() } finally { scanWork.resolve('late') debianWork.resolve('debian') diff --git a/src/main/native-chat/wsl-transcript-fs-gate.ts b/src/main/native-chat/wsl-transcript-fs-gate.ts index 84848e13f11..d4413c91323 100644 --- a/src/main/native-chat/wsl-transcript-fs-gate.ts +++ b/src/main/native-chat/wsl-transcript-fs-gate.ts @@ -1,4 +1,16 @@ -import { wslTranscriptFsRouteKey } from './wsl-transcript-fs-route' +import { wslTranscriptFsLaneKey, wslTranscriptFsRouteKey } from './wsl-transcript-fs-route' +import { + WslTranscriptFsError, + wslTranscriptFsCapacityError as capacityError, + wslTranscriptFsTimeoutError as timeoutError, + wslTranscriptFsUnavailableError as unavailableError +} from './wsl-transcript-fs-error' +import { + liftRouteQuarantine, + quarantineRoute, + resetRouteQuarantinesForTests, + routeIsBlocked +} from './wsl-transcript-fs-route-quarantine' const MAX_CONCURRENT_WSL_TRANSCRIPT_FS_TASKS = 2 export const WSL_TRANSCRIPT_FS_EXACT_TIMEOUT_MS = 30_000 @@ -6,30 +18,14 @@ export const WSL_TRANSCRIPT_FS_SCAN_TIMEOUT_MS = 60_000 // Burst bounds keep polling fan-out from growing retained tasks or callers indefinitely. export const WSL_TRANSCRIPT_FS_MAX_PENDING_TASKS = 64 export const WSL_TRANSCRIPT_FS_MAX_WAITERS_PER_TASK = 64 -export const WSL_TRANSCRIPT_FS_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.' -export const WSL_TRANSCRIPT_FS_CAPACITY_MESSAGE = - 'WSL transcript discovery is temporarily unavailable because too many filesystem requests are already waiting. Try again shortly or restart Orca if the issue continues.' - -export type WslTranscriptFsFailureCode = 'timeout' | 'capacity' | 'unavailable' - -export class WslTranscriptFsError extends Error { - constructor( - readonly code: WslTranscriptFsFailureCode, - message: string - ) { - super(message) - this.name = 'WslTranscriptFsError' - } -} - -/** Narrow a caught error to a gate refusal, rethrowing anything else. */ -export function wslTranscriptFsRefusal(error: unknown): WslTranscriptFsError { - if (error instanceof WslTranscriptFsError) { - return error - } - throw error -} +export { + WSL_TRANSCRIPT_FS_CAPACITY_MESSAGE, + WSL_TRANSCRIPT_FS_SLOW_MESSAGE, + WslTranscriptFsError, + wslTranscriptFsRefusal, + type WslTranscriptFsFailureCode +} from './wsl-transcript-fs-error' +export { WSL_TRANSCRIPT_FS_ROUTE_QUARANTINE_BASE_MS } from './wsl-transcript-fs-route-quarantine' export type WslTranscriptFsTaskPriority = 'exact' | 'scan' @@ -52,10 +48,9 @@ type ScheduledTask = { controller: AbortController waiters: Set> state: 'queued' | 'running' | 'settled' - // Set by the deadline timer, sharing the waiters' monotonic clock — wall time - // would misjudge stuckness across laptop sleep or NTP steps. - stuck: boolean - stuckTimer?: ReturnType + deadlineTimer?: ReturnType + /** When the pump admitted the task; tells the quarantine which incident it saw. */ + startedAt?: number } type UnknownScheduledTask = ScheduledTask @@ -88,10 +83,11 @@ function abandonTaskIfUnused(task: UnknownScheduledTask, reason?: unknown): void if (task.waiters.size > 0 || task.state === 'settled') { return } - task.controller.abort(reason) - // Running I/O keeps its permit; new callers need a reusable controller. + // Running I/O keeps its permit and its process: an abort here would kill a + // healthy child and pre-empt the deadline's quarantine. Only the deadline aborts. clearTask(task) if (task.state === 'queued') { + task.controller.abort(reason) task.state = 'settled' removeQueuedTask(task) pumpTasks() @@ -117,35 +113,19 @@ function timeoutMs(priority: WslTranscriptFsTaskPriority): number { : WSL_TRANSCRIPT_FS_SCAN_TIMEOUT_MS } -function timeoutError(): WslTranscriptFsError { - return new WslTranscriptFsError('timeout', WSL_TRANSCRIPT_FS_SLOW_MESSAGE) -} - -function capacityError(): WslTranscriptFsError { - return new WslTranscriptFsError('capacity', WSL_TRANSCRIPT_FS_CAPACITY_MESSAGE) -} - -function unavailableError(): WslTranscriptFsError { - return new WslTranscriptFsError('unavailable', WSL_TRANSCRIPT_FS_SLOW_MESSAGE) -} - -/** - * The stuck running task that would doom a new task's admission: one on the - * same route (the whole distro mount is hung, whatever the priority — spending - * the other permit on it escalates one bad distro into a global outage), one - * holding the single scan slot, or (with every permit stuck) any permit. - */ -function stuckBlocker( - route: string, - priority: WslTranscriptFsTaskPriority -): UnknownScheduledTask | undefined { - const stuck = [...activeTasks].filter((task) => task.stuck) - if (stuck.length === MAX_CONCURRENT_WSL_TRANSCRIPT_FS_TASKS) { - return stuck[0] +// Why: a task queued behind a quarantined route would otherwise strand until +// its own waiter deadline — one full deadline per file in a sequential scan. +function failQueuedRouteTasks(route: string): void { + const doomed = queuedTasks.filter((task) => task.route === route && task.state === 'queued') + for (const task of doomed) { + task.state = 'settled' + removeQueuedTask(task) + clearTask(task) + for (const waiter of task.waiters) { + removeWaiter(task, waiter) + waiter.reject(unavailableError()) + } } - return stuck.find( - (task) => task.route === route || (priority === 'scan' && task.priority === 'scan') - ) } // Caller-abort pre-checks live in runWslTranscriptFsTask; nothing here yields @@ -170,8 +150,7 @@ function attachWaiter(task: ScheduledTask, signal?: AbortSignal): Promise< } signal.addEventListener('abort', waiter.onAbort, { once: true }) } - // UNC/9P calls cannot be aborted; two blocked calls can retain both slots until settling or - // Orca restarts, while caller and queue retention remains bounded. + // The task deadline also replaces the isolated process; this bounds each caller's own wait. waiter.timeout = setTimeout(() => { const error = timeoutError() if (!removeWaiter(task, waiter)) { @@ -186,15 +165,34 @@ function attachWaiter(task: ScheduledTask, signal?: AbortSignal): Promise< function settleTask(task: ScheduledTask, result: { value: T } | { error: unknown }): void { if (task.state !== 'running') { + // A result that only lands past the deadline is the stall the quarantine + // was set for, so it never lifts the back-off. A late value also has no + // owner: dispose it. + if ('value' in result) { + try { + task.onAbandonedResult?.(result.value) + } catch { + // Best-effort teardown; nothing left to report it to. + } + } return } + // Only a settle the deadline did not force proves the mount answered; a + // transport fault (WslTranscriptFsError from a dead helper) proves nothing. + if ( + 'value' in result || + (result.error !== task.controller.signal.reason && + !(result.error instanceof WslTranscriptFsError)) + ) { + liftRouteQuarantine(task.route) + } task.state = 'settled' if (task.priority === 'scan') { activeScanCount -= 1 } activeLaneKeys.delete(task.laneKey) activeTasks.delete(task as UnknownScheduledTask) - clearTimeout(task.stuckTimer) + clearTimeout(task.deadlineTimer) clearTask(task as UnknownScheduledTask) // Why: an unabortable syscall can still succeed after its last waiter timed // out or cancelled. A resource-valued result (open's FileHandle) then has no @@ -222,11 +220,6 @@ function settleTask(task: ScheduledTask, result: { value: T } | { error: u } function nextTaskIndex(): number { - // Tasks queued before their route hung must not be fed the other permit; - // their waiters drain by deadline while healthy routes use the slot. - const stuckRoutes = new Set( - [...activeTasks].filter((task) => task.stuck).map((task) => task.route) - ) for (const priority of ['exact', 'scan'] as const) { // Why: keep one libuv slot available for a live transcript probe. if (priority === 'scan' && activeScanCount > 0) { @@ -236,7 +229,7 @@ function nextTaskIndex(): number { (task) => task.priority === priority && !activeLaneKeys.has(task.laneKey) && - !stuckRoutes.has(task.route) + !routeIsBlocked(task.route) ) if (index !== -1) { return index @@ -256,21 +249,25 @@ function pumpTasks(): void { continue } task.state = 'running' + task.startedAt = performance.now() if (task.priority === 'scan') { activeScanCount += 1 } activeLaneKeys.add(task.laneKey) activeTasks.add(task) - // Anchored at run start: queue wait says nothing about the I/O itself, so - // the fail-fast may lag admission by at most one deadline period. - task.stuckTimer = setTimeout(() => { - task.stuck = true + task.deadlineTimer = setTimeout(() => { + const error = timeoutError() console.warn( - `[wsl-transcript-fs-gate] ${task.priority} filesystem task still running after ` + - `${timeoutMs(task.priority)}ms and holding a permit: ${task.key}` + `[wsl-transcript-fs-gate] ${task.priority} filesystem task exceeded ` + + `${timeoutMs(task.priority)}ms; replacing its I/O process: ${task.key}` ) + // Keep polling from churning replacement processes on the same stalled mount. + quarantineRoute(task.route, timeoutMs(task.priority), task.startedAt ?? performance.now()) + failQueuedRouteTasks(task.route) + task.controller.abort(error) + settleTask(task, { error }) }, timeoutMs(task.priority)) - task.stuckTimer.unref?.() + task.deadlineTimer.unref?.() void Promise.resolve() .then(() => { task.controller.signal.throwIfAborted() @@ -283,16 +280,11 @@ 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. - */ +/** Test-only: drop every task, route quarantine, and counter. */ export function resetWslTranscriptFsGateForTests(): void { for (const task of [...activeTasks, ...queuedTasks]) { task.state = 'settled' - clearTimeout(task.stuckTimer) + clearTimeout(task.deadlineTimer) for (const waiter of task.waiters) { removeWaiter(task, waiter) } @@ -301,6 +293,7 @@ export function resetWslTranscriptFsGateForTests(): void { queuedTasks.length = 0 inFlightTasks.clear() activeLaneKeys.clear() + resetRouteQuarantinesForTests() activeScanCount = 0 } @@ -334,21 +327,14 @@ export function runWslTranscriptFsTask( : JSON.stringify([options.operation, options.path, options.priority]) const existing = inFlightTasks.get(key) as ScheduledTask | undefined if (existing) { - // A stuck target — or a queued target that stuck I/O keeps from ever - // running — dooms every joiner to a slow timeout, and each fresh joiner - // would keep the queued task alive past its own abandonment. Fail fast. - if ( - existing.stuck || - (existing.state === 'queued' && stuckBlocker(existing.route, existing.priority)) - ) { - return Promise.reject(unavailableError()) - } + // Join even under a route quarantine: the in-flight task costs no new I/O, + // is bounded by its own deadline, and its settle may itself lift the + // quarantine. A queued task cannot linger on a quarantined route — + // failQueuedRouteTasks cleared it when the quarantine was set. return attachWaiter(existing, options.signal) } 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)) { + if (routeIsBlocked(route)) { return Promise.reject(unavailableError()) } if (queuedTasks.length >= WSL_TRANSCRIPT_FS_MAX_PENDING_TASKS) { @@ -357,15 +343,14 @@ export function runWslTranscriptFsTask( const scheduled: ScheduledTask = { key, - laneKey: `${route}:${options.priority}`, + laneKey: wslTranscriptFsLaneKey(options.path, options.priority), route, priority: options.priority, operation: task, onAbandonedResult: options.onAbandonedResult, controller: new AbortController(), waiters: new Set(), - state: 'queued', - stuck: false + state: 'queued' } inFlightTasks.set(key, scheduled as UnknownScheduledTask) const result = attachWaiter(scheduled, options.signal) diff --git a/src/main/native-chat/wsl-transcript-fs-process-client.test.ts b/src/main/native-chat/wsl-transcript-fs-process-client.test.ts new file mode 100644 index 00000000000..177b9270076 --- /dev/null +++ b/src/main/native-chat/wsl-transcript-fs-process-client.test.ts @@ -0,0 +1,510 @@ +import { EventEmitter } from 'node:events' +import type { ChildProcess } from 'node:child_process' +import type { Dirent, Stats } from 'node:fs' +import { join } from 'node:path' +import { describe, expect, it, vi } from 'vitest' +import { WslTranscriptFsProcessClient } from './wsl-transcript-fs-process-client' +import { + WSL_TRANSCRIPT_FS_PROCESS_CLOSE_TIMEOUT_MS, + WSL_TRANSCRIPT_FS_PROCESS_IDLE_REAP_MS +} from './wsl-transcript-fs-process-slot' +import { + resolveWslTranscriptFsProcessEntryPath, + wslTranscriptFsProcessForkEnv +} from './wsl-transcript-fs-process-spawn' +import type { + WslTranscriptFsProcessRequest, + WslTranscriptFsProcessResponse +} from './wsl-transcript-fs-process-protocol' + +class FakeProcess extends EventEmitter { + readonly sent: WslTranscriptFsProcessRequest[] = [] + readonly kill = vi.fn(() => true) + readonly unref = vi.fn() + readonly channel = { unref: vi.fn() } + + send(message: WslTranscriptFsProcessRequest, callback?: (error: Error | null) => void): boolean { + this.sent.push(message) + callback?.(null) + return true + } + + respond(response: WslTranscriptFsProcessResponse): void { + this.emit('message', response) + } +} + +function fakeChild(process: FakeProcess): ChildProcess { + return process as unknown as ChildProcess +} + +describe('WSL transcript filesystem process client', () => { + it('reuses a healthy process for sequential operations', async () => { + const child = new FakeProcess() + const factory = vi.fn(() => fakeChild(child)) + const client = new WslTranscriptFsProcessClient(factory) + + const first = client.run( + { operation: 'access', path: '\\\\wsl.localhost\\Ubuntu\\one' }, + new AbortController().signal + ) + child.respond({ id: child.sent[0].id, ok: true, value: true }) + await expect(first).resolves.toBe(true) + + const second = client.run( + { + operation: 'readfile', + path: '\\\\wsl.localhost\\Ubuntu\\two', + encoding: 'utf8' + }, + new AbortController().signal + ) + child.respond({ id: child.sent[1].id, ok: true, value: 'body' }) + + await expect(second).resolves.toBe('body') + expect(factory).toHaveBeenCalledOnce() + expect(child.kill).not.toHaveBeenCalled() + client.dispose() + }) + + it('kills an aborted process and uses a replacement for later work', async () => { + const firstChild = new FakeProcess() + const replacement = new FakeProcess() + const factory = vi + .fn<() => ChildProcess>() + .mockReturnValueOnce(fakeChild(firstChild)) + .mockReturnValueOnce(fakeChild(replacement)) + const client = new WslTranscriptFsProcessClient(factory) + const controller = new AbortController() + const reason = new Error('deadline expired') + + const stalled = client.run( + { operation: 'stat', path: '\\\\wsl.localhost\\Ubuntu\\stalled' }, + controller.signal + ) + controller.abort(reason) + + await expect(stalled).rejects.toBe(reason) + expect(firstChild.kill).toHaveBeenCalledWith('SIGKILL') + + const later = client.run( + { operation: 'access', path: '\\\\wsl.localhost\\Fedora\\later' }, + new AbortController().signal + ) + replacement.respond({ id: replacement.sent[0].id, ok: true, value: true }) + await expect(later).resolves.toBe(true) + expect(factory).toHaveBeenCalledTimes(2) + client.dispose() + }) + + it('multiplexes sequential opened handles through one process until close', async () => { + const owner = new FakeProcess() + const factory = vi.fn(() => fakeChild(owner)) + const client = new WslTranscriptFsProcessClient(factory) + const signal = new AbortController().signal + + const opening = client.open('\\\\wsl.localhost\\Ubuntu\\transcript', signal) + owner.respond({ id: owner.sent[0].id, ok: true, value: 41 }) + const handle = await opening + + const secondOpening = client.open('\\\\wsl.localhost\\Ubuntu\\second', signal) + owner.respond({ id: owner.sent[1].id, ok: true, value: 42 }) + const secondHandle = await secondOpening + + const reading = client.read(handle, 8, 4, signal) + expect(owner.sent[2]).toMatchObject({ operation: 'read', handleId: 41, position: 8 }) + owner.respond({ id: owner.sent[2].id, ok: true, value: Buffer.from('old!') }) + await expect(reading).resolves.toEqual(Buffer.from('old!')) + + const secondReading = client.read(secondHandle, 0, 3, signal) + expect(owner.sent[3]).toMatchObject({ operation: 'read', handleId: 42, position: 0 }) + owner.respond({ id: owner.sent[3].id, ok: true, value: Buffer.from('new') }) + await expect(secondReading).resolves.toEqual(Buffer.from('new')) + + const closing = client.close(handle) + expect(owner.sent[4]).toMatchObject({ operation: 'close', handleId: 41 }) + owner.respond({ id: owner.sent[4].id, ok: true, value: true }) + await expect(closing).resolves.toBeUndefined() + + const secondClosing = client.close(secondHandle) + expect(owner.sent[5]).toMatchObject({ operation: 'close', handleId: 42 }) + owner.respond({ id: owner.sent[5].id, ok: true, value: true }) + await expect(secondClosing).resolves.toBeUndefined() + + const later = client.run( + { operation: 'access', path: '\\\\wsl.localhost\\Ubuntu\\later' }, + signal + ) + owner.respond({ id: owner.sent[6].id, ok: true, value: true }) + await expect(later).resolves.toBe(true) + expect(factory).toHaveBeenCalledOnce() + client.dispose() + }) + + it('queues concurrent opens instead of creating another helper', async () => { + const child = new FakeProcess() + const factory = vi.fn(() => fakeChild(child)) + const client = new WslTranscriptFsProcessClient(factory) + const signal = new AbortController().signal + + const firstOpening = client.open('\\\\wsl.localhost\\Ubuntu\\first', signal) + const secondOpening = client.open('\\\\wsl.localhost\\Ubuntu\\second', signal) + expect(factory).toHaveBeenCalledOnce() + expect(child.sent).toHaveLength(1) + + child.respond({ id: child.sent[0].id, ok: true, value: 1 }) + const firstHandle = await firstOpening + await vi.waitFor(() => expect(child.sent).toHaveLength(2)) + child.respond({ id: child.sent[1].id, ok: true, value: 2 }) + const secondHandle = await secondOpening + + const firstClose = client.close(firstHandle) + child.respond({ id: child.sent[2].id, ok: true, value: true }) + await firstClose + const secondClose = client.close(secondHandle) + child.respond({ id: child.sent[3].id, ok: true, value: true }) + await secondClose + + expect(factory).toHaveBeenCalledOnce() + client.dispose() + }) + + it('returns a granted slot when its queued caller aborts before sending', async () => { + const child = new FakeProcess() + const client = new WslTranscriptFsProcessClient(() => fakeChild(child)) + const controller = new AbortController() + const signal = new AbortController().signal + + const first = client.run({ operation: 'access', path: 'first' }, signal) + const aborted = client.run({ operation: 'access', path: 'aborted' }, controller.signal) + child.respond({ id: child.sent[0].id, ok: true, value: true }) + controller.abort(new Error('cancelled after grant')) + + await expect(first).resolves.toBe(true) + await expect(aborted).rejects.toThrow('cancelled after grant') + const later = client.run({ operation: 'access', path: 'later' }, signal) + expect(child.sent).toHaveLength(2) + child.respond({ id: child.sent[1].id, ok: true, value: true }) + await expect(later).resolves.toBe(true) + client.dispose() + }) + + it('rejects a granted request if the helper exits before it sends', async () => { + const child = new FakeProcess() + const replacement = new FakeProcess() + const factory = vi + .fn<() => ChildProcess>() + .mockReturnValueOnce(fakeChild(child)) + .mockReturnValueOnce(fakeChild(replacement)) + const client = new WslTranscriptFsProcessClient(factory) + const signal = new AbortController().signal + + const first = client.run({ operation: 'access', path: 'first' }, signal) + const stranded = client.run({ operation: 'access', path: 'stranded' }, signal) + child.respond({ id: child.sent[0].id, ok: true, value: true }) + child.emit('disconnect') + + await expect(first).resolves.toBe(true) + await expect(stranded).rejects.toMatchObject({ code: 'unavailable' }) + expect(replacement.sent).toHaveLength(0) + client.dispose() + }) + + // Why: the gate waiter can give up mid-read and the caller's finally closes + // right away; a refused close would strand the pinned slot (and its child) + // forever once that read settles. + it('defers a close issued while a read is in flight instead of stranding the slot', async () => { + const owner = new FakeProcess() + const factory = vi.fn(() => fakeChild(owner)) + const client = new WslTranscriptFsProcessClient(factory) + const signal = new AbortController().signal + + const opening = client.open('\\\\wsl.localhost\\Ubuntu\\transcript', signal) + owner.respond({ id: owner.sent[0].id, ok: true, value: 5 }) + const handle = await opening + + const reading = client.read(handle, 0, 4, signal) + const closing = client.close(handle) + // The close waits for the in-flight read; nothing extra was sent yet. + expect(owner.sent).toHaveLength(2) + + owner.respond({ id: owner.sent[1].id, ok: true, value: Buffer.from('data') }) + await expect(reading).resolves.toEqual(Buffer.from('data')) + await vi.waitFor(() => expect(owner.sent).toHaveLength(3)) + expect(owner.sent[2]).toMatchObject({ operation: 'close', handleId: 5 }) + owner.respond({ id: owner.sent[2].id, ok: true, value: true }) + await expect(closing).resolves.toBeUndefined() + + // The slot returned to the pool instead of leaking pinned. + const later = client.run( + { operation: 'access', path: '\\\\wsl.localhost\\Ubuntu\\after-close' }, + signal + ) + owner.respond({ id: owner.sent[3].id, ok: true, value: true }) + await expect(later).resolves.toBe(true) + expect(factory).toHaveBeenCalledOnce() + expect(owner.kill).not.toHaveBeenCalled() + client.dispose() + }) + + it('invalidates lane handles and serves queued work from a replacement', async () => { + const owner = new FakeProcess() + const healthy = new FakeProcess() + const factory = vi + .fn<() => ChildProcess>() + .mockReturnValueOnce(fakeChild(owner)) + .mockReturnValueOnce(fakeChild(healthy)) + const client = new WslTranscriptFsProcessClient(factory) + const opening = client.open( + '\\\\wsl.localhost\\Ubuntu\\transcript', + new AbortController().signal + ) + owner.respond({ id: owner.sent[0].id, ok: true, value: 9 }) + const handle = await opening + const controller = new AbortController() + const reason = new Error('read deadline') + + const stalled = client.read(handle, 0, 1, controller.signal) + const other = client.run( + { operation: 'access', path: '\\\\wsl.localhost\\Fedora\\healthy' }, + new AbortController().signal + ) + controller.abort(reason) + + await expect(stalled).rejects.toBe(reason) + await vi.waitFor(() => expect(healthy.sent).toHaveLength(1)) + healthy.respond({ id: healthy.sent[0].id, ok: true, value: true }) + await expect(other).resolves.toBe(true) + // The handle died with its killed process — a transport condition, so later + // reads surface as a retryable refusal rather than a caller EBADF bug. + await expect(client.read(handle, 0, 1, new AbortController().signal)).rejects.toMatchObject({ + name: 'WslTranscriptFsError', + code: 'unavailable' + }) + expect(owner.kill).toHaveBeenCalledWith('SIGKILL') + expect(healthy.kill).not.toHaveBeenCalled() + + const later = client.run( + { operation: 'access', path: '\\\\wsl.localhost\\Fedora\\later' }, + new AbortController().signal + ) + healthy.respond({ id: healthy.sent[1].id, ok: true, value: true }) + await expect(later).resolves.toBe(true) + client.dispose() + }) + + it('retires a process whose handle close does not settle', async () => { + vi.useFakeTimers() + const owner = new FakeProcess() + const healthy = new FakeProcess() + const factory = vi + .fn<() => ChildProcess>() + .mockReturnValueOnce(fakeChild(owner)) + .mockReturnValueOnce(fakeChild(healthy)) + const client = new WslTranscriptFsProcessClient(factory) + try { + const opening = client.open( + '\\\\wsl.localhost\\Ubuntu\\transcript', + new AbortController().signal + ) + owner.respond({ id: owner.sent[0].id, ok: true, value: 12 }) + const handle = await opening + const closing = client.close(handle) + const closeFailure = expect(closing).rejects.toThrow('close timed out') + + const unrelated = client.run( + { operation: 'access', path: '\\\\wsl.localhost\\Fedora\\healthy' }, + new AbortController().signal + ) + + await vi.advanceTimersByTimeAsync(WSL_TRANSCRIPT_FS_PROCESS_CLOSE_TIMEOUT_MS) + await closeFailure + await vi.waitFor(() => expect(healthy.sent).toHaveLength(1)) + healthy.respond({ id: healthy.sent[0].id, ok: true, value: true }) + await expect(unrelated).resolves.toBe(true) + expect(owner.kill).toHaveBeenCalledWith('SIGKILL') + expect(healthy.kill).not.toHaveBeenCalled() + } finally { + client.dispose() + vi.useRealTimers() + } + }) + + it('reaps an idle process after the idle deadline and forks a fresh one later', async () => { + vi.useFakeTimers() + const child = new FakeProcess() + const replacement = new FakeProcess() + const factory = vi + .fn<() => ChildProcess>() + .mockReturnValueOnce(fakeChild(child)) + .mockReturnValueOnce(fakeChild(replacement)) + const client = new WslTranscriptFsProcessClient(factory) + try { + const first = client.run( + { operation: 'access', path: '\\\\wsl.localhost\\Ubuntu\\one' }, + new AbortController().signal + ) + child.respond({ id: child.sent[0].id, ok: true, value: true }) + await expect(first).resolves.toBe(true) + + await vi.advanceTimersByTimeAsync(WSL_TRANSCRIPT_FS_PROCESS_IDLE_REAP_MS) + expect(child.kill).toHaveBeenCalledWith('SIGKILL') + + const later = client.run( + { operation: 'access', path: '\\\\wsl.localhost\\Ubuntu\\two' }, + new AbortController().signal + ) + replacement.respond({ id: replacement.sent[0].id, ok: true, value: true }) + await expect(later).resolves.toBe(true) + expect(factory).toHaveBeenCalledTimes(2) + } finally { + client.dispose() + vi.useRealTimers() + } + }) + + it('surfaces child transport faults as unavailable gate refusals', async () => { + const child = new FakeProcess() + const client = new WslTranscriptFsProcessClient(() => fakeChild(child)) + const pending = client.run( + { operation: 'access', path: '\\\\wsl.localhost\\Ubuntu\\dead' }, + new AbortController().signal + ) + child.emit('exit', 9) + await expect(pending).rejects.toMatchObject({ + name: 'WslTranscriptFsError', + code: 'unavailable' + }) + client.dispose() + }) + + it('names the killing signal instead of a null exit code', async () => { + const child = new FakeProcess() + const client = new WslTranscriptFsProcessClient(() => fakeChild(child)) + const pending = client.run( + { operation: 'access', path: '\\\\wsl.localhost\\Ubuntu\\killed' }, + new AbortController().signal + ) + // A signal-terminated child reports code null; the message must carry the signal. + child.emit('exit', null, 'SIGKILL') + await expect(pending).rejects.toMatchObject({ + name: 'WslTranscriptFsError', + code: 'unavailable', + message: expect.stringContaining('exited (SIGKILL)') + }) + client.dispose() + }) + + it('reconstructs filesystem errors with their Node error code', async () => { + const child = new FakeProcess() + const client = new WslTranscriptFsProcessClient(() => fakeChild(child)) + const pending = client.run( + { operation: 'stat', path: '\\\\wsl.localhost\\Ubuntu\\missing' }, + new AbortController().signal + ) + + child.respond({ + id: child.sent[0].id, + ok: false, + error: { + name: 'Error', + message: 'not found', + code: 'ENOENT', + syscall: 'stat', + path: '\\\\wsl.localhost\\Ubuntu\\missing' + } + }) + + await expect(pending).rejects.toMatchObject({ + message: 'not found', + code: 'ENOENT', + syscall: 'stat' + }) + client.dispose() + }) + + it('restores Stats and Dirent methods after IPC serialization', async () => { + const child = new FakeProcess() + const client = new WslTranscriptFsProcessClient(() => fakeChild(child)) + const stats = client.run( + { operation: 'stat', path: '\\\\wsl.localhost\\Ubuntu\\file' }, + new AbortController().signal + ) + child.respond({ + id: child.sent[0].id, + ok: true, + value: { mode: 0o100644, size: 12, mtime: new Date(0) } + }) + + expect((await stats).isFile()).toBe(true) + + const entries = client.run( + { operation: 'readdir', path: '\\\\wsl.localhost\\Ubuntu\\dir' }, + new AbortController().signal + ) + child.respond({ + id: child.sent[1].id, + ok: true, + value: [ + { + name: 'child', + parentPath: '\\\\wsl.localhost\\Ubuntu\\dir', + isBlockDevice: false, + isCharacterDevice: false, + isDirectory: false, + isFIFO: false, + isFile: true, + isSocket: false, + isSymbolicLink: false + } + ] + }) + + expect((await entries)[0].isFile()).toBe(true) + client.dispose() + }) +}) + +describe('WSL transcript filesystem process entry resolution', () => { + it('prefers the unpacked sibling for a packaged main bundle', () => { + const exists = vi.fn((path: string) => path.includes('app.asar.unpacked')) + const moduleDir = join('root', 'resources', 'app.asar', 'out', 'main') + + expect( + resolveWslTranscriptFsProcessEntryPath(moduleDir, join('root', 'resources'), exists) + ).toBe( + join(moduleDir.replace('app.asar', 'app.asar.unpacked'), 'wsl-transcript-fs-process-entry.js') + ) + }) + + // Why: this module compiles into a shared rollup chunk under out/main/chunks, + // and the scanner service child has no process.resourcesPath to fall back on. + it('resolves the entry from a shared chunk one level below out/main', () => { + const moduleDir = join('root', 'out', 'main', 'chunks') + const target = join('root', 'out', 'main', 'wsl-transcript-fs-process-entry.js') + const exists = vi.fn((path: string) => path === target) + + expect(resolveWslTranscriptFsProcessEntryPath(moduleDir, undefined, exists)).toBe(target) + }) + + it('excludes ambient NODE_OPTIONS and secrets from the fork env', () => { + const env = wslTranscriptFsProcessForkEnv( + { + PATH: 'C:\\bin', + SYSTEMROOT: 'C:\\WINDOWS', + NODE_OPTIONS: '--inspect-brk', + SECRET_TOKEN: 'shh' + }, + 'win32' + ) + + expect(env).toMatchObject({ + ELECTRON_RUN_AS_NODE: '1', + PATH: 'C:\\bin', + // The shared allowlist emits the canonical Windows casing. + SystemRoot: 'C:\\WINDOWS' + }) + expect(env).not.toHaveProperty('NODE_OPTIONS') + expect(env).not.toHaveProperty('SECRET_TOKEN') + }) +}) diff --git a/src/main/native-chat/wsl-transcript-fs-process-client.ts b/src/main/native-chat/wsl-transcript-fs-process-client.ts new file mode 100644 index 00000000000..b93079e4542 --- /dev/null +++ b/src/main/native-chat/wsl-transcript-fs-process-client.ts @@ -0,0 +1,301 @@ +import type { + WslTranscriptFsProcessResponse, + WslTranscriptFsReusableProcessCall +} from './wsl-transcript-fs-process-protocol' +import { + decodeWslTranscriptFsProcessError, + decodeWslTranscriptFsProcessValue +} from './wsl-transcript-fs-process-decode' +// Transport faults (spawn failure, child death) mean nothing was consulted — +// surfacing them as plain errors would read as "path missing"/"empty tree" to +// the discovery layers, which only rethrow WslTranscriptFsError. +import { wslTranscriptFsProcessFailureError } from './wsl-transcript-fs-error' +import { + attachSlotChild, + WSL_TRANSCRIPT_FS_PROCESS_CLOSE_TIMEOUT_MS, + type HandleState, + type ProcessSlot, + type SlotDisposition, + type WslTranscriptFsProcessFactory, + type WslTranscriptFsProcessHandle +} from './wsl-transcript-fs-process-slot' +import { WslTranscriptFsProcessLanePool } from './wsl-transcript-fs-process-lane-pool' +import { sendWslTranscriptFsProcessRequest } from './wsl-transcript-fs-process-send' +import { + processHandleUnavailableError, + wslTranscriptFsHandleOwners +} from './wsl-transcript-fs-process-handle-owner' + +export type { WslTranscriptFsProcessHandle } from './wsl-transcript-fs-process-slot' + +export class WslTranscriptFsProcessClient { + private readonly pool: WslTranscriptFsProcessLanePool + private readonly handles = new WeakMap() + /** Handles retired by a slot fault/kill, not by a clean close: reads on them + * are a transport condition, never a caller bug. */ + private readonly faultedHandles = new WeakSet() + private nextId = 1 + + constructor(private readonly processFactory: WslTranscriptFsProcessFactory) { + this.pool = new WslTranscriptFsProcessLanePool( + () => this.createSlot(), + (slot) => this.destroySlot(slot) + ) + } + + async run(request: WslTranscriptFsReusableProcessCall, signal: AbortSignal): Promise { + signal.throwIfAborted() + const acquired = this.takeSlotOrThrow(signal) + return acquired instanceof Promise + ? acquired.then((slot) => + this.send(this.pool.claim(slot, signal), request, signal, 'idle') + ) + : this.send(acquired, request, signal, 'idle') + } + + async open(path: string, signal: AbortSignal): Promise { + signal.throwIfAborted() + const acquired = this.takeSlotOrThrow(signal) + const slot = acquired instanceof Promise ? await acquired : acquired + this.pool.claim(slot, signal) + const handleId = await this.send(slot, { operation: 'open', path }, signal, 'pin') + if (!this.pool.has(slot)) { + throw wslTranscriptFsProcessFailureError('the process exited while opening a file') + } + const handle = Object.freeze({ + wslTranscriptFsProcessHandle: true as const + }) + slot.handles.add(handle) + this.handles.set(handle, { slot, handleId }) + wslTranscriptFsHandleOwners.set(handle, this) + this.pool.park(slot) + return handle + } + + async read( + handle: WslTranscriptFsProcessHandle, + position: number, + length: number, + signal: AbortSignal + ): Promise { + signal.throwIfAborted() + const state = this.handles.get(handle) + if (!state) { + throw processHandleUnavailableError(handle, this.faultedHandles) + } + const acquired = this.takeSlotOrThrow(signal) + const slot = acquired instanceof Promise ? await acquired : acquired + this.pool.claim(slot, signal) + if (this.handles.get(handle) !== state || state.slot !== slot) { + this.pool.park(slot) + throw processHandleUnavailableError(handle, this.faultedHandles) + } + return this.send( + slot, + { operation: 'read', handleId: state.handleId, position, length }, + signal, + 'pinned' + ) + } + + close(handle: WslTranscriptFsProcessHandle): Promise { + const state = this.handles.get(handle) + if (!state) { + return Promise.resolve() + } + if (!state.closePromise) { + state.closePromise = this.performClose(handle, state) + // The stored promise may reject before any caller chains onto it. + void state.closePromise.catch(() => {}) + } + return state.closePromise + } + + // Why: closes bypass the gate and can arrive during a read. Queue them ahead + // of later lane work so teardown cannot fork around the one-process bound. + private async performClose( + handle: WslTranscriptFsProcessHandle, + state: HandleState + ): Promise { + const controller = new AbortController() + const timer = setTimeout(() => { + controller.abort(new Error('WSL transcript file handle close timed out')) + if (this.pool.has(state.slot) && state.slot.active) { + this.rejectActive( + state.slot, + wslTranscriptFsProcessFailureError('the process was retired by a stuck close') + ) + this.destroySlot(state.slot) + } + }, WSL_TRANSCRIPT_FS_PROCESS_CLOSE_TIMEOUT_MS) + timer.unref?.() + try { + const acquired = this.takeSlotOrThrow(controller.signal, true) + const slot = acquired instanceof Promise ? await acquired : acquired + this.pool.claim(slot, controller.signal, () => this.destroySlot(slot)) + if (this.handles.get(handle) !== state) { + this.pool.park(slot) + return + } + await this.send( + slot, + { operation: 'close', handleId: state.handleId }, + controller.signal, + 'close', + handle + ) + } finally { + clearTimeout(timer) + } + } + + private send( + slot: ProcessSlot, + request: Parameters[0]['request'], + signal: AbortSignal, + disposition: SlotDisposition, + handle?: WslTranscriptFsProcessHandle + ): Promise { + return sendWslTranscriptFsProcessRequest({ + slot, + id: this.nextId++, + request, + signal, + disposition, + handle, + onAbort: (reason) => { + this.rejectActive(slot, reason) + this.destroySlot(slot) + }, + onTransportFailure: (error) => { + this.rejectActive(slot, wslTranscriptFsProcessFailureError(error)) + this.destroySlot(slot) + } + }) + } + + dispose(): void { + const error = wslTranscriptFsProcessFailureError('the client was disposed') + this.pool.beginDispose(error) + for (const slot of this.pool.snapshot()) { + this.rejectActive(slot, error) + this.destroySlot(slot) + } + } + + private takeSlotOrThrow( + signal: AbortSignal, + prioritize = false + ): ProcessSlot | Promise { + return this.pool.acquire(signal, prioritize) + } + + private createSlot(): ProcessSlot { + try { + const child = this.processFactory() + const slot = attachSlotChild(child, { + onResponse: (response) => this.onResponse(slot, response), + onFault: (error) => this.onFault(slot, error) + }) + return slot + } catch (error) { + throw wslTranscriptFsProcessFailureError(error) + } + } + + private onResponse(slot: ProcessSlot, response: WslTranscriptFsProcessResponse): void { + const call = slot.active + if (!call || call.id !== response.id) { + return + } + this.clearActive(slot) + call.signal.removeEventListener('abort', call.onAbort) + if (!response.ok) { + call.reject(decodeWslTranscriptFsProcessError(response.error)) + } else { + try { + call.resolve(decodeWslTranscriptFsProcessValue(call.operation, response.value)) + } catch (error) { + // An undecodable ok-response means the protocol is corrupt: fail the + // call and retire the slot rather than leave the promise unsettled. + call.reject(error) + this.destroySlot(slot) + return + } + } + switch (call.disposition) { + case 'idle': + this.pool.park(slot) + break + case 'pin': + if (!response.ok) { + this.pool.park(slot) + } + break + case 'pinned': + this.pool.park(slot) + break + case 'close': + if (!response.ok) { + this.destroySlot(slot) + } else { + this.releaseHandle(slot, call.handle!) + this.pool.park(slot) + } + break + } + } + + private onFault(slot: ProcessSlot, error: Error): void { + if (!this.pool.has(slot)) { + return + } + this.rejectActive(slot, wslTranscriptFsProcessFailureError(error)) + this.destroySlot(slot) + } + + private clearActive(slot: ProcessSlot): void { + slot.active = null + } + + private rejectActive(slot: ProcessSlot, error: unknown): void { + const call = slot.active + this.clearActive(slot) + if (!call) { + return + } + call.signal.removeEventListener('abort', call.onAbort) + call.reject(error) + } + + private destroySlot(slot: ProcessSlot): void { + if (!this.pool.retire(slot)) { + return + } + for (const handle of slot.handles) { + this.faultedHandles.add(handle) + } + this.releaseAllHandles(slot) + slot.child.removeAllListeners() + try { + slot.child.kill('SIGKILL') + } catch { + // Teardown race: kill of an already-terminating child emits 'error' with + // no listeners left, which throws synchronously; the child dies anyway. + } + } + + private releaseHandle(slot: ProcessSlot, handle: WslTranscriptFsProcessHandle): void { + slot.handles.delete(handle) + this.handles.delete(handle) + // The owners entry stays (WeakMap, collected with the handle) so late + // cross-module reads still reach this client for a classified rejection. + } + + private releaseAllHandles(slot: ProcessSlot): void { + for (const handle of slot.handles) { + this.handles.delete(handle) + } + slot.handles.clear() + } +} diff --git a/src/main/native-chat/wsl-transcript-fs-process-decode.ts b/src/main/native-chat/wsl-transcript-fs-process-decode.ts new file mode 100644 index 00000000000..aad754ea115 --- /dev/null +++ b/src/main/native-chat/wsl-transcript-fs-process-decode.ts @@ -0,0 +1,43 @@ +import { Stats, type Dirent } from 'node:fs' +import type { + WslTranscriptFsDirent, + WslTranscriptFsProcessError, + WslTranscriptFsProcessRequest +} from './wsl-transcript-fs-process-protocol' + +export function decodeWslTranscriptFsProcessError(value: WslTranscriptFsProcessError): Error { + const error = new Error(value.message) as NodeJS.ErrnoException + error.name = value.name + Object.assign(error, value) + return error +} + +function decodeDirent(value: WslTranscriptFsDirent): Dirent { + return { + name: value.name, + parentPath: value.parentPath, + isBlockDevice: () => value.isBlockDevice, + isCharacterDevice: () => value.isCharacterDevice, + isDirectory: () => value.isDirectory, + isFIFO: () => value.isFIFO, + isFile: () => value.isFile, + isSocket: () => value.isSocket, + isSymbolicLink: () => value.isSymbolicLink + } as Dirent +} + +/** Revive prototype-dependent results the structured clone stripped. */ +export function decodeWslTranscriptFsProcessValue( + operation: WslTranscriptFsProcessRequest['operation'], + value: unknown +): unknown { + if (operation === 'stat' || operation === 'lstat') { + // Copy instead of setPrototypeOf: the in-process vitest arm passes the + // suite's own fixture here, which may be shared or frozen. + return Object.assign(Object.create(Stats.prototype) as Stats, value) + } + if (operation === 'readdir') { + return (value as WslTranscriptFsDirent[]).map(decodeDirent) + } + return value +} diff --git a/src/main/native-chat/wsl-transcript-fs-process-dispatch.ts b/src/main/native-chat/wsl-transcript-fs-process-dispatch.ts new file mode 100644 index 00000000000..fda412aa0fc --- /dev/null +++ b/src/main/native-chat/wsl-transcript-fs-process-dispatch.ts @@ -0,0 +1,99 @@ +import { open, type FileHandle } from 'node:fs/promises' +import { + invalidTranscriptHandleError, + type WslTranscriptFsReusableProcessCall +} from './wsl-transcript-fs-process-protocol' +import { decodeWslTranscriptFsProcessValue } from './wsl-transcript-fs-process-decode' +import { + WslTranscriptFsProcessClient, + type WslTranscriptFsProcessHandle +} from './wsl-transcript-fs-process-client' +import { wslTranscriptFsHandleOwners } from './wsl-transcript-fs-process-handle-owner' +import { WslTranscriptFsProcessOperations } from './wsl-transcript-fs-process-operations' +import { forkWslTranscriptFsProcess } from './wsl-transcript-fs-process-spawn' + +export type { WslTranscriptFsProcessHandle } from './wsl-transcript-fs-process-client' + +// Why: an env-only check would let a leaked VITEST=true (harnesses spreading +// process.env into a real app) silently revert production to in-process UNC +// syscalls; the worker global only exists inside an actual vitest runtime. +function inVitestWorker(): boolean { + return process.env.VITEST === 'true' && '__vitest_worker__' in globalThis +} + +// Unit suites run the child's own dispatcher (decode included) in-process, so +// there is exactly one request implementation to drift; production never +// bypasses the process boundary. +let inProcessOperations: WslTranscriptFsProcessOperations | null = null + +function runInProcess(request: WslTranscriptFsReusableProcessCall): Promise { + inProcessOperations ??= new WslTranscriptFsProcessOperations() + return inProcessOperations + .execute({ ...request, id: 0 }) + .then((value) => decodeWslTranscriptFsProcessValue(request.operation, value)) as Promise +} + +const clientsByLane = new Map() + +function getLaneClient(laneKey: string): WslTranscriptFsProcessClient { + const existing = clientsByLane.get(laneKey) + if (existing) { + return existing + } + const client = new WslTranscriptFsProcessClient(forkWslTranscriptFsProcess) + clientsByLane.set(laneKey, client) + return client +} + +export function runWslTranscriptFsProcess( + request: WslTranscriptFsReusableProcessCall, + signal: AbortSignal, + laneKey: string +): Promise { + if (inVitestWorker()) { + return runInProcess(request) + } + return getLaneClient(laneKey).run(request, signal) +} + +export function openWslTranscriptFsProcess( + path: string, + signal: AbortSignal, + laneKey: string +): Promise { + if (inVitestWorker()) { + // A real FileHandle: suites drive reads and closes through the plain + // handle branch, mirroring non-UNC ownership. + return open(path, 'r') + } + return getLaneClient(laneKey).open(path, signal) +} + +export function readWslTranscriptFsProcess( + handle: WslTranscriptFsProcessHandle, + position: number, + length: number, + signal: AbortSignal +): Promise { + const owner = wslTranscriptFsHandleOwners.get(handle) + return owner + ? owner.read(handle, position, length, signal) + : Promise.reject(invalidTranscriptHandleError()) +} + +export function closeWslTranscriptFsProcess(handle: WslTranscriptFsProcessHandle): Promise { + return wslTranscriptFsHandleOwners.get(handle)?.close(handle) ?? Promise.resolve() +} + +export function isWslTranscriptFsProcessHandle( + value: object +): value is WslTranscriptFsProcessHandle { + return 'wslTranscriptFsProcessHandle' in value +} + +export function resetWslTranscriptFsProcessClientForTests(): void { + for (const client of clientsByLane.values()) { + client.dispose() + } + clientsByLane.clear() +} diff --git a/src/main/native-chat/wsl-transcript-fs-process-entry.ts b/src/main/native-chat/wsl-transcript-fs-process-entry.ts new file mode 100644 index 00000000000..c3d946f067c --- /dev/null +++ b/src/main/native-chat/wsl-transcript-fs-process-entry.ts @@ -0,0 +1,40 @@ +// Forked so a killed WSL UNC syscall cannot retain a libuv thread in Orca. +import type { + WslTranscriptFsProcessError, + WslTranscriptFsProcessRequest, + WslTranscriptFsProcessResponse +} from './wsl-transcript-fs-process-protocol' +import { WslTranscriptFsProcessOperations } from './wsl-transcript-fs-process-operations' + +const operations = new WslTranscriptFsProcessOperations() + +function serializeError(error: unknown): WslTranscriptFsProcessError { + const value = error as NodeJS.ErrnoException | null + return { + name: value?.name ?? 'Error', + message: value?.message ?? String(error), + ...(typeof value?.code === 'string' ? { code: value.code } : {}), + ...(typeof value?.errno === 'number' ? { errno: value.errno } : {}), + ...(typeof value?.syscall === 'string' ? { syscall: value.syscall } : {}), + ...(typeof value?.path === 'string' ? { path: value.path } : {}) + } +} + +function respond(response: WslTranscriptFsProcessResponse): void { + try { + // The callback absorbs an async send failure (channel torn mid-write) that + // would otherwise surface as an unhandled 'error' event and crash the child. + process.send?.(response, () => {}) + } catch { + // Channel closed while the op settled (parent teardown); disconnect exits. + } +} + +process.on('message', (request: WslTranscriptFsProcessRequest) => { + void operations.execute(request).then( + (value) => respond({ id: request.id, ok: true, value }), + (error: unknown) => respond({ id: request.id, ok: false, error: serializeError(error) }) + ) +}) + +process.on('disconnect', () => process.exit(0)) diff --git a/src/main/native-chat/wsl-transcript-fs-process-handle-owner.ts b/src/main/native-chat/wsl-transcript-fs-process-handle-owner.ts new file mode 100644 index 00000000000..5b11fc208b6 --- /dev/null +++ b/src/main/native-chat/wsl-transcript-fs-process-handle-owner.ts @@ -0,0 +1,18 @@ +import { wslTranscriptFsProcessFailureError } from './wsl-transcript-fs-error' +import { invalidTranscriptHandleError } from './wsl-transcript-fs-process-protocol' +import type { WslTranscriptFsProcessClient } from './wsl-transcript-fs-process-client' +import type { WslTranscriptFsProcessHandle } from './wsl-transcript-fs-process-slot' + +export const wslTranscriptFsHandleOwners = new WeakMap< + WslTranscriptFsProcessHandle, + WslTranscriptFsProcessClient +>() + +export function processHandleUnavailableError( + handle: WslTranscriptFsProcessHandle, + faultedHandles: WeakSet +): Error { + return faultedHandles.has(handle) + ? wslTranscriptFsProcessFailureError('the process owning this file handle exited') + : invalidTranscriptHandleError() +} diff --git a/src/main/native-chat/wsl-transcript-fs-process-lane-pool.ts b/src/main/native-chat/wsl-transcript-fs-process-lane-pool.ts new file mode 100644 index 00000000000..31f060c9498 --- /dev/null +++ b/src/main/native-chat/wsl-transcript-fs-process-lane-pool.ts @@ -0,0 +1,156 @@ +import { + WSL_TRANSCRIPT_FS_PROCESS_IDLE_REAP_MS, + type ProcessSlot +} from './wsl-transcript-fs-process-slot' +import { wslTranscriptFsProcessFailureError } from './wsl-transcript-fs-error' + +type SlotWaiter = { + resolve: (slot: ProcessSlot) => void + reject: (error: unknown) => void + signal: AbortSignal + onAbort: () => void +} + +export class WslTranscriptFsProcessLanePool { + private readonly available: ProcessSlot[] = [] + private readonly slots = new Set() + private readonly waiters: SlotWaiter[] = [] + private disposed = false + private disposeError: unknown = new Error('WSL transcript filesystem process pool is disposed') + + constructor( + private readonly createSlot: () => ProcessSlot, + private readonly retireIdleSlot: (slot: ProcessSlot) => void + ) {} + + acquire(signal: AbortSignal, prioritize = false): ProcessSlot | Promise { + signal.throwIfAborted() + if (this.disposed) { + return Promise.reject(this.disposeError) + } + const slot = this.available.pop() + if (slot) { + clearTimeout(slot.idleTimer) + return slot + } + if (this.slots.size === 0) { + return this.addSlot() + } + return new Promise((resolve, reject) => { + const waiter: SlotWaiter = { + resolve, + reject, + signal, + onAbort: () => { + const index = this.waiters.indexOf(waiter) + if (index !== -1) { + this.waiters.splice(index, 1) + } + reject(signal.reason ?? new Error('WSL filesystem process acquisition aborted')) + } + } + signal.addEventListener('abort', waiter.onAbort, { once: true }) + if (prioritize) { + this.waiters.unshift(waiter) + } else { + this.waiters.push(waiter) + } + }) + } + + park(slot: ProcessSlot): void { + if (!this.slots.has(slot)) { + return + } + const waiter = this.waiters.shift() + if (waiter) { + waiter.signal.removeEventListener('abort', waiter.onAbort) + waiter.resolve(slot) + return + } + if (this.available.includes(slot)) { + return + } + this.available.push(slot) + if (slot.handles.size === 0) { + slot.idleTimer = setTimeout( + () => this.retireIdleSlot(slot), + WSL_TRANSCRIPT_FS_PROCESS_IDLE_REAP_MS + ) + slot.idleTimer.unref?.() + } + } + + has(slot: ProcessSlot): boolean { + return this.slots.has(slot) + } + + claim(slot: ProcessSlot, signal: AbortSignal, release?: () => void): ProcessSlot { + if (!this.slots.has(slot)) { + throw this.disposed + ? this.disposeError + : wslTranscriptFsProcessFailureError('the process exited before the queued request started') + } + if (signal.aborted) { + if (release) { + release() + } else { + this.park(slot) + } + signal.throwIfAborted() + } + return slot + } + + snapshot(): ProcessSlot[] { + return [...this.slots] + } + + beginDispose(error: unknown): void { + this.disposed = true + this.disposeError = error + for (const waiter of this.waiters.splice(0)) { + waiter.signal.removeEventListener('abort', waiter.onAbort) + waiter.reject(error) + } + } + + retire(slot: ProcessSlot): boolean { + if (!this.slots.delete(slot)) { + return false + } + clearTimeout(slot.idleTimer) + const availableIndex = this.available.indexOf(slot) + if (availableIndex !== -1) { + this.available.splice(availableIndex, 1) + } + this.replaceForWaiter() + return true + } + + private addSlot(): ProcessSlot { + const slot = this.createSlot() + this.slots.add(slot) + return slot + } + + private replaceForWaiter(): void { + if (this.disposed || this.slots.size > 0) { + return + } + while (this.waiters.length > 0) { + const waiter = this.waiters.shift()! + waiter.signal.removeEventListener('abort', waiter.onAbort) + if (waiter.signal.aborted) { + waiter.reject(waiter.signal.reason) + continue + } + try { + waiter.resolve(this.addSlot()) + return + } catch (error) { + waiter.reject(error) + } + } + } +} diff --git a/src/main/native-chat/wsl-transcript-fs-process-operations.test.ts b/src/main/native-chat/wsl-transcript-fs-process-operations.test.ts new file mode 100644 index 00000000000..f38a04462ab --- /dev/null +++ b/src/main/native-chat/wsl-transcript-fs-process-operations.test.ts @@ -0,0 +1,48 @@ +import { mkdtemp, rename, rm, writeFile } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { afterEach, describe, expect, it } from 'vitest' +import { WslTranscriptFsProcessOperations } from './wsl-transcript-fs-process-operations' + +const temporaryDirectories: string[] = [] + +afterEach(async () => { + await Promise.all( + temporaryDirectories + .splice(0) + .map((directory) => rm(directory, { recursive: true, force: true })) + ) +}) + +describe('WSL transcript filesystem process operations', () => { + // Local NTFS rejects replacing an open file; WSL/9P follows Linux rename semantics. + it.skipIf(process.platform === 'win32')( + 'keeps positional reads on the opened inode after atomic path replacement', + async () => { + const directory = await mkdtemp(join(tmpdir(), 'orca-wsl-transcript-')) + temporaryDirectories.push(directory) + const transcriptPath = join(directory, 'session.jsonl') + const replacementPath = join(directory, 'replacement.jsonl') + await writeFile(transcriptPath, 'original transcript') + await writeFile(replacementPath, 'replacement bytes') + const operations = new WslTranscriptFsProcessOperations() + + const handleId = (await operations.execute({ + id: 1, + operation: 'open', + path: transcriptPath + })) as number + await rename(replacementPath, transcriptPath) + const body = await operations.execute({ + id: 2, + operation: 'read', + handleId, + position: 0, + length: 64 + }) + await operations.execute({ id: 3, operation: 'close', handleId }) + + expect(Buffer.from(body as Buffer).toString('utf8')).toBe('original transcript') + } + ) +}) diff --git a/src/main/native-chat/wsl-transcript-fs-process-operations.ts b/src/main/native-chat/wsl-transcript-fs-process-operations.ts new file mode 100644 index 00000000000..b76ee84e4d6 --- /dev/null +++ b/src/main/native-chat/wsl-transcript-fs-process-operations.ts @@ -0,0 +1,63 @@ +import { access, lstat, open, readdir, readFile, stat, type FileHandle } from 'node:fs/promises' +import type { Dirent } from 'node:fs' +import { + invalidTranscriptHandleError, + type WslTranscriptFsDirent, + type WslTranscriptFsProcessRequest +} from './wsl-transcript-fs-process-protocol' + +function serializeDirent(entry: Dirent): WslTranscriptFsDirent { + return { + name: entry.name, + parentPath: entry.parentPath, + isBlockDevice: entry.isBlockDevice(), + isCharacterDevice: entry.isCharacterDevice(), + isDirectory: entry.isDirectory(), + isFIFO: entry.isFIFO(), + isFile: entry.isFile(), + isSocket: entry.isSocket(), + isSymbolicLink: entry.isSymbolicLink() + } +} + +export class WslTranscriptFsProcessOperations { + private readonly handles = new Map() + private nextHandleId = 1 + + async execute(request: WslTranscriptFsProcessRequest): Promise { + switch (request.operation) { + case 'access': + await access(request.path) + return true + case 'stat': + return stat(request.path) + case 'lstat': + return lstat(request.path) + case 'readdir': + return (await readdir(request.path, { withFileTypes: true })).map(serializeDirent) + case 'readfile': + return readFile(request.path, request.encoding) + case 'open': { + const handle = await open(request.path, 'r') + const handleId = this.nextHandleId++ + this.handles.set(handleId, handle) + return handleId + } + case 'read': { + const handle = this.handles.get(request.handleId) + if (!handle) { + throw invalidTranscriptHandleError() + } + const buffer = Buffer.allocUnsafe(request.length) + const { bytesRead } = await handle.read(buffer, 0, request.length, request.position) + return buffer.subarray(0, bytesRead) + } + case 'close': { + const handle = this.handles.get(request.handleId) + this.handles.delete(request.handleId) + await handle?.close() + return true + } + } + } +} diff --git a/src/main/native-chat/wsl-transcript-fs-process-protocol.ts b/src/main/native-chat/wsl-transcript-fs-process-protocol.ts new file mode 100644 index 00000000000..db9580cdb92 --- /dev/null +++ b/src/main/native-chat/wsl-transcript-fs-process-protocol.ts @@ -0,0 +1,51 @@ +export type WslTranscriptFsProcessCall = + | { operation: 'access'; path: string } + | { operation: 'stat' | 'lstat' | 'readdir'; path: string } + // Kept as its own member so the reusable-call Exclude below can strip it: + // Exclude compares whole union members, not individual operation literals. + | { operation: 'open'; path: string } + | { operation: 'readfile'; path: string; encoding: BufferEncoding } + | { operation: 'read'; handleId: number; position: number; length: number } + | { operation: 'close'; handleId: number } + +// The intersection distributes over the union, so `{ ...call, id }` composes +// a request without casts at the IPC boundary. +export type WslTranscriptFsProcessRequest = WslTranscriptFsProcessCall & { id: number } + +/** Calls a pooled process may serve; open/read/close manage a pinned handle. */ +export type WslTranscriptFsReusableProcessCall = Exclude< + WslTranscriptFsProcessCall, + { operation: 'open' | 'read' | 'close' } +> + +export type WslTranscriptFsDirent = { + name: string + parentPath: string + isBlockDevice: boolean + isCharacterDevice: boolean + isDirectory: boolean + isFIFO: boolean + isFile: boolean + isSocket: boolean + isSymbolicLink: boolean +} + +export type WslTranscriptFsProcessError = { + name: string + message: string + code?: string + errno?: number + syscall?: string + path?: string +} + +export type WslTranscriptFsProcessResponse = + | { id: number; ok: true; value: unknown } + | { id: number; ok: false; error: WslTranscriptFsProcessError } + +/** The owning process, client, or entry no longer knows this handle. */ +export function invalidTranscriptHandleError(): NodeJS.ErrnoException { + return Object.assign(new Error('WSL transcript file handle is no longer available'), { + code: 'EBADF' + }) +} diff --git a/src/main/native-chat/wsl-transcript-fs-process-send.ts b/src/main/native-chat/wsl-transcript-fs-process-send.ts new file mode 100644 index 00000000000..6bddf859233 --- /dev/null +++ b/src/main/native-chat/wsl-transcript-fs-process-send.ts @@ -0,0 +1,51 @@ +import type { + WslTranscriptFsProcessCall, + WslTranscriptFsProcessRequest +} from './wsl-transcript-fs-process-protocol' +import type { + ProcessSlot, + SlotDisposition, + WslTranscriptFsProcessHandle +} from './wsl-transcript-fs-process-slot' + +export function sendWslTranscriptFsProcessRequest(args: { + slot: ProcessSlot + id: number + request: WslTranscriptFsProcessCall + signal: AbortSignal + disposition: SlotDisposition + handle?: WslTranscriptFsProcessHandle + onAbort: (reason: unknown) => void + onTransportFailure: (error: unknown) => void +}): Promise { + return new Promise((resolve, reject) => { + const onAbort = (): void => { + if (args.slot.active?.id === args.id) { + args.onAbort(args.signal.reason ?? new Error('WSL filesystem process aborted')) + } + } + args.slot.active = { + id: args.id, + resolve: resolve as (value: unknown) => void, + reject, + signal: args.signal, + onAbort, + operation: args.request.operation, + disposition: args.disposition, + handle: args.handle + } + args.signal.addEventListener('abort', onAbort, { once: true }) + try { + args.slot.child.send( + { ...args.request, id: args.id } as WslTranscriptFsProcessRequest, + (error) => { + if (error && args.slot.active?.id === args.id) { + args.onTransportFailure(error) + } + } + ) + } catch (error) { + args.onTransportFailure(error) + } + }) +} diff --git a/src/main/native-chat/wsl-transcript-fs-process-slot.ts b/src/main/native-chat/wsl-transcript-fs-process-slot.ts new file mode 100644 index 00000000000..b6fc3bf77a2 --- /dev/null +++ b/src/main/native-chat/wsl-transcript-fs-process-slot.ts @@ -0,0 +1,69 @@ +import type { ChildProcess } from 'node:child_process' +import type { + WslTranscriptFsProcessRequest, + WslTranscriptFsProcessResponse +} from './wsl-transcript-fs-process-protocol' + +/** + * The data model one pooled helper child is tracked by: at most one in-flight + * call, with any opened handles owned by that child. + */ + +export type SlotDisposition = 'idle' | 'pin' | 'pinned' | 'close' + +export type ActiveCall = { + id: number + resolve: (value: unknown) => void + reject: (error: unknown) => void + signal: AbortSignal + onAbort: () => void + operation: WslTranscriptFsProcessRequest['operation'] + disposition: SlotDisposition + handle?: WslTranscriptFsProcessHandle +} + +export type WslTranscriptFsProcessHandle = { + readonly wslTranscriptFsProcessHandle: true +} + +export type ProcessSlot = { + child: ChildProcess + active: ActiveCall | null + handles: Set + idleTimer?: ReturnType +} + +export type HandleState = { + slot: ProcessSlot + handleId: number + closePromise?: Promise +} + +export type WslTranscriptFsProcessFactory = () => ChildProcess + +export const WSL_TRANSCRIPT_FS_PROCESS_CLOSE_TIMEOUT_MS = 30_000 +// Why: idle Electron-as-Node children are tens of MB; reap them instead of +// holding RSS for the app session. +// Longer than the close deadline so a pending close never outlives its slot. +export const WSL_TRANSCRIPT_FS_PROCESS_IDLE_REAP_MS = 60_000 + +export function attachSlotChild( + child: ChildProcess, + handlers: { + onResponse: (response: WslTranscriptFsProcessResponse) => void + onFault: (error: Error) => void + } +): ProcessSlot { + const slot: ProcessSlot = { child, active: null, handles: new Set() } + child.on('message', (response: WslTranscriptFsProcessResponse) => handlers.onResponse(response)) + child.on('error', (error) => handlers.onFault(error)) + child.on('disconnect', () => handlers.onFault(new Error('WSL filesystem process disconnected'))) + // A signal-killed child has code null; name the signal, not "(null)". + child.on('exit', (code, signal) => + handlers.onFault(new Error(`WSL filesystem process exited (${signal ?? code})`)) + ) + // Neither the child nor its channel may keep the parent's event loop alive. + child.unref() + child.channel?.unref?.() + return slot +} diff --git a/src/main/native-chat/wsl-transcript-fs-process-spawn.ts b/src/main/native-chat/wsl-transcript-fs-process-spawn.ts new file mode 100644 index 00000000000..5d67cba8bed --- /dev/null +++ b/src/main/native-chat/wsl-transcript-fs-process-spawn.ts @@ -0,0 +1,58 @@ +import { fork, type ChildProcess } from 'node:child_process' +import { existsSync } from 'node:fs' +import { join } from 'node:path' +import { pickAllowedEnv, RUNTIME_ENV_ALLOWLIST } from '../ai-vault/session-scanner-service-env' + +const PROCESS_ENTRY_FILENAME = 'wsl-transcript-fs-process-entry.js' + +// Why: never `...process.env` into a forked transcript reader — an ambient +// NODE_OPTIONS would halt (--inspect-brk) or --require code into every child, +// and shell-exported secrets have no business in one. Shares the AI Vault +// runtime allowlist: only what Node/libuv need to start. +export function wslTranscriptFsProcessForkEnv( + baseEnv: NodeJS.ProcessEnv = process.env, + platform: NodeJS.Platform = process.platform +): NodeJS.ProcessEnv { + const env = pickAllowedEnv(RUNTIME_ENV_ALLOWLIST, baseEnv, platform) + env.ELECTRON_RUN_AS_NODE = '1' + return env +} + +export function resolveWslTranscriptFsProcessEntryPath( + moduleDir: string, + resourcesPath: string | undefined = process.resourcesPath, + pathExists: (path: string) => boolean = existsSync +): string { + // Why: this module compiles into out/main or out/main/chunks, so probe both + // levels. ELECTRON_RUN_AS_NODE children (the scanner service) bypass asar and + // have no process.resourcesPath, so the __dirname legs must succeed there. + const toUnpackedDir = (dir: string): string => + dir.replace(/([\\/])app\.asar(?=([\\/]|$))/, '$1app.asar.unpacked') + for (const baseDir of [moduleDir, join(moduleDir, '..')].map(toUnpackedDir)) { + const candidate = join(baseDir, PROCESS_ENTRY_FILENAME) + if (pathExists(candidate)) { + return candidate + } + } + if (resourcesPath) { + const packaged = join(resourcesPath, 'app.asar.unpacked', 'out', 'main', PROCESS_ENTRY_FILENAME) + if (pathExists(packaged)) { + return packaged + } + } + return join(process.cwd(), 'out', 'main', PROCESS_ENTRY_FILENAME) +} + +export function forkWslTranscriptFsProcess(): ChildProcess { + const entryPath = resolveWslTranscriptFsProcessEntryPath(__dirname) + if (!existsSync(entryPath)) { + throw new Error(`WSL transcript filesystem process entry not found: ${entryPath}`) + } + return fork(entryPath, [], { + env: wslTranscriptFsProcessForkEnv(), + execArgv: [], + serialization: 'advanced', + stdio: ['ignore', 'ignore', 'ignore', 'ipc'], + ...(process.platform === 'win32' ? { windowsHide: true } : {}) + }) +} diff --git a/src/main/native-chat/wsl-transcript-fs-route-quarantine.test.ts b/src/main/native-chat/wsl-transcript-fs-route-quarantine.test.ts new file mode 100644 index 00000000000..2a8a7668b57 --- /dev/null +++ b/src/main/native-chat/wsl-transcript-fs-route-quarantine.test.ts @@ -0,0 +1,151 @@ +import { afterEach, beforeEach, describe, expect, it, vi, type MockInstance } from 'vitest' +import { + resetWslTranscriptFsGateForTests, + runWslTranscriptFsTask, + WSL_TRANSCRIPT_FS_EXACT_TIMEOUT_MS, + WSL_TRANSCRIPT_FS_ROUTE_QUARANTINE_BASE_MS, + WSL_TRANSCRIPT_FS_SCAN_TIMEOUT_MS +} from './wsl-transcript-fs-gate' +import { WSL_TRANSCRIPT_FS_ROUTE_STRIKE_DECAY_MS } from './wsl-transcript-fs-route-quarantine' + +function deferred(): { + promise: Promise + resolve: (value: T) => void + reject: (error: unknown) => void +} { + let resolve!: (value: T) => void + let reject!: (error: unknown) => void + return { + promise: new Promise((res, rej) => ((resolve = res), (reject = rej))), + resolve, + reject + } +} + +function run( + path: string, + priority: 'exact' | 'scan', + task: () => Promise, + signal?: AbortSignal +): Promise { + return runWslTranscriptFsTask( + { operation: priority === 'exact' ? 'access' : 'readdir', path, priority, signal }, + task + ) +} + +describe('WSL transcript fs route quarantine strike accounting', () => { + let warnSpy: MockInstance + + beforeEach(() => { + resetWslTranscriptFsGateForTests() + warnSpy = vi.spyOn(console, 'warn').mockImplementation(() => {}) + }) + + afterEach(() => { + warnSpy.mockRestore() + }) + + // Why: joining costs no new I/O — the in-flight task is bounded by its own + // deadline and its settle may itself lift the quarantine. Refusing would fail + // pollers whose answer is already seconds away. + it('joins a live in-flight task on a quarantined route instead of refusing', async () => { + vi.useFakeTimers() + const scanWork = deferred() + try { + const path = '\\\\wsl.localhost\\Ubuntu\\join-during-quarantine' + const scanTask = vi.fn(() => scanWork.promise) + const scanned = runWslTranscriptFsTask( + { operation: 'stat', path, priority: 'scan' }, + scanTask + ) + await vi.advanceTimersByTimeAsync(0) + expect(scanTask).toHaveBeenCalledOnce() + + const stalled = run( + '\\\\wsl.localhost\\Ubuntu\\join-hung', + 'exact', + () => new Promise(() => {}) + ) + const stalledRejected = expect(stalled).rejects.toMatchObject({ code: 'timeout' }) + await vi.advanceTimersByTimeAsync(WSL_TRANSCRIPT_FS_EXACT_TIMEOUT_MS) + await stalledRejected + + // New work on the route is refused, but joining the live stat is free. + await expect( + run('\\\\wsl.localhost\\Ubuntu\\join-fresh', 'exact', async () => 'fresh') + ).rejects.toMatchObject({ code: 'unavailable' }) + const joinerTask = vi.fn(async () => 'never') + const joined = runWslTranscriptFsTask( + { operation: 'stat', path, priority: 'scan' }, + joinerTask + ) + + scanWork.resolve('shared') + await expect(Promise.all([scanned, joined])).resolves.toEqual(['shared', 'shared']) + expect(joinerTask).not.toHaveBeenCalled() + } finally { + scanWork.resolve('shared') + await vi.advanceTimersByTimeAsync(0) + vi.useRealTimers() + } + }) + + // Why: one hung mount usually stalls the exact and scan lanes together; two + // deadline strikes for one incident would double-step the back-off. + it('counts concurrent lane deadlines on one stall as a single strike', async () => { + vi.useFakeTimers({ toFake: ['setTimeout', 'clearTimeout', 'Date', 'performance'] }) + try { + const path = '\\\\wsl.localhost\\Ubuntu\\two-lane-stall' + const exact = run(path, 'exact', () => new Promise(() => {})) + const scan = run( + '\\\\wsl.localhost\\Ubuntu\\two-lane-tree', + 'scan', + () => new Promise(() => {}) + ) + const exactRejected = expect(exact).rejects.toMatchObject({ code: 'timeout' }) + const scanRejected = expect(scan).rejects.toMatchObject({ code: 'timeout' }) + await vi.advanceTimersByTimeAsync(WSL_TRANSCRIPT_FS_SCAN_TIMEOUT_MS) + await Promise.all([exactRejected, scanRejected]) + + // Still a first strike: admitted again after one base window (a second + // strike would demand two). + await vi.advanceTimersByTimeAsync(WSL_TRANSCRIPT_FS_ROUTE_QUARANTINE_BASE_MS) + await expect(run(path, 'exact', async () => 'recovered')).resolves.toBe('recovered') + } finally { + vi.useRealTimers() + } + }) + + it('starts the back-off from the base window again after strike history decays', async () => { + vi.useFakeTimers({ toFake: ['setTimeout', 'clearTimeout', 'Date', 'performance'] }) + const path = '\\\\wsl.localhost\\Ubuntu\\daily-slow-wake' + const stallOnce = (): Promise => + runWslTranscriptFsTask( + { operation: 'open', path, priority: 'exact', dedupe: false }, + (signal) => + new Promise((_resolve, reject) => + signal.addEventListener('abort', () => reject(signal.reason), { once: true }) + ) + ) + try { + const first = stallOnce() + const firstRejected = expect(first).rejects.toMatchObject({ code: 'timeout' }) + await vi.advanceTimersByTimeAsync(WSL_TRANSCRIPT_FS_EXACT_TIMEOUT_MS) + await firstRejected + + // A quiet stretch beyond the decay window forgets the strike history. + await vi.advanceTimersByTimeAsync(WSL_TRANSCRIPT_FS_ROUTE_STRIKE_DECAY_MS + 1) + const second = stallOnce() + const secondRejected = expect(second).rejects.toMatchObject({ code: 'timeout' }) + await vi.advanceTimersByTimeAsync(WSL_TRANSCRIPT_FS_EXACT_TIMEOUT_MS) + await secondRejected + + // First-strike back-off again, not an escalated second strike. + await vi.advanceTimersByTimeAsync(WSL_TRANSCRIPT_FS_ROUTE_QUARANTINE_BASE_MS) + await expect(run(path, 'exact', async () => 'fresh-start')).resolves.toBe('fresh-start') + } finally { + vi.useRealTimers() + } + }) +}) diff --git a/src/main/native-chat/wsl-transcript-fs-route-quarantine.ts b/src/main/native-chat/wsl-transcript-fs-route-quarantine.ts new file mode 100644 index 00000000000..affce12f0c5 --- /dev/null +++ b/src/main/native-chat/wsl-transcript-fs-route-quarantine.ts @@ -0,0 +1,58 @@ +/** + * Route-level quarantine with exponential back-off. A killed helper cannot + * report late success, so a stalled route's recovery is only probeable: block + * admissions briefly, let the next real task be the probe, escalate on repeat. + */ + +const ROUTE_RETRY_DELAY_MULTIPLIER = 2 +// A short first strike lets a cold-booting distro recover on the next poll. +export const WSL_TRANSCRIPT_FS_ROUTE_QUARANTINE_BASE_MS = 5_000 +// Strikes older than this stop escalating: a distro that wakes slowly once a +// day must restart from the base window, not resume yesterday's back-off. +export const WSL_TRANSCRIPT_FS_ROUTE_STRIKE_DECAY_MS = 5 * 60_000 + +type RouteQuarantine = { until: number; strikes: number; setAt: number } +// Monotonic clock: wall time would misjudge the window across sleep/NTP steps. +const blockedRoutes = new Map() + +export function routeIsBlocked(route: string): boolean { + const blocked = blockedRoutes.get(route) + // Expired entries persist: their strike count seeds the next back-off. + return blocked !== undefined && performance.now() < blocked.until +} + +/** + * One more strike: block the route with doubled back-off, capped by deadline. + * `taskStartedAt` identifies the incident — the exact and scan lanes usually + * both stall on one hung mount, and a task admitted before the current + * quarantine was set is the sibling lane reporting that same stall, so it + * re-arms the window without escalating the strike count. + */ +export function quarantineRoute(route: string, deadlineMs: number, taskStartedAt: number): void { + const now = performance.now() + const previous = blockedRoutes.get(route) + const seed = + previous !== undefined && now - previous.until <= WSL_TRANSCRIPT_FS_ROUTE_STRIKE_DECAY_MS + ? previous + : undefined + const sameIncident = seed !== undefined && taskStartedAt <= seed.setAt + const strikes = seed === undefined ? 1 : sameIncident ? seed.strikes : seed.strikes + 1 + const quarantineMs = Math.min( + WSL_TRANSCRIPT_FS_ROUTE_QUARANTINE_BASE_MS * 2 ** (strikes - 1), + deadlineMs * ROUTE_RETRY_DELAY_MULTIPLIER + ) + blockedRoutes.set(route, { + until: Math.max(seed?.until ?? 0, now + quarantineMs), + strikes, + setAt: sameIncident ? seed.setAt : now + }) +} + +/** A real filesystem answer proves the mount is alive: forget the strikes. */ +export function liftRouteQuarantine(route: string): void { + blockedRoutes.delete(route) +} + +export function resetRouteQuarantinesForTests(): void { + blockedRoutes.clear() +} diff --git a/src/main/native-chat/wsl-transcript-fs-route.test.ts b/src/main/native-chat/wsl-transcript-fs-route.test.ts new file mode 100644 index 00000000000..f49c47c705a --- /dev/null +++ b/src/main/native-chat/wsl-transcript-fs-route.test.ts @@ -0,0 +1,18 @@ +import { describe, expect, it } from 'vitest' +import { wslTranscriptFsLaneKey } from './wsl-transcript-fs-route' + +describe('WSL transcript filesystem lane keys', () => { + it('shares one process within a route and priority while isolating other lanes', () => { + const ubuntu = '\\\\wsl.localhost\\Ubuntu\\home\\ada' + + expect(wslTranscriptFsLaneKey(`${ubuntu}\\one`, 'scan')).toBe( + wslTranscriptFsLaneKey(`${ubuntu}\\two`, 'scan') + ) + expect(wslTranscriptFsLaneKey(`${ubuntu}\\one`, 'exact')).not.toBe( + wslTranscriptFsLaneKey(`${ubuntu}\\one`, 'scan') + ) + expect(wslTranscriptFsLaneKey(`${ubuntu}\\one`, 'scan')).not.toBe( + wslTranscriptFsLaneKey('\\\\wsl.localhost\\Debian\\home\\ada\\one', 'scan') + ) + }) +}) diff --git a/src/main/native-chat/wsl-transcript-fs-route.ts b/src/main/native-chat/wsl-transcript-fs-route.ts index dc576c066d7..130171672d6 100644 --- a/src/main/native-chat/wsl-transcript-fs-route.ts +++ b/src/main/native-chat/wsl-transcript-fs-route.ts @@ -2,8 +2,8 @@ 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. + * must not fast-fail or strand work on any other. Gate admission and helper + * ownership share this key so scheduling and process faults isolate alike. * * wsl$ and wsl.localhost spellings of one distro stay distinct routes on * purpose (provider spelling can change behavior), so a stuck spelling never @@ -17,3 +17,7 @@ export function wslTranscriptFsRouteKey(path: string): string { } return parseWslUncPath(path)?.distro.trim().toLowerCase() ?? path } + +export function wslTranscriptFsLaneKey(path: string, priority: 'exact' | 'scan'): string { + return `${wslTranscriptFsRouteKey(path)}:${priority}` +} diff --git a/src/main/opencode-usage/scanner-wsl-gate.test.ts b/src/main/opencode-usage/scanner-wsl-gate.test.ts index 44d6463d9f9..c8bb123c476 100644 --- a/src/main/opencode-usage/scanner-wsl-gate.test.ts +++ b/src/main/opencode-usage/scanner-wsl-gate.test.ts @@ -22,6 +22,7 @@ vi.mock('node:fs/promises', async (importOriginal) => ({ import { listOpenCodeDatabases } from './opencode-database-discovery' import { + resetWslTranscriptFsGateForTests, WSL_TRANSCRIPT_FS_SCAN_TIMEOUT_MS, WslTranscriptFsError } from '../native-chat/wsl-transcript-fs-gate' @@ -36,8 +37,19 @@ function stalls(): Promise { }) } +// Complete: UNC readdir results pass through the child dispatcher's dirent +// serializer, which reads every kind flag. function dirent(name: string) { - return { name, isFile: () => true } + return { + name, + isBlockDevice: () => false, + isCharacterDevice: () => false, + isDirectory: () => false, + isFIFO: () => false, + isFile: () => true, + isSocket: () => false, + isSymbolicLink: () => false + } } // The point of the gate: an ungated syscall on a stalled 9P mount never returns, @@ -57,6 +69,9 @@ async function settlesOnlyAtTheScanDeadline(pending: Promise): Promise let originalDatabaseOverride: string | undefined beforeEach(() => { + // blockedRoutes is persistent gate state: a prior stall must not quarantine + // this test's route. + resetWslTranscriptFsGateForTests() originalDatabaseOverride = process.env.OPENCODE_DB delete process.env.OPENCODE_DB mocks.resolveDataDirectory.mockReset()