Preserve History scan ownership and publication across cancellation

This commit is contained in:
Brennan Benson
2026-11-13 23:41:55 -08:00
parent 062a17dec1
commit 66887fdcdc
15 changed files with 639 additions and 160 deletions
@@ -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<typeof EMPTY>()
const entered = Promise.withResolvers<void>()
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<typeof EMPTY>()
const entered = Promise.withResolvers<void>()
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<string, typeof firstRow>()
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<typeof EMPTY>()
const oldEntered = Promise.withResolvers<void>()
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<typeof EMPTY>()
const entered = Promise.withResolvers<void>()
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)
})
@@ -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<string, CachedAiVaultList>()
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
}
})
@@ -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()
})
@@ -98,7 +98,8 @@ export async function flushSessionParseCachePersist(): Promise<void> {
}
}
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()
}
@@ -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<T>() {
@@ -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<typeof EMPTY>()
const start = vi.fn(() => pending.promise)
const first = coordinator.run({ key: 'scope', start })
await Promise.resolve()
const survivors: Promise<typeof EMPTY>[] = []
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)
})
@@ -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<string, Entry>()
private readonly entries = new Map<string, ExecutionOwner>()
run(args: Run): Promise<AiVaultListResult> {
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<AiVaultListResult>()
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<AiVaultListResult>): 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<typeof Promise.withResolvers<AiVaultListResult>>,
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<AiVaultListResult> {
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<AiVaultListResult> {
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
}
@@ -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)
}
}
@@ -5,6 +5,7 @@ type Observer = (rows: AiVaultScanRows) => void
export class AiVaultScanRowObservers {
private readonly owners = new Map<Observer, number>()
private readonly observed = new Set<Observer>()
private readonly stops = new Map<Observer, () => void>()
private readonly rows = new Map<string, AiVaultSession>()
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()
}
}
}
}
+10 -4
View File
@@ -24,6 +24,16 @@ export async function readFirstCwd(
agent: AiVaultAgent,
signal?: AbortSignal
): Promise<string | null> {
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<string | null> {
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()
@@ -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' })
})
@@ -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)
}
+19 -2
View File
@@ -19,9 +19,26 @@ export async function scanLocalAiVaultSessionsAsIssue(
publish?: (delta: AiVaultRowDelta) => void
): Promise<AiVaultListResult> {
try {
const forwarded = new Set<string>()
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)
@@ -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?.()
}
@@ -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
}
}
@@ -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()
})
})