Honor coalesced refresh requirements without retaining detached recovery owners

This commit is contained in:
Brennan Benson
2026-11-13 21:10:12 -08:00
parent dc519bd4a4
commit b6b4bef01b
7 changed files with 201 additions and 49 deletions
@@ -115,6 +115,7 @@ export async function listAiVaultSessions(
onReject: (error, scanId) =>
isAiVaultScanCancelledError(error) ? error : new AiVaultLocalScanError(scanId, error),
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.')
@@ -157,14 +158,14 @@ export async function listAiVaultSessions(
) {
const current = cachedLists.get(key)
if (
args?.force === true ||
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 (args?.force === true) {
if (force) {
cachedLists.clear()
forcedRefreshEpoch++
}
@@ -140,3 +140,28 @@ it('runs a fresh pass for a force that arrives after a recovered successor began
])
expect(start).toHaveBeenCalledOnce()
})
it('drops a detached newest requirement so a recovered observation can serve surviving forces', async () => {
const coordinator = new AiVaultScanCoordinator()
const pending = Promise.withResolvers<typeof EMPTY>()
let signal!: AbortSignal
const first = coordinator.run({
key: 'scope',
start: (s) => {
signal = s
return pending.promise
}
})
await Promise.resolve()
const start = vi.fn(async () => EMPTY)
const force = coordinator.run({ key: 'scope', force: true, start })
const interest = scanRecoveryInterest(signal)!
interest.observation = nextScanObservationOrder()
const controller = new AbortController()
const late = coordinator.run({ key: 'scope', force: true, signal: controller.signal, start })
controller.abort()
await expect(late).rejects.toMatchObject({ name: 'AbortError' })
pending.resolve(EMPTY)
await Promise.all([first, force])
expect(start).not.toHaveBeenCalled()
})
@@ -167,10 +167,10 @@ export class AiVaultScanCoordinator {
private attach(args: Run, entry: Entry): Promise<AiVaultListResult> {
entry.waiters++
const interest = scanRecoveryInterest(entry.controller.signal)
const forced = !!args.requiredOrder
const forced = args.force === true && scanRecoveryInterest(args.signal) === undefined
const releaseParent = inheritScanRecoveryInterest(args.signal, entry.controller.signal)
const obligation = Symbol()
if (forced) {
if (args.requiredOrder) {
entry.requirements.set(obligation, args.requiredOrder ?? 0)
}
if (forced && interest) {
@@ -18,7 +18,7 @@ let order = 0
export const nextScanObservationOrder = (): number => ++order
export function scanForceOrder(force: boolean | undefined, parent?: AbortSignal): number {
return force ? scanRecoveryInterest(parent)?.required || nextScanObservationOrder() : 0
return scanRecoveryInterest(parent)?.required || (force ? nextScanObservationOrder() : 0)
}
export function createScanRecoveryInterest(signal: AbortSignal): Interest {
@@ -0,0 +1,111 @@
import { afterEach, expect, it, vi } from 'vitest'
const mocks = vi.hoisted(() => ({ scan: vi.fn(), runtime: vi.fn() }))
vi.mock('electron', () => ({ app: { on: vi.fn() }, ipcMain: { handle: vi.fn(), on: vi.fn() } }))
vi.mock('../../ai-vault/scan-service/session-scanner-background', () => ({
scanAiVaultSessionsInBackground: mocks.scan,
clearAiVaultBackgroundRestartCircuit: vi.fn(),
resetAiVaultScannerBackgroundForTests: vi.fn()
}))
vi.mock('../../ai-vault/discovery/ai-vault-scan-roots', () => ({
configureAiVaultSessionSources: vi.fn(),
resetAiVaultSessionSourcesForTests: vi.fn(),
resolveLocalAiVaultScanRoots: vi.fn(async () => ({
roots: { additionalCodexSessionsDirs: [], wslHomeDirs: [] },
wslConfirmed: true
}))
}))
vi.mock('../../ai-vault/runtime-hosts/ai-vault-runtime-scan', () => ({
scanRuntimeAiVaultSessions: mocks.runtime
}))
import { _internals, registerAiVaultHandlers } from './ai-vault'
import { listAiVaultSessions as localList } from '../../ai-vault/listing/cached-session-list'
import { listResult } from '../../ai-vault/structured-sessions/structured-session-ownership.test-support'
const empty = { sessions: [], issues: [], scannedAt: 'empty' }
afterEach(() => {
_internals.resetAiVaultCacheForTests()
vi.resetAllMocks()
vi.useRealTimers()
})
it.each(['local', 'all'] as const)(
'honors a force absorbed before an ordinary %s cohort starts, bypassing both caches',
async (scope) => {
registerAiVaultHandlers()
mocks.scan.mockResolvedValue(empty)
mocks.runtime.mockResolvedValue(empty)
const args = { executionHostScope: scope }
await _internals.listAiVaultSessions(args)
const changed = { ...listResult(), scannedAt: 'fresh' }
mocks.scan.mockResolvedValue(changed)
await expect(_internals.listAiVaultSessions(args)).resolves.toMatchObject({
scannedAt: expect.any(String)
})
expect(mocks.scan).toHaveBeenCalledOnce()
const ordinary = _internals.listAiVaultSessions(args)
const forced = _internals.listAiVaultSessions({ ...args, force: true })
const [, result] = await Promise.all([ordinary, forced])
expect(result.sessions.map((row) => row.id)).toEqual(changed.sessions.map((row) => row.id))
expect(mocks.scan).toHaveBeenCalledTimes(2)
}
)
it('disarms nested recovery after the last force detaches while an ordinary owner remains', async () => {
vi.useFakeTimers()
registerAiVaultHandlers()
let signal!: AbortSignal
const pending = Promise.withResolvers<typeof empty>()
mocks.scan.mockImplementation((_options, attempt: AbortSignal) => {
signal = attempt
return pending.promise
})
const controller = new AbortController()
const forced = _internals.listAiVaultSessions({ force: true }, { signal: controller.signal })
const cancelled = expect(forced).rejects.toMatchObject({ name: 'AbortError' })
await vi.advanceTimersByTimeAsync(0)
const ordinary = _internals.listAiVaultSessions()
controller.abort()
await cancelled
const { SessionScanInactivity } =
await import('../../ai-vault/scan-service/session-scanner-scan-inactivity')
const stalled = vi.fn()
const inactivity = new SessionScanInactivity(signal, stalled)
inactivity.activity(0)
await vi.advanceTimersByTimeAsync(120000)
expect(signal.aborted).toBe(false)
expect(stalled).not.toHaveBeenCalled()
pending.resolve(empty)
await ordinary
inactivity.dispose()
})
it('refreshes an old remote leg while reusing recovered local completion and its terminal issue', async () => {
vi.useFakeTimers()
const { AiVaultScanStalledError } =
await import('../../ai-vault/scan-service/session-scanner-scan-inactivity')
registerAiVaultHandlers({
getActiveRuntimeAiVaultHostInfos: () => [
{ environmentId: 'peer', executionHostId: 'runtime:peer' }
]
})
const entered = Promise.withResolvers<void>()
const stalled = Promise.withResolvers<typeof empty>()
mocks.scan
.mockImplementationOnce(() => {
entered.resolve()
return stalled.promise
})
.mockRejectedValue(new AiVaultScanStalledError())
mocks.runtime.mockResolvedValue(empty)
const runtime = localList()
const runtimeError = expect(runtime).rejects.toThrow('stopped making progress')
await entered.promise
const all = _internals.listAiVaultSessions({ executionHostScope: 'all' })
await vi.advanceTimersByTimeAsync(0)
const forced = _internals.listAiVaultSessions({ executionHostScope: 'all', force: true })
stalled.reject(new AiVaultScanStalledError())
await runtimeError
const results = await Promise.all([all, forced])
expect(results.every((result) => result.issues.length > 0)).toBe(true)
expect(mocks.scan).toHaveBeenCalledTimes(2)
expect(mocks.runtime).toHaveBeenCalledTimes(2)
})
@@ -0,0 +1,44 @@
import { forwardRowInterest } from '../../ai-vault/scan-lifecycle/ai-vault-row-interest'
import type { AiVaultRowDelta, AiVaultScanRows } from '../../../shared/ai-vault-scan-rows'
import {
listAiVaultSessions as listCachedLocalAiVaultSessions,
AiVaultLocalScanError
} from '../../ai-vault/listing/cached-session-list'
import { aiVaultScanIssueResult } from '../../ai-vault/listing/session-list-results'
import { classifyAiVaultScanError } from '../../../shared/ai-vault/scanning/ai-vault-scan-error-message'
import {
isAiVaultScanCancelledError,
type AiVaultListArgs,
type AiVaultListResult
} from '../../../shared/ai-vault/contracts/ai-vault-types'
import { LOCAL_EXECUTION_HOST_ID } from '../../../shared/execution-host'
// Like remote failures, local scanner/cache/WSL failures must degrade to issues without discarding healthy hosts.
export async function scanLocalAiVaultSessionsAsIssue(
args: AiVaultListArgs | undefined,
signal: AbortSignal | undefined,
publish?: (delta: AiVaultRowDelta) => void
): Promise<AiVaultListResult> {
try {
const observe = publish
? (rows: AiVaultScanRows) =>
publish(rows.kind === 'delta' ? rows.delta : { upserts: [...rows.sessions], removed: [] })
: undefined
if (observe && publish) {
forwardRowInterest(observe, publish)
}
return await listCachedLocalAiVaultSessions(args, { signal, observeRows: observe })
} catch (error) {
if (isAiVaultScanCancelledError(error)) {
throw error
}
// Keep raw supervision details in logs; show an actionable host issue.
const raw = error instanceof Error ? error.message : 'Local session scan failed.'
console.error('[ai-vault] local session scan failed:', raw)
const result = aiVaultScanIssueResult({
executionHostId: LOCAL_EXECUTION_HOST_ID,
path: 'this computer',
...classifyAiVaultScanError(raw)
})
return error instanceof AiVaultLocalScanError ? { ...result, scanId: error.scanId } : result
}
}
+15 -44
View File
@@ -1,8 +1,7 @@
import { scanLocalAiVaultSessionsAsIssue } from './ai-vault-local-list'
import { scanRemoteLeg } from './ai-vault-remote-scan-observation'
import {
forwardRowInterest,
rowsAreObserved
} from '../../ai-vault/scan-lifecycle/ai-vault-row-interest'
import { scanRecoveryInterest } from '../../ai-vault/scan-lifecycle/ai-vault-scan-recovery'
import { rowsAreObserved } from '../../ai-vault/scan-lifecycle/ai-vault-row-interest'
import type { AiVaultRowDelta, AiVaultScanRows } from '../../../shared/ai-vault-scan-rows'
import { registerAiVaultListHandler } from './ai-vault-list-handler'
import {
@@ -11,11 +10,7 @@ import {
} from './ai-vault-projection-dependency'
import { app, ipcMain } from 'electron'
import { configureAiVaultSessionSources } from '../../ai-vault/discovery/ai-vault-scan-roots'
import {
listAiVaultSessions as listCachedLocalAiVaultSessions,
AiVaultLocalScanError,
resetAiVaultSessionListCacheForTests
} from '../../ai-vault/listing/cached-session-list'
import { resetAiVaultSessionListCacheForTests } from '../../ai-vault/listing/cached-session-list'
import type { AiVaultSessionSources } from '../../ai-vault/discovery/ai-vault-scan-roots'
import { deleteAiVaultSession, registerAiVaultDeleteHandler } from './ai-vault-delete'
import { listAiVaultSubagentSessions } from './ai-vault-subagent-list'
@@ -24,10 +19,8 @@ import {
mergeAiVaultListResults
} from '../../ai-vault/listing/session-list-results'
import type { AiVaultDeleteSessionArgs } from '../../../shared/ai-vault/contracts/ai-vault-session-deletion'
import { classifyAiVaultScanError } from '../../../shared/ai-vault/scanning/ai-vault-scan-error-message'
import {
AI_VAULT_SCOPE_PATHS_MAX_COUNT,
isAiVaultScanCancelledError,
type AiVaultFirstUserPromptArgs,
type AiVaultListArgs,
type AiVaultListResult,
@@ -136,8 +129,17 @@ async function listAiVaultSessions(
observeRows: options.observeRows,
signal: options.signal,
start: (scanSignal, _scanId, publish) => {
const force = (scanRecoveryInterest(scanSignal)?.required ?? 0) > 0
const effectiveArgs = force ? { ...args, force: true } : args
const scan = () =>
scanAiVaultSessionsByHostScope(args, hostScope, scanSignal, key, dependency, publish)
scanAiVaultSessionsByHostScope(
effectiveArgs,
hostScope,
scanSignal,
key,
dependency,
publish
)
if (executionHostScope === LOCAL_EXECUTION_HOST_ID) {
return scan()
}
@@ -146,7 +148,7 @@ async function listAiVaultSessions(
dependency,
depth,
scopePaths,
force: args?.force === true,
force,
afterFingerprint,
scan
})
@@ -238,37 +240,6 @@ export function getActiveRuntimeAiVaultHostInfosResult(): AiVaultHostDiscoveryRe
})
}
// Like remote failures, local scanner/cache/WSL failures must degrade to issues without discarding healthy hosts.
async function scanLocalAiVaultSessionsAsIssue(
args: AiVaultListArgs | undefined,
signal: AbortSignal | undefined,
publish?: (delta: AiVaultRowDelta) => void
): Promise<AiVaultListResult> {
try {
const observe = publish
? (rows: AiVaultScanRows) =>
publish(rows.kind === 'delta' ? rows.delta : { upserts: [...rows.sessions], removed: [] })
: undefined
if (observe && publish) {
forwardRowInterest(observe, publish)
}
return await listCachedLocalAiVaultSessions(args, { signal, observeRows: observe })
} catch (error) {
if (isAiVaultScanCancelledError(error)) {
throw error
}
// Keep raw supervision details in logs; show an actionable host issue.
const raw = error instanceof Error ? error.message : 'Local session scan failed.'
console.error('[ai-vault] local session scan failed:', raw)
const result = aiVaultScanIssueResult({
executionHostId: LOCAL_EXECUTION_HOST_ID,
path: 'this computer',
...classifyAiVaultScanError(raw)
})
return error instanceof AiVaultLocalScanError ? { ...result, scanId: error.scanId } : result
}
}
export function registerAiVaultHandlers(options: AiVaultHandlerOptions = {}): void {
handlerOptions = options
// Configure the same source/cache instance used by runtime RPC and headless serve.