diff --git a/src/main/ai-vault/ai-vault-local-scan-progress.ts b/src/main/ai-vault/ai-vault-local-scan-progress.ts new file mode 100644 index 00000000000..2d468ca03aa --- /dev/null +++ b/src/main/ai-vault/ai-vault-local-scan-progress.ts @@ -0,0 +1,134 @@ +import type { + AiVaultScanProgress, + AiVaultScanProgressUpdate, + AiVaultScanWork +} from '../../shared/ai-vault-scan-progress' +import { + AiVaultServiceDeadlineError, + type AiVaultServiceScanListener +} from './session-scanner-service-request-deadline' + +type Listener = (update: AiVaultScanProgressUpdate) => void + +/** One local scan from coordinator creation through its service reply. */ +export type LocalAiVaultScanProgressHandle = AiVaultServiceScanListener & { + finish: (work: AiVaultScanWork | null) => void + fail: (error: unknown) => void +} + +const MAX_PROGRESS_ENTRIES = 8 +const latest = new Map() +const currentHandles = new Map() +const listeners = new Set() + +export function localAiVaultScanProgress(scanKey: string): AiVaultScanProgress | null { + return latest.get(scanKey) ?? null +} + +export function onLocalAiVaultScanProgress(listener: Listener): () => void { + listeners.add(listener) + return () => listeners.delete(listener) +} + +function publish(scanKey: string, next: AiVaultScanProgress | null): void { + latest.delete(scanKey) + if (next) { + latest.set(scanKey, next) + } + while (latest.size > MAX_PROGRESS_ENTRIES) { + const oldest = latest.keys().next().value + if (oldest === undefined) { + break + } + latest.delete(oldest) + for (const listener of listeners) { + listener({ scanKey: oldest, progress: null }) + } + } + for (const listener of listeners) { + listener({ scanKey, progress: next }) + } +} + +/** A replacement clears its previous entry before roots preparation can yield. */ +export function startLocalAiVaultScanProgress( + scanKey: string, + scanId: string, + now: () => number = Date.now +): LocalAiVaultScanProgressHandle { + let serviceRequestId: number | null = null + let current: AiVaultScanProgress = { + scanId, + stage: 'finding', + agents: [], + read: { done: 0, total: 0 }, + startedAt: now(), + settledAt: null + } + const handle: LocalAiVaultScanProgressHandle = { + sent: (requestId) => { + serviceRequestId = requestId + if (currentHandles.get(scanKey) === handle) { + publish(scanKey, current) + } + }, + work: (work) => { + if (currentHandles.get(scanKey) === handle) { + current = { ...current, ...work } + publish(scanKey, current) + } + }, + finish: (work) => { + if (currentHandles.get(scanKey) === handle) { + currentHandles.delete(scanKey) + publish(scanKey, { ...current, ...work, stage: 'finished', settledAt: now() }) + } + }, + fail: (error) => { + if (currentHandles.get(scanKey) === handle) { + currentHandles.delete(scanKey) + if ( + error instanceof AiVaultServiceDeadlineError && + error.operation === 'scan' && + error.requestId === serviceRequestId + ) { + publish(scanKey, { ...current, stage: 'stopped', settledAt: now() }) + } else { + publish(scanKey, null) + } + } + } + } + currentHandles.set(scanKey, handle) + publish(scanKey, null) + return handle +} + +/** Runs the scanner service request using the coordinator's progress ownership. */ +export async function requestLocalAiVaultScan( + request: (listener: AiVaultServiceScanListener) => Promise, + options: { + now?: () => number + progress?: LocalAiVaultScanProgressHandle + scanKey?: string + scanId?: string + } = {} +): Promise { + const progress = + options.progress ?? + startLocalAiVaultScanProgress(options.scanKey ?? '', options.scanId ?? '', options.now) + try { + const value = await request(progress) + progress.finish(value.work) + return value + } catch (error) { + progress.fail(error) + throw error + } +} + +export function resetLocalAiVaultScanProgressForTests(): void { + latest.clear() + currentHandles.clear() + listeners.clear() +} diff --git a/src/main/ai-vault/ai-vault-scan-coordinator.test.ts b/src/main/ai-vault/ai-vault-scan-coordinator.test.ts index 6464770d516..fd35bae6939 100644 --- a/src/main/ai-vault/ai-vault-scan-coordinator.test.ts +++ b/src/main/ai-vault/ai-vault-scan-coordinator.test.ts @@ -1,9 +1,60 @@ import { describe, expect, it, vi } from 'vitest' import { AiVaultScanCoordinator } from './ai-vault-scan-coordinator' +import { + localAiVaultScanProgress, + onLocalAiVaultScanProgress, + resetLocalAiVaultScanProgressForTests, + startLocalAiVaultScanProgress, + type LocalAiVaultScanProgressHandle +} from './ai-vault-local-scan-progress' +import { AiVaultServiceDeadlineError } from './session-scanner-service-request-deadline' const EMPTY_RESULT = { sessions: [], issues: [], scannedAt: '2026-07-27T00:00:00.000Z' } describe('AiVaultScanCoordinator', () => { + it('keeps the replacement finding while the old scan aborts before root preparation ends', async () => { + const coordinator = new AiVaultScanCoordinator() + const stages: (string | null)[] = [] + onLocalAiVaultScanProgress(({ progress }) => stages.push(progress?.stage ?? null)) + try { + const stopped = startLocalAiVaultScanProgress('scope', 'stopped', () => 500) + stopped.sent(1) + stopped.fail(new AiVaultServiceDeadlineError('scan timed out', 'scan', 1)) + let old: LocalAiVaultScanProgressHandle | null = null + const first = coordinator.run({ + key: 'scope', + onCreate: (id) => { + old = startLocalAiVaultScanProgress('scope', id, () => 1_000) + }, + onError: (error) => old?.fail(error), + start: (signal) => + new Promise((_resolve, reject) => { + signal.addEventListener('abort', () => reject(new Error('cancelled')), { once: true }) + }) + }) + await Promise.resolve() + let finishPreparation: ((result: typeof EMPTY_RESULT) => void) | undefined + const replacement = coordinator.run({ + key: 'scope', + force: true, + onCreate: (id) => { + startLocalAiVaultScanProgress('scope', id, () => 2_000) + }, + start: () => + new Promise((resolve) => { + finishPreparation = resolve + }) + }) + await Promise.resolve() + expect(localAiVaultScanProgress('scope')).toBeNull() + expect(stages.at(-1)).toBeNull() + finishPreparation?.(EMPTY_RESULT) + await expect(Promise.all([first, replacement])).resolves.toEqual([EMPTY_RESULT, EMPTY_RESULT]) + } finally { + resetLocalAiVaultScanProgressForTests() + } + }) + it('keeps a shared scan alive when only one caller cancels', async () => { const coordinator = new AiVaultScanCoordinator() let resolveScan: ((result: typeof EMPTY_RESULT) => void) | undefined diff --git a/src/main/ai-vault/ai-vault-scan-coordinator.ts b/src/main/ai-vault/ai-vault-scan-coordinator.ts index 7e704e4b820..1dd9a966eda 100644 --- a/src/main/ai-vault/ai-vault-scan-coordinator.ts +++ b/src/main/ai-vault/ai-vault-scan-coordinator.ts @@ -1,3 +1,4 @@ +import { randomUUID } from 'node:crypto' import { AI_VAULT_SCAN_CANCELLED_MESSAGE, type AiVaultListResult @@ -8,6 +9,7 @@ import { const FORCED_SCAN_PREEMPT_AFTER_MS = 5_000 type ScanEntry = { + id: string controller: AbortController force: boolean startedAt: number @@ -27,7 +29,10 @@ export class AiVaultScanCoordinator { key: string force?: boolean signal?: AbortSignal - start: (signal: AbortSignal) => Promise + onCreate?: (id: string) => void + onError?: (error: unknown) => void + onReject?: (error: unknown, id: string) => unknown + start: (signal: AbortSignal, id: string) => Promise }): Promise { if (args.signal?.aborted) { return Promise.reject(scanCancellationError()) @@ -39,7 +44,13 @@ export class AiVaultScanCoordinator { entry = undefined } if (!entry) { - entry = this.createEntry(args.key, args.force === true, args.start) + entry = this.createEntry( + args.key, + args.force === true, + args.start, + args.onCreate, + args.onError + ) this.entries.set(args.key, entry) } if (preempted) { @@ -48,16 +59,21 @@ export class AiVaultScanCoordinator { preempted.preemptedBy = entry preempted.controller.abort() } - return this.attach(args.key, entry, args.signal) + return this.attach(args.key, entry, args.signal, args.onReject) } private createEntry( key: string, force: boolean, - start: (signal: AbortSignal) => Promise + start: (signal: AbortSignal, id: string) => Promise, + onCreate?: (id: string) => void, + onError?: (error: unknown) => void ): ScanEntry { const controller = new AbortController() + const id = randomUUID() + onCreate?.(id) const entry: ScanEntry = { + id, controller, force, startedAt: Date.now(), @@ -65,7 +81,7 @@ export class AiVaultScanCoordinator { if (controller.signal.aborted) { throw scanCancellationError() } - return start(controller.signal) + return start(controller.signal, id) }), waiterCount: 0, settled: false, @@ -75,11 +91,19 @@ export class AiVaultScanCoordinator { entry.settled = true this.removeEntry(key, entry) } - void entry.promise.then(onSettled, onSettled) + void entry.promise.then(onSettled, (error) => { + onError?.(error) + onSettled() + }) return entry } - private attach(key: string, entry: ScanEntry, signal?: AbortSignal): Promise { + private attach( + key: string, + entry: ScanEntry, + signal?: AbortSignal, + onReject?: (error: unknown, id: string) => unknown + ): Promise { entry.waiterCount++ return new Promise((resolve, reject) => { let attached = true @@ -109,7 +133,7 @@ export class AiVaultScanCoordinator { reject(scanCancellationError()) return } - void this.attach(key, replacement, signal).then(resolve, reject) + void this.attach(key, replacement, signal, onReject).then(resolve, reject) } signal?.addEventListener('abort', onAbort, { once: true }) entry.controller.signal.addEventListener('abort', onAbort, { once: true }) @@ -127,7 +151,7 @@ export class AiVaultScanCoordinator { (error) => { if (attached) { detach() - reject(error) + reject(onReject ? onReject(error, entry.id) : error) } } ) 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 new file mode 100644 index 00000000000..1a91b6e7b2f --- /dev/null +++ b/src/main/ai-vault/ai-vault-scan-progress-flow.test.ts @@ -0,0 +1,219 @@ +import { link, mkdir, mkdtemp, rm } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { afterEach, describe, expect, it, vi } from 'vitest' +import type { AiVaultScanProgress, AiVaultScanWork } from '../../shared/ai-vault-scan-progress' +import { + localAiVaultScanProgress, + onLocalAiVaultScanProgress, + requestLocalAiVaultScan, + resetLocalAiVaultScanProgressForTests, + startLocalAiVaultScanProgress +} from './ai-vault-local-scan-progress' +import { AiVaultScanTally } from './ai-vault-scan-tally' +import { aiVaultScanWork, trackAiVaultScanProgress } from './ai-vault-scan-progress' +import { scanAiVaultSessions } from './session-scanner' +import { resetSessionParseCacheForTests } from './session-scanner-parse-cache' +import { AiVaultScannerServiceClient } from './session-scanner-service-client' +import { + AI_VAULT_SERVICE_SCAN_TIMEOUT_MS, + AiVaultServiceDeadlineError +} from './session-scanner-service-request-deadline' +import { + AiVaultServiceTestChild, + aiVaultServiceRequestId, + readyAiVaultServiceChild +} from './session-scanner-service-test-child' +import { isolatedScanRoots, writeJsonlFile } from './session-scanner-test-fixtures' + +let tempRoot: string | null = null + +afterEach(async () => { + vi.useRealTimers() + resetSessionParseCacheForTests() + resetLocalAiVaultScanProgressForTests() + if (tempRoot) { + await rm(tempRoot, { recursive: true, force: true }) + tempRoot = null + } +}) + +/** Runs a real scan in this process the way the scanner child does, keeping every report. */ +async function scanInChild(): Promise<{ reports: AiVaultScanWork[]; final: AiVaultScanWork }> { + tempRoot = await mkdtemp(join(tmpdir(), 'orca-scan-strip-')) + const roots = isolatedScanRoots(tempRoot) + for (const name of ['a', 'b', 'c']) { + await writeJsonlFile(join(roots.claudeProjectsDir, 'project', `${name}.jsonl`), [ + { + type: 'user', + sessionId: name, + timestamp: '2026-05-01T10:00:00.000Z', + cwd: '/tmp/claude', + message: { role: 'user', content: `ask ${name}` } + } + ]) + } + // A second path to one transcript, as Codex homes and linked profiles produce: still one file. + await mkdir(join(roots.claudeProjectsDir, 'linked')) + await link( + join(roots.claudeProjectsDir, 'project', 'a.jsonl'), + join(roots.claudeProjectsDir, 'linked', 'a.jsonl') + ) + const signal = new AbortController().signal + const reports: AiVaultScanWork[] = [] + let now = 0 + trackAiVaultScanProgress( + signal, + (_units, work) => reports.push(work), + () => (now += 1_000) + ) + await scanAiVaultSessions({ ...roots, platform: 'darwin', limit: 20, signal }) + return { reports, final: aiVaultScanWork(signal)! } +} + +function startScan(): { + child: AiVaultServiceTestChild + client: AiVaultScannerServiceClient + scan: Promise<{ work: AiVaultScanWork | null }> + published: (AiVaultScanProgress | null)[] +} { + const child = new AiVaultServiceTestChild() + const client = new AiVaultScannerServiceClient({ + processFactory: () => child.asChildProcess(), + init: () => ({ sessionParseCache: null, sessionSearch: null }) + }) + const published: (AiVaultScanProgress | null)[] = [] + onLocalAiVaultScanProgress(({ progress }) => published.push(progress)) + const scan = requestLocalAiVaultScan( + (listener) => + client.request<{ work: AiVaultScanWork | null }>( + { type: 'request', operation: 'scan', options: {} }, + undefined, + listener + ), + { now: () => 5_000 } + ) + return { child, client, scan, published } +} + +describe('Session History scan progress, scanner child to window', () => { + it('counts distinct files while walking, then reads, and publishes the finished totals', async () => { + const { reports, final } = await scanInChild() + 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 } + }) + + const { child, scan, published } = startScan() + readyAiVaultServiceChild(child) + await vi.waitFor(() => expect(localAiVaultScanProgress('')?.stage).toBe('finding')) + const id = aiVaultServiceRequestId(child, 'scan') + child.emit('message', { type: 'progress', id, units: 1, work: reports[0] }) + child.emit('message', { + type: 'result', + id, + operation: 'scan', + value: { result: { sessions: [], issues: [], scannedAt: '' }, durationMs: 1, work: final } + }) + await scan + + expect(published.map((progress) => progress?.stage ?? null)).toEqual([ + null, + 'finding', + reports[0]!.stage, + 'finished' + ]) + expect(localAiVaultScanProgress('')).toEqual({ + ...final, + scanId: '', + stage: 'finished', + startedAt: 5_000, + settledAt: 5_000 + }) + }) + + it('reports a scan the no-progress deadline ended as stopped on the agent still walking', async () => { + vi.useFakeTimers() + const { child, scan, published } = startScan() + const settled = scan.catch((error: Error) => error) + readyAiVaultServiceChild(child) + await vi.advanceTimersByTimeAsync(0) + const walking: AiVaultScanWork = { + stage: 'finding', + agents: [{ agent: 'codex', files: 77_483, databases: 0, walking: true }], + read: { done: 0, total: 0 } + } + child.emit('message', { + type: 'progress', + id: aiVaultServiceRequestId(child, 'scan'), + units: 1, + work: walking + }) + await vi.advanceTimersByTimeAsync(AI_VAULT_SERVICE_SCAN_TIMEOUT_MS) + + expect(await settled).toBeInstanceOf(Error) + expect(published.at(-1)).toMatchObject({ stage: 'stopped', agents: walking.agents }) + }) + + it('replaces an old stop during preparation and clears it for a later ordinary failure', () => { + const old = startLocalAiVaultScanProgress('scope', 'old', () => 1_000) + old.sent(1) + old.fail(new AiVaultServiceDeadlineError('scan timed out', 'scan', 1)) + expect(localAiVaultScanProgress('scope')?.stage).toBe('stopped') + + const replacement = startLocalAiVaultScanProgress('scope', 'replacement', () => 2_000) + old.fail(new Error('cancelled')) + expect(localAiVaultScanProgress('scope')).toBeNull() + replacement.sent(2) + expect(localAiVaultScanProgress('scope')).toMatchObject({ stage: 'finding', startedAt: 2_000 }) + + replacement.fail(new Error('service unavailable')) + expect(localAiVaultScanProgress('scope')).toBeNull() + + const interrupted = startLocalAiVaultScanProgress('scope', 'interrupted', () => 3_000) + interrupted.sent(3) + interrupted.fail(new AiVaultServiceDeadlineError('another scan timed out', 'scan', 2)) + expect(localAiVaultScanProgress('scope')).toBeNull() + }) + + it('does not call a healthy scan stopped when the interactive lane expires', async () => { + vi.useFakeTimers() + const { child, client, scan } = startScan() + const rejectedScan = scan.catch((error: Error) => error) + const interactive = client.request({ + type: 'request', + operation: 'subagents', + request: { agent: 'claude', parentFilePath: '/tmp/parent.jsonl' } + }) + const rejectedInteractive = interactive.catch((error: Error) => error) + readyAiVaultServiceChild(child) + await vi.advanceTimersByTimeAsync(0) + child.emit('message', { + type: 'progress', + id: aiVaultServiceRequestId(child, 'scan'), + units: 1, + work: { stage: 'finding', agents: [], read: { done: 0, total: 0 } } + }) + await vi.advanceTimersByTimeAsync(15_000) + + expect(await rejectedInteractive).toBeInstanceOf(AiVaultServiceDeadlineError) + expect(await rejectedScan).toBeInstanceOf(AiVaultServiceDeadlineError) + expect(localAiVaultScanProgress('')).toBeNull() + client.dispose() + }) + + it('counts inode aliases within a namespace but keeps distinct WSL distros', () => { + const tally = new AiVaultScanTally() + for (const path of [ + '\\\\wsl$\\Ubuntu\\home\\a\\one.jsonl', + '\\\\wsl.localhost\\ubuntu\\home\\a\\alias.jsonl', + '\\\\wsl$\\Debian\\home\\a\\other.jsonl', + 'C:\\Users\\a\\native.jsonl' + ]) { + tally.addFile('codex', { path, dev: 1, ino: 100 }) + } + expect(tally.snapshot().agents).toMatchObject([{ agent: 'codex', files: 3 }]) + }) +}) 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 01249bd737c..6c61b63d9a8 100644 --- a/src/main/ai-vault/ai-vault-scan-progress.test.ts +++ b/src/main/ai-vault/ai-vault-scan-progress.test.ts @@ -1,10 +1,12 @@ import { mkdtemp, rm, writeFile } from 'node:fs/promises' import { tmpdir } from 'node:os' import { join } from 'node:path' -import { afterEach, describe, expect, it } from 'vitest' +import { afterEach, describe, expect, it, vi } from 'vitest' import { AI_VAULT_SCAN_PROGRESS_INTERVAL_MS, + noteAiVaultScanFileFound, noteAiVaultScanProgress, + noteAiVaultScanWorkChanged, trackAiVaultScanProgress } from './ai-vault-scan-progress' import { scanAiVaultSessions } from './session-scanner' @@ -15,6 +17,7 @@ import { isolatedScanRoots, writeJsonlFile } from './session-scanner-test-fixtur let tempRoots: string[] = [] afterEach(async () => { + vi.useRealTimers() resetSessionParseCacheForTests() await Promise.all(tempRoots.map((root) => rm(root, { recursive: true, force: true }))) tempRoots = [] @@ -33,7 +36,7 @@ function steppingClock(): () => number { } describe('scan progress reporting', () => { - it('reports at most once per interval, and only after the counter moved', () => { + it('limits units-only reports to an advancing counter and one-second cadence', () => { let now = 0 const reports: number[] = [] const controller = new AbortController() @@ -71,6 +74,34 @@ describe('scan progress reporting', () => { expect(reports).toEqual([]) }) + it('publishes trailing files when the next read stalls without extending work units', () => { + vi.useFakeTimers() + const signal = new AbortController().signal + const reports: { units: number; files: number }[] = [] + const stop = trackAiVaultScanProgress( + signal, + (units, work) => reports.push({ units, files: work.agents[0]?.files ?? 0 }), + () => Date.now() + ) + noteAiVaultScanProgress(signal) + for (let index = 0; index < 16; index++) { + noteAiVaultScanFileFound(signal, 'codex', { + path: `/sessions/${index}.jsonl`, + dev: 1, + ino: index + 1 + }) + } + noteAiVaultScanProgress(signal) + noteAiVaultScanWorkChanged(signal) + vi.advanceTimersByTime(AI_VAULT_SCAN_PROGRESS_INTERVAL_MS) + + expect(reports).toEqual([ + { units: 1, files: 0 }, + { units: 2, files: 16 } + ]) + stop() + }) + it('counts every chunk a long transcript read consumes', async () => { const root = await tempRoot() const path = join(root, 'long.jsonl') @@ -129,6 +160,8 @@ describe('scan progress reporting', () => { expect(result.sessions).toHaveLength(3) // Two directory listings, a stat batch, then a chunk and a parse per file. expect(reports.length).toBeGreaterThanOrEqual(2 + 1 + 3 * 2) - expect(reports).toEqual(reports.map((_, index) => index + 1)) + expect([...new Set(reports)]).toEqual( + Array.from({ length: Math.max(...reports) }, (_, index) => index + 1) + ) }) }) diff --git a/src/main/ai-vault/ai-vault-scan-progress.ts b/src/main/ai-vault/ai-vault-scan-progress.ts index 16551ce0a3d..6d02238976e 100644 --- a/src/main/ai-vault/ai-vault-scan-progress.ts +++ b/src/main/ai-vault/ai-vault-scan-progress.ts @@ -1,34 +1,74 @@ +import type { AiVaultAgent } from '../../shared/ai-vault-types' +import type { AiVaultScanWork } from '../../shared/ai-vault-scan-progress' +import { AiVaultScanTally } from './ai-vault-scan-tally' + // Why: the parent's scan deadline is a no-progress window, so the child has to // say when a scan's work advanced. Keyed by the request's cancel signal, which // every work site already receives, so the index loop or an interactive read // running through the same readers can never extend a scan's deadline. -const reporters = new WeakMap void>() +const reporters = new WeakMap< + AbortSignal, + { note: () => void; workChanged: () => void; tally: AiVaultScanTally } +>() // Rate-limited because a cold scan consumes thousands of chunks and stats, and // the parent only reads the counter as "still moving" inside a 130 s window. export const AI_VAULT_SCAN_PROGRESS_INTERVAL_MS = 1_000 /** - * Reports advancing work for the scan that owns `signal`, at most once per - * interval and only when the counter moved, until the returned stop is called. + * Reports this scan's work at most once per interval until stop is called. + * A trailing tally can keep the same units; only advancing units re-arm the deadline. */ export function trackAiVaultScanProgress( signal: AbortSignal, - report: (units: number) => void, + report: (units: number, work: AiVaultScanWork) => void, now: () => number = () => performance.now() ): () => void { let units = 0 let lastReportedAt = Number.NEGATIVE_INFINITY - reporters.set(signal, () => { - units++ - const at = now() - if (at - lastReportedAt < AI_VAULT_SCAN_PROGRESS_INTERVAL_MS) { + let trailing: ReturnType | null = null + const tally = new AiVaultScanTally() + const publish = (): void => { + lastReportedAt = now() + report(units, tally.snapshot()) + } + const workChanged = (): void => { + if (units === 0) { return } - lastReportedAt = at - report(units) + if (trailing) { + return + } + const wait = Math.max(0, AI_VAULT_SCAN_PROGRESS_INTERVAL_MS - (now() - lastReportedAt)) + if (wait === 0) { + publish() + return + } + trailing = setTimeout(() => { + trailing = null + publish() + }, wait) + trailing.unref?.() + } + reporters.set(signal, { + tally, + workChanged, + note: () => { + units++ + if (now() - lastReportedAt < AI_VAULT_SCAN_PROGRESS_INTERVAL_MS) { + return + } + if (trailing) { + clearTimeout(trailing) + trailing = null + } + publish() + } }) return () => { + if (trailing) { + clearTimeout(trailing) + } reporters.delete(signal) } } @@ -36,6 +76,70 @@ export function trackAiVaultScanProgress( /** Counts one unit of real work for the scan that owns `signal`; a no-op for any other work. */ export function noteAiVaultScanProgress(signal: AbortSignal | undefined): void { if (signal) { - reporters.get(signal)?.() + reporters.get(signal)?.note() + } +} + +function tallyFor(signal: AbortSignal | undefined): AiVaultScanTally | undefined { + return signal ? reporters.get(signal)?.tally : undefined +} + +/** What the scan owning `signal` has found and read; null for untracked work. */ +export function aiVaultScanWork(signal: AbortSignal | undefined): AiVaultScanWork | null { + return tallyFor(signal)?.snapshot() ?? null +} + +/** Marks one of an agent's folders or databases as being listed until the returned close. */ +export function openAiVaultScanSource( + signal: AbortSignal | undefined, + agent: AiVaultAgent +): () => void { + const reporter = signal ? reporters.get(signal) : undefined + if (!reporter) { + return () => undefined + } + const close = reporter.tally.openSource(agent) + reporter.workChanged() + return () => { + close() + reporter.workChanged() + } +} + +export function noteAiVaultScanFileFound( + signal: AbortSignal | undefined, + agent: AiVaultAgent, + file: { path: string; dev?: number; ino?: number } +): void { + tallyFor(signal)?.addFile(agent, file) +} + +export function noteAiVaultScanDatabases( + signal: AbortSignal | undefined, + agent: AiVaultAgent, + count: number +): void { + const reporter = signal ? reporters.get(signal) : undefined + reporter?.tally.addDatabases(agent, count) + reporter?.workChanged() +} + +/** Discovery is over: the scan now reads `total` more transcripts. */ +export function noteAiVaultScanReading(signal: AbortSignal | undefined, total: number): void { + const reporter = signal ? reporters.get(signal) : undefined + reporter?.tally.startReading(total) + reporter?.workChanged() +} + +export function noteAiVaultScanFilesRead(signal: AbortSignal | undefined, count: number): void { + const reporter = signal ? reporters.get(signal) : undefined + reporter?.tally.addRead(count) + reporter?.workChanged() +} + +/** Sends a trailing tally without advancing the scanner's no-progress deadline. */ +export function noteAiVaultScanWorkChanged(signal: AbortSignal | undefined): void { + if (signal) { + reporters.get(signal)?.workChanged() } } diff --git a/src/main/ai-vault/ai-vault-scan-tally.ts b/src/main/ai-vault/ai-vault-scan-tally.ts new file mode 100644 index 00000000000..18f2acbf766 --- /dev/null +++ b/src/main/ai-vault/ai-vault-scan-tally.ts @@ -0,0 +1,89 @@ +import { AI_VAULT_AGENTS, type AiVaultAgent } from '../../shared/ai-vault-types' +import type { AiVaultScanWork } from '../../shared/ai-vault-scan-progress' +import { sessionFileExecutionNamespace } from './codex-session-root-dedup' + +type AgentTally = { + files: Set + databases: number + openSources: number +} + +/** What one scan has found and read so far, for the progress it reports. */ +export class AiVaultScanTally { + private readonly agents = new Map() + private stage: AiVaultScanWork['stage'] = 'finding' + private readDone = 0 + private readTotal = 0 + + openSource(agent: AiVaultAgent): () => void { + const tally = this.agent(agent) + tally.openSources++ + let open = true + return () => { + if (open) { + open = false + tally.openSources-- + } + } + } + + addFile(agent: AiVaultAgent, file: { path: string; dev?: number; ino?: number }): void { + // Why: a hard link is one file within an execution namespace; WSL distros + // can reuse the same device and inode numbers for unrelated files. + const identity = + typeof file.dev === 'number' && + typeof file.ino === 'number' && + Number.isSafeInteger(file.dev) && + Number.isSafeInteger(file.ino) && + !(file.dev === 0 && file.ino === 0) + ? `${sessionFileExecutionNamespace(file.path)}:${file.dev}:${file.ino}` + : file.path + this.agent(agent).files.add(identity) + } + + addDatabases(agent: AiVaultAgent, count: number): void { + this.agent(agent).databases += count + } + + startReading(total: number): void { + this.stage = 'reading' + this.readTotal += total + } + + addRead(count: number): void { + this.readDone += count + } + + snapshot(): AiVaultScanWork { + const agents: AiVaultScanWork['agents'] = [] + for (const [agent, tally] of this.agents) { + if (tally.files.size > 0 || tally.databases > 0 || tally.openSources > 0) { + agents.push({ + agent, + files: tally.files.size, + databases: tally.databases, + walking: tally.openSources > 0 + }) + } + } + // Discovery settles in any order; a fixed order keeps the rows from jumping. + agents.sort((a, b) => AI_VAULT_AGENTS.indexOf(a.agent) - AI_VAULT_AGENTS.indexOf(b.agent)) + return { + stage: this.stage, + agents, + read: { + done: Math.min(this.readDone, this.readTotal), + total: this.readTotal + } + } + } + + private agent(agent: AiVaultAgent): AgentTally { + let tally = this.agents.get(agent) + if (!tally) { + tally = { files: new Set(), databases: 0, openSources: 0 } + this.agents.set(agent, tally) + } + return tally + } +} diff --git a/src/main/ai-vault/cached-session-list-wsl-probe.test.ts b/src/main/ai-vault/cached-session-list-wsl-probe.test.ts index c44f10eeb2d..30397f26967 100644 --- a/src/main/ai-vault/cached-session-list-wsl-probe.test.ts +++ b/src/main/ai-vault/cached-session-list-wsl-probe.test.ts @@ -70,7 +70,13 @@ describe('AI Vault listing wsl.exe probes', () => { additionalCodexSessionsDirs: [join(NATIVE_CODEX_HOME, 'sessions')], wslHomeDirs: [] }), - expect.anything() + expect.anything(), + expect.objectContaining({ + sent: expect.any(Function), + work: expect.any(Function), + finish: expect.any(Function), + fail: expect.any(Function) + }) ) }) @@ -91,7 +97,13 @@ describe('AI Vault listing wsl.exe probes', () => { ]) expect(scanAiVaultSessionsInService).toHaveBeenCalledWith( expect.objectContaining({ wslHomeDirs: [WSL_HOME] }), - expect.anything() + expect.anything(), + expect.objectContaining({ + sent: expect.any(Function), + work: expect.any(Function), + finish: expect.any(Function), + fail: expect.any(Function) + }) ) runProcessMock.mockResolvedValue(FAILED) diff --git a/src/main/ai-vault/cached-session-list.test.ts b/src/main/ai-vault/cached-session-list.test.ts index 55801c6df83..9fcdeb7dadf 100644 --- a/src/main/ai-vault/cached-session-list.test.ts +++ b/src/main/ai-vault/cached-session-list.test.ts @@ -90,12 +90,14 @@ describe('invalidateAiVaultSessionListCache generation guard', () => { it('caches normally when no invalidation interrupts the scan', async () => { scanAiVaultSessionsInService.mockResolvedValueOnce(scanResult('scan-A')) - await listAiVaultSessions() + const request = { limit: 1000, scopePaths: ['/project'] } + const first = await listAiVaultSessions(request) - // Second non-force call is a cache hit — no second scan. - const cached = await listAiVaultSessions() + // A narrower request reuses the same scan entry and its scan ID. + const cached = await listAiVaultSessions({ ...request, limit: 100 }) expect(cached.scannedAt).toBe('scan-A') + expect(cached.scanId).toBe(first.scanId) expect(scanAiVaultSessionsInService).toHaveBeenCalledTimes(1) expect(listRunningWslHomeDirsAsync).toHaveBeenCalledTimes(1) }) diff --git a/src/main/ai-vault/cached-session-list.ts b/src/main/ai-vault/cached-session-list.ts index 7fd66d7bfe0..08dc763879a 100644 --- a/src/main/ai-vault/cached-session-list.ts +++ b/src/main/ai-vault/cached-session-list.ts @@ -7,11 +7,20 @@ import { } from './session-scanner-background' import { getCachedWslDistros, hasCachedWslDistros, listRunningWslHomeDirsAsync } from '../wsl' import { filterPathsToRunningWslDistrosAsync } from '../wsl-running-path-filter' -import type { AiVaultListArgs, AiVaultListResult } from '../../shared/ai-vault-types' +import { + isAiVaultScanCancelledError, + type AiVaultListArgs, + type AiVaultListResult +} from '../../shared/ai-vault-types' +import { aiVaultLocalScanKey, aiVaultLocalScanScopeKey } from '../../shared/ai-vault-local-scan-key' import type { AiVaultScanOptions } from './session-scanner-types' import { prepareOpenCodeWslReaders } from './opencode-wsl-runtime-preparation' import { LOCAL_EXECUTION_HOST_ID } from '../../shared/execution-host' import { AiVaultScanCoordinator } from './ai-vault-scan-coordinator' +import { + startLocalAiVaultScanProgress, + type LocalAiVaultScanProgressHandle +} from './ai-vault-local-scan-progress' import { aiVaultSessionDepthCovers, requestedAiVaultSessionDepth, @@ -24,6 +33,15 @@ import { // mobile screen for the same scope must not double-scan hundreds of transcripts. const AI_VAULT_CACHE_TTL_MS = 60_000 +export class AiVaultLocalScanError extends Error { + constructor( + readonly scanId: string, + readonly original: unknown + ) { + super(original instanceof Error ? original.message : String(original)) + } +} + // Why: codex-home + WSL home dirs must be sourced from a serve-mode-reachable // seam (the OrcaRuntimeService deps), NOT the window-only registerCoreHandlers // path — `orca serve` never runs that path, so sourcing it there would silently @@ -88,12 +106,9 @@ export async function listAiVaultSessions( options: { signal?: AbortSignal } = {} ): Promise { // Scope paths change the result set, so they must be part of the cache key. - const key = JSON.stringify({ - scopePaths: [...new Set(args?.scopePaths ?? [])].sort(), - includeAntigravityIdeSessions: args?.includeAntigravityIdeSessions === true - }) + const key = aiVaultLocalScanScopeKey(args) const depth = requestedAiVaultSessionDepth(args) - const scanKey = JSON.stringify({ key, depth }) + const scanKey = aiVaultLocalScanKey(args) const now = Date.now() if (args?.force === true) { clearAiVaultBackgroundRestartCircuit() @@ -112,12 +127,23 @@ export async function listAiVaultSessions( // microtask, so an invalidation landing in that gap would otherwise be read // as having happened before this scan and leave the stale result cacheable. const startGeneration = cacheGeneration + let progress: LocalAiVaultScanProgressHandle | null = null return scanCoordinator.run({ key: scanKey, force: args?.force, signal: options.signal, - start: async (scanSignal) => { - const result = await scanAiVaultSessionsInBackground( + onCreate: (scanId) => { + progress = startLocalAiVaultScanProgress(scanKey, scanId) + }, + onError: (error) => progress?.fail(error), + onReject: (error, scanId) => + isAiVaultScanCancelledError(error) ? error : new AiVaultLocalScanError(scanId, error), + start: async (scanSignal, scanId) => { + const scanProgress = progress + if (!scanProgress) { + throw new Error('Local scan progress was not created.') + } + const scanned = await scanAiVaultSessionsInBackground( { includeAntigravityIdeSessions: args?.includeAntigravityIdeSessions, limit: args?.limit, @@ -125,8 +151,10 @@ export async function listAiVaultSessions( scopePaths: args?.scopePaths, ...(await localAiVaultScanRoots()) }, - scanSignal + scanSignal, + scanProgress ) + const result = { ...scanned, scanId } // A delete (or other invalidation) landed while this scan was running: // its result predates the delete, so caching it would resurrect the // deleted session for the TTL. Return it to this caller but don't cache. diff --git a/src/main/ai-vault/codex-session-root-dedup.ts b/src/main/ai-vault/codex-session-root-dedup.ts index 061a05aaf8e..3d63c77ef02 100644 --- a/src/main/ai-vault/codex-session-root-dedup.ts +++ b/src/main/ai-vault/codex-session-root-dedup.ts @@ -23,7 +23,7 @@ function lastPathSegment(filePath: string): string { // Why: local Windows discovery scans both the host and every WSL distro under // one `local` host id, even though hardlinks and resume identity cannot cross // those execution boundaries. -function codexPathExecutionNamespace(filePath: string): string { +export function sessionFileExecutionNamespace(filePath: string): string { const wslPath = parseWslUncPath(filePath) return wslPath ? `wsl:${wslPath.distro.toLowerCase()}` : 'native' } @@ -99,7 +99,7 @@ export function dedupeCodexRolloutFileAliases( if (!hardlinkIdentity) { continue } - const aliasKey = `${codexPathExecutionNamespace(filePath)}\0${fileName}\0${hardlinkIdentity}` + const aliasKey = `${sessionFileExecutionNamespace(filePath)}\0${fileName}\0${hardlinkIdentity}` const rank = codexSessionRootRank(accessors.getCodexHome(candidate)) const best = bestByAlias.get(aliasKey) if (!best || rank < best.rank || (rank === best.rank && filePath < best.filePath)) { @@ -116,7 +116,7 @@ export function dedupeCodexRolloutFileAliases( return true } const best = bestByAlias.get( - `${codexPathExecutionNamespace(accessors.getFilePath(candidate))}\0${fileName}\0${hardlinkIdentity}` + `${sessionFileExecutionNamespace(accessors.getFilePath(candidate))}\0${fileName}\0${hardlinkIdentity}` ) return !best || best.candidate === candidate }) @@ -167,7 +167,7 @@ export async function dedupeCodexRolloutCopyAliases( if (!CODEX_ROLLOUT_FILE_NAME_PATTERN.test(fileName)) { continue } - const key = `${codexPathExecutionNamespace(filePath)}\0${fileName}` + const key = `${sessionFileExecutionNamespace(filePath)}\0${fileName}` const group = groups.get(key) if (group) { group.push(candidate) @@ -231,7 +231,7 @@ export function codexSessionAliasKey(session: AiVaultSession): string | null { if (!CODEX_ROLLOUT_FILE_NAME_PATTERN.test(fileName)) { return null } - return `${session.executionHostId}\0${codexPathExecutionNamespace(session.filePath)}\0${session.sessionId}\0${fileName}` + return `${session.executionHostId}\0${sessionFileExecutionNamespace(session.filePath)}\0${session.sessionId}\0${fileName}` } export function codexSessionAliasBeats(candidate: AiVaultSession, best: AiVaultSession): boolean { diff --git a/src/main/ai-vault/session-cwd-bucket-layouts.ts b/src/main/ai-vault/session-cwd-bucket-layouts.ts index 3f0195cdcd4..59c2abaa31c 100644 --- a/src/main/ai-vault/session-cwd-bucket-layouts.ts +++ b/src/main/ai-vault/session-cwd-bucket-layouts.ts @@ -66,3 +66,11 @@ export const CODEBUDDY_CWD_BUCKET_LAYOUT: CwdBucketLayout = { // This is only a prefilter: each transcript's cwd decides membership, including lossy collisions. isDirInScope: (name, prefixes) => [...prefixes].some((prefix) => name.startsWith(prefix)) } + +// Agents whose on-disk layout names a directory per cwd, so a scope's older +// sessions can be found without reading every transcript's header. +export const CWD_BUCKET_LAYOUTS = [ + CLAUDE_CWD_BUCKET_LAYOUT, + PI_CWD_BUCKET_LAYOUT, + CODEBUDDY_CWD_BUCKET_LAYOUT +] diff --git a/src/main/ai-vault/session-scanner-background.ts b/src/main/ai-vault/session-scanner-background.ts index 3a13ca04c51..92f77004362 100644 --- a/src/main/ai-vault/session-scanner-background.ts +++ b/src/main/ai-vault/session-scanner-background.ts @@ -21,6 +21,7 @@ import type { AiVaultServiceSubagentRequest } from './session-scanner-service-protocol' import { isWslUncPath } from '../../shared/wsl-paths' +import type { LocalAiVaultScanProgressHandle } from './ai-vault-local-scan-progress' // Let forced refreshes retry after a local service circuit opens. export function clearAiVaultBackgroundRestartCircuit(): void { @@ -29,9 +30,10 @@ export function clearAiVaultBackgroundRestartCircuit(): void { export function scanAiVaultSessionsInBackground( options: AiVaultServiceScanOptions, - signal?: AbortSignal + signal?: AbortSignal, + progress?: LocalAiVaultScanProgressHandle ): Promise { - return scanAiVaultSessionsInService(options, signal) + return scanAiVaultSessionsInService(options, signal, progress) } export function resolveAiVaultSessionTitlesInBackground( diff --git a/src/main/ai-vault/session-scanner-discovery.ts b/src/main/ai-vault/session-scanner-discovery.ts index 24ba97ebf4c..bfa3a9603b6 100644 --- a/src/main/ai-vault/session-scanner-discovery.ts +++ b/src/main/ai-vault/session-scanner-discovery.ts @@ -12,7 +12,12 @@ import { recordSessionScanIssue } from './session-scan-issues' import type { FileWithMtime, SessionFileDiscovery } from './session-scanner-types' import { errorMessage } from './session-scanner-values' import { throwIfAiVaultScanCancelled } from './ai-vault-scan-cancellation' -import { noteAiVaultScanProgress } from './ai-vault-scan-progress' +import { + noteAiVaultScanFileFound, + noteAiVaultScanProgress, + noteAiVaultScanWorkChanged, + openAiVaultScanSource +} from './ai-vault-scan-progress' const NATIVE_DISCOVERY_CONCURRENCY = 4 const NATIVE_DISCOVERY_BATCH_SIZE = 16 @@ -47,16 +52,20 @@ export async function discoverFiles(args: { let paths: string[] = [] let pending: Promise | null = null const native = !isWslUncPath(args.rootDir) + const source = { close: (): void => undefined } function consume(observations: readonly SessionFileObservation[]): void { // A stat the cancel cut short is not an unreadable file; never record it. throwIfAiVaultScanCancelled(args.signal) - noteAiVaultScanProgress(args.signal) for (const observation of observations) { if ('issue' in observation) { - recordSessionScanIssue(args.issues, { agent: args.agent, ...observation.issue }) + recordSessionScanIssue(args.issues, { + agent: args.agent, + ...observation.issue + }) continue } + noteAiVaultScanFileFound(args.signal, args.agent, observation.file) if (observation.file.sidecar === 'unknown' && !refusedSidecar) { refusedSidecar = true recordSessionScanIssue(args.issues, { @@ -68,7 +77,10 @@ export async function discoverFiles(args: { try { const modifiedAt = new Date(observation.file.mtimeMs) if (Number.isNaN(modifiedAt.getTime()) || files.wouldRetain(observation.file.mtimeMs)) { - files.add({ ...observation.file, modifiedAt: modifiedAt.toISOString() }) + files.add({ + ...observation.file, + modifiedAt: modifiedAt.toISOString() + }) } } catch (err) { recordSessionScanIssue(args.issues, { @@ -78,6 +90,8 @@ export async function discoverFiles(args: { }) } } + noteAiVaultScanProgress(args.signal) + noteAiVaultScanWorkChanged(args.signal) } async function finishPending(): Promise { @@ -121,6 +135,17 @@ export async function discoverFiles(args: { return observations.flat() } + async function readDirectory(path: string): Promise { + const entries = native + ? await withNativeDiscoverySlot(() => wslGatedReaddir(path, 'scan', args.signal)) + : await wslGatedReaddir(path, 'scan', args.signal) + // A listed root is a history folder being walked; a missing one never shows. + if (path === args.rootDir) { + source.close = openAiVaultScanSource(args.signal, args.agent) + } + return entries + } + try { try { await forEachSessionFile( @@ -132,12 +157,7 @@ export async function discoverFiles(args: { filePredicate: args.filePredicate, directoryPredicate: args.directoryPredicate, signal: args.signal, - ...(native - ? { - readDirectory: (path: string) => - withNativeDiscoverySlot(() => wslGatedReaddir(path, 'scan', args.signal)) - } - : {}) + readDirectory }, native ? queueNativeFile @@ -165,6 +185,8 @@ export async function discoverFiles(args: { message: err.message }) return { agent: args.agent, rootDir: args.rootDir, files: [] } + } finally { + source.close() } return { agent: args.agent, rootDir: args.rootDir, files: files.newest() } } @@ -211,7 +233,11 @@ async function observeSessionSidecar( } try { const fileStat = await wslGatedStat(filePath, 'scan', signal) - return { path: filePath, mtimeMs: fileStat.mtimeMs, sizeBytes: fileStat.size } + return { + path: filePath, + mtimeMs: fileStat.mtimeMs, + sizeBytes: fileStat.size + } } catch (error) { return isMissingSidecarError(error) ? 'none' : 'unknown' } diff --git a/src/main/ai-vault/session-scanner-opencode-sqlite-worker-spawn.ts b/src/main/ai-vault/session-scanner-opencode-sqlite-worker-spawn.ts index 7e6f5eb31e0..d8d51aab1fb 100644 --- a/src/main/ai-vault/session-scanner-opencode-sqlite-worker-spawn.ts +++ b/src/main/ai-vault/session-scanner-opencode-sqlite-worker-spawn.ts @@ -17,6 +17,7 @@ import { } from './session-scanner-opencode-wsl-client' import { findForeignSqliteReaderEntry } from '../foreign-sqlite-readers/foreign-sqlite-reader-entry-path' import { runOpenCodeSqliteScanRequest } from './session-scanner-opencode-sqlite-scan-scope' +import { noteAiVaultScanDatabases, openAiVaultScanSource } from './ai-vault-scan-progress' // Why: resolve the built worker entry + own the process-wide shared client so // the client class stays free of Electron (require'd lazily here) and the @@ -43,7 +44,9 @@ function defaultWorkerFactory(): Worker { let sharedClient: OpenCodeSqliteWorkerClient | null = null function getSharedClient(): OpenCodeSqliteWorkerClient { - sharedClient ??= new OpenCodeSqliteWorkerClient({ workerFactory: defaultWorkerFactory }) + sharedClient ??= new OpenCodeSqliteWorkerClient({ + workerFactory: defaultWorkerFactory + }) return sharedClient } @@ -159,6 +162,22 @@ export function captureZcodeSqliteSessionViaWorker(args: { async function listForHost( args: Parameters[0] +): Promise { + const agent = args.agent ?? 'opencode' + const closeSource = args.dbPaths.length > 0 ? openAiVaultScanSource(args.signal, agent) : null + try { + const candidates = await listCandidatesForHost(args) + // A database counts once it holds this agent's sessions: OpenCode 1 and 2 share one file. + const dbPaths = candidates.map((c) => splitOpenCodeSqliteCandidate(c.file.path, agent)?.dbPath) + noteAiVaultScanDatabases(args.signal, agent, new Set(dbPaths.filter(Boolean)).size) + return candidates + } finally { + closeSource?.() + } +} + +async function listCandidatesForHost( + args: Parameters[0] ): Promise { throwIfSignalAborted(args.signal) const native: string[] = [] @@ -170,7 +189,10 @@ async function listForHost( continue } const key = wsl.distro.toLowerCase() - const group = groups.get(key) ?? { distro: wsl.distro, paths: new Map() } + const group = groups.get(key) ?? { + distro: wsl.distro, + paths: new Map() + } group.paths.set(wsl.linuxPath, path) groups.set(key, group) } @@ -184,7 +206,12 @@ async function listForHost( args.signal, async (signal) => { const client = await openCodeWslClient(distro, first, signal) - return client.list({ ...args, signal, dbPaths: [...paths.keys()], issues }) + return client.list({ + ...args, + signal, + dbPaths: [...paths.keys()], + issues + }) }, args.agent ) @@ -214,7 +241,10 @@ async function listForHost( return [] } finally { args.issues.push( - ...issues.map((issue) => ({ ...issue, path: paths.get(issue.path) ?? issue.path })) + ...issues.map((issue) => ({ + ...issue, + path: paths.get(issue.path) ?? issue.path + })) ) } }) @@ -234,7 +264,12 @@ async function parseForHost( args.signal, async (signal) => { const client = await openCodeWslClient(wsl.distro, args.dbPath, signal) - return client.parse({ ...args, signal, dbPath: wsl.linuxPath, platform: 'linux' }) + return client.parse({ + ...args, + signal, + dbPath: wsl.linuxPath, + platform: 'linux' + }) }, args.agent ) @@ -252,11 +287,19 @@ async function captureForHost( args.signal, async (signal) => { const client = await openCodeWslClient(wsl.distro, args.dbPath, signal) - return client.capture({ ...args, signal, dbPath: wsl.linuxPath, platform: 'linux' }) + return client.capture({ + ...args, + signal, + dbPath: wsl.linuxPath, + platform: 'linux' + }) }, args.agent ) - return { ...capture, session: mapOpenCodeWslSession(capture.session, args.dbPath) } + return { + ...capture, + session: mapOpenCodeWslSession(capture.session, args.dbPath) + } } export async function readOpenCodeTranscriptPageViaWorker( diff --git a/src/main/ai-vault/session-scanner-service-client-state.ts b/src/main/ai-vault/session-scanner-service-client-state.ts index f758595d560..9a2f830bc4a 100644 --- a/src/main/ai-vault/session-scanner-service-client-state.ts +++ b/src/main/ai-vault/session-scanner-service-client-state.ts @@ -1,6 +1,7 @@ import type { SessionSearchScanRoots } from '../ai-vault-search/session-search-scan-roots' import type { ChildProcess } from 'node:child_process' import { createAiVaultScanCancelledError } from './ai-vault-scan-cancellation' +import type { AiVaultServiceScanListener } from './session-scanner-service-request-deadline' import { AI_VAULT_SERVICE_PROTOCOL_VERSION, aiVaultServiceLane, @@ -143,10 +144,12 @@ export function createAiVaultServicePendingCall( signal: AbortSignal | undefined, resolve: (value: unknown) => void, reject: (error: Error) => void, - cancel: (call: AiVaultServicePendingCall) => void + cancel: (call: AiVaultServicePendingCall) => void, + scanListener?: AiVaultServiceScanListener ): AiVaultServicePendingCall { const call: AiVaultServicePendingCall = { request, + scanListener, lane: aiVaultServiceLane(request.operation), signal, resolve, @@ -292,6 +295,7 @@ export type AiVaultServicePendingCall = { startRetried: boolean /** Highest progress counter the child reported for this scan. */ progressUnits: number + scanListener?: AiVaultServiceScanListener } export type AiVaultServiceReadyWaiter = { diff --git a/src/main/ai-vault/session-scanner-service-client.ts b/src/main/ai-vault/session-scanner-service-client.ts index 5367c6fbc0d..a61369ca49c 100644 --- a/src/main/ai-vault/session-scanner-service-client.ts +++ b/src/main/ai-vault/session-scanner-service-client.ts @@ -23,8 +23,10 @@ import { } from './session-scanner-service-client-state' import { noteAiVaultServiceScanProgress, - sendAiVaultServiceCall + sendAiVaultServiceCall, + type AiVaultServiceScanListener } from './session-scanner-service-request-deadline' +import { answerAiVaultServiceSessionSearchRoots } from './session-scanner-service-root-answer' import { AiVaultServiceRestartPolicy } from './session-scanner-service-restart-policy' import { isAiVaultServiceChildMessage, @@ -48,7 +50,11 @@ export class AiVaultScannerServiceClient { constructor(private readonly options: AiVaultServiceClientOptions) {} - request(body: AiVaultServiceRequestBody, signal?: AbortSignal): Promise { + request( + body: AiVaultServiceRequestBody, + signal?: AbortSignal, + scanListener?: AiVaultServiceScanListener + ): Promise { if (this.disposed) { return Promise.reject(new Error('AI Vault service client was disposed.')) } @@ -66,7 +72,8 @@ export class AiVaultScannerServiceClient { // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: callers choose T as the result value of the operation they send, which is what the child answers with. resolve as (value: unknown) => void, reject, - (aborted) => this.cancel(aborted) + (aborted) => this.cancel(aborted), + scanListener ) this.queue.push(call) this.idleRetirement.clear() @@ -218,16 +225,7 @@ export class AiVaultScannerServiceClient { return } if (message.type === 'sessionSearchRoots') { - const child = this.child - const resolve = this.options.resolveSessionSearchRoots - void Promise.resolve() - .then(() => (resolve ? resolve() : (this.options.init().sessionSearch?.roots ?? null))) - .catch(() => null) - .then((roots) => { - if (child && this.child === child && child.connected) { - child.send({ type: 'sessionSearchRoots', id: message.id, roots }, () => undefined) - } - }) + answerAiVaultServiceSessionSearchRoots(this.child, message.id, this.options, () => this.child) return } if (message.type === 'ready') { @@ -251,7 +249,7 @@ export class AiVaultScannerServiceClient { return } if (message.type === 'progress') { - noteAiVaultServiceScanProgress(call, message.units, this.onDeadline) + noteAiVaultServiceScanProgress(call, message, this.onDeadline) return } this.active.delete(call.lane) diff --git a/src/main/ai-vault/session-scanner-service-entry.test.ts b/src/main/ai-vault/session-scanner-service-entry.test.ts index 458ba6f5e0c..d45eaffbe53 100644 --- a/src/main/ai-vault/session-scanner-service-entry.test.ts +++ b/src/main/ai-vault/session-scanner-service-entry.test.ts @@ -114,9 +114,10 @@ describe('AI Vault service entry cache invalidation', () => { now += 1_000 noteAiVaultScanProgress(signal) const progress = (): typeof sent => sent.filter((message) => message.type === 'progress') + const work = expect.objectContaining({ stage: 'finding' }) expect(progress()).toEqual([ - { type: 'progress', id: 20, units: 1 }, - { type: 'progress', id: 20, units: 3 } + { type: 'progress', id: 20, units: 1, work }, + { type: 'progress', id: 20, units: 3, work } ]) releaseScan?.() diff --git a/src/main/ai-vault/session-scanner-service-entry.ts b/src/main/ai-vault/session-scanner-service-entry.ts index b1069757ae9..ad77bbd2c9d 100644 --- a/src/main/ai-vault/session-scanner-service-entry.ts +++ b/src/main/ai-vault/session-scanner-service-entry.ts @@ -1,5 +1,6 @@ import { requestSessionSearchRoots } from './session-scanner-service-root-request' -import { trackAiVaultScanProgress } from './ai-vault-scan-progress' +import { aiVaultScanWork, trackAiVaultScanProgress } from './ai-vault-scan-progress' +import type { AiVaultScanWork } from '../../shared/ai-vault-scan-progress' import type { AiVaultSessionTitle } from '../../shared/ai-vault-session-title' import type { AiVaultListResult } from '../../shared/ai-vault-types' import { readAiVaultFirstUserPrompt } from './session-first-user-prompt-read' @@ -50,12 +51,13 @@ function titleKey(request: { agent: string; sessionId: string }): string { async function scanReportingProgress( request: Extract, signal: AbortSignal -): Promise { - const stopProgress = trackAiVaultScanProgress(signal, (units) => - send({ type: 'progress', id: request.id, units }) +): Promise<{ result: AiVaultListResult; work: AiVaultScanWork | null }> { + const stopProgress = trackAiVaultScanProgress(signal, (units, work) => + send({ type: 'progress', id: request.id, units, work }) ) try { - return await scanAiVaultSessions({ ...request.options, signal }) + const result = await scanAiVaultSessions({ ...request.options, signal }) + return { result, work: aiVaultScanWork(signal) } } finally { stopProgress() } @@ -106,7 +108,7 @@ async function executeRequest(request: AiVaultServiceRequest): Promise { if (raw.sessionSearch) { sessionSearch.apply(raw.sessionSearch) } - send({ type: 'ready', protocol: AI_VAULT_SERVICE_PROTOCOL_VERSION, pid: process.pid }) + send({ + type: 'ready', + protocol: AI_VAULT_SERVICE_PROTOCOL_VERSION, + pid: process.pid + }) return } if (!initialized || shuttingDown) { diff --git a/src/main/ai-vault/session-scanner-service-protocol.ts b/src/main/ai-vault/session-scanner-service-protocol.ts index 5c494c1e56f..9273f6c1ebe 100644 --- a/src/main/ai-vault/session-scanner-service-protocol.ts +++ b/src/main/ai-vault/session-scanner-service-protocol.ts @@ -15,6 +15,7 @@ import type { SessionSearchScanRoots } from '../ai-vault-search/session-search-s import type { ReadAiVaultFirstUserPromptArgs } from './session-first-user-prompt-read' import type { SessionParseCachePersistenceOptions } from './session-parse-cache-persistence' import type { AiVaultScanOptions } from './session-scanner-types' +import type { AiVaultScanWork } from '../../shared/ai-vault-scan-progress' export const AI_VAULT_SERVICE_PROTOCOL_VERSION = 1 @@ -105,11 +106,23 @@ export type AiVaultServiceParentMessage = | { type: 'invalidate'; generation: number; paths: string[] } // Fire-and-forget: the child closes the live pair and constructs from this. | { type: 'sessionSearch'; init: AiVaultSessionSearchInit } - | { type: 'sessionSearchRoots'; id: number; roots: SessionSearchScanRoots | null } + | { + type: 'sessionSearchRoots' + id: number + roots: SessionSearchScanRoots | null + } | { type: 'shutdown' } export type AiVaultServiceResultValue = - | { operation: 'scan'; value: { result: AiVaultListResult; durationMs: number } } + | { + operation: 'scan' + // `work` is the scan's final tally, which the throttled progress reports may have skipped. + value: { + result: AiVaultListResult + durationMs: number + work: AiVaultScanWork | null + } + } | { operation: 'titles'; value: AiVaultSessionTitlesResult } | { operation: 'subagents'; value: AiVaultSubagentListResult } | { operation: 'firstPrompt'; value: { prompt: string | null } } @@ -129,7 +142,7 @@ export type AiVaultServiceChildMessage = | { type: 'error'; id: number; message: string; retryable: boolean } | { type: 'invalidated'; generation: number } // `units` only grows within one scan; the parent re-arms that scan's deadline on an advance. - | { type: 'progress'; id: number; units: number } + | { type: 'progress'; id: number; units: number; work?: AiVaultScanWork } /** Everything but the two bulk reads is interactive: a search must not queue behind a scan. */ export function aiVaultServiceLane(operation: AiVaultServiceOperation): AiVaultServiceLane { diff --git a/src/main/ai-vault/session-scanner-service-request-deadline.test.ts b/src/main/ai-vault/session-scanner-service-request-deadline.test.ts index 164ed0a67c8..d180b449734 100644 --- a/src/main/ai-vault/session-scanner-service-request-deadline.test.ts +++ b/src/main/ai-vault/session-scanner-service-request-deadline.test.ts @@ -108,6 +108,29 @@ describe('AI Vault service scan watchdog', () => { client.dispose() }) + it('forwards a trailing work snapshot without extending the scan deadline', async () => { + const { children, client } = setup() + const work = vi.fn() + const scan = client.request(SCAN, undefined, { sent: () => undefined, work }) + const child = children[0]! + readyAiVaultServiceChild(child) + await Promise.resolve() + const id = aiVaultServiceRequestId(child, 'scan') + const rejected = expect(scan).rejects.toThrow(TIMED_OUT) + progress(child, id, 1) + vi.advanceTimersByTime(AI_VAULT_SERVICE_SCAN_TIMEOUT_MS - 1) + child.emit('message', { + type: 'progress', + id, + units: 1, + work: { stage: 'finding', agents: [], read: { done: 0, total: 0 } } + }) + expect(work).toHaveBeenCalledOnce() + vi.advanceTimersByTime(1) + await rejected + client.dispose() + }) + it('ignores progress for another request id', async () => { const { children, client } = setup() const { scan, child, id } = await startScan(client, children) diff --git a/src/main/ai-vault/session-scanner-service-request-deadline.ts b/src/main/ai-vault/session-scanner-service-request-deadline.ts index d8816b81249..255d260e987 100644 --- a/src/main/ai-vault/session-scanner-service-request-deadline.ts +++ b/src/main/ai-vault/session-scanner-service-request-deadline.ts @@ -1,10 +1,28 @@ import type { AiVaultServicePendingCall } from './session-scanner-service-client-state' import type { AiVaultServiceRequest } from './session-scanner-service-protocol' +import type { AiVaultScanWork } from '../../shared/ai-vault-scan-progress' /** No-progress window: a scan whose work keeps advancing re-arms it, however long it runs. */ export const AI_VAULT_SERVICE_SCAN_TIMEOUT_MS = 130_000 export const AI_VAULT_SERVICE_INTERACTIVE_TIMEOUT_MS = 15_000 +/** The named call exhausted its deadline; a shared child fault may reject other calls too. */ +export class AiVaultServiceDeadlineError extends Error { + constructor( + message: string, + readonly operation: AiVaultServiceRequest['operation'], + readonly requestId: number + ) { + super(message) + } +} + +/** Follows one scan request: when the child received it, and each report of its work. */ +export type AiVaultServiceScanListener = { + sent: (requestId: number) => void + work: (work: AiVaultScanWork) => void +} + function armAiVaultServiceRequestDeadline( call: AiVaultServicePendingCall, onTimeout: (error: Error) => void @@ -19,7 +37,13 @@ function armAiVaultServiceRequestDeadline( // Same text for both meanings: remote clients humanize it, so a new wording // would reach older ones as "stopped unexpectedly". call.timer = setTimeout(() => { - onTimeout(new Error(`AI Vault service timed out after ${timeoutMs}ms.`)) + onTimeout( + new AiVaultServiceDeadlineError( + `AI Vault service timed out after ${timeoutMs}ms.`, + call.request.operation, + call.request.id + ) + ) }, timeoutMs) call.timer.unref?.() } @@ -37,6 +61,7 @@ export function sendAiVaultServiceCall( } armAiVaultServiceRequestDeadline(call, onTimeout) call.sent = true + call.scanListener?.sent(call.request.id) child.send(call.request) } @@ -47,17 +72,18 @@ export function sendAiVaultServiceCall( */ export function noteAiVaultServiceScanProgress( call: AiVaultServicePendingCall, - units: number, + progress: { units: number; work?: AiVaultScanWork }, onTimeout: (error: Error) => void ): void { - if ( - call.cancelled || - !call.sent || - call.request.operation !== 'scan' || - units <= call.progressUnits - ) { + if (call.cancelled || !call.sent || call.request.operation !== 'scan') { return } - call.progressUnits = units + if (progress.work) { + call.scanListener?.work(progress.work) + } + if (progress.units <= call.progressUnits) { + return + } + call.progressUnits = progress.units armAiVaultServiceRequestDeadline(call, onTimeout) } diff --git a/src/main/ai-vault/session-scanner-service-root-answer.ts b/src/main/ai-vault/session-scanner-service-root-answer.ts new file mode 100644 index 00000000000..6f9a4d4116e --- /dev/null +++ b/src/main/ai-vault/session-scanner-service-root-answer.ts @@ -0,0 +1,24 @@ +import type { AiVaultServiceClientOptions } from './session-scanner-service-client-state' + +type RootAnswerChild = Pick< + ReturnType, + 'connected' | 'send' +> + +/** The parent's half of `requestSessionSearchRoots`; a child replaced meanwhile gets no answer. */ +export function answerAiVaultServiceSessionSearchRoots( + child: RootAnswerChild | null, + id: number, + options: AiVaultServiceClientOptions, + currentChild: () => RootAnswerChild | null +): void { + const resolve = options.resolveSessionSearchRoots + void Promise.resolve() + .then(() => (resolve ? resolve() : (options.init().sessionSearch?.roots ?? null))) + .catch(() => null) + .then((roots) => { + if (child && currentChild() === child && child.connected) { + child.send({ type: 'sessionSearchRoots', id, roots }, () => undefined) + } + }) +} diff --git a/src/main/ai-vault/session-scanner-service-spawn.ts b/src/main/ai-vault/session-scanner-service-spawn.ts index eeb575b7b52..a22f3b3a27a 100644 --- a/src/main/ai-vault/session-scanner-service-spawn.ts +++ b/src/main/ai-vault/session-scanner-service-spawn.ts @@ -24,10 +24,15 @@ import { AiVaultScannerServiceClient } from './session-scanner-service-client' import { getAiVaultServiceEntryPath } from './session-scanner-service-entry-path' import { lowerAiVaultServicePriority } from './session-scanner-service-priority' import type { + AiVaultServiceResultValue, AiVaultServiceScanOptions, AiVaultServiceSubagentRequest, AiVaultSessionSearchInit } from './session-scanner-service-protocol' +import { + requestLocalAiVaultScan, + type LocalAiVaultScanProgressHandle +} from './ai-vault-local-scan-progress' export function spawnAiVaultServiceProcess(): ChildProcess { const entryPath = getAiVaultServiceEntryPath() @@ -62,13 +67,17 @@ function getSharedClient(): AiVaultScannerServiceClient { export function scanAiVaultSessionsInService( options: AiVaultServiceScanOptions, - signal?: AbortSignal + signal?: AbortSignal, + progress?: LocalAiVaultScanProgressHandle ): Promise { return withSpan('aiVault.scan.service', async (span) => { - const value = await getSharedClient().request<{ - result: AiVaultListResult - durationMs: number - }>({ type: 'request', operation: 'scan', options }, signal) + const value = await requestLocalAiVaultScan( + (listener) => + getSharedClient().request< + Extract['value'] + >({ type: 'request', operation: 'scan', options }, signal, listener), + { progress } + ) span.setAttribute('serviceDurationMs', value.durationMs) span.setAttribute('sessions', value.result.sessions.length) return value.result @@ -109,15 +118,24 @@ export function searchSessionsInService( } export function sessionSearchStatusInService(): Promise { - return getSharedClient().request({ type: 'request', operation: 'searchStatus' }) + return getSharedClient().request({ + type: 'request', + operation: 'searchStatus' + }) } export function reconcileSessionSearchInService(): Promise { - return getSharedClient().request({ type: 'request', operation: 'searchReconcile' }) + return getSharedClient().request({ + type: 'request', + operation: 'searchReconcile' + }) } export function clearSessionSearchInService(): Promise { - return getSharedClient().request({ type: 'request', operation: 'searchClear' }) + return getSharedClient().request({ + type: 'request', + operation: 'searchClear' + }) } /** Boot and every settings change: push the policy and keep a child while the index runs. */ diff --git a/src/main/ai-vault/session-scanner.ts b/src/main/ai-vault/session-scanner.ts index 0f9d4d6ef83..bf467f16793 100644 --- a/src/main/ai-vault/session-scanner.ts +++ b/src/main/ai-vault/session-scanner.ts @@ -25,11 +25,7 @@ import { getSessionParseCacheEntry } from './session-parse-cache-store' import { describeSkippedTranscriptRecords } from './session-transcript-record-budget' import { canStopParsingSessions } from './session-scan-cutoff' import { discoverInScopeCwdBucketFiles } from './session-scanner-scope-discovery' -import { - CLAUDE_CWD_BUCKET_LAYOUT, - PI_CWD_BUCKET_LAYOUT, - CODEBUDDY_CWD_BUCKET_LAYOUT -} from './session-cwd-bucket-layouts' +import { CWD_BUCKET_LAYOUTS } from './session-cwd-bucket-layouts' import { discoverAiVaultSessionSources } from './session-scanner-source-discovery' import { cursorChatMetaRefusals, withCursorChatMetaScan } from './session-scanner-cursor-chat-meta' import type { @@ -40,6 +36,7 @@ import type { } from './session-scanner-types' import { clampPositiveInteger, errorMessage } from './session-scanner-values' import { throwIfAiVaultScanCancelled } from './ai-vault-scan-cancellation' +import { noteAiVaultScanFilesRead, noteAiVaultScanReading } from './ai-vault-scan-progress' import { DEFAULT_AI_VAULT_SCAN_LIMIT } from '../../shared/ai-vault-session-depth' import { withDevinSessionsDbScan } from './session-scanner-devin-db' import { withOpenCodeSqliteScanScope } from './session-scanner-opencode-sqlite-scan-scope' @@ -88,13 +85,19 @@ async function scanAiVaultSessionStores(options: AiVaultScanOptions): Promise sessionSortTime(right) - sessionSortTime(left)) } -// Agents whose on-disk layout names a directory per cwd, so a scope's older -// sessions can be found without reading every transcript's header. -const CWD_BUCKET_LAYOUTS = [ - CLAUDE_CWD_BUCKET_LAYOUT, - PI_CWD_BUCKET_LAYOUT, - CODEBUDDY_CWD_BUCKET_LAYOUT -] - async function scanInScopeSessions(args: { discoveries: SessionFileDiscovery[] scopePaths: readonly string[] @@ -207,6 +202,7 @@ async function scanInScopeSessions(args: { if (candidates.length === 0) { return [] } + noteAiVaultScanReading(args.signal, candidates.length) // Parse every in-scope candidate (limit === candidate count never early-stops). return parseSessionCandidates({ candidates, @@ -258,6 +254,7 @@ 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, batch.length) for (const result of results) { if (result.issue) { diff --git a/src/main/ipc/ai-vault-scan-coalescing.test.ts b/src/main/ipc/ai-vault-scan-coalescing.test.ts index 4ae36ad756f..49e86a7f9a8 100644 --- a/src/main/ipc/ai-vault-scan-coalescing.test.ts +++ b/src/main/ipc/ai-vault-scan-coalescing.test.ts @@ -9,7 +9,11 @@ const mocks = vi.hoisted(() => ({ ipcHandle: vi.fn() })) -vi.mock('electron', () => ({ app: { on: vi.fn() }, ipcMain: { handle: mocks.ipcHandle } })) +vi.mock('electron', () => ({ + app: { on: vi.fn() }, + ipcMain: { handle: mocks.ipcHandle }, + BrowserWindow: { getAllWindows: () => [] } +})) vi.mock('../ai-vault/session-scanner-service-spawn', async (importOriginal) => ({ ...(await importOriginal()), scanAiVaultSessionsInService: mocks.scanAiVaultSessionsInService, @@ -96,7 +100,7 @@ describe('Agent Session History scan coalescing', () => { await expect(first).rejects.toMatchObject({ name: 'AbortError' }) expect(scan).toHaveBeenCalledTimes(1) resolveScan?.(EMPTY_RESULT) - await expect(second).resolves.toEqual(EMPTY_RESULT) + await expect(second).resolves.toMatchObject(EMPTY_RESULT) }) it('reports a failed local scan as a host issue rather than rejecting', async () => { @@ -112,6 +116,7 @@ describe('Agent Session History scan coalescing', () => { ) expect(result).toMatchObject({ sessions: [], + scanId: expect.any(String), issues: [expect.objectContaining({ message: 'transcript root is unreadable', kind: 'host' })] }) expect(result).not.toHaveProperty('cancelled') diff --git a/src/main/ipc/ai-vault-scan-progress-broadcast.ts b/src/main/ipc/ai-vault-scan-progress-broadcast.ts new file mode 100644 index 00000000000..98acaec1ce6 --- /dev/null +++ b/src/main/ipc/ai-vault-scan-progress-broadcast.ts @@ -0,0 +1,23 @@ +import { BrowserWindow, ipcMain } from 'electron' +import { AI_VAULT_SCAN_PROGRESS_CHANNEL } from '../../shared/ai-vault-scan-progress' +import { + localAiVaultScanProgress, + onLocalAiVaultScanProgress +} from '../ai-vault/ai-vault-local-scan-progress' + +let unsubscribe: (() => void) | null = null + +/** Windows receive keyed changes; a late opener asks for its own scan entry. */ +export function registerAiVaultScanProgressBroadcast(): void { + ipcMain.handle('aiVault:getScanProgress', (_event, scanKey: string) => + localAiVaultScanProgress(scanKey) + ) + unsubscribe?.() + unsubscribe = onLocalAiVaultScanProgress((update) => { + for (const window of BrowserWindow.getAllWindows()) { + if (!window.isDestroyed()) { + window.webContents.send(AI_VAULT_SCAN_PROGRESS_CHANNEL, update) + } + } + }) +} diff --git a/src/main/ipc/ai-vault.test.ts b/src/main/ipc/ai-vault.test.ts index ec4295ac5e4..5e242555f19 100644 --- a/src/main/ipc/ai-vault.test.ts +++ b/src/main/ipc/ai-vault.test.ts @@ -87,7 +87,13 @@ describe('listAiVaultSessions host routing', () => { includeAntigravityIdeSessions: true, executionHostId: 'local' }), - expect.any(AbortSignal) + expect.any(AbortSignal), + expect.objectContaining({ + sent: expect.any(Function), + work: expect.any(Function), + finish: expect.any(Function), + fail: expect.any(Function) + }) ) }) diff --git a/src/main/ipc/ai-vault.ts b/src/main/ipc/ai-vault.ts index 7de9ee84360..c6a350bacab 100644 --- a/src/main/ipc/ai-vault.ts +++ b/src/main/ipc/ai-vault.ts @@ -2,6 +2,7 @@ import { app, ipcMain } from 'electron' import { configureAiVaultSessionSources, listAiVaultSessions as listCachedLocalAiVaultSessions, + AiVaultLocalScanError, resetAiVaultSessionListCacheForTests, type AiVaultSessionSources } from '../ai-vault/cached-session-list' @@ -59,6 +60,7 @@ import { projectStructuredAiVaultSessions } from '../ai-vault/structured-session-ownership' import { AI_VAULT_ALL_HOST_TIMEOUT_MS } from './ai-vault-all-host-timeouts' +import { registerAiVaultScanProgressBroadcast } from './ai-vault-scan-progress-broadcast' type AiVaultHandlerOptions = AiVaultSessionSources & AiVaultResumeHandlerOptions & { @@ -152,7 +154,10 @@ async function scanAiVaultSessionsByHostScope( hostInfo, scanner: handlerOptions.scanRuntimeAiVaultSessions, listArgs: args ? { ...args, afterFingerprint: undefined } : undefined, - options: { signal, timeoutMs: AI_VAULT_ALL_HOST_TIMEOUT_MS.runtimeScan } + options: { + signal, + timeoutMs: AI_VAULT_ALL_HOST_TIMEOUT_MS.runtimeScan + } }) }) ) @@ -210,11 +215,12 @@ async function scanLocalAiVaultSessionsAsIssue( // so the row carries actionable copy and the log keeps the original. const raw = error instanceof Error ? error.message : 'Local session scan failed.' console.error('[ai-vault] local session scan failed:', raw) - return aiVaultScanIssueResult({ + const result = aiVaultScanIssueResult({ executionHostId: LOCAL_EXECUTION_HOST_ID, path: 'this computer', message: describeAiVaultScanError(raw) }) + return error instanceof AiVaultLocalScanError ? { ...result, scanId: error.scanId } : result } } @@ -243,7 +249,9 @@ export function registerAiVaultHandlers(options: AiVaultHandlerOptions = {}): vo const controller = listCancellations.begin(event, requestToken) try { await handlerOptions.ensureStructuredSessionOwnership?.() - const result = await listAiVaultSessions(args, { signal: controller?.signal }) + const result = await listAiVaultSessions(args, { + signal: controller?.signal + }) return projectStructuredAiVaultSessions(result, OPENS_EVERY_STRUCTURED_CHAT) } catch (error) { // Why: superseding a scan is normal control flow, but Electron logs every @@ -279,6 +287,7 @@ export function registerAiVaultHandlers(options: AiVaultHandlerOptions = {}): vo handleAiVaultGetFirstUserPrompt(args) ) registerAiVaultDeleteHandler(aiVaultDeleteDeps) + registerAiVaultScanProgressBroadcast() // macOS app activation skips DOM focus events, so emit the refresh signal here. app.on('browser-window-focus', (_event, window) => { if (!window.isDestroyed()) { diff --git a/src/main/runtime/rpc/methods/ai-vault.test.ts b/src/main/runtime/rpc/methods/ai-vault.test.ts index cea3f989ed5..93e195c60b4 100644 --- a/src/main/runtime/rpc/methods/ai-vault.test.ts +++ b/src/main/runtime/rpc/methods/ai-vault.test.ts @@ -574,7 +574,8 @@ describe('aiVault.listSessions handler + shared cache', () => { expect(response).toMatchObject({ ok: true }) expect(scanAiVaultSessionsInService).toHaveBeenCalledWith( expect.objectContaining({ limit: undefined, unlimited: true }), - expect.any(AbortSignal) + expect.any(AbortSignal), + expect.objectContaining({ sent: expect.any(Function), work: expect.any(Function) }) ) }) diff --git a/src/preload/api/ai-vault-api.ts b/src/preload/api/ai-vault-api.ts index 34b6357c814..7314ca78251 100644 --- a/src/preload/api/ai-vault-api.ts +++ b/src/preload/api/ai-vault-api.ts @@ -24,6 +24,10 @@ import type { AiVaultPrepareSessionResumeResult } from '../../shared/ai-vault-resume-preparation' import type { ExecutionHostId, ExecutionHostScope } from '../../shared/execution-host' +import type { + AiVaultScanProgress, + AiVaultScanProgressUpdate +} from '../../shared/ai-vault-scan-progress' export type AiVaultApi = { /** Omitted host means this host; `all` is merged by this desktop across every enumerated host. */ @@ -58,4 +62,7 @@ export type AiVaultApi = { deleteSession: (args: AiVaultDeleteSessionArgs) => Promise /** Fires when any app window regains OS focus; returns an unsubscribe. */ onWindowFocused: (callback: () => void) => () => void + /** The local scan entry for a request key, if it is still retained. */ + getScanProgress: (scanKey: string) => Promise + onScanProgress: (callback: (update: AiVaultScanProgressUpdate) => void) => () => void } diff --git a/src/preload/api/ai-vault-bridge.ts b/src/preload/api/ai-vault-bridge.ts index 9647f72d5dd..425b0dfbc53 100644 --- a/src/preload/api/ai-vault-bridge.ts +++ b/src/preload/api/ai-vault-bridge.ts @@ -14,6 +14,11 @@ import type { import type { AiVaultSessionTitlesArgs } from '../../shared/ai-vault-session-title' import type { AiVaultPrepareSessionResumeArgs } from '../../shared/ai-vault-resume-preparation' import type { PreloadApi } from '../api-types' +import { + AI_VAULT_SCAN_PROGRESS_CHANNEL, + type AiVaultScanProgress, + type AiVaultScanProgressUpdate +} from '../../shared/ai-vault-scan-progress' function searchClient( executionHostScope?: ExecutionHostScope @@ -56,5 +61,13 @@ export const aiVaultApi = { const listener = (_event: Electron.IpcRendererEvent) => callback() ipcRenderer.on('aiVault:windowFocused', listener) return () => ipcRenderer.removeListener('aiVault:windowFocused', listener) + }, + getScanProgress: (scanKey: string): Promise => + ipcRenderer.invoke('aiVault:getScanProgress', scanKey), + onScanProgress: (callback: (update: AiVaultScanProgressUpdate) => void): (() => void) => { + const listener = (_event: Electron.IpcRendererEvent, update: AiVaultScanProgressUpdate) => + callback(update) + ipcRenderer.on(AI_VAULT_SCAN_PROGRESS_CHANNEL, listener) + return () => ipcRenderer.removeListener(AI_VAULT_SCAN_PROGRESS_CHANNEL, listener) } } satisfies PreloadApi['aiVault'] diff --git a/src/renderer/src/web/preload-api/web-ai-vault-api.ts b/src/renderer/src/web/preload-api/web-ai-vault-api.ts index 8576b3d39d2..fe852ecd844 100644 --- a/src/renderer/src/web/preload-api/web-ai-vault-api.ts +++ b/src/renderer/src/web/preload-api/web-ai-vault-api.ts @@ -97,7 +97,10 @@ export function createWebAiVaultApi(): NonNullable['aiVault' agent: args.agent, reason: 'non-local-host' as const }), - onWindowFocused: () => noopUnsubscribe + onWindowFocused: () => noopUnsubscribe, + // Why: the runtime RPC has no progress stream, so the panel keeps its plain spinner here. + getScanProgress: () => Promise.resolve(null), + onScanProgress: () => noopUnsubscribe } } diff --git a/src/shared/ai-vault-local-scan-key.ts b/src/shared/ai-vault-local-scan-key.ts new file mode 100644 index 00000000000..de7f2ffba67 --- /dev/null +++ b/src/shared/ai-vault-local-scan-key.ts @@ -0,0 +1,17 @@ +import type { AiVaultListArgs } from './ai-vault-types' +import { requestedAiVaultSessionDepth } from './ai-vault-session-depth' + +/** The local cache and coordinator entry for one scope and scan depth. */ +export function aiVaultLocalScanKey(args?: AiVaultListArgs): string { + return JSON.stringify({ + key: aiVaultLocalScanScopeKey(args), + depth: requestedAiVaultSessionDepth(args) + }) +} + +export function aiVaultLocalScanScopeKey(args?: AiVaultListArgs): string { + return JSON.stringify({ + scopePaths: [...new Set(args?.scopePaths ?? [])].sort(), + includeAntigravityIdeSessions: args?.includeAntigravityIdeSessions === true + }) +} diff --git a/src/shared/ai-vault-scan-progress.ts b/src/shared/ai-vault-scan-progress.ts new file mode 100644 index 00000000000..697d8a78359 --- /dev/null +++ b/src/shared/ai-vault-scan-progress.ts @@ -0,0 +1,38 @@ +import type { AiVaultAgent } from './ai-vault-types' + +/** One agent's history as counted while its folders and databases are listed. */ +export type AiVaultScanAgentProgress = { + agent: AiVaultAgent + /** Distinct files: a hard link or a linked folder reaching the same file counts once. */ + files: number + /** SQLite stores holding this agent's sessions. */ + databases: number + /** A folder or database of this agent is still being listed. */ + walking: boolean +} + +/** 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 } +} + +/** + * One local Session History scan entry, as main pushes it to windows. + * When `stopped`, the agents still `walking` are the ones the scan stopped on. + */ +export type AiVaultScanProgress = Omit & { + scanId: string + stage: AiVaultScanWork['stage'] | 'finished' | 'stopped' + startedAt: number + settledAt: number | null +} + +export type AiVaultScanProgressUpdate = { + scanKey: string + progress: AiVaultScanProgress | null +} + +export const AI_VAULT_SCAN_PROGRESS_CHANNEL = 'aiVault:scanProgress' diff --git a/src/shared/ai-vault-types.ts b/src/shared/ai-vault-types.ts index 38e6c9a6c7e..821952e557c 100644 --- a/src/shared/ai-vault-types.ts +++ b/src/shared/ai-vault-types.ts @@ -233,6 +233,8 @@ export type AiVaultListArgs = { } export type AiVaultListResult = { + /** Local scan entry that produced this result; absent for remote and merged results. */ + scanId?: string fingerprint?: string /** Requires the caller to reuse the exact previously received body. */ unchanged?: true