mirror of
https://github.com/stablyai/orca.git
synced 2026-10-22 00:03:47 +00:00
fix: consume pre-spawn proof before retry
This commit is contained in:
@@ -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),
|
||||
|
||||
+61
@@ -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<StructuredAgentSessionAdapter['acquire']>()
|
||||
.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
|
||||
})
|
||||
})
|
||||
})
|
||||
|
||||
@@ -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
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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<AgentSessionRecord> =>
|
||||
this.mutate(args.sessionId, (record) =>
|
||||
markAgentSessionReservationProcessless({ ...args, record })
|
||||
setAgentSessionReservationProcesslessProof({ ...args, record })
|
||||
)
|
||||
|
||||
async proveOwner(args: {
|
||||
|
||||
Reference in New Issue
Block a user