consolidate scan issues and gate session scanner I/O (STA-4049)

Both local and remote session scans hit stalled WSL distros identically:
one failed probe per discovered path. Unifying issue recording and gate
refusal handling prevents duplication and ensures consistent behavior.

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