mirror of
https://github.com/stablyai/orca.git
synced 2026-10-23 08:03:39 +00:00
fix(native-chat): give latched recovery stages evidence-driven exits (#15147)
A transient failure could latch a session in recovering/manual-recovery with every documented exit gated on tui+ownerProcess, leaving native sessions permanently wedged behind a switching spinner: - adjudication now releases an abandoned native reservation at restart (the reserving runtime is proven gone and a never-proven native child has no request channel left), and routes a surviving native owner to recovery instead of readopting a transport that died with the previous process - a recovery resolver re-probes latched native records at attach and during startup restore; a live identity-matched orphan gets a stop request and is evicted only after a later probe proves it absent, never on a timeout - the lease renewer no longer extends leases for native records parked in recovery, and the host runtime-state probe no longer fabricates reservation-unused for a live reservation - codex spawn identity retries the start-time read and refuses an owner no probe could ever verify; on hosts that expose process environments the probe now reads the spawn token back as a second identity element
This commit is contained in:
@@ -0,0 +1,48 @@
|
||||
import { describe, expect, it, vi } from 'vitest'
|
||||
import { codexProcessIdentity } from './codex-structured-owner-identity'
|
||||
|
||||
const IDENTITY = {
|
||||
sessionId: 'session-identity',
|
||||
workspaceId: 'workspace-1',
|
||||
hostId: 'local',
|
||||
agent: 'codex' as const,
|
||||
providerHandle: { kind: 'codex' as const, threadId: 'thread-1' }
|
||||
}
|
||||
|
||||
describe('codex process identity', () => {
|
||||
it('records the observed start time alongside the spawn token', async () => {
|
||||
await expect(
|
||||
codexProcessIdentity(
|
||||
{ identity: IDENTITY, spawnToken: 'spawn-a', pid: 4242 },
|
||||
async () => 123
|
||||
)
|
||||
).resolves.toEqual({
|
||||
hostId: 'local',
|
||||
pid: 4242,
|
||||
processStartTimeMs: 123,
|
||||
spawnToken: 'spawn-a'
|
||||
})
|
||||
})
|
||||
|
||||
it('retries a failed start-time read before giving up', async () => {
|
||||
const readStartTime = vi
|
||||
.fn<(pid: number) => Promise<number | null>>()
|
||||
.mockResolvedValueOnce(null)
|
||||
.mockResolvedValueOnce(null)
|
||||
.mockResolvedValueOnce(456)
|
||||
await expect(
|
||||
codexProcessIdentity({ identity: IDENTITY, spawnToken: 'spawn-a', pid: 4242 }, readStartTime)
|
||||
).resolves.toMatchObject({ processStartTimeMs: 456 })
|
||||
expect(readStartTime).toHaveBeenCalledTimes(3)
|
||||
})
|
||||
|
||||
it('refuses an owner whose start time is unreadable rather than record one no probe can verify', async () => {
|
||||
// A null start time guarantees every later owner probe answers indeterminate, which is
|
||||
// a durable latch; refusing here is a retryable failure instead.
|
||||
const readStartTime = vi.fn(async () => null)
|
||||
await expect(
|
||||
codexProcessIdentity({ identity: IDENTITY, spawnToken: 'spawn-a', pid: 4242 }, readStartTime)
|
||||
).rejects.toThrow('start time')
|
||||
expect(readStartTime).toHaveBeenCalledTimes(3)
|
||||
})
|
||||
})
|
||||
@@ -12,6 +12,8 @@ import { readProcessStartTimeMs } from '../runtime/agent-session-process-identit
|
||||
* child of THIS reservation from a same-pid stranger. */
|
||||
export const CODEX_SPAWN_TOKEN_ENV = 'ORCA_AGENT_SESSION_SPAWN_TOKEN'
|
||||
|
||||
const START_TIME_READ_ATTEMPTS = 3
|
||||
|
||||
export async function codexProcessIdentity(
|
||||
input: {
|
||||
identity: AgentSessionJournalIdentity
|
||||
@@ -23,10 +25,23 @@ export async function codexProcessIdentity(
|
||||
if (input.pid === undefined) {
|
||||
throw new Error('codex app-server started without a pid')
|
||||
}
|
||||
let processStartTimeMs: number | null = null
|
||||
for (
|
||||
let attempt = 0;
|
||||
attempt < START_TIME_READ_ATTEMPTS && processStartTimeMs === null;
|
||||
attempt += 1
|
||||
) {
|
||||
processStartTimeMs = await readStartTime(input.pid)
|
||||
}
|
||||
if (processStartTimeMs === null) {
|
||||
// Why: recording null makes every later owner probe indeterminate — a durable latch.
|
||||
// Failing here reaps the child and leaves a retryable refusal instead.
|
||||
throw new Error(`codex app-server start time for pid ${input.pid} could not be read`)
|
||||
}
|
||||
return {
|
||||
hostId: input.identity.hostId,
|
||||
pid: input.pid,
|
||||
processStartTimeMs: await readStartTime(input.pid),
|
||||
processStartTimeMs,
|
||||
spawnToken: input.spawnToken
|
||||
}
|
||||
}
|
||||
|
||||
+101
@@ -0,0 +1,101 @@
|
||||
import { describe, expect, it, vi } from 'vitest'
|
||||
import {
|
||||
AGENT_SESSION_RECORD_SCHEMA_VERSION,
|
||||
type AgentSessionRecord
|
||||
} from '../../../shared/agent-session-record'
|
||||
import type { AgentSessionRecordStore } from '../../runtime/agent-session-record-store'
|
||||
import { StructuredAgentSessionHostRuntimeState } from './structured-agent-session-host-runtime-state'
|
||||
import type { StructuredAgentSessionHostDeps } from './structured-agent-session-host'
|
||||
|
||||
const NOW = 1_800_000_000_000
|
||||
|
||||
function reservedRecord(): AgentSessionRecord {
|
||||
return {
|
||||
schemaVersion: AGENT_SESSION_RECORD_SCHEMA_VERSION,
|
||||
sessionId: 'session-probe',
|
||||
location: {
|
||||
executionHostId: 'local',
|
||||
wslDistro: null,
|
||||
workspaceId: 'workspace-1',
|
||||
workspaceKind: 'folder'
|
||||
},
|
||||
provider: 'codex',
|
||||
providerHandleChain: [],
|
||||
accountHome: { variable: 'CODEX_HOME', path: '/tmp/codex' },
|
||||
lease: {
|
||||
sessionId: 'session-probe',
|
||||
runtimeKind: 'native',
|
||||
runtimeFence: 3,
|
||||
handoffStage: 'new-owner-proving',
|
||||
provenHandleLinkId: null,
|
||||
ownerProcess: null,
|
||||
reservedSpawnToken: 'spawn-probe',
|
||||
leaseDeadlineAt: NOW + 30_000,
|
||||
lastRenewedAt: NOW,
|
||||
handoffOperationId: 'op-1',
|
||||
journalCheckpoint: null,
|
||||
claimKeyId: 'key-1',
|
||||
claimStatus: 'reserved',
|
||||
unreconciled: false,
|
||||
deathEvidence: null
|
||||
},
|
||||
createdAt: NOW,
|
||||
updatedAt: NOW
|
||||
}
|
||||
}
|
||||
|
||||
function runtimeState(
|
||||
record: AgentSessionRecord | null,
|
||||
probeOwner: NonNullable<StructuredAgentSessionHostDeps['probeOwner']>
|
||||
) {
|
||||
const deps = {
|
||||
store: { getRecord: () => record } as unknown as AgentSessionRecordStore,
|
||||
adapter: {},
|
||||
journalRoot: '/tmp',
|
||||
claimKeyId: 'key-1',
|
||||
probeOwner
|
||||
} as StructuredAgentSessionHostDeps
|
||||
return new StructuredAgentSessionHostRuntimeState(deps)
|
||||
}
|
||||
|
||||
describe('host runtime-state owner probe', () => {
|
||||
it('routes an ownerless reservation through the strict probe instead of fabricating proof', async () => {
|
||||
// Fabricating `reservation-unused` here skipped the processless-proof rule the runtime
|
||||
// probe enforces — the exact answer that mints a second writer on one provider session.
|
||||
const probeOwner = vi.fn(async () => ({
|
||||
outcome: 'indeterminate' as const,
|
||||
reason: 'reservation named no process'
|
||||
}))
|
||||
const state = runtimeState(reservedRecord(), probeOwner)
|
||||
|
||||
await expect(state.probeOwner('session-probe')).resolves.toEqual({
|
||||
outcome: 'indeterminate',
|
||||
reason: 'reservation named no process'
|
||||
})
|
||||
expect(probeOwner).toHaveBeenCalledTimes(1)
|
||||
})
|
||||
|
||||
it('skips the probe for a released ownerless record acquisition never consults it for', async () => {
|
||||
const released = reservedRecord()
|
||||
released.lease.claimStatus = 'released'
|
||||
released.lease.handoffStage = null
|
||||
released.lease.reservedSpawnToken = null
|
||||
const probeOwner = vi.fn(async () => ({ outcome: 'indeterminate' as const, reason: 'x' }))
|
||||
const state = runtimeState(released, probeOwner)
|
||||
|
||||
await expect(state.probeOwner('session-probe')).resolves.toEqual({
|
||||
outcome: 'reservation-unused'
|
||||
})
|
||||
expect(probeOwner).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it('treats a session with no record at all as an unused reservation', async () => {
|
||||
const probeOwner = vi.fn(async () => ({ outcome: 'indeterminate' as const, reason: 'x' }))
|
||||
const state = runtimeState(null, probeOwner)
|
||||
|
||||
await expect(state.probeOwner('session-probe')).resolves.toEqual({
|
||||
outcome: 'reservation-unused'
|
||||
})
|
||||
expect(probeOwner).not.toHaveBeenCalled()
|
||||
})
|
||||
})
|
||||
+21
-1
@@ -6,6 +6,7 @@ import {
|
||||
} from './structured-agent-session-event-sink'
|
||||
import type { StructuredAgentSessionHostDeps } from './structured-agent-session-host'
|
||||
import { StructuredAgentSessionLeaseRenewer } from './structured-agent-session-lease-renewer'
|
||||
import { resolveStructuredSessionRecovery } from './structured-agent-session-recovery-resolution'
|
||||
|
||||
export class StructuredAgentSessionHostRuntimeState {
|
||||
private readonly eventSinks = new Map<string, DeferredStructuredAgentSessionEventSink>()
|
||||
@@ -58,11 +59,30 @@ export class StructuredAgentSessionHostRuntimeState {
|
||||
await Promise.all([...this.eventSinks.values()].map((sink) => sink.drained()))
|
||||
}
|
||||
|
||||
/** Exit from a latched recovery stage when present-time evidence permits one. */
|
||||
resolveRecovery(sessionId: string): Promise<'resolved' | 'unresolved' | 'not-applicable'> {
|
||||
return resolveStructuredSessionRecovery(
|
||||
{
|
||||
store: this.deps.store,
|
||||
probeRecord: (record) => this.probeRecord(record),
|
||||
now: () => this.deps.now?.() ?? Date.now(),
|
||||
...(this.deps.stopOwnerProcess ? { stopOwnerProcess: this.deps.stopOwnerProcess } : {})
|
||||
},
|
||||
sessionId
|
||||
)
|
||||
}
|
||||
|
||||
probeOwner(sessionId: string): Promise<AgentSessionOwnerProbe> {
|
||||
const record = this.deps.store.getRecord(sessionId)
|
||||
if (!record || record.lease.ownerProcess === null) {
|
||||
if (
|
||||
!record ||
|
||||
(record.lease.ownerProcess === null && record.lease.claimStatus !== 'reserved')
|
||||
) {
|
||||
// Acquisition only consults the probe against a recorded owner or a live reservation.
|
||||
return Promise.resolve({ outcome: 'reservation-unused' })
|
||||
}
|
||||
// A live reservation goes through the strict probe: calling it unused without its
|
||||
// processless proof is the answer that mints a second writer.
|
||||
return this.probeRecord(record)
|
||||
}
|
||||
|
||||
|
||||
@@ -20,6 +20,8 @@ export type StructuredAgentSessionHostDeps = {
|
||||
journalRoot: string
|
||||
claimKeyId: string
|
||||
probeOwner?: (record: AgentSessionRecord) => Promise<AgentSessionOwnerProbe>
|
||||
/** Recovery-exit stop requests only; a lease moves only on a later proven-absent probe. */
|
||||
stopOwnerProcess?: (pid: number, signal: 'SIGTERM' | 'SIGKILL') => void
|
||||
mintSpawnToken?: () => string
|
||||
resolveLaunchEnv?: (
|
||||
provider: AgentSessionRecord['provider']
|
||||
|
||||
@@ -98,6 +98,7 @@ export class StructuredAgentSessionHost {
|
||||
journalRoot: deps.journalRoot,
|
||||
supportsRecord: (record) => providerSupport.adapterSupportsRecord(deps.adapter, record),
|
||||
reconcile: this.reconcileLeases,
|
||||
resolveRecovery: (sessionId) => this.runtimeState.resolveRecovery(sessionId),
|
||||
resume: (params) =>
|
||||
this.attach({ callerKey: 'trusted-local:host-restart' }, params).then(
|
||||
(result) => result.ok
|
||||
@@ -149,6 +150,7 @@ export class StructuredAgentSessionHost {
|
||||
if (unreconciled) {
|
||||
return refuseAgentSessionMutation(unreconciled)
|
||||
}
|
||||
await this.runtimeState.resolveRecovery(sessionId)
|
||||
const eventSink = this.runtimeState.eventSinkFor(sessionId)
|
||||
const attached = await performAttach({
|
||||
store: this.deps.store,
|
||||
|
||||
+24
@@ -133,6 +133,30 @@ describe('structured agent-session lease renewal', () => {
|
||||
})
|
||||
})
|
||||
|
||||
it('never extends the lease of a native record parked in recovery', async () => {
|
||||
// The host cannot vouch for a native child it holds no transport to; renewing while
|
||||
// recovering keeps an orphan pid's lease alive and reads as a healthy owner.
|
||||
const store = await liveStore()
|
||||
await store.transitionHandoff('session-renewal', (record) => ({
|
||||
...record,
|
||||
lease: { ...record.lease, handoffStage: 'recovering' }
|
||||
}))
|
||||
const probe = vi.fn(async () => ({
|
||||
outcome: 'identity-matched' as const,
|
||||
matchedOn: ['process-start-time' as const]
|
||||
}))
|
||||
const renewer = new StructuredAgentSessionLeaseRenewer({
|
||||
store,
|
||||
probe,
|
||||
now: () => NOW + 10_000
|
||||
})
|
||||
|
||||
await renewer.renewNow()
|
||||
|
||||
expect(probe).not.toHaveBeenCalled()
|
||||
expect(store.getRecord('session-renewal')?.lease.lastRenewedAt).toBe(NOW)
|
||||
})
|
||||
|
||||
it('routes a proven dead TUI owner into handoff recovery', async () => {
|
||||
const store = await liveStore()
|
||||
await store.transitionHandoff('session-renewal', (record) => ({
|
||||
|
||||
@@ -54,7 +54,14 @@ export class StructuredAgentSessionLeaseRenewer {
|
||||
(record) =>
|
||||
!record.lease.unreconciled &&
|
||||
record.lease.claimStatus === 'live' &&
|
||||
record.lease.ownerProcess !== null
|
||||
record.lease.ownerProcess !== null &&
|
||||
// A native record parked in recovery has no transport the host can vouch
|
||||
// for; renewing it keeps an orphan pid's lease reading as a healthy owner.
|
||||
!(
|
||||
record.lease.runtimeKind === 'native' &&
|
||||
(record.lease.handoffStage === 'recovering' ||
|
||||
record.lease.handoffStage === 'manual-recovery')
|
||||
)
|
||||
)
|
||||
.map((record) => this.renewRecord(record))
|
||||
)
|
||||
|
||||
+1
@@ -19,6 +19,7 @@ describe('legacy structured session restart', () => {
|
||||
journalRoot: '/isolated/no-journal',
|
||||
records: [loaded.record],
|
||||
reconcile: async () => null,
|
||||
resolveRecovery: async () => 'not-applicable',
|
||||
operationId: () => 'legacy-upgrade-operation',
|
||||
resume,
|
||||
serialize: async () => undefined as never,
|
||||
|
||||
+7
-4
@@ -157,11 +157,14 @@ describe('processless structured session reservation', () => {
|
||||
probe: createStructuredAgentSessionOwnerProbe('local'),
|
||||
now: NOW + 1
|
||||
})
|
||||
// The consumed proof cannot come back, but restart itself proves the reserving runtime
|
||||
// died before any owner was proven — the reservation is released, never latched.
|
||||
expect(reopened.getRecord(SESSION)?.lease).toMatchObject({
|
||||
claimStatus: 'reserved',
|
||||
handoffStage: 'recovering',
|
||||
runtimeFence: 1,
|
||||
processlessAt: null
|
||||
claimStatus: 'released',
|
||||
handoffStage: null,
|
||||
runtimeFence: 2,
|
||||
ownerProcess: null,
|
||||
reservedSpawnToken: null
|
||||
})
|
||||
})
|
||||
})
|
||||
|
||||
@@ -15,6 +15,7 @@ export class StructuredAgentSessionReadableRestorer {
|
||||
journalRoot: string
|
||||
supportsRecord: (record: AgentSessionRecord) => boolean
|
||||
reconcile: (sessionId: string) => Promise<AgentSessionWireRefusal | null>
|
||||
resolveRecovery: (sessionId: string) => Promise<unknown>
|
||||
resume: (params: AgentSessionAttachParams) => Promise<boolean>
|
||||
serialize: <T>(sessionId: string, task: () => Promise<T>) => Promise<T>
|
||||
hasSession: (sessionId: string) => boolean
|
||||
|
||||
+171
@@ -0,0 +1,171 @@
|
||||
// End-to-end exits from latched recovery states: no shape a user cannot get out of.
|
||||
|
||||
import { mkdtemp, rm } from 'node:fs/promises'
|
||||
import { tmpdir } from 'node:os'
|
||||
import { join } from 'node:path'
|
||||
import { afterEach, beforeEach, describe, expect, it, vi, type Mock } from 'vitest'
|
||||
import { AgentSessionRecordStore } from '../../runtime/agent-session-record-store'
|
||||
import type { StructuredAgentSessionAdapter } from './structured-agent-session-adapter'
|
||||
import { StructuredAgentSessionHost } from './structured-agent-session-host'
|
||||
import type { StructuredAgentSessionHostDeps } from './structured-agent-session-host-types'
|
||||
import {
|
||||
HOST_TEST_NOW as NOW,
|
||||
HOST_TEST_SESSION as SESSION,
|
||||
HOST_TEST_THREAD as THREAD,
|
||||
hostTestAttachParams,
|
||||
resetHostTestOperationIds
|
||||
} from './structured-agent-session-host-test-data'
|
||||
|
||||
const CALLER = { callerKey: 'client-1' }
|
||||
|
||||
let root: string
|
||||
let store: AgentSessionRecordStore
|
||||
let host: StructuredAgentSessionHost
|
||||
let acquire: Mock<StructuredAgentSessionAdapter['acquire']>
|
||||
|
||||
function adapter(): StructuredAgentSessionAdapter {
|
||||
return {
|
||||
acquire,
|
||||
releaseAcquisition: vi.fn(async () => undefined),
|
||||
dispatch: vi.fn(),
|
||||
cancelTurn: vi.fn(),
|
||||
answerPrompt: vi.fn(),
|
||||
setOption: vi.fn()
|
||||
} as unknown as StructuredAgentSessionAdapter
|
||||
}
|
||||
|
||||
function openHost(overrides: Partial<StructuredAgentSessionHostDeps> = {}): void {
|
||||
host = new StructuredAgentSessionHost({
|
||||
store,
|
||||
adapter: adapter(),
|
||||
journalRoot: root,
|
||||
claimKeyId: 'key-1',
|
||||
mintSpawnToken: () => 'spawn-a',
|
||||
now: () => NOW,
|
||||
...overrides
|
||||
})
|
||||
}
|
||||
|
||||
async function reopenStore(): Promise<void> {
|
||||
await host.flushAllStreamedEvents()
|
||||
store = await AgentSessionRecordStore.open({ directory: join(root, 'store'), hostId: 'local' })
|
||||
}
|
||||
|
||||
beforeEach(async () => {
|
||||
root = await mkdtemp(join(tmpdir(), 'orca-recovery-exits-'))
|
||||
resetHostTestOperationIds()
|
||||
acquire = vi.fn(async ({ fence }) => ({
|
||||
process: {
|
||||
hostId: 'local',
|
||||
pid: 4242,
|
||||
processStartTimeMs: 1_700_000_000_000,
|
||||
spawnToken: store.getRecord(SESSION)?.lease.reservedSpawnToken ?? 'spawn-a'
|
||||
},
|
||||
link: {
|
||||
linkId: `link-${fence}`,
|
||||
handle: { provider: 'codex', threadId: THREAD },
|
||||
origin: store.getRecord(SESSION)?.providerHandleChain.length ? 'resumed' : 'created',
|
||||
mintedAtFence: fence,
|
||||
observedAt: NOW
|
||||
}
|
||||
}))
|
||||
store = await AgentSessionRecordStore.open({ directory: join(root, 'store'), hostId: 'local' })
|
||||
openHost()
|
||||
})
|
||||
|
||||
afterEach(async () => {
|
||||
await host.flushAllStreamedEvents()
|
||||
await rm(root, { recursive: true, force: true })
|
||||
})
|
||||
|
||||
describe('recovery exits', () => {
|
||||
it('frees a reservation stranded by a crash between reserve and identity commit', async () => {
|
||||
acquire.mockRejectedValueOnce(new Error('simulated crash before identity commit'))
|
||||
await expect(host.attach(CALLER, hostTestAttachParams(null))).rejects.toThrow('simulated crash')
|
||||
expect(store.getRecord(SESSION)?.lease).toMatchObject({
|
||||
claimStatus: 'reserved',
|
||||
ownerProcess: null
|
||||
})
|
||||
|
||||
await reopenStore()
|
||||
openHost({ mintSpawnToken: () => 'spawn-b' })
|
||||
|
||||
// The abandoned reservation is released at reconcile, so the stale client fence gets
|
||||
// a retryable refusal that names the current fence instead of a dead-end latch.
|
||||
const stale = await host.attach(CALLER, hostTestAttachParams(1))
|
||||
expect(stale).toMatchObject({
|
||||
ok: false,
|
||||
refusal: { code: 'agent_session_checkpoint_stale', currentFence: 2 }
|
||||
})
|
||||
const retried = await host.attach(CALLER, hostTestAttachParams(2))
|
||||
expect(retried).toMatchObject({ ok: true })
|
||||
expect(store.getRecord(SESSION)?.lease).toMatchObject({
|
||||
claimStatus: 'live',
|
||||
handoffStage: null,
|
||||
ownerProcess: { pid: 4242 }
|
||||
})
|
||||
})
|
||||
|
||||
it('stops a surviving native child after restart instead of readopting its dead transport', async () => {
|
||||
expect((await host.attach(CALLER, hostTestAttachParams(null))).ok).toBe(true)
|
||||
await reopenStore()
|
||||
|
||||
let orphanAlive = true
|
||||
const stopOwnerProcess = vi.fn((_pid: number, _signal: 'SIGTERM' | 'SIGKILL') => {
|
||||
orphanAlive = false
|
||||
})
|
||||
openHost({
|
||||
mintSpawnToken: () => 'spawn-b',
|
||||
probeOwner: async () =>
|
||||
orphanAlive
|
||||
? { outcome: 'identity-matched', matchedOn: ['process-start-time'] }
|
||||
: { outcome: 'pid-absent' },
|
||||
stopOwnerProcess
|
||||
})
|
||||
|
||||
const stale = await host.attach(CALLER, hostTestAttachParams(1))
|
||||
expect(stale).toMatchObject({
|
||||
ok: false,
|
||||
refusal: { code: 'agent_session_checkpoint_stale', currentFence: 2 }
|
||||
})
|
||||
expect(stopOwnerProcess).toHaveBeenCalledWith(4242, 'SIGTERM')
|
||||
const retried = await host.attach(CALLER, hostTestAttachParams(2))
|
||||
expect(retried).toMatchObject({ ok: true })
|
||||
// A fresh child was spawned; the orphan pid's lease did not survive as the owner.
|
||||
expect(acquire).toHaveBeenCalledTimes(2)
|
||||
expect(store.getRecord(SESSION)?.lease).toMatchObject({
|
||||
runtimeFence: 3,
|
||||
claimStatus: 'live',
|
||||
handoffStage: null
|
||||
})
|
||||
})
|
||||
|
||||
it('heals a stranded native owner during startup restore without waiting for a client', async () => {
|
||||
expect((await host.attach(CALLER, hostTestAttachParams(null))).ok).toBe(true)
|
||||
await reopenStore()
|
||||
|
||||
let orphanAlive = true
|
||||
const stopOwnerProcess = vi.fn(() => {
|
||||
orphanAlive = false
|
||||
})
|
||||
openHost({
|
||||
mintSpawnToken: () => 'spawn-b',
|
||||
probeOwner: async () =>
|
||||
orphanAlive
|
||||
? { outcome: 'identity-matched', matchedOn: ['process-start-time'] }
|
||||
: { outcome: 'pid-absent' },
|
||||
stopOwnerProcess
|
||||
})
|
||||
|
||||
await host.restoreReadableSessions()
|
||||
|
||||
expect(stopOwnerProcess).toHaveBeenCalledWith(4242, 'SIGTERM')
|
||||
expect(acquire).toHaveBeenCalledTimes(2)
|
||||
expect(store.getRecord(SESSION)?.lease).toMatchObject({
|
||||
runtimeFence: 3,
|
||||
claimStatus: 'live',
|
||||
handoffStage: null,
|
||||
ownerProcess: { spawnToken: 'spawn-b' }
|
||||
})
|
||||
})
|
||||
})
|
||||
+249
@@ -0,0 +1,249 @@
|
||||
import { mkdtemp, rm } from 'node:fs/promises'
|
||||
import { tmpdir } from 'node:os'
|
||||
import { join } from 'node:path'
|
||||
import { afterEach, describe, expect, it, vi } from 'vitest'
|
||||
import type { AgentSessionOwnerProbe } from '../../../shared/agent-session-lease-adjudication'
|
||||
import { AgentSessionRecordStore } from '../../runtime/agent-session-record-store'
|
||||
import {
|
||||
resolveStructuredSessionRecovery,
|
||||
type StructuredSessionRecoveryResolutionDeps
|
||||
} from './structured-agent-session-recovery-resolution'
|
||||
|
||||
const NOW = 1_800_000_000_000
|
||||
const SESSION = 'session-recovery'
|
||||
const roots: string[] = []
|
||||
let operations = 0
|
||||
|
||||
afterEach(async () => {
|
||||
await Promise.all(roots.splice(0).map((root) => rm(root, { recursive: true, force: true })))
|
||||
})
|
||||
|
||||
async function openStore(): Promise<AgentSessionRecordStore> {
|
||||
const root = await mkdtemp(join(tmpdir(), 'orca-recovery-resolution-'))
|
||||
roots.push(root)
|
||||
return AgentSessionRecordStore.open({ directory: root, hostId: 'local' })
|
||||
}
|
||||
|
||||
async function reserve(store: AgentSessionRecordStore, runtimeKind: 'native' | 'tui' = 'native') {
|
||||
operations += 1
|
||||
return store.reserveOwner({
|
||||
sessionId: SESSION,
|
||||
location: {
|
||||
executionHostId: 'local',
|
||||
wslDistro: null,
|
||||
workspaceId: 'workspace-1',
|
||||
workspaceKind: 'folder'
|
||||
},
|
||||
provider: 'codex',
|
||||
accountHome: { variable: 'CODEX_HOME', path: '/tmp/codex' },
|
||||
runtimeKind,
|
||||
expectedFence: null,
|
||||
spawnToken: 'spawn-recovery',
|
||||
claimKeyId: 'key-1',
|
||||
handoffOperationId: null,
|
||||
probe: { outcome: 'reservation-unused' },
|
||||
operation: {
|
||||
callerKey: 'test',
|
||||
operationId: `${NOW}-${String(operations).padStart(32, '0')}`,
|
||||
fingerprint: 'create'
|
||||
},
|
||||
now: NOW
|
||||
})
|
||||
}
|
||||
|
||||
async function liveOwner(store: AgentSessionRecordStore, runtimeKind: 'native' | 'tui' = 'native') {
|
||||
const reserved = await reserve(store, runtimeKind)
|
||||
const fence = reserved.record.lease.runtimeFence
|
||||
await store.commitProcessIdentity({
|
||||
sessionId: SESSION,
|
||||
fence,
|
||||
process: {
|
||||
hostId: 'local',
|
||||
pid: 4242,
|
||||
processStartTimeMs: NOW - 1_000,
|
||||
spawnToken: 'spawn-recovery'
|
||||
},
|
||||
now: NOW
|
||||
})
|
||||
return store.proveOwner({
|
||||
sessionId: SESSION,
|
||||
fence,
|
||||
link: {
|
||||
linkId: 'link-recovery',
|
||||
handle: { provider: 'codex', threadId: 'thread-recovery' },
|
||||
origin: 'created',
|
||||
mintedAtFence: fence,
|
||||
observedAt: NOW
|
||||
},
|
||||
now: NOW
|
||||
})
|
||||
}
|
||||
|
||||
async function latch(store: AgentSessionRecordStore, stage: 'recovering' | 'manual-recovery') {
|
||||
return store.transitionHandoff(SESSION, (record) => ({
|
||||
...record,
|
||||
lease: { ...record.lease, handoffStage: stage }
|
||||
}))
|
||||
}
|
||||
|
||||
function deps(
|
||||
store: AgentSessionRecordStore,
|
||||
probe: (calls: number) => AgentSessionOwnerProbe,
|
||||
overrides: Partial<StructuredSessionRecoveryResolutionDeps> = {}
|
||||
): StructuredSessionRecoveryResolutionDeps & { probes: () => number } {
|
||||
let calls = 0
|
||||
return {
|
||||
store,
|
||||
probeRecord: async () => {
|
||||
calls += 1
|
||||
return probe(calls)
|
||||
},
|
||||
now: () => NOW + 10_000,
|
||||
delay: async () => {},
|
||||
probes: () => calls,
|
||||
...overrides
|
||||
}
|
||||
}
|
||||
|
||||
describe('structured session recovery resolution', () => {
|
||||
it('releases a latched ownerless native reservation', async () => {
|
||||
const store = await openStore()
|
||||
await reserve(store)
|
||||
await latch(store, 'recovering')
|
||||
|
||||
const result = await resolveStructuredSessionRecovery(
|
||||
deps(store, () => ({ outcome: 'indeterminate', reason: 'no scan' })),
|
||||
SESSION
|
||||
)
|
||||
|
||||
expect(result).toBe('resolved')
|
||||
expect(store.getRecord(SESSION)?.lease).toMatchObject({
|
||||
claimStatus: 'released',
|
||||
handoffStage: null,
|
||||
runtimeFence: 2,
|
||||
reservedSpawnToken: null
|
||||
})
|
||||
})
|
||||
|
||||
it('evicts a latched owner the probe now proves dead, without a stop request', async () => {
|
||||
const store = await openStore()
|
||||
await liveOwner(store)
|
||||
await latch(store, 'manual-recovery')
|
||||
const stopOwnerProcess = vi.fn()
|
||||
|
||||
const result = await resolveStructuredSessionRecovery(
|
||||
deps(store, () => ({ outcome: 'pid-absent' }), { stopOwnerProcess }),
|
||||
SESSION
|
||||
)
|
||||
|
||||
expect(result).toBe('resolved')
|
||||
expect(stopOwnerProcess).not.toHaveBeenCalled()
|
||||
expect(store.getRecord(SESSION)?.lease).toMatchObject({
|
||||
claimStatus: 'released',
|
||||
handoffStage: null,
|
||||
runtimeFence: 2,
|
||||
ownerProcess: null
|
||||
})
|
||||
})
|
||||
|
||||
it('stops a live identity-matched orphan and evicts only after absence is proven', async () => {
|
||||
const store = await openStore()
|
||||
await liveOwner(store)
|
||||
await latch(store, 'recovering')
|
||||
let alive = true
|
||||
const stopOwnerProcess = vi.fn(() => {
|
||||
alive = false
|
||||
})
|
||||
|
||||
const result = await resolveStructuredSessionRecovery(
|
||||
deps(
|
||||
store,
|
||||
() =>
|
||||
alive
|
||||
? { outcome: 'identity-matched', matchedOn: ['process-start-time'] }
|
||||
: { outcome: 'pid-absent' },
|
||||
{ stopOwnerProcess }
|
||||
),
|
||||
SESSION
|
||||
)
|
||||
|
||||
expect(result).toBe('resolved')
|
||||
expect(stopOwnerProcess).toHaveBeenCalledTimes(1)
|
||||
expect(stopOwnerProcess).toHaveBeenCalledWith(4242, 'SIGTERM')
|
||||
expect(store.getRecord(SESSION)?.lease).toMatchObject({
|
||||
claimStatus: 'released',
|
||||
handoffStage: null,
|
||||
runtimeFence: 2
|
||||
})
|
||||
})
|
||||
|
||||
it('escalates the stop request but never evicts an owner that stays alive', async () => {
|
||||
const store = await openStore()
|
||||
await liveOwner(store)
|
||||
await latch(store, 'recovering')
|
||||
const stopOwnerProcess = vi.fn()
|
||||
|
||||
const result = await resolveStructuredSessionRecovery(
|
||||
deps(store, () => ({ outcome: 'identity-matched', matchedOn: ['process-start-time'] }), {
|
||||
stopOwnerProcess
|
||||
}),
|
||||
SESSION
|
||||
)
|
||||
|
||||
expect(result).toBe('unresolved')
|
||||
expect(stopOwnerProcess).toHaveBeenCalledWith(4242, 'SIGTERM')
|
||||
expect(stopOwnerProcess).toHaveBeenCalledWith(4242, 'SIGKILL')
|
||||
// The latch is preserved verbatim: no fence move, no cleared owner, no lost state.
|
||||
expect(store.getRecord(SESSION)?.lease).toMatchObject({
|
||||
claimStatus: 'live',
|
||||
handoffStage: 'recovering',
|
||||
runtimeFence: 1,
|
||||
ownerProcess: { pid: 4242 }
|
||||
})
|
||||
})
|
||||
|
||||
it('leaves an unverifiable owner latched and requests no stop', async () => {
|
||||
const store = await openStore()
|
||||
await liveOwner(store)
|
||||
await latch(store, 'recovering')
|
||||
const stopOwnerProcess = vi.fn()
|
||||
|
||||
const result = await resolveStructuredSessionRecovery(
|
||||
deps(store, () => ({ outcome: 'indeterminate', reason: 'probe timed out' }), {
|
||||
stopOwnerProcess
|
||||
}),
|
||||
SESSION
|
||||
)
|
||||
|
||||
expect(result).toBe('unresolved')
|
||||
expect(stopOwnerProcess).not.toHaveBeenCalled()
|
||||
expect(store.getRecord(SESSION)?.lease).toMatchObject({
|
||||
handoffStage: 'recovering',
|
||||
runtimeFence: 1
|
||||
})
|
||||
})
|
||||
|
||||
it('never touches a TUI record or a conflicted claim', async () => {
|
||||
const store = await openStore()
|
||||
await liveOwner(store, 'tui')
|
||||
await latch(store, 'recovering')
|
||||
expect(
|
||||
await resolveStructuredSessionRecovery(
|
||||
deps(store, () => ({ outcome: 'pid-absent' })),
|
||||
SESSION
|
||||
)
|
||||
).toBe('not-applicable')
|
||||
|
||||
await store.transitionHandoff(SESSION, (record) => ({
|
||||
...record,
|
||||
lease: { ...record.lease, runtimeKind: 'native', claimStatus: 'conflicted' }
|
||||
}))
|
||||
expect(
|
||||
await resolveStructuredSessionRecovery(
|
||||
deps(store, () => ({ outcome: 'pid-absent' })),
|
||||
SESSION
|
||||
)
|
||||
).toBe('not-applicable')
|
||||
expect(store.getRecord(SESSION)?.lease.handoffStage).toBe('recovering')
|
||||
})
|
||||
})
|
||||
+106
@@ -0,0 +1,106 @@
|
||||
/**
|
||||
* Exits from latched recovery stages. A session lands in `recovering` / `manual-recovery`
|
||||
* when evidence about its owner was UNAVAILABLE; this module re-asks with present-time
|
||||
* evidence and releases the lease only on proof. A stop is a request — the lease moves only
|
||||
* after a later probe proves the process absent, never on a timeout.
|
||||
*/
|
||||
|
||||
import {
|
||||
isProvenAliveProbe,
|
||||
isProvenDeadProbe,
|
||||
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'
|
||||
|
||||
export type StructuredSessionRecoveryStopSignal = 'SIGTERM' | 'SIGKILL'
|
||||
|
||||
export type StructuredSessionRecoveryResolutionDeps = {
|
||||
store: AgentSessionRecordStore
|
||||
probeRecord: (record: AgentSessionRecord) => Promise<AgentSessionOwnerProbe>
|
||||
now: () => number
|
||||
stopOwnerProcess?: (pid: number, signal: StructuredSessionRecoveryStopSignal) => void
|
||||
delay?: (ms: number) => Promise<void>
|
||||
}
|
||||
|
||||
const STOP_PROBES_PER_SIGNAL = 4
|
||||
const STOP_PROBE_INTERVAL_MS = 250
|
||||
|
||||
const UNRESOLVED_REFUSALS: ReadonlySet<string> = new Set([
|
||||
'agent_session_ownership_unknown',
|
||||
'agent_session_checkpoint_stale',
|
||||
'execution_owner_reconciling',
|
||||
'agent_session_identity_required'
|
||||
])
|
||||
|
||||
/** Only native records resolve here: a TUI owner has its own recovery transport, and a
|
||||
* conflicted claim is a user decision, not missing evidence. */
|
||||
export function structuredSessionRecoveryIsResolvable(record: AgentSessionRecord): boolean {
|
||||
return (
|
||||
record.lease.runtimeKind === 'native' &&
|
||||
record.lease.claimStatus !== 'conflicted' &&
|
||||
(record.lease.handoffStage === 'recovering' || record.lease.handoffStage === 'manual-recovery')
|
||||
)
|
||||
}
|
||||
|
||||
export async function resolveStructuredSessionRecovery(
|
||||
deps: StructuredSessionRecoveryResolutionDeps,
|
||||
sessionId: string
|
||||
): Promise<'resolved' | 'unresolved' | 'not-applicable'> {
|
||||
const record = deps.store.getRecord(sessionId)
|
||||
if (!record || !structuredSessionRecoveryIsResolvable(record)) {
|
||||
return 'not-applicable'
|
||||
}
|
||||
let probe = await deps.probeRecord(record)
|
||||
const owner = record.lease.ownerProcess
|
||||
if (owner && owner.hostId === deps.store.hostId && isProvenAliveProbe(probe)) {
|
||||
// The owner is a live child of a runtime that no longer exists; its transport cannot be
|
||||
// reconstructed, so the only way forward is to stop it and prove it gone.
|
||||
probe = await stopOwnerAndReprobe(deps, record, owner.pid)
|
||||
}
|
||||
try {
|
||||
await deps.store.evictProvenDeadOwner({
|
||||
sessionId,
|
||||
expectedFence: record.lease.runtimeFence,
|
||||
probe,
|
||||
now: deps.now()
|
||||
})
|
||||
return 'resolved'
|
||||
} catch (error) {
|
||||
const code = error instanceof Error ? error.message : String(error)
|
||||
if (UNRESOLVED_REFUSALS.has(code)) {
|
||||
// No proof yet; the record is preserved untouched and the next attempt re-asks.
|
||||
return 'unresolved'
|
||||
}
|
||||
throw error
|
||||
}
|
||||
}
|
||||
|
||||
async function stopOwnerAndReprobe(
|
||||
deps: StructuredSessionRecoveryResolutionDeps,
|
||||
record: AgentSessionRecord,
|
||||
pid: number
|
||||
): Promise<AgentSessionOwnerProbe> {
|
||||
const stop = deps.stopOwnerProcess ?? defaultStopOwnerProcess
|
||||
const delay = deps.delay ?? ((ms: number) => new Promise((resolve) => setTimeout(resolve, ms)))
|
||||
let probe: AgentSessionOwnerProbe = { outcome: 'indeterminate', reason: 'owner stop requested' }
|
||||
for (const signal of ['SIGTERM', 'SIGKILL'] as const) {
|
||||
stop(pid, signal)
|
||||
for (let attempt = 0; attempt < STOP_PROBES_PER_SIGNAL; attempt += 1) {
|
||||
probe = await deps.probeRecord(record)
|
||||
if (isProvenDeadProbe(probe)) {
|
||||
return probe
|
||||
}
|
||||
await delay(STOP_PROBE_INTERVAL_MS)
|
||||
}
|
||||
}
|
||||
return probe
|
||||
}
|
||||
|
||||
function defaultStopOwnerProcess(pid: number, signal: StructuredSessionRecoveryStopSignal): void {
|
||||
try {
|
||||
process.kill(pid, signal)
|
||||
} catch {
|
||||
// Already gone or not ours to signal; the next probe supplies the actual proof.
|
||||
}
|
||||
}
|
||||
@@ -13,6 +13,7 @@ export async function restoreStructuredAgentSessionsOnRestart(input: {
|
||||
journalRoot: string
|
||||
records: AgentSessionRecord[]
|
||||
reconcile: (sessionId: string) => Promise<AgentSessionWireRefusal | null>
|
||||
resolveRecovery: (sessionId: string) => Promise<unknown>
|
||||
operationId: () => string
|
||||
resume: (params: AgentSessionAttachParams) => Promise<boolean>
|
||||
serialize: <T>(sessionId: string, task: () => Promise<T>) => Promise<T>
|
||||
@@ -23,6 +24,10 @@ export async function restoreStructuredAgentSessionsOnRestart(input: {
|
||||
await Promise.all(
|
||||
input.records.map(async ({ sessionId }) => {
|
||||
const unreconciled = await input.reconcile(sessionId)
|
||||
if (!unreconciled) {
|
||||
// A session latched in recovery exits here at startup, without waiting for a client.
|
||||
await input.resolveRecovery(sessionId)
|
||||
}
|
||||
const current = input.store.getRecord(sessionId)
|
||||
if (
|
||||
!unreconciled &&
|
||||
|
||||
@@ -65,6 +65,29 @@ describe('restarted handoff adjudication', () => {
|
||||
})
|
||||
})
|
||||
|
||||
it('fully releases an abandoned native reservation instead of resuming it as a handoff', () => {
|
||||
// A native attach reservation also parks at new-owner-proving; rolling it into the
|
||||
// handoff-retry machinery would strand journal-less sessions behind a stale operation.
|
||||
const abandoned = record('new-owner-proving')
|
||||
abandoned.lease.runtimeKind = 'native'
|
||||
abandoned.lease.ownerProcess = null
|
||||
expect(
|
||||
adjudicateRestartedAgentSessionHandoff(
|
||||
abandoned,
|
||||
{ outcome: 'indeterminate', reason: 'no spawn-token scan' },
|
||||
NOW + 1_000
|
||||
).lease
|
||||
).toMatchObject({
|
||||
runtimeKind: 'native',
|
||||
runtimeFence: 5,
|
||||
handoffStage: null,
|
||||
handoffOperationId: null,
|
||||
claimStatus: 'released',
|
||||
ownerProcess: null,
|
||||
reservedSpawnToken: null
|
||||
})
|
||||
})
|
||||
|
||||
it('rolls a dead proving target back to the source kind before retry', () => {
|
||||
expect(
|
||||
adjudicateRestartedAgentSessionHandoff(
|
||||
|
||||
@@ -27,6 +27,27 @@ export function adjudicateRestartedAgentSessionHandoff(
|
||||
lastRenewedAt: now
|
||||
})
|
||||
}
|
||||
if (
|
||||
record.lease.ownerProcess === null &&
|
||||
record.lease.runtimeKind === 'native' &&
|
||||
record.lease.claimStatus === 'reserved'
|
||||
) {
|
||||
// Why: an abandoned native reservation has no stoppable owner to hand back; parking it
|
||||
// at old-owner-stopped would strand journal-less sessions behind a stale operation id.
|
||||
return updateLease(record, {
|
||||
...record.lease,
|
||||
runtimeFence: adjudication.nextFence,
|
||||
handoffStage: null,
|
||||
handoffOperationId: null,
|
||||
ownerProcess: null,
|
||||
reservedSpawnToken: null,
|
||||
processlessAt: null,
|
||||
claimStatus: 'released',
|
||||
unreconciled: false,
|
||||
lastRenewedAt: now,
|
||||
deathEvidence: adjudication.evidence
|
||||
})
|
||||
}
|
||||
const provingTarget = record.lease.handoffStage === 'new-owner-proving'
|
||||
return updateLease(record, {
|
||||
...record.lease,
|
||||
|
||||
@@ -0,0 +1,41 @@
|
||||
import { describe, expect, it } from 'vitest'
|
||||
import {
|
||||
readEchoedAgentSessionSpawnToken,
|
||||
spawnTokenFromEnvironBlock
|
||||
} from './agent-session-spawn-token-readback'
|
||||
|
||||
const IDENTITY = {
|
||||
hostId: 'local',
|
||||
pid: process.pid,
|
||||
processStartTimeMs: null,
|
||||
spawnToken: 'spawn-a'
|
||||
}
|
||||
|
||||
describe('spawn token read-back', () => {
|
||||
it('finds the token in a NUL-separated environ block', () => {
|
||||
const block = [
|
||||
'PATH=/usr/bin',
|
||||
'ORCA_AGENT_SESSION_SPAWN_TOKEN=tok-123',
|
||||
'HOME=/home/dev'
|
||||
].join('\0')
|
||||
expect(spawnTokenFromEnvironBlock(block)).toBe('tok-123')
|
||||
})
|
||||
|
||||
it('answers null for an absent or empty token instead of guessing', () => {
|
||||
expect(spawnTokenFromEnvironBlock(['PATH=/usr/bin', 'HOME=/home/dev'].join('\0'))).toBeNull()
|
||||
expect(spawnTokenFromEnvironBlock('ORCA_AGENT_SESSION_SPAWN_TOKEN=')).toBeNull()
|
||||
// A prefix collision is not a match.
|
||||
expect(spawnTokenFromEnvironBlock('ORCA_AGENT_SESSION_SPAWN_TOKEN_EXTRA=x')).toBeNull()
|
||||
})
|
||||
|
||||
it('answers null on platforms that hide process environments', async () => {
|
||||
await expect(readEchoedAgentSessionSpawnToken(IDENTITY, 'darwin')).resolves.toBeNull()
|
||||
await expect(readEchoedAgentSessionSpawnToken(IDENTITY, 'win32')).resolves.toBeNull()
|
||||
})
|
||||
|
||||
it('answers null when the environ file is unreadable', async () => {
|
||||
await expect(
|
||||
readEchoedAgentSessionSpawnToken({ ...IDENTITY, pid: 2 ** 30 }, 'linux')
|
||||
).resolves.toBeNull()
|
||||
})
|
||||
})
|
||||
@@ -0,0 +1,37 @@
|
||||
/**
|
||||
* Reads the spawn token a live child carries in its environment, giving the owner probe a
|
||||
* PID-reuse-safe identity element even when no start time was recorded. Only Linux exposes
|
||||
* another process's environment (/proc/<pid>/environ); macOS and Windows answer null, which
|
||||
* the probe treats as "no answer" — never as proof in either direction.
|
||||
*/
|
||||
|
||||
import { readFile } from 'node:fs/promises'
|
||||
import type { AgentSessionProcessIdentity } from '../../shared/agent-session-record'
|
||||
import { CODEX_SPAWN_TOKEN_ENV } from '../codex/codex-structured-owner-identity'
|
||||
|
||||
export function spawnTokenFromEnvironBlock(
|
||||
block: string,
|
||||
variable: string = CODEX_SPAWN_TOKEN_ENV
|
||||
): string | null {
|
||||
for (const entry of block.split('\0')) {
|
||||
if (entry.startsWith(`${variable}=`)) {
|
||||
const value = entry.slice(variable.length + 1)
|
||||
return value.length > 0 ? value : null
|
||||
}
|
||||
}
|
||||
return null
|
||||
}
|
||||
|
||||
export async function readEchoedAgentSessionSpawnToken(
|
||||
identity: AgentSessionProcessIdentity,
|
||||
platform: NodeJS.Platform = process.platform
|
||||
): Promise<string | null> {
|
||||
if (platform !== 'linux') {
|
||||
return null
|
||||
}
|
||||
try {
|
||||
return spawnTokenFromEnvironBlock(await readFile(`/proc/${identity.pid}/environ`, 'utf-8'))
|
||||
} catch {
|
||||
return null
|
||||
}
|
||||
}
|
||||
@@ -313,7 +313,8 @@ beforeEach(async () => {
|
||||
EXAMPLE_GATEWAY_TOKEN: 'shell-exported',
|
||||
CODEX_HOME: '/shell/home'
|
||||
}),
|
||||
openCodexConnection: codex.openConnection
|
||||
openCodexConnection: codex.openConnection,
|
||||
readProcessStartTime: async () => 1_700_000_000_000
|
||||
}).then(() => undefined),
|
||||
registerSubscriptionCleanup: vi.fn((id: string, dispose: () => void) =>
|
||||
cleanups.set(id, dispose)
|
||||
@@ -779,7 +780,8 @@ describe('a structured codex session over agentSession.*', () => {
|
||||
claimKeyId: 'key-1',
|
||||
resolveWorkspacePath: async (workspaceId) => `/repos/${workspaceId}`,
|
||||
resolveCodexCommand: () => '/usr/local/bin/codex',
|
||||
openCodexConnection: codex.openConnection
|
||||
openCodexConnection: codex.openConnection,
|
||||
readProcessStartTime: async () => 1_700_000_000_000
|
||||
})
|
||||
const adapter = (host as unknown as { deps: { adapter: CodexStructuredSessionAdapter } }).deps
|
||||
.adapter
|
||||
|
||||
@@ -26,7 +26,10 @@ describe('structured agent-session owner probe', () => {
|
||||
const probe = vi.fn(async () => ({ outcome: 'pid-absent' }) as const)
|
||||
const result = await createStructuredAgentSessionOwnerProbe(HOST_ID, probe)(record(OWNER))
|
||||
|
||||
expect(probe).toHaveBeenCalledWith({ identity: OWNER })
|
||||
expect(probe).toHaveBeenCalledWith({
|
||||
identity: OWNER,
|
||||
deps: { readEchoedSpawnToken: expect.any(Function) }
|
||||
})
|
||||
expect(result).toEqual({ outcome: 'pid-absent' })
|
||||
})
|
||||
|
||||
|
||||
@@ -20,6 +20,7 @@ import type { StructuredAgentSessionHandoffTransport } from '../native-chat/agen
|
||||
import { setStructuredAgentSessionHost } from '../native-chat/agent-session-wire/structured-agent-session-registry'
|
||||
import { AgentSessionRecordStore } from './agent-session-record-store'
|
||||
import { probeAgentSessionProcessIdentity } from './agent-session-process-identity-probe'
|
||||
import { readEchoedAgentSessionSpawnToken } from './agent-session-spawn-token-readback'
|
||||
import { agentSessionPtyWriteGate } from './agent-session-pty-write-gate'
|
||||
import { resolveLoginShellEnvironment } from '../startup/login-shell-environment'
|
||||
|
||||
@@ -40,6 +41,8 @@ export type StructuredAgentSessionRuntimeDeps = {
|
||||
/** Transport for the Codex child. Overridden only to drive the whole runtime
|
||||
* against a scripted app-server; production spawns the real one. */
|
||||
openCodexConnection?: CodexStructuredSessionAdapterDeps['openConnection']
|
||||
/** Scripted app-servers carry fake pids the real start-time read cannot answer for. */
|
||||
readProcessStartTime?: CodexStructuredSessionAdapterDeps['readProcessStartTime']
|
||||
resolveLaunchEnv?: () => Promise<NodeJS.ProcessEnv>
|
||||
onError?: (input: { scope: string; error: unknown }) => void
|
||||
handoffTransport?: StructuredAgentSessionHandoffTransport
|
||||
@@ -97,7 +100,8 @@ async function install(deps: StructuredAgentSessionRuntimeDeps): Promise<Install
|
||||
resolveWorkspacePath: deps.resolveWorkspacePath,
|
||||
...(deps.resolveCodexCommand ? { resolveCommand: deps.resolveCodexCommand } : {})
|
||||
}),
|
||||
...(deps.openCodexConnection ? { openConnection: deps.openCodexConnection } : {})
|
||||
...(deps.openCodexConnection ? { openConnection: deps.openCodexConnection } : {}),
|
||||
...(deps.readProcessStartTime ? { readProcessStartTime: deps.readProcessStartTime } : {})
|
||||
})
|
||||
const host = new StructuredAgentSessionHost({
|
||||
store,
|
||||
@@ -149,6 +153,11 @@ export function createStructuredAgentSessionOwnerProbe(
|
||||
reason: `owner runs on ${owner.hostId}, which this host cannot probe`
|
||||
}
|
||||
}
|
||||
return probe({ identity: owner })
|
||||
// The env read-back answers on hosts that expose it and null elsewhere, giving the
|
||||
// probe a PID-reuse-safe element even when no start time was recorded.
|
||||
return probe({
|
||||
identity: owner,
|
||||
deps: { readEchoedSpawnToken: readEchoedAgentSessionSpawnToken }
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
@@ -183,10 +183,22 @@ describe('acquisition compare-and-swap', () => {
|
||||
})
|
||||
|
||||
describe('restart reconciliation', () => {
|
||||
it('re-adopts a proven-live owner without moving the fence', () => {
|
||||
it('re-adopts a proven-live TUI owner without moving the fence', () => {
|
||||
expect(
|
||||
adjudicateAgentSessionRestart({
|
||||
lease: lease({ runtimeKind: 'tui' }),
|
||||
probe: MATCHED,
|
||||
observedAt: 9_000
|
||||
})
|
||||
).toEqual({ disposition: 'readopt' })
|
||||
})
|
||||
|
||||
it('routes a surviving native owner to recovery instead of readopting a dead transport', () => {
|
||||
// The native child's stdio belonged to the runtime that died; readoption would extend
|
||||
// a lease no process can drive. Recovery stops the orphan and respawns at fence + 1.
|
||||
expect(
|
||||
adjudicateAgentSessionRestart({ lease: lease(), probe: MATCHED, observedAt: 9_000 })
|
||||
).toEqual({ disposition: 'readopt' })
|
||||
).toMatchObject({ disposition: 'recovering', stage: 'recovering' })
|
||||
})
|
||||
|
||||
it('bumps the fence exactly once for a proven-dead owner and records the evidence', () => {
|
||||
@@ -226,8 +238,18 @@ describe('restart reconciliation', () => {
|
||||
).toEqual({ disposition: 'conflicted', reason: 'claim conflicted before restart' })
|
||||
})
|
||||
|
||||
it('frees a reservation only when nothing ever spawned', () => {
|
||||
it('frees an abandoned native reservation even without a spawn-token scan', () => {
|
||||
// Restart proves the reserving runtime is gone, and a never-proven native child lost
|
||||
// its only request channel with it — nothing under this token can write again.
|
||||
const reserved = lease({ ownerProcess: null, claimStatus: 'reserved' })
|
||||
expect(
|
||||
adjudicateAgentSessionRestart({ lease: reserved, probe: INDETERMINATE, observedAt: 9_000 })
|
||||
).toMatchObject({ disposition: 'evicted', nextFence: 8 })
|
||||
})
|
||||
|
||||
it('frees a TUI reservation only when a probe proves nothing ever spawned', () => {
|
||||
// A TUI child lives in a terminal that outlives the runtime, so absence needs proof.
|
||||
const reserved = lease({ ownerProcess: null, claimStatus: 'reserved', runtimeKind: 'tui' })
|
||||
expect(
|
||||
adjudicateAgentSessionRestart({
|
||||
lease: reserved,
|
||||
|
||||
@@ -191,6 +191,21 @@ export function adjudicateAgentSessionRestart(args: {
|
||||
evidence: { kind: 'pid-absent', detail: 'reservation never spawned', observedAt }
|
||||
}
|
||||
}
|
||||
if (lease.runtimeKind === 'native' && lease.claimStatus === 'reserved') {
|
||||
// Why: adjudication itself proves the reserving runtime died before an owner was
|
||||
// proven, and a never-proven native child lost its only request channel with that
|
||||
// runtime — nothing under this token can reach the provider session again. A TUI
|
||||
// child outlives the runtime inside its terminal, so it stays latched below.
|
||||
return {
|
||||
disposition: 'evicted',
|
||||
nextFence: lease.runtimeFence + 1,
|
||||
evidence: {
|
||||
kind: 'pid-absent',
|
||||
detail: 'native reservation abandoned before an owner was proven',
|
||||
observedAt
|
||||
}
|
||||
}
|
||||
}
|
||||
return {
|
||||
disposition: 'recovering',
|
||||
stage: 'recovering',
|
||||
@@ -198,6 +213,15 @@ export function adjudicateAgentSessionRestart(args: {
|
||||
}
|
||||
}
|
||||
if (isProvenAliveProbe(probe)) {
|
||||
if (lease.runtimeKind === 'native') {
|
||||
// Why: the surviving child's stdio died with the previous runtime, so readoption
|
||||
// would renew a lease no host can drive. Recovery stops it and respawns at fence + 1.
|
||||
return {
|
||||
disposition: 'recovering',
|
||||
stage: 'recovering',
|
||||
reason: 'native owner outlived the runtime that held its transport'
|
||||
}
|
||||
}
|
||||
// Why: re-adoption is not a new generation, so the fence does not move.
|
||||
return { disposition: 'readopt' }
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user