diff --git a/src/main/ai-vault/listing/cached-session-list-recovery.test.ts b/src/main/ai-vault/listing/cached-session-list-recovery.test.ts index cd4258dfadd..a892ee9ca9b 100644 --- a/src/main/ai-vault/listing/cached-session-list-recovery.test.ts +++ b/src/main/ai-vault/listing/cached-session-list-recovery.test.ts @@ -1,4 +1,7 @@ import { afterEach, expect, it, vi } from 'vitest' +import { scanLocalAiVaultSessionsAsIssue } from '../../ipc/ai-vault/ai-vault-local-list' +import type { AiVaultRowDelta, AiVaultScanRows } from '../../../shared/ai-vault-scan-rows' +import { listResult } from '../structured-sessions/structured-session-ownership.test-support' import { AiVaultScanCoordinator } from '../scan-lifecycle/ai-vault-scan-coordinator' import { nextScanObservationOrder, @@ -97,3 +100,131 @@ it('shares the probe and child retry budget, and preserves the terminal outcome expect(mocks.roots).toHaveBeenCalledTimes(2) expect(mocks.scan).toHaveBeenCalledOnce() }) + +it('invalidates an older scope cache when a recovered ordinary result satisfies an outer force', async () => { + vi.useFakeTimers() + const project = { scopePaths: ['/repo'], scopeOnly: true } + mocks.roots.mockResolvedValue(ROOTS) + mocks.scan.mockResolvedValueOnce({ ...EMPTY, scannedAt: 'stale-scope' }) + await listAiVaultSessions(project) + const blocked = Promise.withResolvers() + const entered = Promise.withResolvers() + const fresh = { ...EMPTY, scannedAt: 'recovered' } + mocks.scan + .mockImplementationOnce(() => { + entered.resolve() + return blocked.promise + }) + .mockResolvedValue(fresh) + const outer = new AiVaultScanCoordinator() + const start = vi.fn((signal: AbortSignal) => + listAiVaultSessions({ scopePaths: ['/repo'] }, { signal }) + ) + const ordinary = outer.run({ key: 'All', aggregate: true, start }) + await entered.promise + const forced = outer.run({ key: 'All', force: true, aggregate: true, start }) + blocked.reject(new AiVaultScanStalledError()) + await expect(Promise.all([ordinary, forced])).resolves.toHaveLength(2) + expect(start).toHaveBeenCalledOnce() + await expect(listAiVaultSessions(project)).resolves.toMatchObject({ scannedAt: 'recovered' }) + expect(mocks.scan).toHaveBeenCalledTimes(3) +}) + +it('replays retained local rows on nested observer activation without resurrecting removals', async () => { + mocks.roots.mockResolvedValue(ROOTS) + const pending = Promise.withResolvers() + const entered = Promise.withResolvers() + let publish!: (delta: AiVaultRowDelta) => void + mocks.scan.mockImplementation((_args, _signal, _progress, emit) => { + publish = emit + entered.resolve() + return pending.promise + }) + const outer = new AiVaultScanCoordinator() + const start = (signal: AbortSignal, _id: string, emit: (delta: AiVaultRowDelta) => void) => + scanLocalAiVaultSessionsAsIssue(undefined, signal, emit) + const all = outer.run({ key: 'All', start, observeRows: () => undefined }) + await entered.promise + const local = outer.run({ key: 'Local', start }) + await Promise.resolve() + const firstRow = { ...listResult().sessions[0]!, id: 'first-row' } + const nextRow = { ...listResult().sessions[0]!, id: 'next-row' } + publish({ upserts: [firstRow], removed: [] }) + const rows = new Map() + const observe = vi.fn((event: AiVaultScanRows) => { + if (event.kind === 'snapshot') { + rows.clear() + for (const row of event.sessions) { + rows.set(row.id, row) + } + } else { + for (const id of event.delta.removed) { + rows.delete(id) + } + for (const row of event.delta.upserts) { + rows.set(row.id, row) + } + } + }) + const controller = new AbortController() + const stream = outer.run({ key: 'Local', start, observeRows: observe, signal: controller.signal }) + expect([...rows.keys()]).toEqual([firstRow.id]) + controller.abort() + await expect(stream).rejects.toMatchObject({ name: 'AbortError' }) + publish({ upserts: [nextRow], removed: [firstRow.id] }) + const resumed = outer.run({ key: 'Local', start, observeRows: observe }) + const shared = outer.run({ key: 'Local', start, observeRows: observe }) + expect([...rows.keys()]).toEqual([nextRow.id]) + const calls = observe.mock.calls.length + publish({ upserts: [], removed: [nextRow.id] }) + expect(observe).toHaveBeenCalledTimes(calls + 1) + expect(rows.size).toBe(0) + pending.resolve(EMPTY) + await Promise.all([all, local, resumed, shared]) +}) + +it('keeps a newer accepted cache when an older forced scan finishes later', async () => { + mocks.roots.mockResolvedValue(ROOTS) + const older = Promise.withResolvers() + const oldEntered = Promise.withResolvers() + mocks.scan.mockImplementationOnce(() => { + oldEntered.resolve() + return older.promise + }) + const first = listAiVaultSessions({ scopePaths: ['/old'], force: true }) + await oldEntered.promise + mocks.scan.mockResolvedValueOnce({ ...EMPTY, scannedAt: 'newer' }) + await listAiVaultSessions({ scopePaths: ['/new'], force: true }) + older.resolve({ ...EMPTY, scannedAt: 'older' }) + await first + await expect(listAiVaultSessions({ scopePaths: ['/new'] })).resolves.toMatchObject({ + scannedAt: 'newer' + }) + expect(mocks.scan).toHaveBeenCalledTimes(2) +}) + +it('does not invalidate sibling cache policy after the sole force detaches', async () => { + mocks.roots.mockResolvedValue(ROOTS) + const project = { scopePaths: ['/repo'], scopeOnly: true } + mocks.scan.mockResolvedValueOnce({ ...EMPTY, scannedAt: 'scope-cache' }) + await listAiVaultSessions(project) + const pending = Promise.withResolvers() + const entered = Promise.withResolvers() + mocks.scan.mockImplementationOnce(() => { + entered.resolve() + return pending.promise + }) + const controller = new AbortController() + const forced = listAiVaultSessions( + { scopePaths: ['/repo'], force: true }, + { signal: controller.signal } + ) + await entered.promise + const ordinary = listAiVaultSessions({ scopePaths: ['/repo'] }) + controller.abort() + await expect(forced).rejects.toMatchObject({ name: 'AbortError' }) + pending.resolve({ ...EMPTY, scannedAt: 'ordinary' }) + await ordinary + await expect(listAiVaultSessions(project)).resolves.toMatchObject({ scannedAt: 'scope-cache' }) + expect(mocks.scan).toHaveBeenCalledTimes(2) +}) diff --git a/src/main/ai-vault/listing/cached-session-list.ts b/src/main/ai-vault/listing/cached-session-list.ts index ea4257bc01c..dbc76d999ee 100644 --- a/src/main/ai-vault/listing/cached-session-list.ts +++ b/src/main/ai-vault/listing/cached-session-list.ts @@ -3,7 +3,6 @@ import type { AiVaultScanRows } from '../../../shared/ai-vault-scan-rows' import { recordScanOutcome, scanForceOrder, - scanRecoveryInterest, type ScanOutcome } from '../scan-lifecycle/ai-vault-scan-recovery' import { @@ -71,8 +70,8 @@ const cachedLists = new Map() let scanCoordinator = new AiVaultScanCoordinator() // Reject cache writes from scans that began before invalidation, or late results can resurrect deleted sessions. let cacheGeneration = 0 -// Advances on each forced refresh, so a scan that began before one cannot cache its older result. -let forcedRefreshEpoch = 0 +// Older observations cannot repopulate caches after an accepted Refresh. +let acceptedForceOrder = 0 export async function listAiVaultSessions( args?: AiVaultListArgs, @@ -90,6 +89,7 @@ export async function listAiVaultSessions( const fullKey = args?.scopeOnly ? aiVaultLocalScanScopeKey({ ...args, scopeOnly: false }) : key const cachedList = servingCachedList([key, fullKey], depth, args, now, requiredOrder) if (cachedList) { + cachedList.outcome.accept?.(requiredOrder) recordScanOutcome( options.signal, JSON.stringify({ scanKey, generation: cacheGeneration }), @@ -99,7 +99,6 @@ export async function listAiVaultSessions( } // Capture before the coordinator's start microtask so intervening invalidation still rejects the result. const startGeneration = cacheGeneration - const startForcedRefresh = forcedRefreshEpoch let progress: LocalAiVaultScanProgressHandle | null = null return scanCoordinator.run({ // Isolate invalidated scans without changing the key used by progress readers. @@ -114,8 +113,38 @@ export async function listAiVaultSessions( onError: () => progress?.fail(), onReject: (error, scanId) => isAiVaultScanCancelledError(error) ? error : new AiVaultLocalScanError(scanId, error), + onAccept: (outcome, required) => { + if (!outcome.ok || startGeneration !== cacheGeneration || outcome.order <= required) { + return + } + if (required > acceptedForceOrder) { + acceptedForceOrder = required + for (const [cachedKey, cached] of cachedLists) { + if (cached.outcome.order <= required) { + cachedLists.delete(cachedKey) + } + } + } + if (outcome.order <= acceptedForceOrder) { + return + } + const current = cachedLists.get(key) + if ( + !current || + current.expiresAt <= Date.now() || + (outcome.order >= current.outcome.order && + !aiVaultCachedDepthServes(current.depth, depth, args?.scopePaths)) + ) { + rememberCachedList({ + key, + depth, + result: outcome.result, + outcome, + expiresAt: Date.now() + AI_VAULT_CACHE_TTL_MS + }) + } + }, start: async (scanSignal, scanId, publish) => { - const force = (scanRecoveryInterest(scanSignal)?.required ?? 0) > 0 const scanProgress = progress if (!scanProgress) { throw new Error('Local scan progress was not created.') @@ -150,38 +179,6 @@ export async function listAiVaultSessions( { ...attachAiVaultScopeCwdAliases(scanned, scope), scanId }, wslConfirmed ) - // Return pre-invalidation results to their caller, but never repopulate the cache with them. - if ( - !scanSignal.aborted && - startGeneration === cacheGeneration && - startForcedRefresh === forcedRefreshEpoch - ) { - const current = cachedLists.get(key) - if ( - force || - !current || - current.expiresAt <= Date.now() || - !aiVaultCachedDepthServes(current.depth, depth, args?.scopePaths) - ) { - // A forced refresh supersedes every list: an older one for another scope (the - // scope-only view a full list also serves) must not be served after it. - if (force) { - cachedLists.clear() - forcedRefreshEpoch++ - } - rememberCachedList({ - key, - depth, - result, - outcome: { - ok: true, - result, - order: scanRecoveryInterest(scanSignal)?.observation ?? 0 - }, - expiresAt: Date.now() + AI_VAULT_CACHE_TTL_MS - }) - } - } return result } }) diff --git a/src/main/ai-vault/parse-cache/session-parse-cache-persistence.test.ts b/src/main/ai-vault/parse-cache/session-parse-cache-persistence.test.ts index 11a6c9e4c95..21c87e27f00 100644 --- a/src/main/ai-vault/parse-cache/session-parse-cache-persistence.test.ts +++ b/src/main/ai-vault/parse-cache/session-parse-cache-persistence.test.ts @@ -22,8 +22,10 @@ import { } from './session-scanner-parse-cache' import { invalidateSessionParseCacheEntry, - resetSessionParseCacheForTests + resetSessionParseCacheForTests, + storeSessionScopeCwd } from './session-parse-cache-store' +import { readSessionCopyProof } from './session-copy-proof' import type * as ParseCacheModule from './session-scanner-parse-cache' import { parseSessionCandidate } from '../parsing/session-scanner-candidate-parse' import { MAX_SESSION_TRANSCRIPT_RECORD_BYTES } from '../transcripts/session-transcript-record-budget' @@ -294,3 +296,28 @@ describe('session parse cache persistence', () => { expect(await readdir(root)).toEqual([]) }) }) + +it('drains copy proofs and later scope writes when the database cannot open', async () => { + const root = await makeTempDir() + const blocked = join(root, 'not-a-directory') + await writeFile(blocked, 'blocked') + initSessionParseCachePersistence({ + filePath: join(blocked, 'parse.json'), + appVersion: APP_VERSION + }) + const file = { + path: join(root, 'transcript.jsonl'), + mtimeMs: 1, + modifiedAt: '', + sizeBytes: 2, + dev: 1, + ino: 2, + birthtimeMs: 3 + } + await expect(readSessionCopyProof(file, async () => 'proved-session')).resolves.toBe( + 'proved-session' + ) + storeSessionScopeCwd(file.path, { ...file, platform: process.platform, cwd: '/repo' }) + await flushSessionParseCachePersist() + await flushSessionParseCachePersist() +}) diff --git a/src/main/ai-vault/parse-cache/session-parse-cache-persistence.ts b/src/main/ai-vault/parse-cache/session-parse-cache-persistence.ts index d1904ff5d31..9d06aec6dbe 100644 --- a/src/main/ai-vault/parse-cache/session-parse-cache-persistence.ts +++ b/src/main/ai-vault/parse-cache/session-parse-cache-persistence.ts @@ -98,7 +98,8 @@ export async function flushSessionParseCachePersist(): Promise { } } while (pendingCopyProofs.size > 0) { - openedDatabase()?.writeCopyProofs(takeWriteChunk(pendingCopyProofs), Date.now()) + const chunk = takeWriteChunk(pendingCopyProofs) + openedDatabase()?.writeCopyProofs(chunk, Date.now()) if (pendingCopyProofs.size > 0) { await yieldToEventLoop() } diff --git a/src/main/ai-vault/scan-lifecycle/ai-vault-scan-coordinator.test.ts b/src/main/ai-vault/scan-lifecycle/ai-vault-scan-coordinator.test.ts index d7821366a22..47668c2cdf2 100644 --- a/src/main/ai-vault/scan-lifecycle/ai-vault-scan-coordinator.test.ts +++ b/src/main/ai-vault/scan-lifecycle/ai-vault-scan-coordinator.test.ts @@ -1,6 +1,10 @@ import { describe, expect, it, vi } from 'vitest' import { AiVaultScanCoordinator } from './ai-vault-scan-coordinator' -import { scanRecoveryInterest, nextScanObservationOrder } from './ai-vault-scan-recovery' +import { + scanRecoveryInterest, + nextScanObservationOrder, + createScanRecoveryInterest +} from './ai-vault-scan-recovery' const EMPTY = { sessions: [], issues: [], scannedAt: '2026-11-13' } function deferred() { @@ -165,3 +169,53 @@ it('drops a detached newest requirement so a recovered observation can serve sur await Promise.all([first, force]) expect(start).not.toHaveBeenCalled() }) + +it.each([10, 100])( + 'keeps one live execution across %i cancelled successors and abort re-entry', + async (count) => { + const coordinator = new AiVaultScanCoordinator() + const pending = Promise.withResolvers() + const start = vi.fn(() => pending.promise) + const first = coordinator.run({ key: 'scope', start }) + await Promise.resolve() + const survivors: Promise[] = [] + for (let index = 0; index < count; index++) { + const controller = new AbortController() + const forced = coordinator.run({ + key: 'scope', + start, + force: true, + signal: controller.signal + }) + const cancelled = expect(forced).rejects.toMatchObject({ name: 'AbortError' }) + controller.signal.addEventListener( + 'abort', + () => { + survivors.push(coordinator.run({ key: 'scope', start })) + }, + { once: true } + ) + controller.abort() + await cancelled + } + expect(start).toHaveBeenCalledOnce() + pending.resolve(EMPTY) + await expect(Promise.all([first, ...survivors])).resolves.toEqual( + Array.from({ length: count + 1 }, () => EMPTY) + ) + expect(start).toHaveBeenCalledOnce() + } +) + +it('carries accepted-outcome policy through a completed parent receipt without another start', async () => { + const coordinator = new AiVaultScanCoordinator() + const parent = new AbortController() + const interest = createScanRecoveryInterest(parent.signal) + interest.required = nextScanObservationOrder() + const start = vi.fn(async () => EMPTY) + const onAccept = vi.fn() + await coordinator.run({ key: 'scope', force: true, signal: parent.signal, start, onAccept }) + await coordinator.run({ key: 'scope', force: true, signal: parent.signal, start }) + expect(start).toHaveBeenCalledOnce() + expect(onAccept).toHaveBeenCalledTimes(2) +}) diff --git a/src/main/ai-vault/scan-lifecycle/ai-vault-scan-coordinator.ts b/src/main/ai-vault/scan-lifecycle/ai-vault-scan-coordinator.ts index 7d42cf82639..1de8ca82e39 100644 --- a/src/main/ai-vault/scan-lifecycle/ai-vault-scan-coordinator.ts +++ b/src/main/ai-vault/scan-lifecycle/ai-vault-scan-coordinator.ts @@ -1,11 +1,10 @@ import { AiVaultScanRowObservers } from './ai-vault-scan-row-observers' import { randomUUID } from 'node:crypto' -import { - AI_VAULT_SCAN_CANCELLED_MESSAGE, - type AiVaultListResult -} from '../../../shared/ai-vault/contracts/ai-vault-types' +import type { AiVaultListResult } from '../../../shared/ai-vault/contracts/ai-vault-types' +import { createAiVaultScanCancelledError as scanCancellationError } from './ai-vault-scan-cancellation' import type { AiVaultRowDelta, AiVaultScanRows } from '../../../shared/ai-vault-scan-rows' import { + acceptScanOutcome, createScanRecoveryInterest, forwardScanRecoveryInterest, inheritScanRecoveryInterest, @@ -26,6 +25,7 @@ type Run = { observeRows?: (rows: AiVaultScanRows) => void onCreate?: (id: string) => void onError?: (error: unknown) => void + onAccept?: (outcome: ScanOutcome, requiredOrder: number) => void onReject?: (error: unknown, id: string) => unknown start: ( signal: AbortSignal, @@ -45,9 +45,11 @@ type Entry = { rows: AiVaultScanRowObservers } +type ExecutionOwner = { active: Entry; queued?: Entry } + /** A force arriving during observation owns one trailing pass, without aborting healthy work. */ export class AiVaultScanCoordinator { - private readonly entries = new Map() + private readonly entries = new Map() run(args: Run): Promise { if (args.signal?.aborted) { @@ -56,29 +58,64 @@ export class AiVaultScanCoordinator { args = { ...args, requiredOrder: args.requiredOrder ?? scanForceOrder(args.force, args.signal) } const receipt = scanRecoveryInterest(args.signal)?.receipts.get(args.key) if (args.requiredOrder && receipt && receipt.order > args.requiredOrder) { + acceptScanOutcome(receipt, args.requiredOrder) return receipt.ok ? Promise.resolve(receipt.result) : Promise.reject( args.onReject ? args.onReject(receipt.error, receipt.scanId ?? '') : receipt.error ) } - let entry = this.entries.get(args.key) - const leading = entry + let owner = this.entries.get(args.key) + const active = owner?.active const trailing = - !!entry && - !!args.requiredOrder && - entry.started && - (scanRecoveryInterest(entry.controller.signal)?.observation ?? 0) <= args.requiredOrder - if (!entry || trailing) { - entry = this.create(args, trailing ? leading : undefined) - this.entries.set(args.key, entry) + !!active && + (active.controller.signal.aborted || + (!!args.requiredOrder && + active.started && + (scanRecoveryInterest(active.controller.signal)?.observation ?? 0) <= args.requiredOrder)) + let entry = trailing ? owner?.queued : active + if (!entry) { + const completion = Promise.withResolvers() + entry = this.create(completion.promise) + const leading = trailing ? owner?.active : undefined + if (owner) { + owner.queued = entry + } else { + owner = { active: entry } + this.entries.set(args.key, owner) + } + const attached = this.attach(args, entry, owner) + this.launch(args, entry, owner, completion, leading) + return attached } - return this.attach(args, entry) + return this.attach(args, entry, owner!) } - private create(args: Run, leading?: Entry): Entry { + private create(promise: Promise): Entry { const controller = new AbortController() - const interest = createScanRecoveryInterest(controller.signal) + createScanRecoveryInterest(controller.signal) + const id = randomUUID() + return { + id, + controller, + promise, + waiters: 0, + started: false, + settled: false, + requirements: new Map(), + rows: new AiVaultScanRowObservers(id, controller.signal) + } + } + + private launch( + args: Run, + entry: Entry, + owner: ExecutionOwner, + completion: ReturnType>, + leading?: Entry + ): void { + const { controller, id } = entry + const interest = scanRecoveryInterest(controller.signal)! const leadingInterest = leading ? scanRecoveryInterest(leading.controller.signal) : undefined const releaseLeading = leadingInterest ? forwardScanRecoveryInterest(interest, leadingInterest) @@ -86,23 +123,13 @@ export class AiVaultScanCoordinator { if (leadingInterest) { interest.receipts = leadingInterest.receipts } - const id = randomUUID() - const entry: Entry = { - id, - controller, - promise: Promise.resolve(undefined!), - waiters: 0, - started: false, - settled: false, - requirements: new Map(), - rows: new AiVaultScanRowObservers(id, controller.signal) - } const hold = leading ? new AbortController() : null const held = leading && hold ? this.attach( { ...args, force: false, requiredOrder: 0, signal: hold.signal }, - leading + leading, + owner ).then( () => leading.outcome, () => leading.outcome @@ -110,13 +137,21 @@ export class AiVaultScanCoordinator { : null const releaseHold = () => hold?.abort() controller.signal.addEventListener('abort', releaseHold, { once: true }) - entry.promise = Promise.resolve().then(async () => { + if (controller.signal.aborted) { + releaseHold() + } + let acceptedLeading: ScanOutcome | undefined + const running = Promise.resolve().then(async () => { if (leading) { const outcome = await held releaseHold() + if (controller.signal.aborted) { + throw scanCancellationError() + } const required = Math.max(0, ...entry.requirements.values()) if (outcome && outcome.order > required) { interest.observation = outcome.order + acceptedLeading = outcome if (outcome.ok) { return outcome.result } @@ -142,64 +177,86 @@ export class AiVaultScanCoordinator { }) const settle = () => { entry.settled = true + if (owner.active === entry) { + if (owner.queued) { + owner.active = owner.queued + owner.queued = undefined + } else if (this.entries.get(args.key) === owner) { + this.entries.delete(args.key) + } + } else if (owner.queued === entry) { + owner.queued = undefined + } entry.rows.dispose() releaseLeading() releaseHold() controller.signal.removeEventListener('abort', releaseHold) - if (this.entries.get(args.key) === entry) { - this.entries.delete(args.key) + } + const accept = (requiredOrder: number) => { + if (!entry.outcome?.ok || entry.outcome.order <= requiredOrder) { + return + } + if (acceptedLeading) { + acceptedLeading.accept?.(requiredOrder) + } else { + args.onAccept?.(entry.outcome, requiredOrder) + } + for (const outcome of interest.receipts.values()) { + acceptScanOutcome(outcome, requiredOrder) } } - void entry.promise.then( + void running.then( (result) => { - entry.outcome = { ok: true, result, order: interest.observation, scanId: id } + entry.outcome = { ok: true, result, order: interest.observation, scanId: id, accept } settle() + completion.resolve(result) }, (error) => { entry.outcome = { ok: false, error, order: interest.observation, scanId: id } - args.onError?.(error) settle() + args.onError?.(error) + completion.reject(error) } ) - return entry } - private attach(args: Run, entry: Entry): Promise { - entry.waiters++ - const interest = scanRecoveryInterest(entry.controller.signal) - const forced = args.force === true && scanRecoveryInterest(args.signal) === undefined - const releaseParent = inheritScanRecoveryInterest(args.signal, entry.controller.signal) - const obligation = Symbol() - if (args.requiredOrder) { - entry.requirements.set(obligation, args.requiredOrder ?? 0) - } - if (forced && interest) { - interest.forced++ - notifyScanRecoveryInterest(interest) - } - const releaseRows = entry.rows.attach(args.observeRows) + private attach(args: Run, entry: Entry, owner: ExecutionOwner): Promise { return new Promise((resolve, reject) => { let attached = true + let releaseRows: () => void = () => undefined + let releaseParent: () => void = () => undefined + const interest = scanRecoveryInterest(entry.controller.signal) + const forced = args.force === true && scanRecoveryInterest(args.signal) === undefined + const obligation = Symbol() + entry.waiters++ + if (args.requiredOrder) { + entry.requirements.set(obligation, args.requiredOrder) + } + if (forced && interest) { + interest.forced++ + } const detach = () => { if (!attached) { return } attached = false args.signal?.removeEventListener('abort', abort) - releaseRows() entry.waiters-- entry.requirements.delete(obligation) - releaseParent() if (forced && interest) { interest.forced-- - notifyScanRecoveryInterest(interest) } if (!entry.waiters && !entry.settled) { - if (this.entries.get(args.key) === entry) { - this.entries.delete(args.key) + if (owner.queued === entry) { + owner.queued = undefined } entry.controller.abort() } + releaseRows() + releaseParent() + if (forced && interest) { + notifyScanRecoveryInterest(interest) + } } const abort = () => { detach() @@ -210,10 +267,27 @@ export class AiVaultScanCoordinator { abort() return } + const parent = inheritScanRecoveryInterest(args.signal, entry.controller.signal) + if (attached) { + releaseParent = parent + } else { + parent() + } + const rows = entry.rows.attach(args.observeRows) + if (attached) { + releaseRows = rows + } else { + rows() + } + if (forced && interest) { + notifyScanRecoveryInterest(interest) + } + void entry.promise.then( (result) => { if (attached) { if (entry.outcome) { + entry.outcome.accept?.(args.requiredOrder ?? 0) recordScanOutcome(args.signal, args.key, entry.outcome) } detach() @@ -233,9 +307,3 @@ export class AiVaultScanCoordinator { }) } } - -function scanCancellationError(): Error { - const error = new Error(AI_VAULT_SCAN_CANCELLED_MESSAGE) - error.name = 'AbortError' - return error -} diff --git a/src/main/ai-vault/scan-lifecycle/ai-vault-scan-recovery.ts b/src/main/ai-vault/scan-lifecycle/ai-vault-scan-recovery.ts index e27a5a8f52e..2a2757728eb 100644 --- a/src/main/ai-vault/scan-lifecycle/ai-vault-scan-recovery.ts +++ b/src/main/ai-vault/scan-lifecycle/ai-vault-scan-recovery.ts @@ -1,9 +1,10 @@ import type { AiVaultListResult } from '../../../shared/ai-vault/contracts/ai-vault-types' -export type ScanOutcome = { order: number; scanId?: string } & ( - | { ok: true; result: AiVaultListResult } - | { ok: false; error: unknown } -) +export type ScanOutcome = { + order: number + scanId?: string + accept?: (requiredOrder: number) => void +} & ({ ok: true; result: AiVaultListResult } | { ok: false; error: unknown }) type Interest = { forced: number inherited: number @@ -85,3 +86,10 @@ export function forwardScanRecoveryInterest(source: Interest, target: Interest): updateTotal() } } + +/** A successful observation discharges an obligation only after it actually began. */ +export function acceptScanOutcome(outcome: ScanOutcome, requiredOrder: number): void { + if (outcome.ok && outcome.order > requiredOrder) { + outcome.accept?.(requiredOrder) + } +} diff --git a/src/main/ai-vault/scan-lifecycle/ai-vault-scan-row-observers.ts b/src/main/ai-vault/scan-lifecycle/ai-vault-scan-row-observers.ts index 5bf1e519fb0..1f47f882aa4 100644 --- a/src/main/ai-vault/scan-lifecycle/ai-vault-scan-row-observers.ts +++ b/src/main/ai-vault/scan-lifecycle/ai-vault-scan-row-observers.ts @@ -5,6 +5,7 @@ type Observer = (rows: AiVaultScanRows) => void export class AiVaultScanRowObservers { private readonly owners = new Map() + private readonly observed = new Set() private readonly stops = new Map void>() private readonly rows = new Map() private readonly listeners = new Set<() => void>() @@ -54,18 +55,11 @@ export class AiVaultScanRowObservers { this.stops.set(observer, watchRowInterest(observer, this.changed)) } this.changed() - if (this.rows.size && rowsAreObserved(observer)) { - observer({ - kind: 'snapshot', - scanId: this.id, - revision: this.revision, - sessions: [...this.rows.values()] - }) - } return () => { const owners = this.owners.get(observer) ?? 0 if (owners <= 1) { this.owners.delete(observer) + this.observed.delete(observer) this.stops.get(observer)?.() this.stops.delete(observer) } else { @@ -77,6 +71,7 @@ export class AiVaultScanRowObservers { dispose(): void { this.closed = true this.owners.clear() + this.observed.clear() for (const stop of this.stops.values()) { stop() } @@ -86,16 +81,37 @@ export class AiVaultScanRowObservers { this.listeners.clear() } private readonly changed = (): void => { - const observing = !this.closed && [...this.owners.keys()].some(rowsAreObserved) - if (observing === this.observing) { - return + const activated: Observer[] = [] + for (const observer of this.owners.keys()) { + if (rowsAreObserved(observer)) { + if (!this.observed.has(observer)) { + this.observed.add(observer) + activated.push(observer) + } + } else { + this.observed.delete(observer) + } } + const observing = !this.closed && this.observed.size > 0 + const changed = observing !== this.observing this.observing = observing if (!observing) { this.rows.clear() } - for (const changed of this.listeners) { - changed() + for (const observer of activated) { + if (this.revision && this.observed.has(observer)) { + observer({ + kind: 'snapshot', + scanId: this.id, + revision: this.revision, + sessions: [...this.rows.values()] + }) + } + } + if (changed) { + for (const changed of this.listeners) { + changed() + } } } } diff --git a/src/main/ai-vault/scope/session-bucket-cwd.ts b/src/main/ai-vault/scope/session-bucket-cwd.ts index 61e812e2530..0296512dc5c 100644 --- a/src/main/ai-vault/scope/session-bucket-cwd.ts +++ b/src/main/ai-vault/scope/session-bucket-cwd.ts @@ -24,6 +24,16 @@ export async function readFirstCwd( agent: AiVaultAgent, signal?: AbortSignal ): Promise { + try { + return await readObservedCwd(filePath, signal) + } catch (err) { + throwIfAiVaultScanCancelled(signal) + recordRefusal(issues, filePath, err, agent) + return null + } +} + +async function readObservedCwd(filePath: string, signal?: AbortSignal): Promise { const observed = await wslGatedStat(filePath, 'scan', signal) const file: FileWithMtime = { path: filePath, @@ -62,10 +72,6 @@ export async function readFirstCwd( return cwd } } - } catch (err) { - throwIfAiVaultScanCancelled(signal) - recordRefusal(issues, filePath, err, agent) - return null } finally { // readline.close() leaves its input open; destroy the stream after early exits and read failures. lines.close() diff --git a/src/main/ai-vault/scope/session-scanner-scope-inventory.test.ts b/src/main/ai-vault/scope/session-scanner-scope-inventory.test.ts index 4909b96049e..a4feb0c24f6 100644 --- a/src/main/ai-vault/scope/session-scanner-scope-inventory.test.ts +++ b/src/main/ai-vault/scope/session-scanner-scope-inventory.test.ts @@ -1,4 +1,4 @@ -import { mkdtemp, rm } from 'node:fs/promises' +import { mkdtemp, rm, utimes } from 'node:fs/promises' import { PassThrough } from 'node:stream' import { tmpdir } from 'node:os' import { join } from 'node:path' @@ -9,11 +9,17 @@ import * as transcriptFs from '../../native-chat/transcript-filesystem/wsl-trans import { aiVaultScanWork, trackAiVaultScanProgress } from '../scan-lifecycle/ai-vault-scan-progress' import { CLAUDE_CWD_BUCKET_LAYOUT } from './session-cwd-bucket-layouts' +import { + cachedProjectDirCwd, + resetProjectDirCwdCacheForTests +} from './session-scanner-scope-discovery' +import type { AiVaultScanIssue } from '../../../shared/ai-vault/contracts/ai-vault-types' import { inventoryCwdBucketFiles } from './session-scanner-scope-inventory' import { writeJsonlFile } from '../scanner-fixtures/session-scanner-test-fixtures' let root: string | null = null afterEach(async () => { vi.restoreAllMocks() + resetProjectDirCwdCacheForTests() if (root) { await rm(root, { recursive: true, force: true }) } @@ -48,3 +54,35 @@ it('closes representative streams and inventory sources on cancellation before r stop() } }) + +it('continues cold representative lookup after a post-enumeration stat refusal', async () => { + root = await mkdtemp(join(tmpdir(), 'orca-bucket-refusal-')) + const bad = join(root, 'newest.jsonl') + const healthy = join(root, 'healthy.jsonl') + await writeJsonlFile(bad, [{ cwd: '/missing' }]) + await writeJsonlFile(healthy, [{ cwd: '/healthy' }]) + await utimes(bad, 2000, 2000) + await utimes(healthy, 1000, 1000) + const stat = transcriptFs.wslGatedStat + let observed = false + vi.spyOn(transcriptFs, 'wslGatedStat').mockImplementation((path, priority, signal) => { + if (path === bad) { + if (observed) { + return Promise.reject(Object.assign(new Error('refused'), { code: 'EACCES' })) + } + observed = true + } + return stat(path, priority, signal) + }) + const issues: AiVaultScanIssue[] = [] + await expect(cachedProjectDirCwd(root, (issue) => issues.push(issue), 'claude')).resolves.toBe( + '/healthy' + ) + expect(issues).toHaveLength(1) + expect(issues[0]?.path).toBe(bad) + const cancelled = new AbortController() + cancelled.abort() + await expect( + cachedProjectDirCwd(join(root, 'uncached'), () => undefined, 'claude', cancelled.signal) + ).rejects.toMatchObject({ name: 'AbortError' }) +}) diff --git a/src/main/ai-vault/session-scanner-reuse.test.ts b/src/main/ai-vault/session-scanner-reuse.test.ts index f615bfc379c..29a5c653a73 100644 --- a/src/main/ai-vault/session-scanner-reuse.test.ts +++ b/src/main/ai-vault/session-scanner-reuse.test.ts @@ -113,6 +113,8 @@ it.each([false, true])( const path = join(roots.codexSessionsDir, 'rollout-legacy.jsonl') const cache = { filePath: join(root, 'cache', 'parse.json'), appVersion: 'test' } await writeJsonlFile(path, codex('before')) + const timestamp = new Date('2026-11-13T12:00:00.000Z') + await fs.utimes(path, timestamp, timestamp) const before = await fs.stat(path) initSessionParseCachePersistence(cache) const first = await scanAiVaultSessions({ ...roots, unlimited: true }) @@ -125,8 +127,12 @@ it.each([false, true])( const replacement = join(root, 'replacement.jsonl') await writeJsonlFile(replacement, codex('after!')) expect((await fs.stat(replacement)).size).toBe(before.size) - await fs.utimes(replacement, before.atime, before.mtime) + await fs.utimes(replacement, timestamp, timestamp) await fs.rename(replacement, path) + const replaced = await fs.stat(path) + expect(replaced.mtimeMs).toBe(before.mtimeMs) + expect(replaced.size).toBe(before.size) + expect(replaced.ino).not.toBe(before.ino) if (restart) { initSessionParseCachePersistence(cache) } diff --git a/src/main/ipc/ai-vault/ai-vault-local-list.ts b/src/main/ipc/ai-vault/ai-vault-local-list.ts index 43ebeb21d4f..f0e78e86100 100644 --- a/src/main/ipc/ai-vault/ai-vault-local-list.ts +++ b/src/main/ipc/ai-vault/ai-vault-local-list.ts @@ -19,9 +19,26 @@ export async function scanLocalAiVaultSessionsAsIssue( publish?: (delta: AiVaultRowDelta) => void ): Promise { try { + const forwarded = new Set() const observe = publish - ? (rows: AiVaultScanRows) => - publish(rows.kind === 'delta' ? rows.delta : { upserts: [...rows.sessions], removed: [] }) + ? (rows: AiVaultScanRows) => { + const snapshotIds = + rows.kind === 'snapshot' ? new Set(rows.sessions.map((row) => row.id)) : null + const delta = + rows.kind === 'delta' + ? rows.delta + : { + upserts: [...rows.sessions], + removed: [...forwarded].filter((id) => !snapshotIds?.has(id)) + } + for (const id of delta.removed) { + forwarded.delete(id) + } + for (const row of delta.upserts) { + forwarded.add(row.id) + } + publish(delta) + } : undefined if (observe && publish) { forwardRowInterest(observe, publish) diff --git a/src/renderer/src/components/right-sidebar/ai-vault-provisional-rows.ts b/src/renderer/src/components/right-sidebar/ai-vault-provisional-rows.ts index 8b8f8a0e8ed..6be36194876 100644 --- a/src/renderer/src/components/right-sidebar/ai-vault-provisional-rows.ts +++ b/src/renderer/src/components/right-sidebar/ai-vault-provisional-rows.ts @@ -106,6 +106,7 @@ export function observeAiVaultProvisionalRows(args: { const preview = new AiVaultProvisionalRows(args.baseline, args.deleted, args.limit) let observing = true let rows: readonly AiVaultSession[] = args.baseline + let cancelPublication: (() => void) | undefined const unsubscribe = window.api.aiVault.onScanRows?.((event) => { if (!observing || event.requestToken !== args.requestToken || !args.isCurrent()) { return @@ -113,16 +114,17 @@ export function observeAiVaultProvisionalRows(args: { if (!preview.accept(event)) { return } - args.gate.publishProvisional(() => { - if (!observing || !args.isCurrent()) { - return - } - rows = preview.snapshot(rows) - args.apply(rows) - }) + cancelPublication = args.gate.publishProvisional( + () => { + rows = preview.snapshot(rows) + args.apply(rows) + }, + () => observing && args.isCurrent() + ) }) return () => { observing = false + cancelPublication?.() preview.retire() unsubscribe?.() } diff --git a/src/renderer/src/components/right-sidebar/ai-vault/scanning/ai-vault-session-publication-gate.ts b/src/renderer/src/components/right-sidebar/ai-vault/scanning/ai-vault-session-publication-gate.ts index e24e4dc256f..22c0ff3f24a 100644 --- a/src/renderer/src/components/right-sidebar/ai-vault/scanning/ai-vault-session-publication-gate.ts +++ b/src/renderer/src/components/right-sidebar/ai-vault/scanning/ai-vault-session-publication-gate.ts @@ -25,41 +25,60 @@ export class AiVaultSessionPublicationGate { } publish(result: AiVaultListResult, apply: (result: AiVaultListResult) => void): void { - this.cancel() - const generation = this.generation - // Counted when it lands: a cancelled publication must not drop a saved list still on its way. - const land = () => { - this.published += 1 - startTransition(() => apply(result)) - } - if (hasInputBeenQuietFor(AI_VAULT_PUBLICATION_QUIET_MS)) { - land() - return - } - this.cancelPending = scheduleAfterInputQuiet( - () => { - if (generation !== this.generation) { - return - } - this.cancelPending = null - land() - }, - { - delayMs: 0, - quietMs: AI_VAULT_PUBLICATION_QUIET_MS, - idleTimeoutMs: AI_VAULT_PUBLICATION_QUIET_MS, - maxWaitMs: AI_VAULT_PUBLICATION_MAX_WAIT_MS - } + this.enqueue( + () => apply(result), + () => true ) } - publishProvisional(apply: () => void): void { - this.publish({ sessions: [], issues: [], scannedAt: '' }, apply) + publishProvisional(apply: () => void, eligible: () => boolean): () => void { + return this.enqueue(apply, eligible) + } + + private pending: (() => void) | null = null + + private enqueue(apply: () => void, eligible: () => boolean): () => void { + const land = () => { + if (eligible()) { + this.published += 1 + startTransition(apply) + } + } + this.pending = land + if (hasInputBeenQuietFor(AI_VAULT_PUBLICATION_QUIET_MS)) { + this.cancel() + land() + } else if (!this.cancelPending) { + const generation = this.generation + this.cancelPending = scheduleAfterInputQuiet( + () => { + if (generation !== this.generation) { + return + } + const pending = this.pending + this.pending = null + this.cancelPending = null + pending?.() + }, + { + delayMs: 0, + quietMs: AI_VAULT_PUBLICATION_QUIET_MS, + idleTimeoutMs: AI_VAULT_PUBLICATION_QUIET_MS, + maxWaitMs: AI_VAULT_PUBLICATION_MAX_WAIT_MS + } + ) + } + return () => { + if (this.pending === land) { + this.cancel() + } + } } cancel(): void { this.generation += 1 this.cancelPending?.() this.cancelPending = null + this.pending = null } } diff --git a/src/renderer/src/components/right-sidebar/ai-vault/scanning/ai-vault-session-refresh.test.ts b/src/renderer/src/components/right-sidebar/ai-vault/scanning/ai-vault-session-refresh.test.ts index 80092e19973..e34e57ca868 100644 --- a/src/renderer/src/components/right-sidebar/ai-vault/scanning/ai-vault-session-refresh.test.ts +++ b/src/renderer/src/components/right-sidebar/ai-vault/scanning/ai-vault-session-refresh.test.ts @@ -1,5 +1,11 @@ // @vitest-environment happy-dom +import { AiVaultSessionPublicationGate } from './ai-vault-session-publication-gate' +import { observeAiVaultProvisionalRows } from '../../ai-vault-provisional-rows' +import { + markInputQuietSchedulerInput, + resetInputQuietSchedulerForTest +} from '@/lib/window-activity/input-quiet-scheduler' import { act, createElement } from 'react' import { createRoot, type Root } from 'react-dom/client' import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' @@ -151,6 +157,7 @@ function lastCallArgs(): unknown { beforeEach(() => { vi.useFakeTimers() + resetInputQuietSchedulerForTest() listSessionsMock.mockReset().mockResolvedValue(EMPTY_RESULT) cancelListSessionsMock.mockReset().mockResolvedValue() rowObservers.clear() @@ -388,3 +395,85 @@ describe('useAiVaultSessionRefresh after a restart', () => { }) // Plain HTTP hides crypto.randomUUID; fallback tokens must remain v4 UUIDs (#18096). + +describe('progressive History publication ownership', () => { + function observe( + gate: AiVaultSessionPublicationGate, + apply: (rows: readonly AiVaultSession[]) => void + ) { + return observeAiVaultProvisionalRows({ + requestToken: 'token', + baseline: [], + deleted: new Set(), + gate, + isCurrent: () => true, + apply + }) + } + function emit(sequence: number) { + for (const callback of rowObservers) { + callback({ + requestToken: 'token', + scanId: 'scan', + sequence, + reset: sequence === 1, + upserts: [makeVaultSession(sequence)], + removed: sequence > 1 ? [`session-${sequence - 1}`] : [] + }) + } + } + + it('publishes the latest rows by the first deadline despite continuous input and batches', async () => { + vi.spyOn(performance, 'now').mockImplementation(() => Date.now()) + const gate = new AiVaultSessionPublicationGate() + const reentrant = vi.fn() + const apply = vi.fn((_rows: readonly AiVaultSession[]) => gate.publish(EMPTY_RESULT, reentrant)) + const stop = observe(gate, apply) + markInputQuietSchedulerInput() + emit(1) + for (let sequence = 2; sequence <= 20; sequence++) { + await vi.advanceTimersByTimeAsync(50) + markInputQuietSchedulerInput() + emit(sequence) + } + expect(apply).not.toHaveBeenCalled() + await vi.advanceTimersByTimeAsync(50) + expect(apply).toHaveBeenCalledOnce() + expect(apply.mock.calls[0]?.[0]?.map((row) => row.id)).toEqual(['session-20']) + await vi.advanceTimersByTimeAsync(1000) + expect(reentrant).toHaveBeenCalledOnce() + stop() + gate.cancel() + }) + + it('lets delayed saved rows appear after a queued provisional owner retires', async () => { + const gate = new AiVaultSessionPublicationGate() + const saved = deferredList() + const applySaved = vi.fn() + gate.showUntilPublished(saved.promise, applySaved) + const apply = vi.fn() + const stop = observe(gate, apply) + markInputQuietSchedulerInput() + emit(1) + stop() + await vi.advanceTimersByTimeAsync(1000) + saved.resolve(EMPTY_RESULT) + await flushMicrotasks() + expect(apply).not.toHaveBeenCalled() + expect(applySaved).toHaveBeenCalledWith(EMPTY_RESULT) + gate.cancel() + }) + + it('keeps an authoritative result that replaced a retiring provisional callback', async () => { + const gate = new AiVaultSessionPublicationGate() + const stop = observe(gate, vi.fn()) + const applyFinal = vi.fn() + markInputQuietSchedulerInput() + emit(1) + gate.publish(EMPTY_RESULT, applyFinal) + stop() + await vi.advanceTimersByTimeAsync(1000) + expect(applyFinal).toHaveBeenCalledOnce() + gate.cancel() + }) +})