From b6b4bef01b0b31dfd2ee33d4270e06c1b61e174f Mon Sep 17 00:00:00 2001 From: Brennan Benson <79079362+brennanb2025@users.noreply.github.com> Date: Fri, 13 Nov 2026 21:10:12 -0800 Subject: [PATCH] Honor coalesced refresh requirements without retaining detached recovery owners --- .../ai-vault/listing/cached-session-list.ts | 5 +- .../ai-vault-scan-coordinator.test.ts | 25 ++++ .../ai-vault-scan-coordinator.ts | 4 +- .../scan-lifecycle/ai-vault-scan-recovery.ts | 2 +- .../ai-vault/ai-vault-list-freshness.test.ts | 111 ++++++++++++++++++ src/main/ipc/ai-vault/ai-vault-local-list.ts | 44 +++++++ src/main/ipc/ai-vault/ai-vault.ts | 59 +++------- 7 files changed, 201 insertions(+), 49 deletions(-) create mode 100644 src/main/ipc/ai-vault/ai-vault-list-freshness.test.ts create mode 100644 src/main/ipc/ai-vault/ai-vault-local-list.ts diff --git a/src/main/ai-vault/listing/cached-session-list.ts b/src/main/ai-vault/listing/cached-session-list.ts index 51775fb32ab..ea4257bc01c 100644 --- a/src/main/ai-vault/listing/cached-session-list.ts +++ b/src/main/ai-vault/listing/cached-session-list.ts @@ -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++ } 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 0ca76b17731..d7821366a22 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 @@ -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() + 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() +}) 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 d94345661ec..7d42cf82639 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 @@ -167,10 +167,10 @@ export class AiVaultScanCoordinator { private attach(args: Run, entry: Entry): Promise { 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) { 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 ba37fde8ee0..c3f950741a7 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 @@ -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 { diff --git a/src/main/ipc/ai-vault/ai-vault-list-freshness.test.ts b/src/main/ipc/ai-vault/ai-vault-list-freshness.test.ts new file mode 100644 index 00000000000..b0fc69124f6 --- /dev/null +++ b/src/main/ipc/ai-vault/ai-vault-list-freshness.test.ts @@ -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() + 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() + const stalled = Promise.withResolvers() + 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) +}) diff --git a/src/main/ipc/ai-vault/ai-vault-local-list.ts b/src/main/ipc/ai-vault/ai-vault-local-list.ts new file mode 100644 index 00000000000..43ebeb21d4f --- /dev/null +++ b/src/main/ipc/ai-vault/ai-vault-local-list.ts @@ -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 { + 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 + } +} diff --git a/src/main/ipc/ai-vault/ai-vault.ts b/src/main/ipc/ai-vault/ai-vault.ts index 31d101f0e2b..01cff9275b3 100644 --- a/src/main/ipc/ai-vault/ai-vault.ts +++ b/src/main/ipc/ai-vault/ai-vault.ts @@ -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 { - 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.