From 547e7b5581d4cd6b25007cffca8a3c7b9cc01655 Mon Sep 17 00:00:00 2001 From: Jinjing <6427696+AmethystLiang@users.noreply.github.com> Date: Wed, 12 Aug 2026 20:46:28 -0700 Subject: [PATCH] refactor(native-chat): consolidate transcript I/O and remove fallback te - Move boundaryFingerprint from its own module to transcript-file-version.ts - Extract runPathOperation helper to eliminate duplicate UNC path routing - Remove tests for fallback behaviors when transcripts are unavailable or incomplete - Clean up implementation comments and verbose test documentation --- ...sion-scanner-core-parser-wsl-stall.test.ts | 52 ++----- .../transcript-boundary-fingerprint.ts | 21 --- .../native-chat/transcript-file-version.ts | 17 ++- ...script-read-cache-refusal-recovery.test.ts | 77 +++++----- .../transcript-read-cache-wsl-stall.test.ts | 57 -------- .../native-chat/transcript-watch-engine.ts | 2 +- .../wsl-transcript-fs-access.test.ts | 64 +++------ .../native-chat/wsl-transcript-fs-access.ts | 131 ++++++------------ 8 files changed, 128 insertions(+), 293 deletions(-) delete mode 100644 src/main/native-chat/transcript-boundary-fingerprint.ts delete mode 100644 src/main/native-chat/transcript-read-cache-wsl-stall.test.ts 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 1d5fa7e1c61..72f767ec09a 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 @@ -18,10 +18,7 @@ vi.mock('node:fs/promises', async (importOriginal) => ({ stat: mocks.stat })) -// These parsers derive sibling transcript paths (Kimi's wire file, Grok's -// chat_history, OpenCode's message dir) from the session file. Under POSIX -// semantics a `\\wsl.localhost\...` path has no separators, so every derived -// path would collapse to '.' and silently stop being a UNC path. +// Derived sibling paths must retain Windows separators in this cross-platform test. vi.mock('node:path', async (importOriginal) => { const actual = await importOriginal() return { ...actual.win32, default: actual.win32 } @@ -36,8 +33,6 @@ import { WslTranscriptFsError } from '../native-chat/wsl-transcript-fs-gate' -// One distro per case: route state (stuck-blocker fast-fail) is keyed by distro, -// so cases cannot leak a refusal into each other. function uncPath(distro: string, ...segments: string[]): string { return `\\\\wsl.localhost\\${distro}\\home\\ada\\${segments.join('\\')}` } @@ -84,8 +79,6 @@ const OPENCODE_MESSAGE = JSON.stringify({ type ReadResult = { bytesRead: number; buffer: Buffer } -// A stalled task holds the gate's single scan slot until it settles, so every -// case releases its stall before the next one runs. let releaseStall: (() => void) | undefined function stalls(): Promise { @@ -121,14 +114,17 @@ function dirent(name: string) { function candidate(agent: SessionFileCandidate['agent'], path: string): SessionFileCandidate { return { agent, - file: { path, mtimeMs: 1, modifiedAt: '2026-06-01T10:05:00.000Z', sizeBytes: 128 }, + file: { + path, + mtimeMs: 1, + modifiedAt: '2026-06-01T10:05:00.000Z', + sizeBytes: 128 + }, codexHome: null } } async function expectRefusal(target: SessionFileCandidate): Promise { - // Attach the assertion before the clock moves so the rejection is never - // momentarily unhandled. const refusal = expect(parseAgentSessionFileCached(target, 'linux')).rejects.toBeInstanceOf( WslTranscriptFsError ) @@ -149,8 +145,6 @@ beforeEach(() => { mocks.readdir.mockReset() mocks.stat.mockReset() releaseStall = undefined - // Kimi's session_index.jsonl is absent in these fixtures: sessions still list, - // just without a cwd, and the lookup never reaches the gate's stream path. mocks.stat.mockRejectedValue(missing()) vi.useFakeTimers() }) @@ -176,23 +170,11 @@ describe('Kimi session parse against a stalled WSL transcript', () => { await expectRefusal(candidate('kimi', state)) await releaseAndSettle() - // Same mtime as the refused attempt: a cached metadata-only session would - // still be served here, so a real transcript proves nothing was stored. mocks.open.mockResolvedValue(servingHandle(KIMI_WIRE)) const recovered = await parseAgentSessionFileCached(candidate('kimi', state), 'linux') expect(recovered?.messageCount).toBe(1) }) - - it('still returns a metadata-only session when the wire file is simply missing', async () => { - const state = uncPath('KimiMissing', ...KIMI_HOME, 'state.json') - mocks.readFile.mockResolvedValue(KIMI_STATE) - mocks.open.mockRejectedValue(missing()) - - const session = await parseAgentSessionFileCached(candidate('kimi', state), 'linux') - - expect(session).toMatchObject({ agent: 'kimi', title: 'Kimi session', messageCount: 0 }) - }) }) describe('Grok session parse against a stalled WSL transcript', () => { @@ -207,8 +189,6 @@ describe('Grok session parse against a stalled WSL transcript', () => { mocks.open.mockResolvedValue(servingHandle(GROK_HISTORY)) const recovered = await parseAgentSessionFileCached(candidate('grok', file), 'linux') - // Grok's messageCount comes from the session doc, so the transcript's - // contribution shows up in the previews. expect(recovered?.previewMessages).toEqual([ expect.objectContaining({ role: 'user', text: 'ship it' }) ]) @@ -221,7 +201,11 @@ describe('Grok session parse against a stalled WSL transcript', () => { const session = await parseAgentSessionFileCached(candidate('grok', file), 'linux') - expect(session).toMatchObject({ agent: 'grok', title: 'Grok session', previewMessages: [] }) + expect(session).toMatchObject({ + agent: 'grok', + title: 'Grok session', + previewMessages: [] + }) }) }) @@ -254,16 +238,4 @@ describe('OpenCode session parse against a stalled WSL transcript', () => { expect(recovered?.messageCount).toBe(1) }) - - it('still returns the session when a live process left a half-written message', async () => { - const file = sessionPath('OpenCodeHalfWritten') - mocks.readdir.mockResolvedValue([dirent('msg-1.json')]) - mocks.readFile.mockImplementation((path: string) => - Promise.resolve(path === file ? OPENCODE_SESSION : '{"role":"user",') - ) - - const session = await parseAgentSessionFileCached(candidate('opencode', file), 'linux') - - expect(session).toMatchObject({ agent: 'opencode', title: 'OpenCode session', messageCount: 0 }) - }) }) diff --git a/src/main/native-chat/transcript-boundary-fingerprint.ts b/src/main/native-chat/transcript-boundary-fingerprint.ts deleted file mode 100644 index 45bee32f1ce..00000000000 --- a/src/main/native-chat/transcript-boundary-fingerprint.ts +++ /dev/null @@ -1,21 +0,0 @@ -import { readTranscriptSlice } from './wsl-transcript-fs-access' - -const BOUNDARY_FINGERPRINT_BYTES = 64 - -/** - * Fingerprint the bytes immediately before `offset`, so a same-size rewrite that - * leaves mtime untouched is still detected as replaced content rather than - * treated as an append the reader already consumed. - */ -export async function boundaryFingerprint( - filePath: string, - offset: number, - signal?: AbortSignal -): Promise { - if (offset <= 0) { - return '' - } - const start = Math.max(0, offset - BOUNDARY_FINGERPRINT_BYTES) - const slice = await readTranscriptSlice(filePath, start, offset - start, 'exact', signal) - return slice.toString('base64') -} diff --git a/src/main/native-chat/transcript-file-version.ts b/src/main/native-chat/transcript-file-version.ts index 0ed9a25e3a1..2de3c76821f 100644 --- a/src/main/native-chat/transcript-file-version.ts +++ b/src/main/native-chat/transcript-file-version.ts @@ -1,4 +1,6 @@ -import { wslGatedStat } from './wsl-transcript-fs-access' +import { readTranscriptSlice, wslGatedStat } from './wsl-transcript-fs-access' + +const BOUNDARY_FINGERPRINT_BYTES = 64 export type TranscriptFileVersion = { identity: string @@ -20,6 +22,19 @@ export async function readTranscriptFileVersion( } } +export async function boundaryFingerprint( + filePath: string, + offset: number, + signal?: AbortSignal +): Promise { + if (offset <= 0) { + return '' + } + const start = Math.max(0, offset - BOUNDARY_FINGERPRINT_BYTES) + const slice = await readTranscriptSlice(filePath, start, offset - start, 'exact', signal) + return slice.toString('base64') +} + export function transcriptFileVersionChanged( current: TranscriptFileVersion, previous: TranscriptFileVersion 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 b15f29a7f6d..4e865b306fc 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 @@ -2,16 +2,15 @@ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' import type * as NodeFsPromisesModule from 'node:fs/promises' const UNC_PATH = '\\\\wsl.localhost\\Ubuntu\\home\\ada\\.claude\\projects\\p\\session.jsonl' -const SLOW_MESSAGE = - 'WSL transcript files are temporarily unavailable because filesystem access is taking too long. Try again shortly or restart Orca if the issue continues.' - const mocks = vi.hoisted(() => ({ resolve: vi.fn<() => Promise>(), stat: vi.fn(), open: vi.fn() })) -vi.mock('./session-file-resolver', () => ({ resolveSessionFilePath: mocks.resolve })) +vi.mock('./session-file-resolver', () => ({ + resolveSessionFilePath: mocks.resolve +})) vi.mock('node:fs/promises', async (importOriginal) => ({ ...(await importOriginal()), stat: mocks.stat, @@ -22,7 +21,10 @@ import { clearNativeChatTranscriptCache, readNativeChatTranscriptCached } from './transcript-read-cache' -import { WSL_TRANSCRIPT_FS_EXACT_TIMEOUT_MS } from './wsl-transcript-fs-gate' +import { + WSL_TRANSCRIPT_FS_EXACT_TIMEOUT_MS, + WSL_TRANSCRIPT_FS_SLOW_MESSAGE +} from './wsl-transcript-fs-gate' const BODY = Buffer.from( `${JSON.stringify({ @@ -33,26 +35,18 @@ const BODY = Buffer.from( })}\n` ) -// The stalled body read keeps a permit until it settles, so every case releases -// it before the next gated call runs. let releaseStall: (() => void) | undefined -function stallingHandle() { - return { - read: vi.fn( - (buffer: Buffer) => - new Promise<{ bytesRead: number; buffer: Buffer }>((resolve) => { - releaseStall = () => resolve({ bytesRead: 0, buffer }) - }) - ), - close: vi.fn(async () => {}) - } +function stalls(value: T): Promise { + return new Promise((resolve) => { + releaseStall = () => resolve(value) + }) } -function servingHandle() { +function transcriptHandle(body = BODY) { return { read: vi.fn(async (buffer: Buffer, offset: number, length: number, position: number) => { - const slice = BODY.subarray(position, Math.min(position + length, BODY.length)) + const slice = body.subarray(position, Math.min(position + length, body.length)) slice.copy(buffer, offset) return { bytesRead: slice.length, buffer } }), @@ -79,32 +73,43 @@ afterEach(async () => { vi.useRealTimers() }) -describe('cached native chat transcript read with a refused transcript body', () => { - it('retries the body on the next call even though the mtime is unchanged', async () => { - mocks.open.mockResolvedValue(stallingHandle()) +describe('cached native chat transcript read after WSL gate refusals', () => { + it('retries a refused stat instead of caching it as a missing file', async () => { + mocks.stat.mockReturnValueOnce(stalls({ mtimeMs: 42, size: BODY.length })) const refused = readNativeChatTranscriptCached('claude', 'session-id') await vi.advanceTimersByTimeAsync(WSL_TRANSCRIPT_FS_EXACT_TIMEOUT_MS + 1) - expect(await refused).toEqual({ error: SLOW_MESSAGE }) - // Free the permit the unabortable read still holds, then recover. + expect(await refused).toEqual({ error: WSL_TRANSCRIPT_FS_SLOW_MESSAGE }) releaseStall?.() releaseStall = undefined await vi.advanceTimersByTimeAsync(0) - mocks.open.mockResolvedValue(servingHandle()) + mocks.open.mockResolvedValue(transcriptHandle()) + + await expect(readNativeChatTranscriptCached('claude', 'session-id')).resolves.toMatchObject({ + messages: [expect.objectContaining({ role: 'user' })] + }) + expect(mocks.stat).toHaveBeenCalledTimes(2) + }) + + it('retries the body on the next call even though the mtime is unchanged', async () => { + mocks.open.mockResolvedValue({ + read: vi.fn((buffer: Buffer) => stalls({ bytesRead: 0, buffer })), + close: vi.fn(async () => {}) + }) + const refused = readNativeChatTranscriptCached('claude', 'session-id') + await vi.advanceTimersByTimeAsync(WSL_TRANSCRIPT_FS_EXACT_TIMEOUT_MS + 1) + expect(await refused).toEqual({ error: WSL_TRANSCRIPT_FS_SLOW_MESSAGE }) + + releaseStall?.() + releaseStall = undefined + await vi.advanceTimersByTimeAsync(0) + mocks.open.mockResolvedValue(transcriptHandle()) const recovered = await readNativeChatTranscriptCached('claude', 'session-id') expect(mocks.open).toHaveBeenCalledTimes(2) - expect(recovered).toMatchObject({ messages: [expect.objectContaining({ role: 'user' })] }) - }) - - it('still caches a healthy parse, so recovery does not cost a read per call', async () => { - mocks.open.mockResolvedValue(servingHandle()) - - const first = await readNativeChatTranscriptCached('claude', 'session-id') - const second = await readNativeChatTranscriptCached('claude', 'session-id') - - expect(mocks.open).toHaveBeenCalledTimes(1) - expect(second).toBe(first) + expect(recovered).toMatchObject({ + messages: [expect.objectContaining({ role: 'user' })] + }) }) }) diff --git a/src/main/native-chat/transcript-read-cache-wsl-stall.test.ts b/src/main/native-chat/transcript-read-cache-wsl-stall.test.ts deleted file mode 100644 index b4696fe0dd1..00000000000 --- a/src/main/native-chat/transcript-read-cache-wsl-stall.test.ts +++ /dev/null @@ -1,57 +0,0 @@ -import { beforeEach, describe, expect, it, vi } from 'vitest' -import type * as NodeFsPromisesModule from 'node:fs/promises' - -const UNC_PATH = '\\\\wsl.localhost\\Ubuntu\\home\\ada\\.claude\\projects\\p\\session.jsonl' - -const mocks = vi.hoisted(() => ({ - resolve: vi.fn<() => Promise>(), - stat: vi.fn() -})) - -vi.mock('./session-file-resolver', () => ({ resolveSessionFilePath: mocks.resolve })) -vi.mock('node:fs/promises', async (importOriginal) => ({ - ...(await importOriginal()), - stat: mocks.stat -})) - -import { readNativeChatTranscriptCached } from './transcript-read-cache' -import { WSL_TRANSCRIPT_FS_EXACT_TIMEOUT_MS } from './wsl-transcript-fs-gate' - -const SLOW_MESSAGE = - 'WSL transcript files are temporarily unavailable because filesystem access is taking too long. Try again shortly or restart Orca if the issue continues.' - -beforeEach(() => { - mocks.resolve.mockReset() - mocks.stat.mockReset() - mocks.resolve.mockResolvedValue(UNC_PATH) -}) - -describe('cached native chat transcript read with a stalled post-resolution stat', () => { - it('reports a retryable error without notFound and leaves nothing cached', async () => { - vi.useFakeTimers() - try { - mocks.stat.mockReturnValueOnce(new Promise(() => {})) - const stalled = readNativeChatTranscriptCached('claude', 'session-id') - await vi.advanceTimersByTimeAsync(WSL_TRANSCRIPT_FS_EXACT_TIMEOUT_MS + 1) - - const result = await stalled - expect(result).toEqual({ error: SLOW_MESSAGE }) - expect(result).not.toHaveProperty('notFound') - } finally { - vi.useRealTimers() - } - }) - - it('re-attempts the stat on the next call instead of serving a poisoned entry', async () => { - // The stalled task above still holds its permit, so this fast-fails on the - // stuck route rather than waiting out a second deadline — but it must still - // reach `stat` again once a different distro answers. - const healthyPath = '\\\\wsl.localhost\\Debian\\home\\ada\\.claude\\projects\\p\\session.jsonl' - mocks.resolve.mockResolvedValue(healthyPath) - mocks.stat.mockResolvedValue({ mtimeMs: 5, size: 0 }) - - await readNativeChatTranscriptCached('claude', 'session-id') - - expect(mocks.stat).toHaveBeenCalledWith(healthyPath) - }) -}) diff --git a/src/main/native-chat/transcript-watch-engine.ts b/src/main/native-chat/transcript-watch-engine.ts index 5fa850e033d..3c51319eae4 100644 --- a/src/main/native-chat/transcript-watch-engine.ts +++ b/src/main/native-chat/transcript-watch-engine.ts @@ -1,6 +1,6 @@ import type { NativeChatMessage, NativeChatTurnLifecycle } from '../../shared/native-chat-types' -import { boundaryFingerprint } from './transcript-boundary-fingerprint' import { + boundaryFingerprint, readTranscriptFileVersion, transcriptFileVersionChanged, type TranscriptFileVersion 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 d844b17850e..8b402a3bb98 100644 --- a/src/main/native-chat/wsl-transcript-fs-access.test.ts +++ b/src/main/native-chat/wsl-transcript-fs-access.test.ts @@ -95,7 +95,9 @@ describe('transcript filesystem accessor off WSL UNC', () => { expect(mocks.open).toHaveBeenCalledWith(path, 'r') // Off UNC the raw stream is handed back verbatim, encoding included. expect(stream).toBe('raw-stream') - expect(mocks.createReadStream).toHaveBeenCalledWith(path, { encoding: 'utf-8' }) + expect(mocks.createReadStream).toHaveBeenCalledWith(path, { + encoding: 'utf-8' + }) }) }) @@ -105,12 +107,21 @@ describe('transcript filesystem accessor on WSL UNC', () => { ['wsl$', LEGACY_UNC_PATH] ])('routes %s paths through the gate', async (_label, path) => { mocks.stat.mockResolvedValue({ size: 7 }) - await expect(wslGatedStat(path, 'exact')).resolves.toEqual({ size: 7 }) - expect(mocks.runTask).toHaveBeenCalledWith( - expect.objectContaining({ operation: 'stat', path, priority: 'exact' }), - expect.any(Function) - ) - expect(mocks.createReadStream).not.toHaveBeenCalled() + mocks.lstat.mockResolvedValue({ size: 7 }) + mocks.readdir.mockResolvedValue([]) + mocks.readFile.mockResolvedValue('body') + + await wslGatedStat(path, 'exact') + await wslGatedLstat(path, 'scan') + await wslGatedReaddir(path, 'scan') + await wslGatedReadFile(path, 'utf-8', 'scan') + + expect(mocks.runTask.mock.calls.map(([options]) => options.operation)).toEqual([ + 'stat', + 'lstat', + 'readdir', + 'readfile' + ]) }) it('opts positional reads and opens out of coalescing', async () => { @@ -251,45 +262,6 @@ describe('transcript handle close off WSL UNC', () => { }) describe('per-chunk admission', () => { - it('streams a long healthy-but-slow file past the whole-file deadline', async () => { - vi.useFakeTimers() - try { - const chunkDelayMs = WSL_TRANSCRIPT_FS_EXACT_TIMEOUT_MS / 4 - const totalChunks = 6 - let served = 0 - const handle = fakeHandle() - handle.read.mockImplementation( - (buffer: Buffer) => - new Promise((resolve) => { - setTimeout(() => { - if (served++ >= totalChunks) { - resolve({ bytesRead: 0, buffer }) - return - } - buffer.write('x') - resolve({ bytesRead: 1, buffer }) - }, chunkDelayMs) - }) - ) - mocks.open.mockResolvedValue(handle) - - const stream = openTranscriptReadStream(UNC_PATH, {}, 'exact') - const collected: Promise = (async () => { - const out: Buffer[] = [] - for await (const chunk of stream) { - out.push(chunk as Buffer) - } - return out - })() - - // Elapsed time far exceeds one exact deadline; each chunk stays under it. - await vi.advanceTimersByTimeAsync(chunkDelayMs * (totalChunks + 2)) - await expect(collected).resolves.toHaveLength(totalChunks) - } finally { - vi.useRealTimers() - } - }) - it('carries a codepoint straddling the 1 MiB chunk boundary across chunks', async () => { const chunkBytes = 1024 * 1024 const emoji = Buffer.from('😀', 'utf8') diff --git a/src/main/native-chat/wsl-transcript-fs-access.ts b/src/main/native-chat/wsl-transcript-fs-access.ts index 128876aeb5f..d9ac0170c60 100644 --- a/src/main/native-chat/wsl-transcript-fs-access.ts +++ b/src/main/native-chat/wsl-transcript-fs-access.ts @@ -5,31 +5,28 @@ import { StringDecoder } from 'node:string_decoder' import { isWslUncPath } from '../../shared/wsl-paths' import { runWslTranscriptFsTask, type WslTranscriptFsTaskPriority } from './wsl-transcript-fs-gate' -/** - * Transcript-layer filesystem primitives that route WSL UNC paths through the - * admission gate and everything else straight to `node:fs`. The `isWslUncPath` - * guard is the single switch: macOS, Linux, Windows drive paths, SSH remotes and - * folder workspaces execute the exact call they execute today. - * - * Invariant future call sites must preserve: every gated unit here is a single - * syscall, and `openTranscriptReadStream` composes them OUTSIDE the gate. A - * gated call awaited from inside another gated task's operation would deadlock - * on the single scan slot or the route+priority lane. - */ - -// Why: one deadline per chunk instead of one for the whole file, so a large -// healthy-but-slow transcript is not false-failed by a whole-file timeout. const WSL_TRANSCRIPT_READ_CHUNK_BYTES = 1024 * 1024 +type Operation = Parameters[0]['operation'] + +function runPathOperation( + operation: Operation, + path: string, + priority: WslTranscriptFsTaskPriority, + signal: AbortSignal | undefined, + task: () => Promise, + options?: { dedupe?: boolean; onAbandonedResult?: (value: T) => void } +): Promise { + return isWslUncPath(path) + ? runWslTranscriptFsTask({ operation, path, priority, signal, ...options }, task) + : task() +} export function wslGatedStat( path: string, priority: WslTranscriptFsTaskPriority, signal?: AbortSignal ): Promise { - if (!isWslUncPath(path)) { - return stat(path) - } - return runWslTranscriptFsTask({ operation: 'stat', path, priority, signal }, () => stat(path)) + return runPathOperation('stat', path, priority, signal, () => stat(path)) } export function wslGatedLstat( @@ -37,10 +34,7 @@ export function wslGatedLstat( priority: WslTranscriptFsTaskPriority, signal?: AbortSignal ): Promise { - if (!isWslUncPath(path)) { - return lstat(path) - } - return runWslTranscriptFsTask({ operation: 'lstat', path, priority, signal }, () => lstat(path)) + return runPathOperation('lstat', path, priority, signal, () => lstat(path)) } export function wslGatedReaddir( @@ -48,10 +42,7 @@ export function wslGatedReaddir( priority: WslTranscriptFsTaskPriority, signal?: AbortSignal ): Promise { - if (!isWslUncPath(path)) { - return readdir(path, { withFileTypes: true }) - } - return runWslTranscriptFsTask({ operation: 'readdir', path, priority, signal }, () => + return runPathOperation('readdir', path, priority, signal, () => readdir(path, { withFileTypes: true }) ) } @@ -62,44 +53,20 @@ export function wslGatedReadFile( priority: WslTranscriptFsTaskPriority, signal?: AbortSignal ): Promise { - if (!isWslUncPath(path)) { - return readFile(path, encoding) - } - return runWslTranscriptFsTask({ operation: 'readfile', path, priority, signal }, () => - readFile(path, encoding) - ) + return runPathOperation('readfile', path, priority, signal, () => readFile(path, encoding)) } -// dedupe:false — two joiners would share one FileHandle and both close it. export function wslGatedOpen( path: string, priority: WslTranscriptFsTaskPriority, signal?: AbortSignal ): Promise { - if (!isWslUncPath(path)) { - return open(path, 'r') - } - return runWslTranscriptFsTask( - { - operation: 'open', - path, - priority, - signal, - 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) - }, - () => open(path, 'r') - ) + return runPathOperation('open', path, priority, signal, () => open(path, 'r'), { + dedupe: false, + onAbandonedResult: (handle) => void closeTranscriptHandle(handle, path) + }) } -/** - * `path` is only the gate's route key and UNC guard — the handle carries no - * path and is never re-opened. dedupe:false because `read` fills the CALLER's - * buffer: a joiner would receive the first caller's buffer while its own - * (often `Buffer.allocUnsafe`) stays uninitialized. - */ export function wslGatedRead( handle: FileHandle, path: string, @@ -110,21 +77,16 @@ export function wslGatedRead( priority: WslTranscriptFsTaskPriority, signal?: AbortSignal ): Promise<{ bytesRead: number; buffer: Buffer }> { - if (!isWslUncPath(path)) { - return handle.read(buffer, offset, length, position) - } - return runWslTranscriptFsTask({ operation: 'read', path, priority, signal, dedupe: false }, () => - handle.read(buffer, offset, length, position) + return runPathOperation( + 'read', + path, + priority, + signal, + () => handle.read(buffer, offset, length, position), + { dedupe: false } ) } -/** - * 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() @@ -133,11 +95,6 @@ export function closeTranscriptHandle(handle: FileHandle, path: string): Promise return Promise.resolve() } -/** - * Open, read one slice, close — for callers that want bytes at an offset and - * never touch the handle. Each of the two syscalls is admitted separately, so a - * stalled mount fails at whichever one it blocks on. - */ export async function readTranscriptSlice( path: string, position: number, @@ -148,8 +105,17 @@ export async function readTranscriptSlice( const handle = await wslGatedOpen(path, priority, signal) try { const buffer = Buffer.allocUnsafe(length) - const read = await wslGatedRead(handle, path, buffer, 0, length, position, priority, signal) - return buffer.subarray(0, read.bytesRead) + const { bytesRead } = await wslGatedRead( + handle, + path, + buffer, + 0, + length, + position, + priority, + signal + ) + return buffer.subarray(0, bytesRead) } finally { await closeTranscriptHandle(handle, path) } @@ -157,11 +123,7 @@ export async function readTranscriptSlice( export type TranscriptReadStreamOptions = { start?: number - /** Inclusive, matching `createReadStream`. */ end?: number - /** Set it to get decoded string chunks on both branches; leave it unset and - * the UNC branch yields `Buffer`s the consumer must decode incrementally - * itself, since a chunk boundary can split a multibyte codepoint. */ encoding?: BufferEncoding } @@ -172,10 +134,6 @@ async function* gatedChunks( signal?: AbortSignal ): AsyncGenerator { const handle = await wslGatedOpen(path, priority, signal) - // Why: chunk boundaries fall mid-codepoint, so decoding each slice - // independently would emit U+FFFD on both sides of any straddling character. - // The decoder carries the partial sequence across chunks, as the - // `createReadStream` branch's own decoder does. const decoder = options.encoding ? new StringDecoder(options.encoding) : null try { let position = options.start ?? 0 @@ -212,24 +170,15 @@ async function* gatedChunks( yield decoded } } - // Trailing bytes of an incomplete sequence at EOF, replacement-char'd once. const trailing = decoder?.end() if (trailing) { yield trailing } } finally { - // Runs on `.destroy()` too (Readable.from calls the generator's return()), - // so an aborted head-read cannot leak the gated handle. await closeTranscriptHandle(handle, path) } } -/** - * A read stream whose UNC branch admits and deadlines each 1 MiB chunk - * separately. A gate refusal mid-stream surfaces as an `'error'` event carrying - * the `WslTranscriptFsError`, which every existing consumer funnels into its - * `catch`. - */ export function openTranscriptReadStream( path: string, options: TranscriptReadStreamOptions,