diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-attach-flow.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-attach-flow.ts index 49ce093c199..93d448d04fb 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-attach-flow.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-attach-flow.ts @@ -82,11 +82,13 @@ export async function performAttach( if (reservedRecord && isAgentSessionPreSpawnError(error)) { const spawnToken = reservedRecord.lease.reservedSpawnToken if (spawnToken) { - await store.markReservationProcessless({ + const processlessAt = input.now() + await store.setReservationProcesslessProof({ sessionId, fence: reservedRecord.lease.runtimeFence, spawnToken, - now: input.now() + processlessAt, + now: processlessAt }) } } @@ -136,6 +138,14 @@ async function acquireOwner( if (!spawnToken) { throw new Error('agent_session_ownership_unknown') } + // Pre-spawn proof is single-use: this retry may create a child after the durable clear. + record = await input.store.setReservationProcesslessProof({ + sessionId: record.sessionId, + fence, + spawnToken, + processlessAt: null, + now: input.now() + }) await input.onAcquiring?.() const acquired = await input.adapter.acquire({ identity: journalIdentityFor(record, input.params), diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-processless-reservation.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-processless-reservation.test.ts index d33980d9ff5..a3451ff9037 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-processless-reservation.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-processless-reservation.test.ts @@ -4,6 +4,7 @@ import { join } from 'node:path' import { afterEach, describe, expect, it, vi } from 'vitest' import { computeAgentSessionPayloadFingerprint } from '../../../shared/agent-session-mutation-envelope' import { AgentSessionRecordStore } from '../../runtime/agent-session-record-store' +import { createStructuredAgentSessionOwnerProbe } from '../../runtime/structured-agent-session-runtime' import { AgentSessionPreSpawnError, type StructuredAgentSessionAdapter @@ -103,4 +104,64 @@ describe('processless structured session reservation', () => { reservedSpawnToken: null }) }) + + it('consumes stale pre-spawn proof before a retry can spawn', async () => { + root = await mkdtemp(join(tmpdir(), 'orca-processless-retry-')) + const storeDir = join(root, 'store') + const store = await AgentSessionRecordStore.open({ directory: storeDir, hostId: 'local' }) + const adapter = { + acquire: vi + .fn() + .mockRejectedValueOnce(new AgentSessionPreSpawnError(new Error('launch not ready'))) + .mockResolvedValueOnce({ + process: { + hostId: 'local', + pid: 4242, + processStartTimeMs: NOW, + spawnToken: 'spawn-a' + }, + link: { + linkId: 'link-1', + handle: { provider: 'codex', threadId: 'thread-1' }, + origin: 'created', + mintedAtFence: 1, + observedAt: NOW + } + }) + } as unknown as StructuredAgentSessionAdapter + const input = { + store, + adapter, + journalRoot: root, + authority: { + spawnToken: 'spawn-a', + claimKeyId: 'key-1', + handoffOperationId: OPERATION, + probe: { outcome: 'reservation-unused' as const } + }, + callerKey: 'client-1', + params: attachParams(), + now: () => NOW, + onAttached: () => {} + } + + await expect(performAttach(input)).rejects.toThrow('launch not ready') + expect(store.getRecord(SESSION)?.lease.processlessAt).toBe(NOW) + vi.spyOn(store, 'commitProcessIdentity').mockRejectedValueOnce(new Error('simulated crash')) + + await expect(performAttach(input)).rejects.toThrow('simulated crash') + + const reopened = await AgentSessionRecordStore.open({ directory: storeDir, hostId: 'local' }) + expect(reopened.getRecord(SESSION)?.lease.processlessAt).toBeNull() + await reopened.reconcileOnRestart({ + probe: createStructuredAgentSessionOwnerProbe('local'), + now: NOW + 1 + }) + expect(reopened.getRecord(SESSION)?.lease).toMatchObject({ + claimStatus: 'reserved', + handoffStage: 'recovering', + runtimeFence: 1, + processlessAt: null + }) + }) }) diff --git a/src/main/runtime/agent-session-processless-reservation.ts b/src/main/runtime/agent-session-processless-reservation.ts index 52ab279e01d..9415ba8fcb7 100644 --- a/src/main/runtime/agent-session-processless-reservation.ts +++ b/src/main/runtime/agent-session-processless-reservation.ts @@ -7,23 +7,38 @@ export type AgentSessionReservationProcesslessProof = { now: number } -export function markAgentSessionReservationProcessless( - args: AgentSessionReservationProcesslessProof & { record: AgentSessionRecord } -): AgentSessionRecord { - const { record } = args +function assertReservation( + record: AgentSessionRecord, + args: AgentSessionReservationProcesslessProof +): void { if (record.lease.runtimeFence !== args.fence || record.lease.unreconciled) { throw new Error('agent_session_checkpoint_stale') } if ( record.lease.claimStatus !== 'reserved' || - record.lease.ownerProcess !== null || record.lease.reservedSpawnToken !== args.spawnToken ) { throw new Error('agent_session_ownership_unknown') } +} + +export function setAgentSessionReservationProcesslessProof( + args: AgentSessionReservationProcesslessProof & { + record: AgentSessionRecord + processlessAt: number | null + } +): AgentSessionRecord { + const { record } = args + assertReservation(record, args) + if (args.processlessAt === null && record.lease.processlessAt == null) { + return record + } + if (record.lease.ownerProcess !== null) { + throw new Error('agent_session_ownership_unknown') + } return { ...record, - lease: { ...record.lease, processlessAt: args.now }, + lease: { ...record.lease, processlessAt: args.processlessAt }, updatedAt: args.now } } diff --git a/src/main/runtime/agent-session-record-store.ts b/src/main/runtime/agent-session-record-store.ts index f148d0064c9..f4d829698ac 100644 --- a/src/main/runtime/agent-session-record-store.ts +++ b/src/main/runtime/agent-session-record-store.ts @@ -42,7 +42,7 @@ import { type AgentSessionProcessIdentityCommit } from './agent-session-lease-transitions' import { - markAgentSessionReservationProcessless, + setAgentSessionReservationProcesslessProof, type AgentSessionReservationProcesslessProof } from './agent-session-processless-reservation' import { @@ -178,11 +178,11 @@ export class AgentSessionRecordStore { ) } - markReservationProcessless = ( - args: AgentSessionReservationProcesslessProof + setReservationProcesslessProof = ( + args: AgentSessionReservationProcesslessProof & { processlessAt: number | null } ): Promise => this.mutate(args.sessionId, (record) => - markAgentSessionReservationProcessless({ ...args, record }) + setAgentSessionReservationProcesslessProof({ ...args, record }) ) async proveOwner(args: {