From 865cdaa7062624040ca3d86d0c74c8c4e15411f7 Mon Sep 17 00:00:00 2001 From: Brennan Benson <79079362+brennanb2025@users.noreply.github.com> Date: Thu, 15 Oct 2026 18:28:43 -0700 Subject: [PATCH] fix(native-chat): acting on one chat no longer waits on checks of every other chat (#26555) * fix(native-chat): acting on one chat no longer waits on checks of every other chat Starting, resuming or switching a chat's agent first ran restart reconciliation, which probed every unreconciled session and committed them together, so one slow or failing probe for another chat delayed or failed this chat's action. Acquisition now reconciles only the addressed session; the explicit startup pass still covers all of them, and an owner that isn't proven still can't be granted a writer. Co-Authored-By: OpenAI Codex gpt-6.1-sol Co-Authored-By: Claude Opus 5.5 * fix(native-chat): settle each restored chat on its own; register new sqlite tests - Startup restore no longer lets one chat's failed lease check skip recovery for the others. - Run the two new SQLite-backed tests in the Node runtime project. - Startup reconcile runs the whole-host pass without a placeholder session id. - Test names describe an injected failing/stalling owner check, not an SSH host. * fix(native-chat): restore keeps one whole-host lease check per startup The restore opens with the whole-host pass again and stops per-chat bookkeeping after the first failure, as on main: on a failing store it makes one lease-check write, not one per chat. User actions stay targeted to their own chat. --------- Co-authored-by: OpenAI Codex gpt-6.1-sol Co-authored-by: Claude Opus 5.5 --- .../scripts/vitest-sqlite-runtime-files.mjs | 2 + ...ed-agent-session-acquire-isolation.test.ts | 46 ++++ .../structured-agent-session-host.ts | 1 + ...ed-agent-session-readable-restorer.test.ts | 1 + ...ed-agent-session-restart-reconcile.test.ts | 16 +- ...uctured-agent-session-restart-reconcile.ts | 68 +++--- ...ured-agent-session-restart-restore.test.ts | 34 ++- ...tructured-agent-session-restart-restore.ts | 14 +- .../structured-agent-session-reveal.ts | 9 +- ...red-agent-session-target-reconcile.test.ts | 199 ++++++++++++++++++ .../runtime/agent-session-record-store.ts | 7 +- .../agent-session-restart-reconciliation.ts | 12 +- 12 files changed, 353 insertions(+), 56 deletions(-) create mode 100644 src/main/native-chat/agent-session-wire/structured-agent-session-acquire-isolation.test.ts create mode 100644 src/main/native-chat/agent-session-wire/structured-agent-session-target-reconcile.test.ts diff --git a/config/scripts/vitest-sqlite-runtime-files.mjs b/config/scripts/vitest-sqlite-runtime-files.mjs index fb1f69c5f87..4971734c4ff 100644 --- a/config/scripts/vitest-sqlite-runtime-files.mjs +++ b/config/scripts/vitest-sqlite-runtime-files.mjs @@ -73,6 +73,7 @@ export const SQLITE_RUNTIME_INCLUDE = [ 'src/main/native-chat/agent-session-wire/agent-session-history-page.test.ts', 'src/main/native-chat/agent-session-wire/agent-session-stop-note-projection.test.ts', 'src/main/native-chat/agent-session-wire/structured-agent-session-accept-then-deliver.test.ts', + 'src/main/native-chat/agent-session-wire/structured-agent-session-acquire-isolation.test.ts', 'src/main/native-chat/agent-session-wire/structured-agent-session-acquisition-options.test.ts', 'src/main/native-chat/agent-session-wire/structured-agent-session-acquisition-runtime-stamp.test.ts', 'src/main/native-chat/agent-session-wire/structured-agent-session-adopted-import.test.ts', @@ -199,6 +200,7 @@ export const SQLITE_RUNTIME_INCLUDE = [ 'src/main/native-chat/agent-session-wire/structured-agent-session-subagent-rows.test.ts', 'src/main/native-chat/agent-session-wire/structured-agent-session-subscriber-background-tasks.test.ts', 'src/main/native-chat/agent-session-wire/structured-agent-session-subscribers.test.ts', + 'src/main/native-chat/agent-session-wire/structured-agent-session-target-reconcile.test.ts', 'src/main/native-chat/agent-session-wire/structured-agent-session-thread-goal.test.ts', 'src/main/native-chat/agent-session-wire/structured-agent-session-transition.test.ts', 'src/main/native-chat/agent-session-wire/structured-agent-session-turns.test.ts', diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-acquire-isolation.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-acquire-isolation.test.ts new file mode 100644 index 00000000000..f04c3f49b13 --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-acquire-isolation.test.ts @@ -0,0 +1,46 @@ +import { expect, it, vi } from 'vitest' +import { openTestAgentSessionRecordStore } from '../../runtime/agent-session-record-store-test-harness' +import { StructuredAgentSessionHost } from './structured-agent-session-host' +import { codexProviderHandle } from '../../../shared/agent-session-provider-handle-encoding' +import { HOST_TEST_NOW, HOST_TEST_THREAD } from './structured-agent-session-host-test-data' +import { + CALLER, + attach, + attachParams, + hostTestState, + replaceHostTestState +} from './structured-agent-session-host-test-harness' + +it("creates a chat while another saved chat's owner check fails", async () => { + await attach() + const prior = hostTestState() + prior.acquire.mockImplementation(async ({ fence, spawnToken }) => ({ + process: { hostId: 'local', pid: 4242, processStartTimeMs: HOST_TEST_NOW, spawnToken }, + link: { + linkId: 'new-link', + handle: codexProviderHandle(HOST_TEST_THREAD), + origin: 'created', + mintedAtFence: fence, + observedAt: HOST_TEST_NOW + } + })) + const peerId = attachParams().envelope.sessionId + const store = await openTestAgentSessionRecordStore(prior.root) + const probeOwners = vi.fn(async () => { + throw new Error('owner check failed') + }) + const host = new StructuredAgentSessionHost({ ...prior.host.deps, store, probeOwners }) + replaceHostTestState({ store, host }) + const request = attachParams() + const params = attachParams({ + envelope: { ...request.envelope, sessionId: 'new-chat' }, + accountHome: { variable: 'CODEX_HOME', path: prior.root } + }) + + const result = await host.attach(CALLER, params) + + expect(result.ok, JSON.stringify(result)).toBe(true) + expect(probeOwners).not.toHaveBeenCalled() + expect(store.getRecord(peerId)?.lease.unreconciled).toBe(true) + expect(store.getRecord('new-chat')?.lease.claimStatus).toBe('live') +}) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-host.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-host.ts index 18781369de4..2395ac67d83 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-host.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-host.ts @@ -130,6 +130,7 @@ export class StructuredAgentSessionHost { }) this.restore = reveal.createStructuredAgentSessionHostRestore(deps, { reconcileLeases: this.reconcileLeases, + reconcileRestartLeases: () => this.reconcileLeases(), resolveRecovery: (sessionId) => this.runtimeState.resolveRecovery(sessionId), serialize: (sessionId, task) => this.serialize(sessionId, task), hasSession: this.hasSession, diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-readable-restorer.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-readable-restorer.test.ts index 87c03c071ea..dd682d0e276 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-readable-restorer.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-readable-restorer.test.ts @@ -43,6 +43,7 @@ describe('StructuredAgentSessionReadableRestorer', () => { journalDatabase: openTestJournalHostDatabase(stateDirectory), logger: recordingStructuredAgentSessionLogger().logger }, + reconcileAll: async () => true, reconcile: async () => true, resolveRecovery: async () => true, serialize: async (_sessionId, task) => task(), diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-restart-reconcile.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-restart-reconcile.test.ts index e61587a6598..595fc8672f1 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-restart-reconcile.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-restart-reconcile.test.ts @@ -27,7 +27,7 @@ describe('createRestartReconciler', () => { expect(reconcileOnRestart).toHaveBeenCalledTimes(2) }) - it('passes every pending record through the batch owner probe', async () => { + it('passes only the addressed record through the batch owner probe', async () => { let records = [ { sessionId: 'session-1', lease: { unreconciled: true } }, { sessionId: 'session-2', lease: { unreconciled: true } } @@ -40,14 +40,18 @@ describe('createRestartReconciler', () => { }) const reconcileOnRestart = vi.fn( async (args: { + sessionId?: string probeMany?: ( pending: readonly AgentSessionRecord[] ) => Promise> }) => { - await args.probeMany?.(records) + await args.probeMany?.(records.filter((record) => record.sessionId === args.sessionId)) records = records.map((record) => ({ ...record, - lease: { ...record.lease, unreconciled: false } + lease: { + ...record.lease, + unreconciled: record.sessionId !== args.sessionId + } })) return new Map() } @@ -64,10 +68,8 @@ describe('createRestartReconciler', () => { ).resolves.toBeNull() expect(probeMany).toHaveBeenCalledOnce() - expect(probeMany.mock.calls[0]?.[0].map((record) => record.sessionId)).toEqual([ - 'session-1', - 'session-2' - ]) + expect(probeMany.mock.calls[0]?.[0].map((record) => record.sessionId)).toEqual(['session-1']) + expect(records[1]?.lease.unreconciled).toBe(true) expect(probe).not.toHaveBeenCalled() }) }) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-restart-reconcile.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-restart-reconcile.ts index 29f0c9f1047..15ba7cfe9f9 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-restart-reconcile.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-restart-reconcile.ts @@ -21,27 +21,27 @@ export function createRestartReconciler(deps: { records: readonly AgentSessionRecord[] ) => Promise> now: () => number -}): (sessionId: string) => Promise { - let pending: Promise | null = null +}): (sessionId?: string) => Promise { + const pending = new Map>() return async (sessionId) => { - if (!deps.store.listRecords().some((record) => record.lease.unreconciled)) { + if (!hasUnreconciledLease(deps.store, sessionId)) { return null } - if (!pending) { - const run = reconcileCurrentLeases(deps) - pending = run.finally(() => { - pending = null - }) + if (!pending.has(sessionId)) { + const run = reconcileCurrentLease(deps, sessionId) + pending.set( + sessionId, + run.finally(() => { + pending.delete(sessionId) + }) + ) } try { - await pending + await pending.get(sessionId) return null } catch (error) { - return classifyStoreFailure( - error, - deps.store.getRecord(sessionId)?.lease.runtimeFence ?? null, - deps.store.getRecord(sessionId) - ) + const record = sessionId === undefined ? undefined : deps.store.getRecord(sessionId) + return classifyStoreFailure(error, record?.lease.runtimeFence ?? null, record) } } } @@ -79,15 +79,15 @@ export function reportEachFailureOnce( /** The reconcile a reader runs, at startup and before each restored read: it never throws, since * an unreconciled lease grants no writer and the next send reconciles again before it acts. - * Answers whether every lease is settled. */ -export function createReaderReconcile( - reconcile: (sessionId: string) => Promise, + * Answers whether the leases it checked are settled. */ +export function createReaderReconcile( + reconcile: (...args: Args) => Promise, failures: ReaderBookkeepingFailures -): (sessionId: string) => Promise { - return async (sessionId) => { +): (...args: Args) => Promise { + return async (...args) => { let failure: unknown try { - const refusal = await reconcile(sessionId) + const refusal = await reconcile(...args) if (!refusal) { failures.clear() return true @@ -113,24 +113,34 @@ function failureKey(failure: unknown): string { return String(failure) } -async function reconcileCurrentLeases(deps: { - store: AgentSessionRecordStore - probe: (record: AgentSessionRecord) => Promise - probeMany?: ( - records: readonly AgentSessionRecord[] - ) => Promise> - now: () => number -}): Promise { +async function reconcileCurrentLease( + deps: { + store: AgentSessionRecordStore + probe: (record: AgentSessionRecord) => Promise + probeMany?: ( + records: readonly AgentSessionRecord[] + ) => Promise> + now: () => number + }, + sessionId?: string +): Promise { for (let pass = 0; pass < MAX_RECONCILIATION_PASSES; pass += 1) { await deps.store.reconcileOnRestart({ + ...(sessionId === undefined ? {} : { sessionId }), probe: deps.probe, ...(deps.probeMany ? { probeMany: deps.probeMany } : {}), now: deps.now() }) - if (!deps.store.listRecords().some((record) => record.lease.unreconciled)) { + if (!hasUnreconciledLease(deps.store, sessionId)) { return } } // An outgoing runtime can still be writing during restart; preserve the record and retry later. throw new Error('execution_owner_reconciling') } + +function hasUnreconciledLease(store: AgentSessionRecordStore, sessionId?: string): boolean { + return sessionId === undefined + ? store.listRecords().some((record) => record.lease.unreconciled) + : store.getRecord(sessionId)?.lease.unreconciled === true +} diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-restart-restore.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-restart-restore.test.ts index 1b74a23911f..018f826a0b4 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-restart-restore.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-restart-restore.test.ts @@ -62,6 +62,7 @@ describe('restart journal restoration', () => { const restoration = restoreStructuredAgentSessionsOnRestart({ openDeps: NO_OPEN_DEPS, records, + reconcileAll: async () => true, reconcile: async () => true, resolveRecovery: async () => true, serialize: async (_sessionId, task) => task(), @@ -103,6 +104,7 @@ describe('restart journal restoration', () => { await restoreStructuredAgentSessionsOnRestart({ openDeps: NO_OPEN_DEPS, records, + reconcileAll: async () => true, reconcile: async () => true, resolveRecovery: async () => true, serialize: async (_sessionId, task) => task(), @@ -149,6 +151,7 @@ describe('restart journal restoration', () => { // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: the restore reads only the record's session id here. records: [{ sessionId: 'session-1' } as AgentSessionRecord], openDeps: NO_OPEN_DEPS, + reconcileAll: async () => true, reconcile: async () => true, resolveRecovery: async () => { calls.push('resolveRecovery') @@ -174,7 +177,7 @@ describe('restart journal restoration', () => { const restore = ( bookkeeping: Pick< Parameters[0], - 'reconcile' | 'resolveRecovery' + 'reconcileAll' | 'reconcile' | 'resolveRecovery' > ) => restoreStructuredAgentSessionsOnRestart({ @@ -194,13 +197,28 @@ describe('restart journal restoration', () => { beforeEach(() => restoreRead.mockResolvedValue(null)) + it('opens with one whole-host check, then checks each chat while it holds', async () => { + const reconcileAll = vi.fn(async () => true) + const reconcile = vi.fn(async (_sessionId: string) => true) + + await restore({ reconcileAll, reconcile, resolveRecovery: async () => true }) + + expect(reconcileAll).toHaveBeenCalledOnce() + expect(reconcileAll).toHaveBeenCalledWith() + expect(reconcile.mock.calls.map(([sessionId]) => sessionId).sort()).toEqual( + records.map((record) => record.sessionId).sort() + ) + }) + it('skips it for every chat when the pass check fails, and still opens them all', async () => { - const reconcile = vi.fn(slowFailure) + const reconcileAll = vi.fn(slowFailure) + const reconcile = vi.fn(async () => true) const resolveRecovery = vi.fn(async () => true) - await restore({ reconcile, resolveRecovery }) + await restore({ reconcileAll, reconcile, resolveRecovery }) - expect(reconcile).toHaveBeenCalledOnce() + expect(reconcileAll).toHaveBeenCalledOnce() + expect(reconcile).not.toHaveBeenCalled() expect(resolveRecovery).not.toHaveBeenCalled() expect(restoreRead).toHaveBeenCalledTimes(records.length) }) @@ -208,7 +226,11 @@ describe('restart journal restoration', () => { it('starts no more after the first failed recovery, and still opens every chat', async () => { const resolveRecovery = vi.fn(slowFailure) - await restore({ reconcile: async () => true, resolveRecovery }) + await restore({ + reconcileAll: async () => true, + reconcile: async () => true, + resolveRecovery + }) // Only those already started when the first failed: at most one per chat open at once. expect(resolveRecovery.mock.calls.length).toBeLessThanOrEqual(4) @@ -235,6 +257,7 @@ describe('restart journal restoration', () => { restoreStructuredAgentSessionsOnRestart({ openDeps: { ...NO_OPEN_DEPS, logger: log.logger }, records, + reconcileAll: async () => true, reconcile: async () => true, resolveRecovery: async () => true, serialize: async (_sessionId, task) => task(), @@ -262,6 +285,7 @@ describe('restart journal restoration', () => { // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: the restore reads only the record's session id here. records: [{ sessionId: 'session-1' } as AgentSessionRecord], openDeps: NO_OPEN_DEPS, + reconcileAll: async () => true, reconcile: async () => true, resolveRecovery: async () => true, serialize: async (_sessionId, task) => task(), diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-restart-restore.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-restart-restore.ts index 6c6efc49d72..f301e519b21 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-restart-restore.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-restart-restore.ts @@ -28,7 +28,9 @@ export type StructuredAgentSessionReadRestoreDeps = { store: Pick } // Lease bookkeeping. Neither throws: a read grants no writer, so bookkeeping must not block it. - /** Whether every lease is settled. */ + /** Whether every lease on this host is settled. */ + reconcileAll: () => Promise + /** Whether this chat's lease is settled. */ reconcile: (sessionId: string) => Promise /** False when its store write failed; the next attach or send resolves it again. */ resolveRecovery: (sessionId: string) => Promise @@ -74,14 +76,12 @@ async function restoreOneStructuredAgentSessionReadUnderSerialize( export async function restoreStructuredAgentSessionsOnRestart( input: StructuredAgentSessionReadRestoreDeps & { records: AgentSessionRecord[] } ): Promise { - const [first] = input.records - if (!first) { + if (input.records.length === 0) { return } - // One check for the pass. Each chat checks again while it holds, since another writer can mark - // leases unreconciled mid-pass; after the first failure, retrying per chat only waits on the - // same store again, and the next attach or send settles those chats instead. - let settled = await input.reconcile(first.sessionId) + // One whole-host check for the pass, which also retries a startup pass that failed. After a + // failure, per-chat retries only hit the same store again; each chat's own action reconciles it. + let settled = await input.reconcileAll() const settleLeases = async (sessionId: string): Promise => { // A session latched in recovery exits here at startup, without waiting for a client. if ( diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-reveal.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-reveal.ts index 9a6e3aca591..5e85f35a43b 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-reveal.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-reveal.ts @@ -58,20 +58,23 @@ export function createStructuredAgentSessionHostRestore( deps: StructuredAgentSessionHostDeps, wiring: Omit< ConstructorParameters[0], - 'openDeps' | 'reconcile' | 'resolveRecovery' + 'openDeps' | 'reconcileAll' | 'reconcile' | 'resolveRecovery' > & { reconcileLeases: (sessionId: string) => Promise + reconcileRestartLeases: () => Promise resolveRecovery: (sessionId: string) => Promise } ): { reconcileRestartLeases: () => Promise restoreReadableSessions: (sessionIds?: readonly string[]) => Promise } { - const { reconcileLeases, resolveRecovery, ...rest } = wiring + const { reconcileLeases, reconcileRestartLeases, resolveRecovery, ...rest } = wiring const failures = reportEachFailureOnce(deps.logger) const reconcile = createReaderReconcile(reconcileLeases, failures) + const reconcileStartup = createReaderReconcile(reconcileRestartLeases, failures) const restorer = new StructuredAgentSessionReadableRestorer({ openDeps: deps, + reconcileAll: reconcileStartup, reconcile, // The next attach or send resolves recovery again, strictly, before it acts. resolveRecovery: (sessionId) => @@ -87,7 +90,7 @@ export function createStructuredAgentSessionHostRestore( const gate = new StructuredAgentSessionRestartRestoreGate() return { reconcileRestartLeases: async () => { - await reconcile('startup') + await reconcileStartup() }, restoreReadableSessions: (sessionIds) => gate.run(() => restorer.restore(sessionIds)) } diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-target-reconcile.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-target-reconcile.test.ts new file mode 100644 index 00000000000..7ecb8403194 --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-target-reconcile.test.ts @@ -0,0 +1,199 @@ +import { mkdtemp, rm } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import type { AgentSessionOwnerProbe } from '../../../shared/agent-session-lease-adjudication' +import type { AgentSessionRecord } from '../../../shared/agent-session-record' +import type { AgentSessionRecordStore } from '../../runtime/agent-session-record-store' +import type { AgentSessionReserveRequest } from '../../runtime/agent-session-reservation-admission' +import { openTestAgentSessionRecordStore } from '../../runtime/agent-session-record-store-test-harness' +import { closeTestJournalHostDatabases } from '../agent-session-journal/journal-host-database-test-support' +import { createRestartReconciler } from './structured-agent-session-restart-reconcile' + +const NOW = 1_800_000_000_000 +const UNUSED: AgentSessionOwnerProbe = { outcome: 'reservation-unused' } +let root: string +let store: AgentSessionRecordStore +let operation = 0 + +function reservation(sessionId: string): AgentSessionReserveRequest { + operation += 1 + return { + sessionId, + location: { + executionHostId: 'local', + wslDistro: null, + workspaceId: 'folder-workspace', + workspaceKind: 'folder' as const + }, + provider: 'codex' as const, + accountHome: { variable: 'CODEX_HOME' as const, path: root }, + expectedFence: null, + spawnToken: sessionId, + claimKeyId: 'key-1', + handoffOperationId: null, + probe: UNUSED, + operation: { + callerKey: 'test', + operationId: `${NOW}-${operation.toString(16).padStart(32, '0')}`, + fingerprint: sessionId + }, + now: NOW + } +} + +beforeEach(async () => { + root = await mkdtemp(join(tmpdir(), 'orca-targeted-reconcile-')) + operation = 0 + const previous = await openTestAgentSessionRecordStore(root) + await previous.reserveOwner(reservation('session-1')) + await previous.reserveOwner(reservation('session-2')) + for (const sessionId of ['session-1', 'session-2']) { + await previous.commitProcessIdentity({ + sessionId, + fence: previous.getRecord(sessionId)?.lease.runtimeFence ?? 0, + process: { + hostId: 'local', + pid: 4242, + processStartTimeMs: NOW, + spawnToken: sessionId + }, + now: NOW + }) + } + store = await openTestAgentSessionRecordStore(root) +}) + +afterEach(async () => { + closeTestJournalHostDatabases() + await rm(root, { recursive: true, force: true }) +}) + +describe('targeted restart reconciliation', () => { + it("does not fail an action because another chat's owner check fails", async () => { + const probe = vi.fn(async (record: AgentSessionRecord) => { + if (record.sessionId === 'session-2') { + throw new Error('owner check failed') + } + return UNUSED + }) + const reconcile = createRestartReconciler({ store, probe, now: () => NOW }) + + expect(await reconcile('session-1')).toBeNull() + expect(probe).toHaveBeenCalledOnce() + expect(store.getRecord('session-2')?.lease.unreconciled).toBe(true) + await expect(reconcile('session-2')).rejects.toThrow('owner check failed') + }) + + it("settles a chat while another chat's record changes under every check of it", async () => { + let renames = 0 + const probe = vi.fn(async (record: AgentSessionRecord) => { + if (record.sessionId === 'session-2') { + renames += 1 + // Each check of session-2 is stale by the time it applies. + await store.setConversationName('session-2', `renamed ${renames}`) + } + return UNUSED + }) + const reconcile = createRestartReconciler({ store, probe, now: () => NOW }) + + expect(await reconcile('session-1')).toBeNull() + expect(store.getRecord('session-1')?.lease.unreconciled).toBe(false) + // The whole-host pass does run out of passes on session-2. + expect(await reconcile()).not.toBeNull() + expect(store.getRecord('session-2')?.lease.unreconciled).toBe(true) + }) + + it("does not wait on another chat's stalled owner check in the startup batch", async () => { + let release: (probe: AgentSessionOwnerProbe) => void = () => undefined + let started: () => void = () => undefined + const probing = new Promise((resolve) => { + started = resolve + }) + const waiting = new Promise((resolve) => { + release = resolve + }) + const probe = vi.fn(async (record: AgentSessionRecord) => { + if (record.sessionId === 'session-2') { + started() + return waiting + } + return UNUSED + }) + const reconcile = createRestartReconciler({ store, probe, now: () => NOW }) + const startup = reconcile() + await probing + const target = reconcile('session-1') + let timeout: ReturnType | undefined + try { + expect( + await Promise.race([ + target, + new Promise((resolve) => { + timeout = setTimeout(() => resolve('blocked'), 1_000) + }) + ]) + ).toBeNull() + expect(store.getRecord('session-2')?.lease.unreconciled).toBe(true) + } finally { + clearTimeout(timeout) + release(UNUSED) + await Promise.all([startup, target]) + } + }) + + it('joins simultaneous checks of the same session', async () => { + const probe = vi.fn(async () => UNUSED) + const reconcile = createRestartReconciler({ store, probe, now: () => NOW }) + expect(await Promise.all([reconcile('session-1'), reconcile('session-1')])).toEqual([ + null, + null + ]) + expect(probe).toHaveBeenCalledOnce() + }) + + it('does no reconciliation for a new chat or a session already reconciled', async () => { + const probe = vi.fn(async () => UNUSED) + const reconcile = createRestartReconciler({ store, probe, now: () => NOW }) + expect(await reconcile('new-session')).toBeNull() + expect(probe).not.toHaveBeenCalled() + expect(await reconcile('session-1')).toBeNull() + expect(await reconcile('session-1')).toBeNull() + expect(probe).toHaveBeenCalledOnce() + }) + + it('keeps an unproven target in recovery and refuses a new writer', async () => { + const reconcile = createRestartReconciler({ + store, + probe: async () => ({ outcome: 'indeterminate', reason: 'owner runs on another host' }), + now: () => NOW + }) + expect(await reconcile('session-1')).toBeNull() + const fence = store.getRecord('session-1')?.lease.runtimeFence ?? 0 + await expect( + store.reserveOwner({ + ...reservation('session-1'), + expectedFence: fence, + probe: { outcome: 'indeterminate', reason: 'owner runs on another host' } + }) + ).rejects.toThrow('agent_session_ownership_unknown') + }) + + it('still batches every pending record for explicit startup reconciliation', async () => { + const probeMany = vi.fn( + async (records: readonly AgentSessionRecord[]) => + new Map(records.map((record) => [record.sessionId, UNUSED])) + ) + const reconcile = createRestartReconciler({ + store, + probe: async () => UNUSED, + probeMany, + now: () => NOW + }) + expect(await reconcile()).toBeNull() + expect(probeMany.mock.calls[0]?.[0].map((record) => record.sessionId)).toEqual([ + 'session-1', + 'session-2' + ]) + }) +}) diff --git a/src/main/runtime/agent-session-record-store.ts b/src/main/runtime/agent-session-record-store.ts index 02a1ab2f63a..585f003a01a 100644 --- a/src/main/runtime/agent-session-record-store.ts +++ b/src/main/runtime/agent-session-record-store.ts @@ -264,15 +264,14 @@ export class AgentSessionRecordStore { } /** - * Adjudicate every lease this host loaded. No lease grants a writer until it appears here. On a - * database a newer Orca wrote, the verdicts are kept in memory only: they are re-derived at every + * Adjudicate loaded leases, or only the addressed session. No lease grants a writer until it + * appears here. On a newer database, verdicts stay in memory only: they are re-derived at every * start, and none grants a writer there, since every grant is a write. */ async reconcileOnRestart( args: AgentSessionRestartProbeArgs ): Promise> { - const pending = this.listRecords().filter((record) => record.lease.unreconciled) - const probes = await collectAgentSessionRestartProbes(pending, args) + const probes = await collectAgentSessionRestartProbes(this, args) return this.transact((draft) => applyAgentSessionRestartProbes(draft, probes, args.now), { inMemoryWhenReadOnly: true }) diff --git a/src/main/runtime/agent-session-restart-reconciliation.ts b/src/main/runtime/agent-session-restart-reconciliation.ts index 3e05914c42e..e864af337c2 100644 --- a/src/main/runtime/agent-session-restart-reconciliation.ts +++ b/src/main/runtime/agent-session-restart-reconciliation.ts @@ -6,6 +6,7 @@ import { agentSessionReconciliationTargetMatches } from './agent-session-reconci import { applyAgentSessionRestartAdjudication } from './agent-session-restart-lease-transitions' export type AgentSessionRestartProbeArgs = { + sessionId?: string probe: (record: AgentSessionRecord) => Promise probeMany?: ( records: readonly AgentSessionRecord[] @@ -16,10 +17,19 @@ export type AgentSessionRestartProbeArgs = { type RestartProbe = { record: AgentSessionRecord; probe: AgentSessionOwnerProbe } export async function collectAgentSessionRestartProbes( - records: readonly AgentSessionRecord[], + store: { + listRecords: () => AgentSessionRecord[] + getRecord: (sessionId: string) => AgentSessionRecord | null + }, args: AgentSessionRestartProbeArgs ): Promise> { const probes = new Map() + const target = args.sessionId === undefined ? null : store.getRecord(args.sessionId) + const candidates = args.sessionId === undefined ? store.listRecords() : target ? [target] : [] + const records = candidates.filter((record) => record.lease.unreconciled) + if (records.length === 0) { + return probes + } const batched = args.probeMany ? await args.probeMany(records) : null for (const record of records) { probes.set(record.sessionId, {