From 6fcd52116519385a852dbda55977d7f6c3f46ec5 Mon Sep 17 00:00:00 2001 From: Brennan Benson <79079362+brennanb2025@users.noreply.github.com> Date: Thu, 15 Oct 2026 13:35:47 -0700 Subject: [PATCH] Session History: the scanner reports its progress to windows (no visible change) (#28788) * feat(session-history): the scanner reports its progress to windows The Session History scanner now counts each agent's history files and databases while it lists them, and how many transcripts it has read. Main keeps one progress entry per local scan, tags list results with the scan that produced them, tells the scan's own no-progress deadline apart from other faults, and pushes updates to every window. The preload API exposes getScanProgress and onScanProgress. Nothing reads them yet. * fix(session-history): drop the redundant scan key from list results and progress The scan ID already names the scan a result or progress entry came from, so the key copy on the list result (sent to paired clients too) and on each progress entry was never needed. The key stays only where it is the lookup. --- .../ai-vault/ai-vault-local-scan-progress.ts | 134 +++++++++++ .../ai-vault-scan-coordinator.test.ts | 51 ++++ .../ai-vault/ai-vault-scan-coordinator.ts | 42 +++- .../ai-vault-scan-progress-flow.test.ts | 219 ++++++++++++++++++ .../ai-vault/ai-vault-scan-progress.test.ts | 39 +++- src/main/ai-vault/ai-vault-scan-progress.ts | 126 +++++++++- src/main/ai-vault/ai-vault-scan-tally.ts | 89 +++++++ .../cached-session-list-wsl-probe.test.ts | 16 +- src/main/ai-vault/cached-session-list.test.ts | 8 +- src/main/ai-vault/cached-session-list.ts | 46 +++- src/main/ai-vault/codex-session-root-dedup.ts | 10 +- .../ai-vault/session-cwd-bucket-layouts.ts | 8 + .../ai-vault/session-scanner-background.ts | 6 +- .../ai-vault/session-scanner-discovery.ts | 48 +++- ...on-scanner-opencode-sqlite-worker-spawn.ts | 57 ++++- .../session-scanner-service-client-state.ts | 6 +- .../session-scanner-service-client.ts | 26 +-- .../session-scanner-service-entry.test.ts | 5 +- .../ai-vault/session-scanner-service-entry.ts | 22 +- .../session-scanner-service-protocol.ts | 19 +- ...n-scanner-service-request-deadline.test.ts | 23 ++ ...ession-scanner-service-request-deadline.ts | 44 +++- .../session-scanner-service-root-answer.ts | 24 ++ .../ai-vault/session-scanner-service-spawn.ts | 34 ++- src/main/ai-vault/session-scanner.ts | 27 +-- src/main/ipc/ai-vault-scan-coalescing.test.ts | 9 +- .../ipc/ai-vault-scan-progress-broadcast.ts | 23 ++ src/main/ipc/ai-vault.test.ts | 8 +- src/main/ipc/ai-vault.ts | 15 +- src/main/runtime/rpc/methods/ai-vault.test.ts | 3 +- src/preload/api/ai-vault-api.ts | 7 + src/preload/api/ai-vault-bridge.ts | 13 ++ .../src/web/preload-api/web-ai-vault-api.ts | 5 +- src/shared/ai-vault-local-scan-key.ts | 17 ++ src/shared/ai-vault-scan-progress.ts | 38 +++ src/shared/ai-vault-types.ts | 2 + 36 files changed, 1139 insertions(+), 130 deletions(-) create mode 100644 src/main/ai-vault/ai-vault-local-scan-progress.ts create mode 100644 src/main/ai-vault/ai-vault-scan-progress-flow.test.ts create mode 100644 src/main/ai-vault/ai-vault-scan-tally.ts create mode 100644 src/main/ai-vault/session-scanner-service-root-answer.ts create mode 100644 src/main/ipc/ai-vault-scan-progress-broadcast.ts create mode 100644 src/shared/ai-vault-local-scan-key.ts create mode 100644 src/shared/ai-vault-scan-progress.ts 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