diff --git a/src/main/ai-vault/ai-vault-scan-progress-flow.test.ts b/src/main/ai-vault/ai-vault-scan-progress-flow.test.ts index 9002fae03bd..52d13dc1cfb 100644 --- a/src/main/ai-vault/ai-vault-scan-progress-flow.test.ts +++ b/src/main/ai-vault/ai-vault-scan-progress-flow.test.ts @@ -1,6 +1,6 @@ import { link, mkdir, mkdtemp, rm, utimes } from 'node:fs/promises' import { tmpdir } from 'node:os' -import { join } from 'node:path' +import { basename, join } from 'node:path' import { afterEach, describe, expect, it, vi } from 'vitest' import type { AiVaultScanProgress, AiVaultScanWork } from '../../shared/ai-vault-scan-progress' import { @@ -10,8 +10,11 @@ import { resetLocalAiVaultScanProgressForTests, startLocalAiVaultScanProgress } from './ai-vault-local-scan-progress' +import * as parseCache from './session-scanner-parse-cache' +import * as transcriptFs from '../native-chat/wsl-transcript-fs-access' import { AiVaultScanTally } from './ai-vault-scan-tally' import { aiVaultScanWork, trackAiVaultScanProgress } from './ai-vault-scan-progress' +import { writeOpenCodeSqliteDatabase } from './session-scanner-opencode-sqlite-fixture' import { scanAiVaultSessions } from './session-scanner' import { resetSessionParseCacheForTests } from './session-scanner-parse-cache' import { AiVaultScannerServiceClient } from './session-scanner-service-client' @@ -22,10 +25,25 @@ import { } from './session-scanner-service-test-child' import { isolatedScanRoots, writeJsonlFile } from './session-scanner-test-fixtures' +// Source fixtures exercise the real SQLite readers without a packaged worker bundle. +vi.mock('./session-scanner-opencode-sqlite-worker-spawn', async () => { + const { listOpenCodeSqliteSessions } = await import('./session-scanner-opencode-sqlite-list') + const { listOpenCode2SqliteSessions } = await import('./session-scanner-opencode2-sqlite-list') + const { parseOpenCodeSqliteSession } = await import('./session-scanner-opencode-sqlite') + return { + listOpenCode2SqliteSessionsViaWorker: listOpenCode2SqliteSessions, + listOpenCodeSqliteSessionsViaWorker: listOpenCodeSqliteSessions, + listZcodeSqliteSessionsViaWorker: (args: Parameters[0]) => + listOpenCodeSqliteSessions({ ...args, agent: 'zcode' }), + parseOpenCodeSqliteSessionViaWorker: parseOpenCodeSqliteSession + } +}) + let tempRoot: string | null = null afterEach(async () => { vi.useRealTimers() + vi.restoreAllMocks() resetSessionParseCacheForTests() resetLocalAiVaultScanProgressForTests() if (tempRoot) { @@ -102,8 +120,16 @@ describe('Session History scan progress, scanner child to window', () => { expect(reports.some((work) => work.stage === 'finding')).toBe(true) expect(final).toEqual({ stage: 'reading', - agents: [{ agent: 'claude', files: 3, databases: 0, walking: false }], - read: { done: 4, total: 4 } + agents: [ + { + agent: 'claude', + files: 3, + databases: 0, + walking: false, + read: { done: 4, total: 4, skipped: 0 } + } + ], + read: { done: 4, total: 4, skipped: 0 } }) const { child, scan, published } = startScan() @@ -135,8 +161,11 @@ describe('Session History scan progress, scanner child to window', () => { it('reaches its total when the first pass stops before reading every file it planned', async () => { // The newest file fills a limit of 1, so the pass never reads the older ones it announced. - const { final } = await scanInChild(1) - expect(final.read).toEqual({ done: 1, total: 1 }) + const { reports, final } = await scanInChild(1) + expect(final.read).toEqual({ done: 1, total: 2, skipped: 1 }) + expect( + reports.filter((work) => work.stage === 'reading').every((work) => work.read.total === 2) + ).toBe(true) }) it('keeps a replacement entry when the scan it replaced fails late', () => { @@ -202,3 +231,204 @@ describe('Session History scan progress, scanner child to window', () => { expect(tally.snapshot().agents).toMatchObject([{ agent: 'codex', files: 3 }]) }) }) + +it.each(['cutoff', 'bucket', 'filtered'] as const)( + 'freezes %s work without changing cold/warm execution', + async (flow) => { + tempRoot = await mkdtemp(join(tmpdir(), 'orca-scan-frozen-')) + const roots = isolatedScanRoots(tempRoot) + for (let index = 0; index < (flow === 'filtered' ? 27 : 6); index++) { + const timestamp = new Date(Date.UTC(2026, 9, 19) - index * 60_000).toISOString() + const cwd = + index >= 3 && (flow === 'bucket' || (index - 3) % 8 === 0) ? '/repo' : '/elsewhere' + const path = + flow === 'filtered' + ? join(roots.codexSessionsDir, `rollout-${index}.jsonl`) + : join( + roots.claudeProjectsDir, + cwd === '/repo' ? '-repo' : '-elsewhere', + `${index}.jsonl` + ) + await writeJsonlFile( + path, + flow === 'filtered' + ? [ + { timestamp, type: 'session_meta', payload: { id: `fixture-${index}`, cwd } }, + { + timestamp, + type: 'event_msg', + payload: { type: 'user_message', message: `Generated ${index}` } + } + ] + : [ + { + type: 'user', + sessionId: `fixture-${index}`, + timestamp, + cwd, + message: { role: 'user', content: `Generated ${index}` } + } + ] + ) + await utimes(path, Date.parse(timestamp) / 1000, Date.parse(timestamp) / 1000) + } + const listed = vi.spyOn(parseCache, 'parseListedSessionFile') + const walked = vi.spyOn(parseCache, 'parseAgentSessionFileCached') + const stats = vi.spyOn(parseCache, 'createSessionParseStats') + const opened = vi.spyOn(transcriptFs, 'openTranscriptReadStream') + const ids = + flow === 'cutoff' + ? [0, 1, 2] + : flow === 'filtered' + ? [0, 1, 2, 3, 11, 19] + : [0, 1, 2, 3, 4, 5] + for (const warm of [false, true]) { + const reports: AiVaultScanWork[] = [] + const signal = new AbortController().signal + let now = 0 + const stop = trackAiVaultScanProgress( + signal, + (work) => reports.push(work), + () => (now += 1000) + ) + try { + const result = await scanAiVaultSessions({ + ...roots, + platform: 'darwin', + limit: 3, + limitPerAgent: 50, + signal, + ...(flow === 'cutoff' ? {} : { scopePaths: ['/repo'] }) + }) + expect(result.issues).toEqual([]) + expect(result.sessions.map((row) => row.sessionId)).toEqual( + ids.map((id) => `fixture-${id}`) + ) + const attempts = [...listed.mock.calls, ...walked.mock.calls].map( + ([candidate]) => candidate.file.path + ) + expect( + attempts.map((path) => + Number(basename(path).replace('rollout-', '').replace('.jsonl', '')) + ) + ).toEqual(ids) + // Captured from the same generated corpus before the change; representatives move earlier. + expect(opened.mock.calls).toHaveLength(warm ? 0 : flow === 'bucket' ? 7 : ids.length) + expect(stats.mock.results.at(-1)?.value).toEqual({ + reused: warm ? ids.length : 0, + incremental: 0, + fullParses: warm ? 0 : ids.length, + earlyStopped: 0, + bytesRead: warm ? 0 : flow === 'filtered' ? 1363 : flow === 'bucket' ? 873 : 444 + }) + const reading = reports.filter((work) => work.stage === 'reading') + const total = flow === 'filtered' ? 27 : 6 + let settled = 0 + for (const work of reading) { + expect(work.read.total).toBe(total) + expect(work.read.done + (work.read.skipped ?? 0)).toBeGreaterThanOrEqual(settled) + settled = work.read.done + (work.read.skipped ?? 0) + expect(work.agents.reduce((sum, agent) => sum + (agent.read?.total ?? 0), 0)).toBe(total) + expect(work.agents.reduce((sum, agent) => sum + (agent.read?.done ?? 0), 0)).toBe( + work.read.done + ) + expect(work.agents.reduce((sum, agent) => sum + (agent.read?.skipped ?? 0), 0)).toBe( + work.read.skipped + ) + } + expect(aiVaultScanWork(signal)?.read).toEqual({ + done: ids.length, + total, + skipped: total - ids.length + }) + } finally { + stop() + } + listed.mockClear() + walked.mockClear() + opened.mockClear() + } + } +) + +it.each([false, true])( + 'counts repeated global visits and database sessions once (scoped=%s)', + async (scoped) => { + tempRoot = await mkdtemp(join(tmpdir(), 'orca-scan-repeat-')) + const roots = isolatedScanRoots(tempRoot) + await writeJsonlFile(join(roots.droidSessionsDir, 'same.jsonl'), [ + { type: 'session_start', id: 'same', cwd: '/repo', timestamp: '2026-10-19T00:00:00Z' }, + { type: 'message', message: { role: 'user', content: 'Generated' } } + ]) + const database = join(tempRoot, 'opencode.db') + writeOpenCodeSqliteDatabase(database, [ + { id: 'db-session', directory: '/repo', turns: [{ role: 'user', parts: ['Generated'] }] } + ]) + const listed = vi.spyOn(parseCache, 'parseListedSessionFile') + const signal = new AbortController().signal + const reports: AiVaultScanWork[] = [] + let pendingAfterParses = false + let now = 0 + const stop = trackAiVaultScanProgress( + signal, + (work) => { + reports.push(work) + if (listed.mock.calls.length === 3 && work.stage === 'reading' && work.read.done === 0) { + pendingAfterParses = true + } + }, + () => (now += 1000) + ) + try { + const result = await scanAiVaultSessions({ + ...roots, + droidProjectsDir: roots.droidSessionsDir, + opencodeDbPaths: [database], + scopePaths: scoped ? ['/repo'] : [], + signal, + limit: 20, + platform: 'darwin' + }) + expect(result.sessions.map((row) => row.sessionId).sort()).toEqual( + scoped ? ['db-session', 'same'] : ['db-session', 'same', 'same'] + ) + expect(listed.mock.calls.filter(([candidate]) => candidate.agent === 'droid')).toHaveLength(2) + expect(aiVaultScanWork(signal)?.read).toEqual({ done: 2, total: 2, skipped: 0 }) + if (scoped) { + expect(pendingAfterParses).toBe(true) + } + expect( + reports.filter((work) => work.stage === 'reading').every((work) => work.read.total === 2) + ).toBe(true) + expect( + aiVaultScanWork(signal)?.agents.find((agent) => agent.agent === 'opencode')?.read + ).toEqual({ done: 1, total: 1, skipped: 0 }) + } finally { + stop() + } + } +) + +it('does not settle a batch interrupted during its real parser invocation', async () => { + tempRoot = await mkdtemp(join(tmpdir(), 'orca-scan-abort-')) + const roots = isolatedScanRoots(tempRoot) + await writeJsonlFile(join(roots.claudeProjectsDir, 'project', 'a.jsonl'), [ + { type: 'user', sessionId: 'a', message: { role: 'user', content: 'Generated' } } + ]) + const controller = new AbortController() + const original = parseCache.parseListedSessionFile + vi.spyOn(parseCache, 'parseListedSessionFile').mockImplementation(async (...args) => { + const result = await original(...args) + controller.abort() + return result + }) + const stop = trackAiVaultScanProgress(controller.signal, () => undefined) + try { + await expect( + scanAiVaultSessions({ ...roots, signal: controller.signal }) + ).rejects.toMatchObject({ name: 'AbortError' }) + expect(aiVaultScanWork(controller.signal)?.read).toEqual({ done: 0, total: 1, skipped: 0 }) + } finally { + stop() + } +}) diff --git a/src/main/ai-vault/ai-vault-scan-progress.test.ts b/src/main/ai-vault/ai-vault-scan-progress.test.ts index c89bd929cf4..76d003517fc 100644 --- a/src/main/ai-vault/ai-vault-scan-progress.test.ts +++ b/src/main/ai-vault/ai-vault-scan-progress.test.ts @@ -6,6 +6,8 @@ import type { AiVaultScanWork } from '../../shared/ai-vault-scan-progress' import { AI_VAULT_SCAN_PROGRESS_INTERVAL_MS, noteAiVaultScanFileFound, + aiVaultScanWork, + openAiVaultScanSource, noteAiVaultScanWorkChanged, trackAiVaultScanProgress } from './ai-vault-scan-progress' @@ -103,3 +105,19 @@ describe('scan progress reporting', () => { expect(reports.at(-1)).toMatchObject({ stage: 'reading', read: { done: 3, total: 3 } }) }) }) + +it('keeps source ownership and final facts independent of immediate and trailing sink faults', () => { + vi.useFakeTimers() + const signal = new AbortController().signal + const sink = vi.fn(() => { + throw new Error('Synthetic reporting sink fault') + }) + const stop = trackAiVaultScanProgress(signal, sink, () => Date.now()) + const close = openAiVaultScanSource(signal, 'claude') + expect(aiVaultScanWork(signal)?.agents[0]?.walking).toBe(true) + close() + expect(() => vi.advanceTimersByTime(AI_VAULT_SCAN_PROGRESS_INTERVAL_MS)).not.toThrow() + expect(sink).toHaveBeenCalledTimes(2) + expect(aiVaultScanWork(signal)?.agents).toEqual([]) + stop() +}) diff --git a/src/main/ai-vault/ai-vault-scan-progress.ts b/src/main/ai-vault/ai-vault-scan-progress.ts index f261ce06bc2..6182945825d 100644 --- a/src/main/ai-vault/ai-vault-scan-progress.ts +++ b/src/main/ai-vault/ai-vault-scan-progress.ts @@ -20,7 +20,12 @@ export function trackAiVaultScanProgress( const tally = new AiVaultScanTally() const publish = (): void => { lastReportedAt = now() - report(tally.snapshot()) + const snapshot = tally.snapshot() + try { + report(snapshot) + } catch { + // Reporting cannot interrupt scanning or strand source ownership. + } } const workChanged = (): void => { if (trailing) { @@ -90,24 +95,15 @@ export function noteAiVaultScanDatabases( reporter?.workChanged() } -/** Discovery is over: the scan now reads `total` more transcripts. */ -export function noteAiVaultScanReading(signal: AbortSignal | undefined, total: number): void { +/** The scan has registered all candidate opportunities. */ +export function noteAiVaultScanReading(signal: AbortSignal | undefined): void { const reporter = signal ? reporters.get(signal) : undefined - reporter?.tally.startReading(total) + reporter?.tally.startReading() reporter?.workChanged() } -/** A pass stopped early: `count` transcripts it announced will not be read. */ -export function noteAiVaultScanUnread(signal: AbortSignal | undefined, count: number): void { - const reporter = signal ? reporters.get(signal) : undefined - reporter?.tally.dropUnread(count) - reporter?.workChanged() -} - -export function noteAiVaultScanFilesRead(signal: AbortSignal | undefined, count: number): void { - const reporter = signal ? reporters.get(signal) : undefined - reporter?.tally.addRead(count) - reporter?.workChanged() +export function aiVaultScanWorkload(signal: AbortSignal | undefined) { + return tallyFor(signal)?.workload } /** Sends the tally after a batch of files was counted. */ diff --git a/src/main/ai-vault/ai-vault-scan-tally.ts b/src/main/ai-vault/ai-vault-scan-tally.ts index b4b2ce971fd..d93c42d554a 100644 --- a/src/main/ai-vault/ai-vault-scan-tally.ts +++ b/src/main/ai-vault/ai-vault-scan-tally.ts @@ -1,5 +1,6 @@ import { AI_VAULT_AGENTS, type AiVaultAgent } from '../../shared/ai-vault-types' import type { AiVaultScanWork } from '../../shared/ai-vault-scan-progress' +import { AiVaultScanWorkload } from './ai-vault-scan-workload' import { sessionFileExecutionNamespace } from './codex-session-root-dedup' type AgentTally = { @@ -12,8 +13,7 @@ type AgentTally = { export class AiVaultScanTally { private readonly agents = new Map() private stage: AiVaultScanWork['stage'] = 'finding' - private readDone = 0 - private readTotal = 0 + readonly workload = new AiVaultScanWorkload() openSource(agent: AiVaultAgent): () => void { const tally = this.agent(agent) @@ -45,28 +45,30 @@ export class AiVaultScanTally { this.agent(agent).databases += count } - startReading(total: number): void { + startReading(): void { this.stage = 'reading' - this.readTotal += total - } - - addRead(count: number): void { - this.readDone += count - } - - dropUnread(count: number): void { - this.readTotal -= count } snapshot(): AiVaultScanWork { const agents: AiVaultScanWork['agents'] = [] + for (const agent of AI_VAULT_AGENTS) { + if (this.workload.has(agent)) { + this.agent(agent) + } + } for (const [agent, tally] of this.agents) { - if (tally.files.size > 0 || tally.databases > 0 || tally.openSources > 0) { + if ( + tally.files.size > 0 || + tally.databases > 0 || + tally.openSources > 0 || + this.workload.has(agent) + ) { agents.push({ agent, files: tally.files.size, databases: tally.databases, - walking: tally.openSources > 0 + walking: tally.openSources > 0, + ...(this.stage === 'reading' ? { read: this.workload.snapshot(agent) } : {}) }) } } @@ -75,10 +77,7 @@ export class AiVaultScanTally { return { stage: this.stage, agents, - read: { - done: Math.min(this.readDone, this.readTotal), - total: this.readTotal - } + read: this.stage === 'reading' ? this.workload.total() : { done: 0, total: 0 } } } diff --git a/src/main/ai-vault/ai-vault-scan-workload.ts b/src/main/ai-vault/ai-vault-scan-workload.ts new file mode 100644 index 00000000000..02479e09170 --- /dev/null +++ b/src/main/ai-vault/ai-vault-scan-workload.ts @@ -0,0 +1,105 @@ +import type { AiVaultAgent } from '../../shared/ai-vault-types' +import type { AiVaultScanReadingProgress } from '../../shared/ai-vault-scan-progress' +import type { SessionFileCandidate } from './session-scanner-types' + +type Ownership = { globalRemaining: number; owners: number; parsed: boolean } +export type ScanReadPass = 'global' | 'walk' | 'bucket' +const OWNER = { walk: 1, bucket: 2 } as const + +/** Logical membership is fixed before execution; repeated execution visits stay independent. */ +export class AiVaultScanWorkload { + private readonly keys = new Map>() + private readonly agents = new Map() + + register(candidate: SessionFileCandidate, pass: ScanReadPass): void { + this.registerPath(candidate.agent, candidate.file.path, pass) + } + + registerPath(agent: AiVaultAgent, path: string, pass: ScanReadPass): void { + let keys = this.keys.get(agent) + let key = keys?.get(path) + if (!key) { + this.counts(agent).total++ + // Walk-only paths are unique and cannot occur in bucket agents. + if (pass === 'walk') { + return + } + keys ??= new Map() + key = { globalRemaining: 0, owners: 0, parsed: false } + keys.set(path, key) + this.keys.set(agent, keys) + } + if (pass === 'global') { + key.globalRemaining++ + } else { + key.owners |= OWNER[pass] + } + } + + settle(candidate: SessionFileCandidate, pass: ScanReadPass, parsed: boolean): void { + this.settlePath(candidate.agent, candidate.file.path, pass, parsed) + } + + settlePath(agent: AiVaultAgent, path: string, pass: ScanReadPass, parsed: boolean): void { + const keys = this.keys.get(agent) + const key = keys?.get(path) + if (!key) { + this.complete(agent, parsed) + return + } + key.parsed ||= parsed + if (pass === 'global') { + key.globalRemaining-- + } else { + key.owners &= ~OWNER[pass] + } + if (key.globalRemaining === 0 && key.owners === 0) { + this.complete(agent, key.parsed) + keys?.delete(path) + } + } + + *bucketPaths(agent: AiVaultAgent): Iterable { + for (const [path, key] of this.keys.get(agent) ?? []) { + if ((key.owners & OWNER.bucket) !== 0) { + yield path + } + } + } + + snapshot(agent: AiVaultAgent): AiVaultScanReadingProgress { + return { ...this.counts(agent) } + } + + has(agent: AiVaultAgent): boolean { + return this.agents.has(agent) + } + + total(): AiVaultScanReadingProgress { + const total = { done: 0, total: 0, skipped: 0 } + for (const count of this.agents.values()) { + total.done += count.done + total.total += count.total + total.skipped += count.skipped ?? 0 + } + return total + } + + private complete(agent: AiVaultAgent, parsed: boolean): void { + const count = this.counts(agent) + if (parsed) { + count.done++ + } else { + count.skipped = (count.skipped ?? 0) + 1 + } + } + + private counts(agent: AiVaultAgent): AiVaultScanReadingProgress { + let counts = this.agents.get(agent) + if (!counts) { + counts = { done: 0, total: 0, skipped: 0 } + this.agents.set(agent, counts) + } + return counts + } +} diff --git a/src/main/ai-vault/session-scan-issues.ts b/src/main/ai-vault/session-scan-issues.ts index b3e21f9c5a8..456e40dd5bb 100644 --- a/src/main/ai-vault/session-scan-issues.ts +++ b/src/main/ai-vault/session-scan-issues.ts @@ -4,7 +4,7 @@ import type { AiVaultScanIssue } from '../../shared/ai-vault-types' // 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 +export 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 { diff --git a/src/main/ai-vault/session-scanner-candidate-parse.ts b/src/main/ai-vault/session-scanner-candidate-parse.ts index 8d070889454..563a3654e6f 100644 --- a/src/main/ai-vault/session-scanner-candidate-parse.ts +++ b/src/main/ai-vault/session-scanner-candidate-parse.ts @@ -21,11 +21,7 @@ import { canStopParsingSessions } from './session-scan-cutoff' import type { SessionFileCandidate, SessionParseResult } from './session-scanner-types' import { errorMessage } from './session-scanner-values' import { throwIfAiVaultScanCancelled } from './ai-vault-scan-cancellation' -import { - noteAiVaultScanFilesRead, - noteAiVaultScanReading, - noteAiVaultScanUnread -} from './ai-vault-scan-progress' +import { noteAiVaultScanWorkChanged } from './ai-vault-scan-progress' const SESSION_PARSE_CONCURRENCY = 8 @@ -42,16 +38,11 @@ export async function parseSessionCandidates(args: { selectCandidate?: (candidate: SessionFileCandidate) => Promise admitSession?: (session: AiVaultSession) => boolean listed?: boolean + settled?: (candidate: SessionFileCandidate, parsed: boolean) => void }): Promise { const sessions = new ScannedSessionCollection() let nextYieldIndex = 256 let nextCutoffIndex = 0 - // Why: the loop alone knows what it will read, so it owns its share of the read total; - // a filtered walk announces each selected batch, an unfiltered pass its whole list. - if (!args.selectCandidate) { - noteAiVaultScanReading(args.signal, args.candidates.length) - } - let index = 0 while (index < args.candidates.length) { if (index >= nextYieldIndex) { @@ -89,9 +80,6 @@ export async function parseSessionCandidates(args: { : batch.map(() => true) throwIfAiVaultScanCancelled(args.signal) const selected = batch.filter((_, batchIndex) => verdicts[batchIndex]) - if (args.selectCandidate) { - noteAiVaultScanReading(args.signal, selected.length) - } const results = await Promise.all( selected.map((candidate) => parseSessionCandidate(candidate, { @@ -107,7 +95,8 @@ export async function parseSessionCandidates(args: { // An abort can land while a batch settles: a read it cut short is not that // file's issue, and a partial parse must never be cached or returned. throwIfAiVaultScanCancelled(args.signal) - noteAiVaultScanFilesRead(args.signal, selected.length) + batch.forEach((candidate, index) => args.settled?.(candidate, verdicts[index])) + noteAiVaultScanWorkChanged(args.signal) results.forEach((result, resultIndex) => { if (result.issue) { @@ -127,9 +116,14 @@ export async function parseSessionCandidates(args: { }) } - if (!args.selectCandidate) { - noteAiVaultScanUnread(args.signal, args.candidates.length - index) + for (; index < args.candidates.length; index++) { + if (index % 256 === 0) { + await yieldToEventLoop() + } + throwIfAiVaultScanCancelled(args.signal) + args.settled?.(args.candidates[index], false) } + noteAiVaultScanWorkChanged(args.signal) return [...sessions.values()] } diff --git a/src/main/ai-vault/session-scanner-scope-cwd-facts.test.ts b/src/main/ai-vault/session-scanner-scope-cwd-facts.test.ts index fd6c22d0d72..6cd0d939abb 100644 --- a/src/main/ai-vault/session-scanner-scope-cwd-facts.test.ts +++ b/src/main/ai-vault/session-scanner-scope-cwd-facts.test.ts @@ -63,10 +63,9 @@ it.each([false, true])('keeps cold authority with read failure=%s', async (failR ) if (index === 0) { const context = jsonLines([{ type: 'turn_context', payload: { cwd: '/repo' } }]) + const { atime, mtime } = await stat(path) await appendFile(path, `\n${context}\n`) - // Why: the append stamps wall-clock mtime, which overtakes the busy files once real time passes their date. - const day1 = new Date(Date.UTC(2026, 9, 1, 10)) - await utimes(path, day1, day1) + await utimes(path, atime, mtime) } const { mtimeMs, size: sizeBytes, mtime } = await stat(path) const parsed = await parseAgentSessionFileCached( diff --git a/src/main/ai-vault/session-scanner-scope-discovery.ts b/src/main/ai-vault/session-scanner-scope-discovery.ts index f74051ca3d2..1b1beead140 100644 --- a/src/main/ai-vault/session-scanner-scope-discovery.ts +++ b/src/main/ai-vault/session-scanner-scope-discovery.ts @@ -6,11 +6,12 @@ import { wslGatedStat } from '../native-chat/wsl-transcript-fs-access' import { WslTranscriptFsError } from '../native-chat/wsl-transcript-fs-gate' -import { createAiVaultScopeCwdMatcher } from '../../shared/ai-vault-scope-cwd-match' +import type { createAiVaultScopeCwdMatcher } from '../../shared/ai-vault-scope-cwd-match' import type { CwdBucketLayout } from './session-cwd-bucket-layouts' import type { AiVaultAgent, AiVaultScanIssue } from '../../shared/ai-vault-types' import { parseWslUncPath } from '../../shared/wsl-paths' -import { recordSessionScanIssue } from './session-scan-issues' +import { throwIfAiVaultScanCancelled } from './ai-vault-scan-cancellation' +import { yieldToEventLoop } from '../../shared/event-loop-yield' import type { FileWithMtime } from './session-scanner-types' import { errorMessage, extractString, parseJsonObject } from './session-scanner-values' @@ -23,6 +24,7 @@ const TRANSCRIPT_EXTENSIONS = new Set(['.jsonl']) // Bucket encodings are lossy; retain Claude’s cache and verify other transcripts individually. const PROJECT_DIR_CWD_CACHE_MAX = 2048 const projectDirCwdCache = new Map() +export type IssueSink = (issue: AiVaultScanIssue) => void export function resetProjectDirCwdCacheForTests(): void { projectDirCwdCache.clear() @@ -30,21 +32,22 @@ export function resetProjectDirCwdCacheForTests(): void { // A gate refusal is a stalled WSL root, not an empty one: keep the existing // degrade-to-empty containment but make the gap visible in the scan issues. -function recordRefusal( - issues: AiVaultScanIssue[], +export function recordRefusal( + issues: IssueSink, path: string, error: unknown, agent: AiVaultAgent ): void { if (error instanceof WslTranscriptFsError) { - recordSessionScanIssue(issues, { agent, path, message: error.message }) + issues({ agent, path, message: error.message }) } } -async function cachedProjectDirCwd( +export async function cachedProjectDirCwd( projectDir: string, - issues: AiVaultScanIssue[], - agent: AiVaultAgent + issues: IssueSink, + agent: AiVaultAgent, + signal?: AbortSignal ): Promise { const cached = projectDirCwdCache.get(projectDir) if (cached !== undefined) { @@ -53,7 +56,7 @@ async function cachedProjectDirCwd( projectDirCwdCache.set(projectDir, cached) return cached } - const cwd = await readProjectDirCwd(projectDir, issues, agent) + const cwd = await readProjectDirCwd(projectDir, issues, agent, signal) if (cwd) { if (projectDirCwdCache.size >= PROJECT_DIR_CWD_CACHE_MAX) { const oldest = projectDirCwdCache.keys().next() @@ -66,62 +69,7 @@ async function cachedProjectDirCwd( return cwd } -type InScopeDiscoveryArgs = { - rootDirs: readonly string[] - scopePaths: readonly string[] - limit: number - excludedFilePaths: ReadonlySet - issues: AiVaultScanIssue[] -} - -/** - * Fully include the transcripts of cwd-bucket directories (Claude, Pi, CodeBuddy) whose cwd - * falls inside the active workspace/project paths. - * - * Why: these agents keep one directory per cwd, e.g. `~/.claude/projects//`. - * The global scan is recency-capped, so a project the user hasn't touched recently - * can drop off the list entirely even though the agent's own resume still finds it. - * For scoped panel views we resolve each bucket's cwd cheaply and bypass the cap for - * the ones that belong to the active scope. - */ -export async function discoverInScopeCwdBucketFiles( - layout: CwdBucketLayout, - args: InScopeDiscoveryArgs -): Promise { - if (args.scopePaths.length === 0 || args.limit <= 0) { - return [] - } - const scopeProjectPrefixes = cwdBucketScopePrefixes(layout, args.scopePaths) - const inScope = createAiVaultScopeCwdMatcher(args.scopePaths) - const collected = new Map() - for (const rootDir of args.rootDirs) { - for (const projectDir of await listProjectDirs( - layout, - rootDir, - scopeProjectPrefixes, - args.issues - )) { - if (layout.agent === 'claude') { - const cwd = await cachedProjectDirCwd(projectDir, args.issues, layout.agent) - if (!inScope(cwd, projectDir)) { - continue - } - } - await collectBucketFiles({ - agent: layout.agent, - inScope: layout.agent !== 'claude' ? inScope : undefined, - projectDir, - issues: args.issues, - collected, - limit: args.limit, - excludedFilePaths: args.excludedFilePaths - }) - } - } - return [...collected.values()].sort((left, right) => right.mtimeMs - left.mtimeMs) -} - -function cwdBucketScopePrefixes( +export function cwdBucketScopePrefixes( layout: CwdBucketLayout, scopePaths: readonly string[] ): Set { @@ -141,16 +89,21 @@ function scopePathCandidates(scopePath: string): string[] { return wslScopePath ? [scopePath, wslScopePath.linuxPath] : [scopePath] } -async function listProjectDirs( +export async function listProjectDirs( layout: CwdBucketLayout, rootDir: string, scopeProjectPrefixes: ReadonlySet, - issues: AiVaultScanIssue[] + issues: IssueSink, + signal?: AbortSignal, + listed?: () => void ): Promise { let entries try { - entries = await wslGatedReaddir(rootDir, 'scan') + entries = await wslGatedReaddir(rootDir, 'scan', signal) + throwIfAiVaultScanCancelled(signal) + listed?.() } catch (err) { + throwIfAiVaultScanCancelled(signal) recordRefusal(issues, rootDir, err, layout.agent) return [] } @@ -161,12 +114,13 @@ async function listProjectDirs( async function readProjectDirCwd( projectDir: string, - issues: AiVaultScanIssue[], - agent: AiVaultAgent + issues: IssueSink, + agent: AiVaultAgent, + signal?: AbortSignal ): Promise { - const files = await newestTranscriptsInDir(projectDir, issues, agent) + const files = await newestTranscriptsInDir(projectDir, issues, agent, signal) for (const file of files.slice(0, REPRESENTATIVE_FILE_LIMIT)) { - const cwd = await readFirstCwd(file, issues, agent) + const cwd = await readFirstCwd(file, issues, agent, signal) if (cwd) { return cwd } @@ -176,18 +130,25 @@ async function readProjectDirCwd( async function newestTranscriptsInDir( projectDir: string, - issues: AiVaultScanIssue[], - agent: AiVaultAgent + issues: IssueSink, + agent: AiVaultAgent, + signal?: AbortSignal ): Promise { let entries try { - entries = await wslGatedReaddir(projectDir, 'scan') + entries = await wslGatedReaddir(projectDir, 'scan', signal) + throwIfAiVaultScanCancelled(signal) } catch (err) { + throwIfAiVaultScanCancelled(signal) recordRefusal(issues, projectDir, err, agent) return [] } const newest: { path: string; mtimeMs: number }[] = [] - for (const entry of entries) { + for (const [index, entry] of entries.entries()) { + if (index % 256 === 0) { + await yieldToEventLoop() + } + throwIfAiVaultScanCancelled(signal) if (!entry.isFile() || !TRANSCRIPT_EXTENSIONS.has(extname(entry.name).toLowerCase())) { continue } @@ -195,9 +156,10 @@ async function newestTranscriptsInDir( try { addBoundedPath(newest, REPRESENTATIVE_FILE_LIMIT, { path, - mtimeMs: (await wslGatedStat(path, 'scan')).mtimeMs + mtimeMs: (await wslGatedStat(path, 'scan', signal)).mtimeMs }) } catch (err) { + throwIfAiVaultScanCancelled(signal) // Best effort: unreadable candidates are ignored here and reported during // full collection if the project directory proves in-scope. recordRefusal(issues, path, err, agent) @@ -208,14 +170,17 @@ async function newestTranscriptsInDir( async function readFirstCwd( filePath: string, - issues: AiVaultScanIssue[], - agent: AiVaultAgent + issues: IssueSink, + agent: AiVaultAgent, + signal?: AbortSignal ): Promise { - const input = openTranscriptReadStream(filePath, { encoding: 'utf-8' }, 'scan') + const input = openTranscriptReadStream(filePath, { encoding: 'utf-8' }, 'scan', signal) const lines = createInterface({ input, crlfDelay: Infinity }) let read = 0 try { + throwIfAiVaultScanCancelled(signal) for await (const line of lines) { + throwIfAiVaultScanCancelled(signal) if (read++ >= REPRESENTATIVE_CWD_LINE_LIMIT) { break } @@ -225,6 +190,7 @@ async function readFirstCwd( } } } catch (err) { + throwIfAiVaultScanCancelled(signal) recordRefusal(issues, filePath, err, agent) return null } finally { @@ -236,38 +202,33 @@ async function readFirstCwd( return null } -async function collectBucketFiles(args: { +export async function collectBucketFiles(args: { + paths: readonly string[] agent: AiVaultAgent inScope?: ReturnType - projectDir: string - issues: AiVaultScanIssue[] + issues: IssueSink collected: Map limit: number excludedFilePaths: ReadonlySet + signal?: AbortSignal }): Promise { - let entries - try { - entries = await wslGatedReaddir(args.projectDir, 'scan') - } catch (err) { - recordRefusal(args.issues, args.projectDir, err, args.agent) - return - } - for (const entry of entries) { - if (!entry.isFile() || !TRANSCRIPT_EXTENSIONS.has(extname(entry.name).toLowerCase())) { - continue + for (const [index, path] of args.paths.entries()) { + if (index % 256 === 0) { + await yieldToEventLoop() } - const path = join(args.projectDir, entry.name) + throwIfAiVaultScanCancelled(args.signal) if (args.collected.has(path) || args.excludedFilePaths.has(path)) { continue } try { if (args.inScope) { - const cwd = await readFirstCwd(path, args.issues, args.agent) + const cwd = await readFirstCwd(path, args.issues, args.agent, args.signal) if (!args.inScope(cwd, path)) { continue } } - const fileStat = await wslGatedStat(path, 'scan') + const fileStat = await wslGatedStat(path, 'scan', args.signal) + throwIfAiVaultScanCancelled(args.signal) addBoundedFile(args.collected, args.limit, { path, mtimeMs: fileStat.mtimeMs, @@ -275,7 +236,8 @@ async function collectBucketFiles(args: { sizeBytes: fileStat.size }) } catch (err) { - recordSessionScanIssue(args.issues, { agent: args.agent, path, message: errorMessage(err) }) + throwIfAiVaultScanCancelled(args.signal) + args.issues({ agent: args.agent, path, message: errorMessage(err) }) } } } diff --git a/src/main/ai-vault/session-scanner-scope-inventory.test.ts b/src/main/ai-vault/session-scanner-scope-inventory.test.ts new file mode 100644 index 00000000000..0008e32c4c5 --- /dev/null +++ b/src/main/ai-vault/session-scanner-scope-inventory.test.ts @@ -0,0 +1,129 @@ +import { mkdtemp, rm, utimes } from 'node:fs/promises' +import { PassThrough } from 'node:stream' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { afterEach, expect, it, vi } from 'vitest' +import type { AiVaultScanIssue } from '../../shared/ai-vault-types' +import * as transcriptFs from '../native-chat/wsl-transcript-fs-access' +import { WslTranscriptFsError } from '../native-chat/wsl-transcript-fs-error' +import { aiVaultScanWork, trackAiVaultScanProgress } from './ai-vault-scan-progress' +import { recordSessionScanIssue } from './session-scan-issues' +import { CLAUDE_CWD_BUCKET_LAYOUT, PI_CWD_BUCKET_LAYOUT } from './session-cwd-bucket-layouts' +import { inventoryCwdBucketFiles, selectCwdBucketFiles } from './session-scanner-scope-inventory' +import { writeJsonlFile } from './session-scanner-test-fixtures' +let root: string | null = null +afterEach(async () => { + vi.restoreAllMocks() + if (root) { + await rm(root, { recursive: true, force: true }) + } +}) + +it('replays tied selection, winner exclusion and evicted paths in original root order', async () => { + root = await mkdtemp(join(tmpdir(), 'orca-bucket-replay-')) + const paths = ['b', 'a', 'c', 'winner'].map((name) => join(root!, '--repo--', `${name}.jsonl`)) + for (const [index, path] of paths.entries()) { + await writeJsonlFile(path, [{ type: 'session', cwd: '/repo' }]) + await utimes(path, index < 2 ? 1 : 2, index < 2 ? 1 : 2) + } + const read = transcriptFs.wslGatedReaddir + vi.spyOn(transcriptFs, 'wslGatedReaddir').mockImplementation(async (path, ...args) => + (await read(path, ...args)).sort( + (left, right) => paths.indexOf(join(path, left.name)) - paths.indexOf(join(path, right.name)) + ) + ) + const opened = vi.spyOn(transcriptFs, 'openTranscriptReadStream') + const inventory = await inventoryCwdBucketFiles(PI_CWD_BUCKET_LAYOUT, { + rootDirs: [root, root], + scopePaths: ['/repo'], + limit: 1 + }) + expect(opened).not.toHaveBeenCalled() + const selected = await selectCwdBucketFiles(inventory, { + scopePaths: ['/repo'], + limit: 1, + excludedFilePaths: new Set([paths[3]]), + issues: [] + }) + expect(selected.map((file) => file.path)).toEqual([paths[2]]) + expect(opened.mock.calls.map(([path]) => path)).toEqual([ + paths[0], + paths[1], + paths[2], + paths[0], + paths[1] + ]) +}) + +it('replays staged errors after global errors, interleaved with bucket errors under the issue cap', async () => { + root = await mkdtemp(join(tmpdir(), 'orca-bucket-issues-')) + const path = join(root, '--repo--', 'a.jsonl') + await writeJsonlFile(path, [{ type: 'session', cwd: '/repo' }]) + const read = transcriptFs.wslGatedReaddir + const refused = join(root, 'refused') + vi.spyOn(transcriptFs, 'wslGatedReaddir').mockImplementation((path, ...args) => { + if (path === refused) { + throw new WslTranscriptFsError('unavailable', 'Generated root failure') + } + return read(path, ...args) + }) + const inventory = await inventoryCwdBucketFiles(PI_CWD_BUCKET_LAYOUT, { + rootDirs: [root, refused], + scopePaths: ['/repo'], + limit: 1 + }) + vi.spyOn(transcriptFs, 'wslGatedStat').mockRejectedValue(new Error('Generated stat failure')) + const issues: AiVaultScanIssue[] = [] + for (let index = 0; index < 498; index++) { + recordSessionScanIssue(issues, { + agent: 'codex', + path: String(index), + message: 'Generated global failure' + }) + } + const expected = [...issues] + recordSessionScanIssue(expected, { agent: 'pi', path, message: 'Generated stat failure' }) + recordSessionScanIssue(expected, { + agent: 'pi', + path: refused, + message: 'Generated root failure' + }) + await selectCwdBucketFiles(inventory, { + scopePaths: ['/repo'], + limit: 1, + excludedFilePaths: new Set(), + issues + }) + expect(issues).toEqual(expected) + expect(issues.at(-1)).toMatchObject({ kind: 'notice' }) +}) + +it('closes representative streams and inventory sources on cancellation before reading', async () => { + root = await mkdtemp(join(tmpdir(), 'orca-bucket-abort-')) + await writeJsonlFile(join(root, '-repo', 'a.jsonl'), [{ cwd: '/repo' }]) + const controller = new AbortController() + const input = new PassThrough() + vi.spyOn(transcriptFs, 'openTranscriptReadStream').mockImplementation(() => { + controller.abort() + return input + }) + const stop = trackAiVaultScanProgress(controller.signal, () => undefined) + try { + await expect( + inventoryCwdBucketFiles(CLAUDE_CWD_BUCKET_LAYOUT, { + rootDirs: [root], + scopePaths: ['/repo'], + limit: 1, + signal: controller.signal + }) + ).rejects.toMatchObject({ name: 'AbortError' }) + expect(input.destroyed).toBe(true) + expect(aiVaultScanWork(controller.signal)).toMatchObject({ + stage: 'finding', + agents: [], + read: { done: 0, total: 0 } + }) + } finally { + stop() + } +}) diff --git a/src/main/ai-vault/session-scanner-scope-inventory.ts b/src/main/ai-vault/session-scanner-scope-inventory.ts new file mode 100644 index 00000000000..1f6121b0e90 --- /dev/null +++ b/src/main/ai-vault/session-scanner-scope-inventory.ts @@ -0,0 +1,150 @@ +import { extname, join } from 'node:path' +import { yieldToEventLoop } from '../../shared/event-loop-yield' +import type { AiVaultScanIssue } from '../../shared/ai-vault-types' +import { wslGatedReaddir } from '../native-chat/wsl-transcript-fs-access' +import { throwIfAiVaultScanCancelled } from './ai-vault-scan-cancellation' +import { openAiVaultScanSource } from './ai-vault-scan-progress' +import { SCAN_ISSUE_LIMIT, recordSessionScanIssue } from './session-scan-issues' +import { createAiVaultScopeCwdMatcher } from '../../shared/ai-vault-scope-cwd-match' +import type { CwdBucketLayout } from './session-cwd-bucket-layouts' +import type { FileWithMtime } from './session-scanner-types' +import { + cachedProjectDirCwd, + cwdBucketScopePrefixes, + listProjectDirs, + collectBucketFiles, + recordRefusal, + type IssueSink +} from './session-scanner-scope-discovery' + +const TRANSCRIPT_EXTENSIONS = new Set(['.jsonl']) +export type CwdBucketInventory = { + layout: CwdBucketLayout + directories: string[][] + issues: { before: number; issue: AiVaultScanIssue }[] +} + +/** Bounded CPU chunks must let cancellation and other scanner requests run. */ +export async function yieldScanInventory(index: number, signal?: AbortSignal): Promise { + if (index % 256 === 0) { + await yieldToEventLoop() + } + throwIfAiVaultScanCancelled(signal) +} + +type InScopeDiscoveryArgs = { + rootDirs: readonly string[] + scopePaths: readonly string[] + limit: number + excludedFilePaths: ReadonlySet + issues: AiVaultScanIssue[] + signal?: AbortSignal +} + +export async function discoverInScopeCwdBucketFiles( + layout: CwdBucketLayout, + args: InScopeDiscoveryArgs +): Promise { + const inventory = await inventoryCwdBucketFiles(layout, args) + return selectCwdBucketFiles(inventory, args) +} + +/** Freeze names only; winner-dependent header/stat decisions remain in selection. */ +export async function inventoryCwdBucketFiles( + layout: CwdBucketLayout, + args: Omit, + issueLimit = SCAN_ISSUE_LIMIT +): Promise { + const inventory: CwdBucketInventory = { layout, directories: [], issues: [] } + if (args.scopePaths.length === 0 || args.limit <= 0) { + return inventory + } + const record: IssueSink = (issue) => { + if (inventory.issues.length < issueLimit) { + inventory.issues.push({ before: inventory.directories.length, issue }) + } + } + const prefixes = cwdBucketScopePrefixes(layout, args.scopePaths) + const inScope = createAiVaultScopeCwdMatcher(args.scopePaths) + let index = 0 + for (const rootDir of args.rootDirs) { + throwIfAiVaultScanCancelled(args.signal) + let close: () => void = () => undefined + try { + const directories = await listProjectDirs( + layout, + rootDir, + prefixes, + record, + args.signal, + () => { + close = openAiVaultScanSource(args.signal, layout.agent) + } + ) + for (const directory of directories) { + await yieldScanInventory(index++, args.signal) + if (layout.agent === 'claude') { + const cwd = await cachedProjectDirCwd(directory, record, layout.agent, args.signal) + if (!inScope(cwd, directory)) { + continue + } + } + const paths: string[] = [] + try { + const entries = await wslGatedReaddir(directory, 'scan', args.signal) + for (const entry of entries) { + await yieldScanInventory(index++, args.signal) + if (entry.isFile() && TRANSCRIPT_EXTENSIONS.has(extname(entry.name).toLowerCase())) { + paths.push(join(directory, entry.name)) + } + } + } catch (err) { + throwIfAiVaultScanCancelled(args.signal) + recordRefusal(record, directory, err, layout.agent) + } + inventory.directories.push(paths) + } + } finally { + close() + } + } + return inventory +} + +/** Replay the original selector, interleaving earlier diagnostics at their original positions. */ +export async function selectCwdBucketFiles( + inventory: CwdBucketInventory, + args: Pick< + InScopeDiscoveryArgs, + 'scopePaths' | 'limit' | 'excludedFilePaths' | 'issues' | 'signal' + > +): Promise { + const { layout } = inventory + const collected = new Map() + const record: IssueSink = (issue) => recordSessionScanIssue(args.issues, issue) + const inScope = createAiVaultScopeCwdMatcher(args.scopePaths) + let issueIndex = 0 + for (let index = 0; index <= inventory.directories.length; index++) { + await yieldScanInventory(index, args.signal) + while (inventory.issues[issueIndex]?.before === index) { + record(inventory.issues[issueIndex++].issue) + } + const paths = inventory.directories[index] + if (!paths) { + break + } + await collectBucketFiles({ + paths, + agent: layout.agent, + inScope: layout.agent !== 'claude' ? inScope : undefined, + issues: record, + collected, + limit: args.limit, + excludedFilePaths: args.excludedFilePaths, + signal: args.signal + }) + } + inventory.directories.length = 0 + inventory.issues.length = 0 + return [...collected.values()].sort((left, right) => right.mtimeMs - left.mtimeMs) +} diff --git a/src/main/ai-vault/session-scanner-scope-walk.test.ts b/src/main/ai-vault/session-scanner-scope-walk.test.ts index a83626009be..55d5b47e0f2 100644 --- a/src/main/ai-vault/session-scanner-scope-walk.test.ts +++ b/src/main/ai-vault/session-scanner-scope-walk.test.ts @@ -352,6 +352,7 @@ it('walks a history larger than the spread-argument limit', async () => { modifiedAt: new Date(1).toISOString() })) const sessions = await scanInScopeSessions({ + inventory: [], discoveries: [ { agent: 'codex', diff --git a/src/main/ai-vault/session-scanner-scope-walk.ts b/src/main/ai-vault/session-scanner-scope-walk.ts index 1f428c530f2..aaea1adb559 100644 --- a/src/main/ai-vault/session-scanner-scope-walk.ts +++ b/src/main/ai-vault/session-scanner-scope-walk.ts @@ -1,3 +1,4 @@ +import { SCAN_ISSUE_LIMIT } from './session-scan-issues' import { createAiVaultScopeCwdMatcher } from '../../shared/ai-vault-scope-cwd-match' import type { AiVaultScanIssue, AiVaultSession } from '../../shared/ai-vault-types' import type { ExecutionHostId } from '../../shared/execution-host' @@ -16,7 +17,13 @@ import { parseSessionCandidates } from './session-scanner-candidate-parse' import { sessionSortTime } from './session-scanner-accumulator' import { discoveryFileCandidates, dropCodexHardlinkAliases } from './session-scanner-candidates' import type { SessionParseStats } from './session-scanner-parse-cache' -import { discoverInScopeCwdBucketFiles } from './session-scanner-scope-discovery' +import { + inventoryCwdBucketFiles, + selectCwdBucketFiles, + yieldScanInventory, + type CwdBucketInventory +} from './session-scanner-scope-inventory' +import { aiVaultScanWorkload, noteAiVaultScanWorkChanged } from './ai-vault-scan-progress' import type { AiVaultScanOptions, FileWithMtime, @@ -28,6 +35,7 @@ import { sidecarUnchanged } from './session-sidecar-stat' import { prioritizeAntigravityTranscriptCandidates } from './antigravity-transcript-candidates' type ScopeScanArgs = { + inventory: CwdBucketInventory[] discoveries: readonly SessionFileDiscovery[] options: AiVaultScanOptions limit: number @@ -50,24 +58,29 @@ export async function scanInScopeSessions(args: ScopeScanArgs): Promise discovery.agent === layout.agent) - .map((discovery) => discovery.rootDir), + const workload = aiVaultScanWorkload(args.options.signal) + for (const inventory of args.inventory) { + const files = await selectCwdBucketFiles(inventory, { scopePaths, limit: args.limit, excludedFilePaths: args.alreadyParsedFilePaths, - issues: args.issues + issues: args.issues, + signal: args.options.signal }) + const selected = new Set(files.map((file) => file.path)) + let index = 0 + for (const path of workload?.bucketPaths(inventory.layout.agent) ?? []) { + await yieldScanInventory(index++, args.options.signal) + if (!selected.has(path)) { + workload?.settlePath(inventory.layout.agent, path, 'bucket', false) + } + } for (const file of files) { - candidates.push({ agent: layout.agent, file, codexHome: null }) + candidates.push({ agent: inventory.layout.agent, file, codexHome: null }) } } + args.inventory.length = 0 + noteAiVaultScanWorkChanged(args.options.signal) const walked = await walkInScopeSessions(args, scopePaths).finally(clearSessionScopeCwdKnowledge) if (candidates.length === 0) { return walked @@ -80,7 +93,8 @@ export async function scanInScopeSessions(args: ScopeScanArgs): Promise workload?.settle(candidate, 'bucket', parsed) }) return [...bucketed, ...walked] } @@ -96,7 +110,7 @@ async function walkInScopeSessions( ): Promise { const signal = args.options.signal const inScope = createAiVaultScopeCwdMatcher(scopePaths) - const candidates = walkCandidates(args, scopePaths) + const candidates = await walkCandidates(args, scopePaths) await lookupScopeCwds(candidates, signal) const sessions = await parseSessionCandidates({ candidates, @@ -108,6 +122,7 @@ async function walkInScopeSessions( antigravityWorkspaceResolver: args.antigravityWorkspaceResolver, signal, listed: false, + settled: (candidate, parsed) => aiVaultScanWorkload(signal)?.settle(candidate, 'walk', parsed), selectCandidate: (candidate) => mayBeInScope(candidate, inScope, args.platform, signal), admitSession: (session) => inScope(session.cwd, session.filePath) }) @@ -116,39 +131,106 @@ async function walkInScopeSessions( .slice(0, args.limit) } -function walkCandidates( - args: ScopeScanArgs, +/** Pure membership enumeration; pruning and ranking remain execution decisions. */ +async function walkInventory( + args: Pick, scopePaths: readonly string[] -): SessionFileCandidate[] { +): Promise<{ candidates: SessionFileCandidate[]; seenPaths: Set }> { const all: SessionFileCandidate[] = [] const seenPaths = new Set() + let index = 0 for (const discovery of args.discoveries) { if (!scopeWalksAgentFiles(discovery.agent, scopePaths)) { continue } const files: FileWithMtime[] = [] - for (const file of [...discovery.files, ...(discovery.scopeFiles ?? [])]) { - if (!seenPaths.has(file.path)) { - seenPaths.add(file.path) - files.push(file) + for (const collection of [discovery.files, discovery.scopeFiles ?? []]) { + for (const file of collection) { + await yieldScanInventory(index++, args.options.signal) + if (!seenPaths.has(file.path)) { + seenPaths.add(file.path) + files.push(file) + } } } - // Why: one item per push; a history-sized list spread into push() exceeds V8's argument limit. for (const candidate of discoveryFileCandidates(discovery, files, args.options)) { all.push(candidate) } } + return { candidates: dropCodexHardlinkAliases(all), seenPaths } +} + +async function walkCandidates( + args: ScopeScanArgs, + scopePaths: readonly string[] +): Promise { + const { candidates, seenPaths } = await walkInventory(args, scopePaths) pruneSessionScopeCwds(seenPaths) - // Aliases collapse with the listed files still present, so a listed rollout's - // hardlink in another Codex home is not parsed again. + const workload = aiVaultScanWorkload(args.options.signal) + const remaining: SessionFileCandidate[] = [] + for (const [index, candidate] of candidates.entries()) { + await yieldScanInventory(index, args.options.signal) + if (args.alreadyParsedFilePaths.has(candidate.file.path)) { + workload?.settle(candidate, 'walk', false) + } else { + remaining.push(candidate) + } + } return prioritizeAntigravityTranscriptCandidates( - dropCodexHardlinkAliases(all).filter( - (candidate) => !args.alreadyParsedFilePaths.has(candidate.file.path) - ), + remaining, (candidate) => candidate.agent === 'antigravity' ) } +export async function prepareScopeScan( + args: Pick +): Promise { + const scopePaths = args.options.scopePaths ?? [] + if (scopePaths.length === 0) { + return [] + } + const workload = aiVaultScanWorkload(args.options.signal) + // H is temporary: walk-only keys settle directly, without a second retained registry. + if (workload) { + const { candidates } = await walkInventory(args, scopePaths) + for (const [index, candidate] of candidates.entries()) { + await yieldScanInventory(index, args.options.signal) + workload.register(candidate, 'walk') + } + } + const inventories: CwdBucketInventory[] = [] + let issueLimit = SCAN_ISSUE_LIMIT + let index = 0 + for (const source of Object.values(AI_VAULT_SCOPE_CWD_SOURCES)) { + if (source.kind !== 'bucket') { + continue + } + const inventory = await inventoryCwdBucketFiles( + source.layout, + { + rootDirs: args.discoveries + .filter((discovery) => discovery.agent === source.layout.agent) + .map((discovery) => discovery.rootDir), + scopePaths, + limit: args.limit, + signal: args.options.signal + }, + issueLimit + ) + issueLimit -= inventory.issues.length + inventories.push(inventory) + if (workload) { + for (const paths of inventory.directories) { + for (const path of paths) { + await yieldScanInventory(index++, args.options.signal) + workload.registerPath(source.layout.agent, path, 'bucket') + } + } + } + } + return inventories +} + // Per lookup batch: one database round trip, then a turn for the event loop. const SCOPE_LOOKUP_BATCH = 500 diff --git a/src/main/ai-vault/session-scanner.ts b/src/main/ai-vault/session-scanner.ts index 3df46b506e4..d5d8c488ee6 100644 --- a/src/main/ai-vault/session-scanner.ts +++ b/src/main/ai-vault/session-scanner.ts @@ -15,7 +15,9 @@ import { sessionCandidatesFromDiscoveries } from './session-scanner-candidates' import { parseSessionCandidates } from './session-scanner-candidate-parse' import { createSessionParseStats } from './session-scanner-parse-cache' import { recordSessionScanIssue } from './session-scan-issues' -import { scanInScopeSessions } from './session-scanner-scope-walk' +import { aiVaultScanWorkload, noteAiVaultScanReading } from './ai-vault-scan-progress' +import { yieldScanInventory } from './session-scanner-scope-inventory' +import { prepareScopeScan, scanInScopeSessions } from './session-scanner-scope-walk' import { discoverAiVaultSessionSources } from './session-scanner-source-discovery' import { cursorChatMetaRefusals, withCursorChatMetaScan } from './session-scanner-cursor-chat-meta' import type { AiVaultScanOptions } from './session-scanner-types' @@ -75,6 +77,17 @@ async function scanAiVaultSessionStores(options: AiVaultScanOptions): Promise workload?.settle(candidate, 'global', parsed) }) const cappedSessions = dedupeScannedSessions(parsedSessions) @@ -91,6 +105,7 @@ async function scanAiVaultSessionStores(options: AiVaultScanOptions): Promise push({ scanKey: 'worktree-a', progress: next })) expect(view.getByText('Scanned', { exact: false })).toBeTruthy() @@ -161,3 +161,18 @@ it('keeps real rows through partial rescans, detached keys and failed retries', act(() => push({ scanKey: 'worktree-c', progress: { ...next, agents: [FINISHED.agents[0]!] } })) expect(rowCount()).toBe(1) }) + +it('shows settled candidate work and keeps older snapshots without skipped compatible', () => { + const reading: AiVaultScanProgress = { + ...FINISHED, + stage: 'reading', + settledAt: null, + read: { done: 3, total: 10, skipped: 4 } + } + const panel = render() + expect(screen.getByRole('button').textContent).toContain('Checking 7 / 10') + panel.rerender( + + ) + expect(screen.getByRole('button').textContent).toContain('Checking 3 / 10') +}) diff --git a/src/renderer/src/components/right-sidebar/AiVaultScanProgressStrip.tsx b/src/renderer/src/components/right-sidebar/AiVaultScanProgressStrip.tsx index 9438b8221e1..00083881cd1 100644 --- a/src/renderer/src/components/right-sidebar/AiVaultScanProgressStrip.tsx +++ b/src/renderer/src/components/right-sidebar/AiVaultScanProgressStrip.tsx @@ -61,8 +61,8 @@ function stripSentence(progress: AiVaultScanProgress): string { ) case 'reading': return translate( - 'auto.components.right.sidebar.AiVaultScanProgressStrip.reading', - 'Reading {{value0}} files', + 'auto.components.right.sidebar.AiVaultScanProgressStrip.checking', + 'Checking {{value0}}', { value0 } ) case 'finished': @@ -81,7 +81,7 @@ function foundCount(agent: AiVaultScanAgentProgress): number { function StripText({ progress, active }: StripActivity): React.JSX.Element { const emphasis = active && progress.stage === 'reading' - ? `${formatCount(progress.read.done)} / ${formatCount(progress.read.total)}` + ? `${formatCount(progress.read.done + (progress.read.skipped ?? 0))} / ${formatCount(progress.read.total)}` : fileCountLabel(progress.agents.reduce((sum, agent) => sum + foundCount(agent), 0)) const sentence = active || progress.stage === 'finished' ? stripSentence(progress) : EMPHASIS_TOKEN diff --git a/src/renderer/src/i18n/locales/en.json b/src/renderer/src/i18n/locales/en.json index 45ab0cfe584..3ac5024b1c2 100644 --- a/src/renderer/src/i18n/locales/en.json +++ b/src/renderer/src/i18n/locales/en.json @@ -14824,7 +14824,7 @@ "databaseCount_one": "{{value0}} database", "databaseCount_other": "{{value0}} databases", "scanning": "Scanning {{value0}}", - "reading": "Reading {{value0}} files", + "checking": "Checking {{value0}}", "scanned": "Scanned {{value0}}" } } diff --git a/src/renderer/src/i18n/locales/es.json b/src/renderer/src/i18n/locales/es.json index d518b933ead..4cf19c6d0ff 100644 --- a/src/renderer/src/i18n/locales/es.json +++ b/src/renderer/src/i18n/locales/es.json @@ -14487,7 +14487,7 @@ "databaseCount_one": "{{value0}} base de datos", "databaseCount_other": "{{value0}} bases de datos", "scanning": "Analizando {{value0}}", - "reading": "Leyendo {{value0}} archivos", + "checking": "Comprobando {{value0}}", "scanned": "Analizados {{value0}}" } } diff --git a/src/renderer/src/i18n/locales/fr.json b/src/renderer/src/i18n/locales/fr.json index 81f5f5b98ad..134cc5e961b 100644 --- a/src/renderer/src/i18n/locales/fr.json +++ b/src/renderer/src/i18n/locales/fr.json @@ -14600,7 +14600,7 @@ "databaseCount_one": "{{value0}} base de données", "databaseCount_other": "{{value0}} bases de données", "scanning": "Analyse de {{value0}}", - "reading": "Lecture de {{value0}} fichiers", + "checking": "Vérification {{value0}}", "scanned": "{{value0}} analysés" } } diff --git a/src/renderer/src/i18n/locales/ja.json b/src/renderer/src/i18n/locales/ja.json index 0de09af9c45..c90b2c5e303 100644 --- a/src/renderer/src/i18n/locales/ja.json +++ b/src/renderer/src/i18n/locales/ja.json @@ -14492,7 +14492,7 @@ "databaseCount_one": "{{value0}} 個のデータベース", "databaseCount_other": "{{value0}} 個のデータベース", "scanning": "{{value0}}をスキャン中", - "reading": "{{value0}} ファイルを読み込み中", + "checking": "確認中 {{value0}}", "scanned": "{{value0}}をスキャン済み" } } diff --git a/src/renderer/src/i18n/locales/ko.json b/src/renderer/src/i18n/locales/ko.json index 790d14126a2..ea569a1c2c2 100644 --- a/src/renderer/src/i18n/locales/ko.json +++ b/src/renderer/src/i18n/locales/ko.json @@ -14492,7 +14492,7 @@ "databaseCount_one": "데이터베이스 {{value0}}개", "databaseCount_other": "데이터베이스 {{value0}}개", "scanning": "{{value0}} 스캔 중", - "reading": "파일 {{value0}}개 읽는 중", + "checking": "확인 중 {{value0}}", "scanned": "{{value0}} 스캔함" } } diff --git a/src/renderer/src/i18n/locales/pt.json b/src/renderer/src/i18n/locales/pt.json index 9648c31a8e1..454b425ad70 100644 --- a/src/renderer/src/i18n/locales/pt.json +++ b/src/renderer/src/i18n/locales/pt.json @@ -10369,7 +10369,6 @@ "databaseCount_one": "{{value0}} banco de dados", "databaseCount_other": "{{value0}} bancos de dados", "scanning": "Verificando {{value0}}", - "reading": "Lendo {{value0}} arquivos", "scanned": "Verificação concluída: {{value0}}" } } diff --git a/src/renderer/src/i18n/locales/zh.json b/src/renderer/src/i18n/locales/zh.json index 2af56b29582..00a78d2d9b9 100644 --- a/src/renderer/src/i18n/locales/zh.json +++ b/src/renderer/src/i18n/locales/zh.json @@ -14492,7 +14492,7 @@ "databaseCount_one": "{{value0}} 个数据库", "databaseCount_other": "{{value0}} 个数据库", "scanning": "正在扫描 {{value0}}", - "reading": "正在读取 {{value0}} 个文件", + "checking": "正在检查 {{value0}}", "scanned": "已扫描 {{value0}}" } } diff --git a/src/shared/ai-vault-scan-progress.ts b/src/shared/ai-vault-scan-progress.ts index efec22acfb4..db12fbe61b0 100644 --- a/src/shared/ai-vault-scan-progress.ts +++ b/src/shared/ai-vault-scan-progress.ts @@ -9,14 +9,19 @@ export type AiVaultScanAgentProgress = { databases: number /** A folder or database of this agent is still being listed. */ walking: boolean + /** Absent until the scan freezes its candidate workload. */ + read?: AiVaultScanReadingProgress } +/** Terminal candidate outcomes; skipped candidates needed no parser invocation. */ +export type AiVaultScanReadingProgress = { done: number; total: number; skipped?: number } + /** The scanner process's view of the scan it is running. */ export type AiVaultScanWork = { stage: 'finding' | 'reading' agents: AiVaultScanAgentProgress[] - /** Transcripts read so far, out of the ones this scan will read. */ - read: { done: number; total: number } + /** Fixed candidate workload; settled work is done + skipped. */ + read: AiVaultScanReadingProgress } /** One local Session History scan entry, as main pushes it to windows. */