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, {