mirror of
https://github.com/stablyai/orca.git
synced 2026-10-23 08:03:39 +00:00
fix(agent-session): give every latched recovery state an exit, and free the profiles already stuck in one
A session could reach `recovering` / `manual-recovery` with no way back out, so the chat behind it never opened again. Three separate latches, all of them permanent: - A lease with no owner AND no reserved spawn token was treated as an unproven reservation. Nothing was ever minted that a child could carry, so it is already free — but every released record re-latched to `recovering` on the next boot. `adjudicateAgentSessionRestart` now answers `free` for that shape and the owner probe answers `reservation-unused`, neither of which moves the fence. - `claimStatus: 'conflicted'` had no clearing path anywhere in the product. The v1 to v2 record upgrade stamps every migrated row conflicted + manual-recovery, and restart adjudication refused to consult the probe at all. It now evicts when present-time evidence proves the process the conflict NAMES has exited; with no proof, and with no named process, the conflict still stands. - Recovery resolution skipped every TUI record because a TUI owner has its own recovery transport. That transport needs a process to talk to, so a TUI reservation that crashed before `commitProcessIdentity` had no exit at all. Records, journals, and provider handle chains are untouched throughout: an exit is a re-adjudication of ownership, never a deletion. Where evidence stays unavailable the refusal now names the process Orca could not prove gone and what to do about it, instead of echoing a store code. Also wires `listOrphanSpawnTokens`, which had no production caller: restart restore now stops children carrying a token no lease claims BEFORE recovery grants a new owner, so a child that went down with its record cannot end up as a second live writer on one provider session. The scan reads `/proc/<pid>/environ` and answers null - "cannot enumerate", never "none" - on macOS and Windows.
This commit is contained in:
@@ -95,7 +95,11 @@ export async function performAttach(
|
||||
}
|
||||
return {
|
||||
ok: false,
|
||||
refusal: classifyStoreFailure(error, store.getRecord(sessionId)?.lease.runtimeFence ?? null)
|
||||
refusal: classifyStoreFailure(
|
||||
error,
|
||||
store.getRecord(sessionId)?.lease.runtimeFence ?? null,
|
||||
store.getRecord(sessionId)
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -38,6 +38,7 @@ import {
|
||||
type AgentSessionJournalRecovery
|
||||
} from './agent-session-journal-recovery'
|
||||
import type { StructuredAgentSessionAdapter } from './structured-agent-session-adapter'
|
||||
import { structuredAgentSessionRefusalMessage } from './structured-agent-session-refusal-message'
|
||||
|
||||
/**
|
||||
* Everything a client may declare about the session it wants. Deliberately no
|
||||
@@ -199,15 +200,20 @@ export function reserveRequestFor(input: {
|
||||
* known set is a defect, not a client error, and is rethrown. */
|
||||
export function classifyStoreFailure(
|
||||
error: unknown,
|
||||
currentFence: number | null
|
||||
currentFence: number | null,
|
||||
record: AgentSessionRecord | null = null
|
||||
): AgentSessionWireRefusal {
|
||||
const code = error instanceof Error ? error.message : String(error)
|
||||
if (!(AGENT_SESSION_WIRE_REFUSAL_CODES as readonly string[]).includes(code)) {
|
||||
const rawCode = error instanceof Error ? error.message : String(error)
|
||||
if (!(AGENT_SESSION_WIRE_REFUSAL_CODES as readonly string[]).includes(rawCode)) {
|
||||
throw error
|
||||
}
|
||||
const code = rawCode as AgentSessionWireRefusalCode
|
||||
return {
|
||||
code: code as AgentSessionWireRefusalCode,
|
||||
message: `The session store refused this call: ${code}.`,
|
||||
code,
|
||||
// Why: a latched session is exactly where a bare store code strands the user.
|
||||
message:
|
||||
structuredAgentSessionRefusalMessage(code, record) ??
|
||||
`The session store refused this call: ${code}.`,
|
||||
...(code === 'agent_session_checkpoint_stale' && currentFence !== null ? { currentFence } : {})
|
||||
}
|
||||
}
|
||||
|
||||
+14
@@ -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 { stopOrphanAgentSessionChildren } from '../../runtime/agent-session-orphan-child-reaper'
|
||||
import { resolveStructuredSessionRecovery } from './structured-agent-session-recovery-resolution'
|
||||
|
||||
export class StructuredAgentSessionHostRuntimeState {
|
||||
@@ -59,6 +60,19 @@ export class StructuredAgentSessionHostRuntimeState {
|
||||
await Promise.all([...this.eventSinks.values()].map((sink) => sink.drained()))
|
||||
}
|
||||
|
||||
/** Stops children carrying a spawn token no lease claims. A failure here must not block
|
||||
* restore: an unreaped orphan is a worse outcome than a session that never comes back. */
|
||||
reapOrphanChildren(): Promise<number[]> {
|
||||
return stopOrphanAgentSessionChildren({
|
||||
store: this.deps.store,
|
||||
...(this.deps.scanSpawnTokenProcesses ? { scan: this.deps.scanSpawnTokenProcesses } : {}),
|
||||
...(this.deps.stopOwnerProcess ? { stop: this.deps.stopOwnerProcess } : {})
|
||||
}).catch((error: unknown) => {
|
||||
this.deps.onEventSinkError?.({ sessionId: 'orphan-spawn-token-reap', error })
|
||||
return []
|
||||
})
|
||||
}
|
||||
|
||||
/** Exit from a latched recovery stage when present-time evidence permits one. */
|
||||
resolveRecovery(sessionId: string): Promise<'resolved' | 'unresolved' | 'not-applicable'> {
|
||||
return resolveStructuredSessionRecovery(
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
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 { AgentSessionSpawnTokenScan } from '../../runtime/agent-session-spawn-token-process-scan'
|
||||
import type { AgentSessionJournal } from '../agent-session-journal/journal-store'
|
||||
import type { StructuredAgentSessionAdapter } from './structured-agent-session-adapter'
|
||||
import type { AgentSessionAttachParams } from './structured-agent-session-attach'
|
||||
@@ -22,6 +23,8 @@ export type StructuredAgentSessionHostDeps = {
|
||||
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
|
||||
/** Host spawn-token process scan; null means the platform cannot enumerate, never "none". */
|
||||
scanSpawnTokenProcesses?: () => Promise<AgentSessionSpawnTokenScan | null>
|
||||
mintSpawnToken?: () => string
|
||||
resolveLaunchEnv?: (
|
||||
provider: AgentSessionRecord['provider']
|
||||
|
||||
@@ -107,6 +107,7 @@ export class StructuredAgentSessionHost {
|
||||
hasSession: (sessionId) => this.sessions.has(sessionId),
|
||||
onReadable: (sessionId, restored) => this.sessions.set(sessionId, restored),
|
||||
restoreHandoff: (sessionId) => this.handoffs.restore(sessionId),
|
||||
reapOrphanChildren: () => this.runtimeState.reapOrphanChildren(),
|
||||
now: this.now
|
||||
})
|
||||
this.runtimeState.startLeaseRenewal()
|
||||
|
||||
+15
-6
@@ -21,20 +21,29 @@ export class StructuredAgentSessionReadableRestorer {
|
||||
hasSession: (sessionId: string) => boolean
|
||||
onReadable: (sessionId: string, restored: RestoredStructuredAgentSessionRead) => void
|
||||
restoreHandoff: (sessionId: string) => Promise<void>
|
||||
/** Reaps children carrying a token no lease claims, before recovery grants a new owner. */
|
||||
reapOrphanChildren?: () => Promise<unknown>
|
||||
now: () => number
|
||||
}
|
||||
) {}
|
||||
|
||||
restore(): Promise<void> {
|
||||
this.restorePromise ??= restoreStructuredAgentSessionsOnRestart({
|
||||
...this.input,
|
||||
records: this.input.store.listRecords().filter(this.input.supportsRecord),
|
||||
operationId: () =>
|
||||
`${Math.trunc(this.input.now()).toString().padStart(13, '0')}-${randomUUID().replaceAll('-', '')}`
|
||||
}).catch((error) => {
|
||||
this.restorePromise ??= this.reapThenRestore().catch((error: unknown) => {
|
||||
this.restorePromise = null
|
||||
throw error
|
||||
})
|
||||
return this.restorePromise
|
||||
}
|
||||
|
||||
private async reapThenRestore(): Promise<void> {
|
||||
// A lost record leaves its child unreferenced while recovery is about to hand the same
|
||||
// provider session a fresh owner, so the orphan has to be stopped before that, not after.
|
||||
await this.input.reapOrphanChildren?.()
|
||||
await restoreStructuredAgentSessionsOnRestart({
|
||||
...this.input,
|
||||
records: this.input.store.listRecords().filter(this.input.supportsRecord),
|
||||
operationId: () =>
|
||||
`${Math.trunc(this.input.now()).toString().padStart(13, '0')}-${randomUUID().replaceAll('-', '')}`
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
+62
-11
@@ -223,21 +223,11 @@ describe('structured session recovery resolution', () => {
|
||||
})
|
||||
})
|
||||
|
||||
it('never touches a TUI record or a conflicted claim', async () => {
|
||||
it('leaves a TUI record that still names an owner to its own recovery transport', 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' })),
|
||||
@@ -246,4 +236,65 @@ describe('structured session recovery resolution', () => {
|
||||
).toBe('not-applicable')
|
||||
expect(store.getRecord(SESSION)?.lease.handoffStage).toBe('recovering')
|
||||
})
|
||||
|
||||
it('resolves a TUI reservation that names nobody, because nothing else can', async () => {
|
||||
// The TUI carve-out exists because a TUI owner has its own recovery transport, and that
|
||||
// transport needs a process to talk to. A reservation that crashed before `commitProcessIdentity`
|
||||
// names none, so skipping it here left the session with no exit at all.
|
||||
const store = await openStore()
|
||||
await reserve(store, 'tui')
|
||||
await latch(store, 'recovering')
|
||||
|
||||
expect(
|
||||
await resolveStructuredSessionRecovery(
|
||||
deps(store, () => ({ outcome: 'reservation-unused' })),
|
||||
SESSION
|
||||
)
|
||||
).toBe('resolved')
|
||||
expect(store.getRecord(SESSION)?.lease).toMatchObject({
|
||||
handoffStage: null,
|
||||
claimStatus: 'released'
|
||||
})
|
||||
})
|
||||
|
||||
it('frees a conflicted claim once its named owner is proven gone', async () => {
|
||||
const store = await openStore()
|
||||
await liveOwner(store)
|
||||
await store.markClaimConflicted(SESSION, NOW)
|
||||
|
||||
expect(
|
||||
await resolveStructuredSessionRecovery(
|
||||
deps(store, () => ({ outcome: 'pid-absent' })),
|
||||
SESSION
|
||||
)
|
||||
).toBe('resolved')
|
||||
expect(store.getRecord(SESSION)?.lease).toMatchObject({
|
||||
handoffStage: null,
|
||||
claimStatus: 'released',
|
||||
deathEvidence: { kind: 'pid-absent' }
|
||||
})
|
||||
})
|
||||
|
||||
it('never stops the process a conflicted claim names, and keeps the conflict without proof', async () => {
|
||||
const store = await openStore()
|
||||
await liveOwner(store)
|
||||
await store.markClaimConflicted(SESSION, NOW)
|
||||
const stopOwnerProcess = vi.fn()
|
||||
|
||||
const result = await resolveStructuredSessionRecovery(
|
||||
deps(store, () => ({ outcome: 'identity-matched', matchedOn: ['spawn-token'] }), {
|
||||
stopOwnerProcess
|
||||
}),
|
||||
SESSION
|
||||
)
|
||||
|
||||
// Ownership was never settled, so the process on the other side of the conflict is not
|
||||
// Orca's to kill; only the user can decide which claimant wins.
|
||||
expect(stopOwnerProcess).not.toHaveBeenCalled()
|
||||
expect(result).toBe('unresolved')
|
||||
expect(store.getRecord(SESSION)?.lease).toMatchObject({
|
||||
claimStatus: 'conflicted',
|
||||
handoffStage: 'manual-recovery'
|
||||
})
|
||||
})
|
||||
})
|
||||
|
||||
+26
-8
@@ -33,14 +33,27 @@ const UNRESOLVED_REFUSALS: ReadonlySet<string> = new Set([
|
||||
'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. */
|
||||
/** Which latched records this module may re-ask about. */
|
||||
export function structuredSessionRecoveryIsResolvable(record: AgentSessionRecord): boolean {
|
||||
return (
|
||||
record.lease.runtimeKind === 'native' &&
|
||||
record.lease.claimStatus !== 'conflicted' &&
|
||||
(record.lease.handoffStage === 'recovering' || record.lease.handoffStage === 'manual-recovery')
|
||||
)
|
||||
const { claimStatus, handoffStage, ownerProcess, runtimeKind } = record.lease
|
||||
if (handoffStage !== 'recovering' && handoffStage !== 'manual-recovery') {
|
||||
return false
|
||||
}
|
||||
if (claimStatus === 'conflicted') {
|
||||
// A conflict names one process. Re-asking is only meaningful against that name; with none
|
||||
// recorded there is nothing present-time evidence could settle, and the user decides.
|
||||
return ownerProcess !== null
|
||||
}
|
||||
// A TUI owner has its own recovery transport — but that transport needs a process to talk to
|
||||
// (`structuredManualRecoveryIsAdmissible` requires one). A TUI reservation that crashed before
|
||||
// its identity was committed names nobody, so nothing else in the system can exit it.
|
||||
return runtimeKind === 'native' || ownerProcess === null
|
||||
}
|
||||
|
||||
/** Stopping a matched owner is only Orca's call when Orca owned its transport. A conflicted claim
|
||||
* means ownership was never settled, and a TUI child is the user's foreground agent. */
|
||||
function recoveryMayStopOwner(record: AgentSessionRecord): boolean {
|
||||
return record.lease.runtimeKind === 'native' && record.lease.claimStatus !== 'conflicted'
|
||||
}
|
||||
|
||||
export async function resolveStructuredSessionRecovery(
|
||||
@@ -53,7 +66,12 @@ export async function resolveStructuredSessionRecovery(
|
||||
}
|
||||
let probe = await deps.probeRecord(record)
|
||||
const owner = record.lease.ownerProcess
|
||||
if (owner && owner.hostId === deps.store.hostId && isProvenAliveProbe(probe)) {
|
||||
if (
|
||||
owner &&
|
||||
owner.hostId === deps.store.hostId &&
|
||||
isProvenAliveProbe(probe) &&
|
||||
recoveryMayStopOwner(record)
|
||||
) {
|
||||
// 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)
|
||||
|
||||
@@ -0,0 +1,45 @@
|
||||
/**
|
||||
* Human-readable text for a lease refusal.
|
||||
*
|
||||
* A latched session is the one place a bare code is worst: the user is looking at a chat that will
|
||||
* not open, and `agent_session_ownership_unknown` tells them neither what Orca could not prove nor
|
||||
* what they can do about it. Every message here names the specific evidence that is missing and
|
||||
* the action that supplies it.
|
||||
*/
|
||||
|
||||
import type { AgentSessionRecord } from '../../../shared/agent-session-record'
|
||||
import type { AgentSessionWireRefusalCode } from '../../../shared/agent-session-wire'
|
||||
|
||||
function ownerDescription(record: AgentSessionRecord): string {
|
||||
const owner = record.lease.ownerProcess
|
||||
return owner ? `process ${owner.pid} on ${owner.hostId}` : 'a process it never got to record'
|
||||
}
|
||||
|
||||
function latchedMessage(record: AgentSessionRecord): string {
|
||||
const owner = record.lease.ownerProcess
|
||||
if (record.lease.claimStatus === 'conflicted') {
|
||||
return owner
|
||||
? `Two runtimes claimed this session and Orca cannot yet prove that ${ownerDescription(record)} has exited. Quit that process, or reopen this chat once it is gone, and Orca will take the session back.`
|
||||
: 'Two runtimes claimed this session and the record names no process to check. Quit any other Orca or agent process using this workspace, then reopen this chat.'
|
||||
}
|
||||
return owner
|
||||
? `Orca cannot prove that ${ownerDescription(record)} — the previous owner of this session — has exited, so it will not start a second agent on the same conversation. Quit that process and reopen this chat.`
|
||||
: 'Orca cannot tell whether an agent started for this session before the app stopped, so it will not start a second one on the same conversation. Quit any leftover agent process for this workspace and reopen this chat.'
|
||||
}
|
||||
|
||||
/** Null when the code has no session-specific story to tell; the caller keeps its own wording. */
|
||||
export function structuredAgentSessionRefusalMessage(
|
||||
code: AgentSessionWireRefusalCode,
|
||||
record: AgentSessionRecord | null
|
||||
): string | null {
|
||||
if (!record) {
|
||||
return null
|
||||
}
|
||||
if (code === 'agent_session_ownership_unknown' || code === 'agent_session_conflict') {
|
||||
return latchedMessage(record)
|
||||
}
|
||||
if (code === 'execution_owner_reconciling') {
|
||||
return 'Orca is still working out who owns this session on this machine. Reopen the chat in a moment.'
|
||||
}
|
||||
return null
|
||||
}
|
||||
+2
-1
@@ -35,7 +35,8 @@ export function createRestartReconciler(deps: {
|
||||
} catch (error) {
|
||||
return classifyStoreFailure(
|
||||
error,
|
||||
deps.store.getRecord(sessionId)?.lease.runtimeFence ?? null
|
||||
deps.store.getRecord(sessionId)?.lease.runtimeFence ?? null,
|
||||
deps.store.getRecord(sessionId)
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
+314
@@ -0,0 +1,314 @@
|
||||
// The profiles already shipped into a dead end.
|
||||
//
|
||||
// Every record here is a shape taken from a real wedged store: a lease that no acquisition, no
|
||||
// handoff restore, and no manual recovery can move, so the chat behind it never opens again. The
|
||||
// contract is that loading the record under this build makes it usable WITHOUT losing the
|
||||
// conversation — the journal, the provider handle chain, and the recorded evidence all survive.
|
||||
|
||||
import { mkdir, mkdtemp, rm, writeFile } 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 type {
|
||||
AgentSessionClaimStatus,
|
||||
AgentSessionHandoffStage,
|
||||
AgentSessionOwnerRuntimeKind,
|
||||
AgentSessionProcessIdentity,
|
||||
AgentSessionRecord
|
||||
} from '../../../shared/agent-session-record'
|
||||
import { AgentSessionRecordStore } from '../../runtime/agent-session-record-store'
|
||||
import { AGENT_SESSION_STORE_FILE_NAME } from '../../runtime/agent-session-record-store-file'
|
||||
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_LOCATION as LOCATION,
|
||||
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' }
|
||||
const DEAD_OWNER: AgentSessionProcessIdentity = {
|
||||
hostId: 'local',
|
||||
pid: 12_546,
|
||||
processStartTimeMs: 1_786_772_085_000,
|
||||
spawnToken: 'spawn-dead'
|
||||
}
|
||||
|
||||
let root: string
|
||||
let store: AgentSessionRecordStore
|
||||
let host: StructuredAgentSessionHost
|
||||
let acquire: Mock<StructuredAgentSessionAdapter['acquire']>
|
||||
|
||||
type WedgeOverrides = {
|
||||
claimStatus: AgentSessionClaimStatus
|
||||
handoffStage: AgentSessionHandoffStage | null
|
||||
runtimeKind?: AgentSessionOwnerRuntimeKind
|
||||
ownerProcess?: AgentSessionProcessIdentity | null
|
||||
reservedSpawnToken?: string | null
|
||||
handoffOperationId?: string | null
|
||||
}
|
||||
|
||||
/** A record in the wedged shape, with real history behind it. */
|
||||
function wedgedRecord(overrides: WedgeOverrides): AgentSessionRecord {
|
||||
const fence = 13
|
||||
return {
|
||||
schemaVersion: 2,
|
||||
sessionId: SESSION,
|
||||
location: LOCATION,
|
||||
provider: 'codex',
|
||||
providerHandleChain: [
|
||||
{
|
||||
linkId: `codex-${fence}-link`,
|
||||
handle: { provider: 'codex', threadId: THREAD },
|
||||
origin: 'created',
|
||||
mintedAtFence: fence,
|
||||
observedAt: NOW - 10_000
|
||||
}
|
||||
],
|
||||
accountHome: { variable: 'CODEX_HOME', path: '/home/dev/.codex' },
|
||||
createdAt: NOW - 100_000,
|
||||
updatedAt: NOW - 10_000,
|
||||
lease: {
|
||||
sessionId: SESSION,
|
||||
runtimeKind: overrides.runtimeKind ?? 'native',
|
||||
runtimeFence: fence,
|
||||
handoffStage: overrides.handoffStage,
|
||||
provenHandleLinkId: `codex-${fence}-link`,
|
||||
ownerProcess: overrides.ownerProcess ?? null,
|
||||
reservedSpawnToken: overrides.reservedSpawnToken ?? null,
|
||||
leaseDeadlineAt: NOW - 9_000,
|
||||
lastRenewedAt: NOW - 10_000,
|
||||
handoffOperationId: overrides.handoffOperationId ?? null,
|
||||
journalCheckpoint: null,
|
||||
claimKeyId: 'key-1',
|
||||
claimStatus: overrides.claimStatus,
|
||||
unreconciled: false,
|
||||
deathEvidence: null
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async function seedStore(record: AgentSessionRecord): Promise<void> {
|
||||
const directory = join(root, 'store')
|
||||
await mkdir(directory, { recursive: true })
|
||||
await writeFile(
|
||||
join(directory, AGENT_SESSION_STORE_FILE_NAME),
|
||||
JSON.stringify({
|
||||
schemaVersion: 2,
|
||||
hostId: 'local',
|
||||
records: { [record.sessionId]: record },
|
||||
operations: {},
|
||||
retiredClaimKeys: [],
|
||||
unusableRecords: {}
|
||||
}),
|
||||
'utf-8'
|
||||
)
|
||||
store = await AgentSessionRecordStore.open({ directory, hostId: 'local' })
|
||||
}
|
||||
|
||||
/** Every recorded owner in these fixtures is long gone; that is the present-time evidence. */
|
||||
function openHost(overrides: Partial<StructuredAgentSessionHostDeps> = {}): void {
|
||||
host = new StructuredAgentSessionHost({
|
||||
store,
|
||||
adapter: {
|
||||
acquire,
|
||||
releaseAcquisition: vi.fn(async () => undefined),
|
||||
dispatch: vi.fn(),
|
||||
cancelTurn: vi.fn(),
|
||||
answerPrompt: vi.fn(),
|
||||
setOption: vi.fn()
|
||||
} as unknown as StructuredAgentSessionAdapter,
|
||||
journalRoot: root,
|
||||
claimKeyId: 'key-1',
|
||||
mintSpawnToken: () => 'spawn-new',
|
||||
now: () => NOW,
|
||||
probeOwner: async () => ({ outcome: 'pid-absent' }),
|
||||
...overrides
|
||||
})
|
||||
}
|
||||
|
||||
beforeEach(async () => {
|
||||
root = await mkdtemp(join(tmpdir(), 'orca-wedged-profile-'))
|
||||
resetHostTestOperationIds()
|
||||
acquire = vi.fn(async ({ fence }) => ({
|
||||
process: {
|
||||
hostId: 'local',
|
||||
pid: 4242,
|
||||
processStartTimeMs: 1_700_000_000_000,
|
||||
spawnToken: store.getRecord(SESSION)?.lease.reservedSpawnToken ?? 'spawn-new'
|
||||
},
|
||||
link: {
|
||||
linkId: `link-${fence}`,
|
||||
handle: { provider: 'codex', threadId: THREAD },
|
||||
origin: 'resumed' as const,
|
||||
mintedAtFence: fence,
|
||||
observedAt: NOW
|
||||
}
|
||||
}))
|
||||
})
|
||||
|
||||
afterEach(async () => {
|
||||
await host?.flushAllStreamedEvents()
|
||||
await rm(root, { recursive: true, force: true })
|
||||
})
|
||||
|
||||
describe('already-wedged profiles become usable on load', () => {
|
||||
it('re-adjudicates a conflicted manual-recovery record whose owner is provably gone', async () => {
|
||||
// The v1 -> v2 record upgrade stamps every migrated row `conflicted` + `manual-recovery`, and
|
||||
// nothing in the system ever clears it: acquisition refuses `agent_session_conflict`, recovery
|
||||
// resolution used to skip conflicted rows, and manual recovery needs a TUI owner.
|
||||
await seedStore(
|
||||
wedgedRecord({
|
||||
claimStatus: 'conflicted',
|
||||
handoffStage: 'manual-recovery',
|
||||
ownerProcess: DEAD_OWNER
|
||||
})
|
||||
)
|
||||
openHost()
|
||||
|
||||
await host.restoreReadableSessions()
|
||||
|
||||
const lease = store.getRecord(SESSION)?.lease
|
||||
expect(lease).toMatchObject({ claimStatus: 'live', handoffStage: null })
|
||||
// A real re-adjudication, not a no-op: the eviction minted a new generation.
|
||||
expect(lease?.runtimeFence).toBeGreaterThan(13)
|
||||
// The conversation survived: the codex thread was resumed, not recreated.
|
||||
expect(store.getRecord(SESSION)?.providerHandleChain[0]).toMatchObject({
|
||||
linkId: 'codex-13-link',
|
||||
handle: { threadId: THREAD }
|
||||
})
|
||||
expect(acquire).toHaveBeenCalledTimes(1)
|
||||
})
|
||||
|
||||
it('leaves a conflicted record alone while its owner cannot be proven gone', async () => {
|
||||
await seedStore(
|
||||
wedgedRecord({
|
||||
claimStatus: 'conflicted',
|
||||
handoffStage: 'manual-recovery',
|
||||
ownerProcess: DEAD_OWNER
|
||||
})
|
||||
)
|
||||
openHost({
|
||||
probeOwner: async () => ({ outcome: 'identity-matched', matchedOn: ['spawn-token'] })
|
||||
})
|
||||
|
||||
await host.restoreReadableSessions()
|
||||
|
||||
expect(store.getRecord(SESSION)?.lease).toMatchObject({
|
||||
claimStatus: 'conflicted',
|
||||
handoffStage: 'manual-recovery',
|
||||
ownerProcess: { pid: DEAD_OWNER.pid }
|
||||
})
|
||||
})
|
||||
|
||||
it('unlatches a released record that reloaded into recovery with nothing outstanding', async () => {
|
||||
// An evicted lease has no owner and no token, so a restart has nothing to probe. Treating that
|
||||
// as an unproven reservation re-latched it to `recovering` on every single boot.
|
||||
await seedStore(wedgedRecord({ claimStatus: 'released', handoffStage: 'recovering' }))
|
||||
openHost()
|
||||
|
||||
await host.restoreReadableSessions()
|
||||
|
||||
expect(store.getRecord(SESSION)?.lease).toMatchObject({
|
||||
claimStatus: 'live',
|
||||
handoffStage: null
|
||||
})
|
||||
})
|
||||
|
||||
it('exits a TUI reservation that crashed before its identity was committed', async () => {
|
||||
// The reviewer's shape: a TUI child launched, the runtime died before `commitProcessIdentity`,
|
||||
// and restart adjudication could not answer, so the lease latched at `recovering` with a null
|
||||
// owner. Handoff restore cannot help (no owner to talk to) and manual recovery requires one,
|
||||
// so recovery resolution is the ONLY exit — and it used to skip every TUI record.
|
||||
await seedStore(
|
||||
wedgedRecord({
|
||||
claimStatus: 'reserved',
|
||||
handoffStage: 'new-owner-proving',
|
||||
runtimeKind: 'tui',
|
||||
reservedSpawnToken: 'spawn-tui',
|
||||
handoffOperationId: 'handoff-op-1'
|
||||
})
|
||||
)
|
||||
// Restart adjudication runs while the host still cannot enumerate the token; the later
|
||||
// recovery pass gets a real answer.
|
||||
let probes = 0
|
||||
openHost({
|
||||
probeOwner: async () => {
|
||||
probes += 1
|
||||
return probes === 1
|
||||
? { outcome: 'indeterminate', reason: 'host could not enumerate spawn tokens' }
|
||||
: { outcome: 'reservation-unused' }
|
||||
}
|
||||
})
|
||||
|
||||
await host.restoreReadableSessions()
|
||||
|
||||
expect(store.getRecord(SESSION)?.lease).toMatchObject({
|
||||
claimStatus: 'live',
|
||||
handoffStage: null
|
||||
})
|
||||
})
|
||||
|
||||
it("stops a child carrying a lost record's spawn token before recovery respawns", async () => {
|
||||
// `listOrphanSpawnTokens` had no production caller, so a child spawned under a reservation
|
||||
// that went down with the primary store file was never stopped — and recovery then granted a
|
||||
// second owner on the same Codex thread.
|
||||
await seedStore(
|
||||
wedgedRecord({
|
||||
claimStatus: 'conflicted',
|
||||
handoffStage: 'manual-recovery',
|
||||
ownerProcess: DEAD_OWNER
|
||||
})
|
||||
)
|
||||
const order: string[] = []
|
||||
acquire.mockImplementation(async ({ fence }) => {
|
||||
order.push('acquire')
|
||||
return {
|
||||
process: { hostId: 'local', pid: 4242, processStartTimeMs: 1, spawnToken: 'spawn-new' },
|
||||
link: {
|
||||
linkId: `link-${fence}`,
|
||||
handle: { provider: 'codex' as const, threadId: THREAD },
|
||||
origin: 'resumed' as const,
|
||||
mintedAtFence: fence,
|
||||
observedAt: NOW
|
||||
}
|
||||
}
|
||||
})
|
||||
openHost({
|
||||
scanSpawnTokenProcesses: async () =>
|
||||
new Map([
|
||||
['spawn-lost', [31_337]],
|
||||
// The surviving record still claims this one, so it must be left alone.
|
||||
[DEAD_OWNER.spawnToken, [12_546]]
|
||||
]),
|
||||
stopOwnerProcess: (pid) => order.push(`stop:${pid}`)
|
||||
})
|
||||
|
||||
await host.restoreReadableSessions()
|
||||
|
||||
expect(order).toEqual(['stop:31337', 'acquire'])
|
||||
})
|
||||
|
||||
it('names the missing evidence when a latched record still cannot be freed', async () => {
|
||||
await seedStore(
|
||||
wedgedRecord({
|
||||
claimStatus: 'conflicted',
|
||||
handoffStage: 'manual-recovery',
|
||||
ownerProcess: DEAD_OWNER
|
||||
})
|
||||
)
|
||||
openHost({ probeOwner: async () => ({ outcome: 'indeterminate', reason: 'no answer' }) })
|
||||
await host.restoreReadableSessions()
|
||||
|
||||
const refused = await host.attach(CALLER, hostTestAttachParams(13))
|
||||
|
||||
expect(refused.ok).toBe(false)
|
||||
const message = refused.ok ? '' : refused.refusal.message
|
||||
expect(message).toContain('process 12546 on local')
|
||||
expect(message).not.toContain('The session store refused this call')
|
||||
})
|
||||
})
|
||||
@@ -4,7 +4,7 @@ import {
|
||||
agentSessionRecordFixture
|
||||
} from '../../shared/agent-session-record.test-fixture'
|
||||
import { recoverDeadTuiOwnerForHandoff } from './agent-session-handoff-lease-transitions'
|
||||
import { applyAgentSessionRestartAdjudication } from './agent-session-lease-transitions'
|
||||
import { applyAgentSessionRestartAdjudication } from './agent-session-restart-lease-transitions'
|
||||
|
||||
describe('agent session handoff restart transitions', () => {
|
||||
it('turns a proven dead TUI owner into one durable retry owner', () => {
|
||||
|
||||
@@ -23,7 +23,6 @@ import type {
|
||||
AgentSessionProcessIdentity,
|
||||
AgentSessionRecord
|
||||
} from '../../shared/agent-session-record'
|
||||
import { adjudicateRestartedAgentSessionHandoff } from './agent-session-restart-handoff-adjudication'
|
||||
|
||||
export type AgentSessionReservation = {
|
||||
runtimeKind: AgentSessionOwnerRuntimeKind
|
||||
@@ -214,6 +213,17 @@ export function evictAgentSessionOwner(args: {
|
||||
probe: args.probe,
|
||||
observedAt: args.now
|
||||
})
|
||||
if (adjudication.disposition === 'free') {
|
||||
// Nothing outstanding to evict; clearing the latched stage IS the resolution, and no new
|
||||
// generation was granted, so the fence and the recorded evidence both stay put.
|
||||
return withLease(record, {
|
||||
...record.lease,
|
||||
handoffStage: null,
|
||||
handoffOperationId: null,
|
||||
processlessAt: null,
|
||||
lastRenewedAt: args.now
|
||||
})
|
||||
}
|
||||
if (adjudication.disposition !== 'evicted') {
|
||||
throw new Error('agent_session_ownership_unknown')
|
||||
}
|
||||
@@ -231,66 +241,6 @@ export function evictAgentSessionOwner(args: {
|
||||
})
|
||||
}
|
||||
|
||||
/** Apply one restart adjudication. Never consults deadlines — only proof moves a lease. */
|
||||
export function applyAgentSessionRestartAdjudication(args: {
|
||||
record: AgentSessionRecord
|
||||
probe: AgentSessionOwnerProbe
|
||||
now: number
|
||||
}): AgentSessionRecord {
|
||||
const { record } = args
|
||||
if (
|
||||
record.lease.handoffStage === 'old-owner-stopped' &&
|
||||
record.lease.claimStatus === 'released' &&
|
||||
record.lease.ownerProcess === null
|
||||
) {
|
||||
return withLease(record, {
|
||||
...record.lease,
|
||||
unreconciled: false,
|
||||
lastRenewedAt: args.now
|
||||
})
|
||||
}
|
||||
if (
|
||||
record.lease.handoffStage === 'preparing' ||
|
||||
record.lease.handoffStage === 'new-owner-proving'
|
||||
) {
|
||||
return adjudicateRestartedAgentSessionHandoff(record, args.probe, args.now)
|
||||
}
|
||||
const adjudication = adjudicateAgentSessionRestart({
|
||||
lease: record.lease,
|
||||
probe: args.probe,
|
||||
observedAt: args.now
|
||||
})
|
||||
if (adjudication.disposition === 'readopt') {
|
||||
// Why: re-adoption is not a new generation, so the fence does not move.
|
||||
return withLease(record, { ...record.lease, unreconciled: false, lastRenewedAt: args.now })
|
||||
}
|
||||
if (adjudication.disposition === 'evicted') {
|
||||
return withLease(record, {
|
||||
...record.lease,
|
||||
runtimeFence: adjudication.nextFence,
|
||||
handoffStage: null,
|
||||
ownerProcess: null,
|
||||
reservedSpawnToken: null,
|
||||
processlessAt: null,
|
||||
claimStatus: 'released',
|
||||
unreconciled: false,
|
||||
lastRenewedAt: args.now,
|
||||
handoffOperationId: null,
|
||||
deathEvidence: adjudication.evidence
|
||||
})
|
||||
}
|
||||
const stage: AgentSessionHandoffStage =
|
||||
adjudication.disposition === 'conflicted' ? 'manual-recovery' : adjudication.stage
|
||||
return withLease(record, {
|
||||
...record.lease,
|
||||
handoffStage: stage,
|
||||
claimStatus:
|
||||
adjudication.disposition === 'conflicted' ? 'conflicted' : record.lease.claimStatus,
|
||||
unreconciled: false,
|
||||
lastRenewedAt: args.now
|
||||
})
|
||||
}
|
||||
|
||||
export function setAgentSessionHandoffStage(args: {
|
||||
record: AgentSessionRecord
|
||||
fence: number
|
||||
|
||||
@@ -0,0 +1,56 @@
|
||||
// A child spawned under a reservation whose record was lost is invisible to the lease, and
|
||||
// recovery is about to grant the same provider session a new owner. Nothing else in the host
|
||||
// stops it, so this is the only thing standing between the user and two live Codex children on
|
||||
// one thread.
|
||||
|
||||
import { describe, expect, it, vi } from 'vitest'
|
||||
import type { AgentSessionRecordStore } from './agent-session-record-store'
|
||||
import { stopOrphanAgentSessionChildren } from './agent-session-orphan-child-reaper'
|
||||
|
||||
function storeWithLeasedTokens(tokens: readonly string[]) {
|
||||
return {
|
||||
listOrphanSpawnTokens: (observed: readonly string[]) =>
|
||||
observed.filter((token) => !tokens.includes(token))
|
||||
} as Pick<AgentSessionRecordStore, 'listOrphanSpawnTokens'>
|
||||
}
|
||||
|
||||
describe('orphan agent-session child reaper', () => {
|
||||
it('stops every process whose spawn token no lease claims', async () => {
|
||||
const stop = vi.fn()
|
||||
|
||||
const stopped = await stopOrphanAgentSessionChildren({
|
||||
store: storeWithLeasedTokens(['token-owned']),
|
||||
scan: async () =>
|
||||
new Map([
|
||||
['token-owned', [101]],
|
||||
['token-lost', [202, 203]]
|
||||
]),
|
||||
stop
|
||||
})
|
||||
|
||||
expect(stopped).toEqual([202, 203])
|
||||
expect(stop).toHaveBeenCalledWith(202, 'SIGTERM')
|
||||
expect(stop).toHaveBeenCalledWith(203, 'SIGTERM')
|
||||
expect(stop).not.toHaveBeenCalledWith(101, 'SIGTERM')
|
||||
})
|
||||
|
||||
it('stops nothing on a host that cannot enumerate spawn tokens', async () => {
|
||||
const stop = vi.fn()
|
||||
|
||||
// Null is "cannot answer", never "no tokens" — treating it as an empty scan would be a
|
||||
// license to signal nothing, but a future empty-map reading would be a license to signal
|
||||
// whatever the caller guessed.
|
||||
const stopped = await stopOrphanAgentSessionChildren({
|
||||
store: {
|
||||
listOrphanSpawnTokens: () => {
|
||||
throw new Error('the reaper must not ask when the host could not answer')
|
||||
}
|
||||
},
|
||||
scan: async () => null,
|
||||
stop
|
||||
})
|
||||
|
||||
expect(stopped).toEqual([])
|
||||
expect(stop).not.toHaveBeenCalled()
|
||||
})
|
||||
})
|
||||
@@ -0,0 +1,46 @@
|
||||
/**
|
||||
* Stops provider children that carry an Orca spawn token no lease claims.
|
||||
*
|
||||
* A child spawned under a reservation whose record was lost — the primary store file went with it,
|
||||
* or the crash beat the durable write — is unreachable but still connected to the provider session.
|
||||
* Recovery will later grant a new owner for that same session, so without this the host ends up
|
||||
* with two live children on one provider thread. Only a token match justifies signalling a process;
|
||||
* neither age nor CPU is evidence, and a host that cannot enumerate tokens stops nothing.
|
||||
*/
|
||||
|
||||
import type { AgentSessionRecordStore } from './agent-session-record-store'
|
||||
import {
|
||||
scanAgentSessionSpawnTokenProcesses,
|
||||
type AgentSessionSpawnTokenScan
|
||||
} from './agent-session-spawn-token-process-scan'
|
||||
|
||||
export type AgentSessionOrphanStopSignal = 'SIGTERM' | 'SIGKILL'
|
||||
|
||||
function defaultStop(pid: number, signal: AgentSessionOrphanStopSignal): void {
|
||||
try {
|
||||
process.kill(pid, signal)
|
||||
} catch {
|
||||
// Already gone, or not ours to signal. Either way there is nothing further to do.
|
||||
}
|
||||
}
|
||||
|
||||
/** Returns the pids signalled, so the caller can report what it reaped. */
|
||||
export async function stopOrphanAgentSessionChildren(input: {
|
||||
store: Pick<AgentSessionRecordStore, 'listOrphanSpawnTokens'>
|
||||
scan?: () => Promise<AgentSessionSpawnTokenScan | null>
|
||||
stop?: (pid: number, signal: AgentSessionOrphanStopSignal) => void
|
||||
}): Promise<number[]> {
|
||||
const observed = await (input.scan ?? scanAgentSessionSpawnTokenProcesses)()
|
||||
if (observed === null || observed.size === 0) {
|
||||
return []
|
||||
}
|
||||
const stop = input.stop ?? defaultStop
|
||||
const stopped: number[] = []
|
||||
for (const token of input.store.listOrphanSpawnTokens([...observed.keys()])) {
|
||||
for (const pid of observed.get(token) ?? []) {
|
||||
stop(pid, 'SIGTERM')
|
||||
stopped.push(pid)
|
||||
}
|
||||
}
|
||||
return stopped
|
||||
}
|
||||
@@ -600,13 +600,16 @@ describe('restart reconciliation', () => {
|
||||
).rejects.toThrow('agent_session_ownership_unknown')
|
||||
})
|
||||
|
||||
it('keeps a conflict conflicted across a restart', async () => {
|
||||
it('keeps a conflict conflicted across a restart that proves nothing', async () => {
|
||||
const first = await open()
|
||||
await establishOwner(first)
|
||||
await first.markClaimConflicted('session-alpha', NOW)
|
||||
|
||||
const reopened = await open()
|
||||
await reopened.reconcileOnRestart({ probe: async () => ({ outcome: 'pid-absent' }), now: NOW })
|
||||
await reopened.reconcileOnRestart({
|
||||
probe: async () => ({ outcome: 'indeterminate', reason: 'no answer' }),
|
||||
now: NOW
|
||||
})
|
||||
expect(reopened.getRecord('session-alpha')?.lease).toMatchObject({
|
||||
claimStatus: 'conflicted',
|
||||
handoffStage: 'manual-recovery'
|
||||
@@ -622,6 +625,32 @@ describe('restart reconciliation', () => {
|
||||
).rejects.toThrow('agent_session_conflict')
|
||||
})
|
||||
|
||||
it('releases a conflict whose named owner is proven gone at restart', async () => {
|
||||
// A conflict with no exit is a session the user can never open again; present-time proof that
|
||||
// the process the conflict names has exited leaves no claimant left to protect.
|
||||
const first = await open()
|
||||
await establishOwner(first)
|
||||
await first.markClaimConflicted('session-alpha', NOW)
|
||||
|
||||
const reopened = await open()
|
||||
await reopened.reconcileOnRestart({ probe: async () => ({ outcome: 'pid-absent' }), now: NOW })
|
||||
|
||||
const lease = reopened.getRecord('session-alpha')?.lease
|
||||
expect(lease).toMatchObject({
|
||||
claimStatus: 'released',
|
||||
handoffStage: null,
|
||||
deathEvidence: { kind: 'pid-absent' }
|
||||
})
|
||||
const reacquired = await reopened.reserveOwner(
|
||||
reserveRequest({
|
||||
expectedFence: lease?.runtimeFence ?? null,
|
||||
probe: { outcome: 'pid-absent' },
|
||||
operation: { callerKey: 'client-1', operationId: operationId(), fingerprint: 'fp-2' }
|
||||
})
|
||||
)
|
||||
expect(reacquired.disposition).toBe('reserved')
|
||||
})
|
||||
|
||||
it('frees a reservation that provably never spawned', async () => {
|
||||
const first = await open()
|
||||
await first.reserveOwner(reserveRequest())
|
||||
|
||||
@@ -23,7 +23,6 @@ import {
|
||||
type AgentSessionRecord
|
||||
} from '../../shared/agent-session-record'
|
||||
import {
|
||||
applyAgentSessionRestartAdjudication,
|
||||
commitAgentSessionProcessIdentity,
|
||||
evictAgentSessionOwner,
|
||||
proveAgentSessionOwner,
|
||||
@@ -31,6 +30,7 @@ import {
|
||||
setAgentSessionJournalCheckpoint,
|
||||
type AgentSessionProcessIdentityCommit
|
||||
} from './agent-session-lease-transitions'
|
||||
import { applyAgentSessionRestartAdjudication } from './agent-session-restart-lease-transitions'
|
||||
import { agentSessionReconciliationTargetMatches } from './agent-session-reconciliation-target'
|
||||
import { replaceAgentSessionRecordOptions } from './agent-session-record-options'
|
||||
import {
|
||||
|
||||
@@ -17,6 +17,16 @@ export function adjudicateRestartedAgentSessionHandoff(
|
||||
if (adjudication.disposition === 'readopt') {
|
||||
return updateLease(record, { ...record.lease, unreconciled: false, lastRenewedAt: now })
|
||||
}
|
||||
if (adjudication.disposition === 'free') {
|
||||
return updateLease(record, {
|
||||
...record.lease,
|
||||
handoffStage: null,
|
||||
handoffOperationId: null,
|
||||
processlessAt: null,
|
||||
unreconciled: false,
|
||||
lastRenewedAt: now
|
||||
})
|
||||
}
|
||||
if (adjudication.disposition !== 'evicted') {
|
||||
return updateLease(record, {
|
||||
...record.lease,
|
||||
|
||||
@@ -0,0 +1,90 @@
|
||||
/**
|
||||
* Turning a restart adjudication into the next record.
|
||||
*
|
||||
* Applies one restart verdict to one loaded lease. Kept apart from the acquisition transitions
|
||||
* because it is the only one that moves a lease WITHOUT a new owner proving anything — which is
|
||||
* exactly the polarity that has to be read carefully.
|
||||
*/
|
||||
|
||||
import {
|
||||
adjudicateAgentSessionRestart,
|
||||
type AgentSessionOwnerProbe
|
||||
} from '../../shared/agent-session-lease-adjudication'
|
||||
import type {
|
||||
AgentSessionHandoffStage,
|
||||
AgentSessionRecord
|
||||
} from '../../shared/agent-session-record'
|
||||
import { adjudicateRestartedAgentSessionHandoff } from './agent-session-restart-handoff-adjudication'
|
||||
import { withLease } from './agent-session-lease-transitions'
|
||||
|
||||
/** Apply one restart adjudication. Never consults deadlines — only proof moves a lease. */
|
||||
export function applyAgentSessionRestartAdjudication(args: {
|
||||
record: AgentSessionRecord
|
||||
probe: AgentSessionOwnerProbe
|
||||
now: number
|
||||
}): AgentSessionRecord {
|
||||
const { record } = args
|
||||
if (
|
||||
record.lease.handoffStage === 'old-owner-stopped' &&
|
||||
record.lease.claimStatus === 'released' &&
|
||||
record.lease.ownerProcess === null
|
||||
) {
|
||||
return withLease(record, {
|
||||
...record.lease,
|
||||
unreconciled: false,
|
||||
lastRenewedAt: args.now
|
||||
})
|
||||
}
|
||||
if (
|
||||
record.lease.handoffStage === 'preparing' ||
|
||||
record.lease.handoffStage === 'new-owner-proving'
|
||||
) {
|
||||
return adjudicateRestartedAgentSessionHandoff(record, args.probe, args.now)
|
||||
}
|
||||
const adjudication = adjudicateAgentSessionRestart({
|
||||
lease: record.lease,
|
||||
probe: args.probe,
|
||||
observedAt: args.now
|
||||
})
|
||||
if (adjudication.disposition === 'readopt') {
|
||||
// Why: re-adoption is not a new generation, so the fence does not move.
|
||||
return withLease(record, { ...record.lease, unreconciled: false, lastRenewedAt: args.now })
|
||||
}
|
||||
if (adjudication.disposition === 'free') {
|
||||
// Why: an already-free lease that reloads into `recovering` is unopenable forever; clearing
|
||||
// the stage restores it without moving the fence or touching the recorded death evidence.
|
||||
return withLease(record, {
|
||||
...record.lease,
|
||||
handoffStage: null,
|
||||
handoffOperationId: null,
|
||||
processlessAt: null,
|
||||
unreconciled: false,
|
||||
lastRenewedAt: args.now
|
||||
})
|
||||
}
|
||||
if (adjudication.disposition === 'evicted') {
|
||||
return withLease(record, {
|
||||
...record.lease,
|
||||
runtimeFence: adjudication.nextFence,
|
||||
handoffStage: null,
|
||||
ownerProcess: null,
|
||||
reservedSpawnToken: null,
|
||||
processlessAt: null,
|
||||
claimStatus: 'released',
|
||||
unreconciled: false,
|
||||
lastRenewedAt: args.now,
|
||||
handoffOperationId: null,
|
||||
deathEvidence: adjudication.evidence
|
||||
})
|
||||
}
|
||||
const stage: AgentSessionHandoffStage =
|
||||
adjudication.disposition === 'conflicted' ? 'manual-recovery' : adjudication.stage
|
||||
return withLease(record, {
|
||||
...record.lease,
|
||||
handoffStage: stage,
|
||||
claimStatus:
|
||||
adjudication.disposition === 'conflicted' ? 'conflicted' : record.lease.claimStatus,
|
||||
unreconciled: false,
|
||||
lastRenewedAt: args.now
|
||||
})
|
||||
}
|
||||
@@ -0,0 +1,25 @@
|
||||
import { describe, expect, it } from 'vitest'
|
||||
import {
|
||||
findAgentSessionSpawnTokenProcesses,
|
||||
scanAgentSessionSpawnTokenProcesses
|
||||
} from './agent-session-spawn-token-process-scan'
|
||||
|
||||
describe('agent-session spawn token process scan', () => {
|
||||
it('answers null on platforms that cannot read another process environment', async () => {
|
||||
// macOS and Windows have no `/proc/<pid>/environ`. Answering "none" there would free a
|
||||
// reservation whose child is alive and mint a second writer on the same provider session.
|
||||
expect(await scanAgentSessionSpawnTokenProcesses('darwin')).toBeNull()
|
||||
expect(await scanAgentSessionSpawnTokenProcesses('win32')).toBeNull()
|
||||
})
|
||||
|
||||
it('propagates the host non-answer rather than reporting an empty pid list', async () => {
|
||||
expect(await findAgentSessionSpawnTokenProcesses('token-1', async () => null)).toBeNull()
|
||||
})
|
||||
|
||||
it('reports the pids carrying one token', async () => {
|
||||
const scan = async () => new Map([['token-1', [11, 12]]])
|
||||
|
||||
expect(await findAgentSessionSpawnTokenProcesses('token-1', scan)).toEqual([11, 12])
|
||||
expect(await findAgentSessionSpawnTokenProcesses('token-2', scan)).toEqual([])
|
||||
})
|
||||
})
|
||||
@@ -0,0 +1,59 @@
|
||||
/**
|
||||
* Host scan for processes carrying an Orca agent-session spawn token.
|
||||
*
|
||||
* The token is the only PID-reuse-safe identity element a child is guaranteed to carry, and it
|
||||
* lives in the child's environment — which only Linux lets another process read (`/proc/<pid>/environ`).
|
||||
* macOS and Windows answer `null`, meaning "this host cannot enumerate", NEVER "no process carries
|
||||
* it": reporting an empty result there would free a reservation whose child is alive and hand a
|
||||
* second writer to the same provider session.
|
||||
*/
|
||||
|
||||
import { readFile, readdir } from 'node:fs/promises'
|
||||
import { CODEX_SPAWN_TOKEN_ENV } from '../codex/codex-structured-owner-identity'
|
||||
import { spawnTokenFromEnvironBlock } from './agent-session-spawn-token-readback'
|
||||
|
||||
export type AgentSessionSpawnTokenScan = ReadonlyMap<string, readonly number[]>
|
||||
|
||||
/** Tokens observed on this host, or null when the platform cannot answer at all. */
|
||||
export async function scanAgentSessionSpawnTokenProcesses(
|
||||
platform: NodeJS.Platform = process.platform,
|
||||
variable: string = CODEX_SPAWN_TOKEN_ENV
|
||||
): Promise<AgentSessionSpawnTokenScan | null> {
|
||||
if (platform !== 'linux') {
|
||||
return null
|
||||
}
|
||||
let entries: string[]
|
||||
try {
|
||||
entries = await readdir('/proc')
|
||||
} catch {
|
||||
return null
|
||||
}
|
||||
const observed = new Map<string, number[]>()
|
||||
for (const entry of entries) {
|
||||
const pid = Number(entry)
|
||||
if (!Number.isSafeInteger(pid) || pid <= 0) {
|
||||
continue
|
||||
}
|
||||
let token: string | null
|
||||
try {
|
||||
token = spawnTokenFromEnvironBlock(await readFile(`/proc/${pid}/environ`, 'utf-8'), variable)
|
||||
} catch {
|
||||
// A process that exited mid-scan, or one this user may not read, is not evidence either way.
|
||||
continue
|
||||
}
|
||||
if (token === null) {
|
||||
continue
|
||||
}
|
||||
observed.set(token, [...(observed.get(token) ?? []), pid])
|
||||
}
|
||||
return observed
|
||||
}
|
||||
|
||||
/** Pids carrying one specific token, or null when the host could not enumerate. */
|
||||
export async function findAgentSessionSpawnTokenProcesses(
|
||||
spawnToken: string,
|
||||
scan: () => Promise<AgentSessionSpawnTokenScan | null> = scanAgentSessionSpawnTokenProcesses
|
||||
): Promise<number[] | null> {
|
||||
const observed = await scan()
|
||||
return observed === null ? null : [...(observed.get(spawnToken) ?? [])]
|
||||
}
|
||||
@@ -1,5 +1,6 @@
|
||||
import { describe, expect, it, vi } from 'vitest'
|
||||
import type {
|
||||
AgentSessionClaimStatus,
|
||||
AgentSessionProcessIdentity,
|
||||
AgentSessionRecord
|
||||
} from '../../shared/agent-session-record'
|
||||
@@ -9,9 +10,24 @@ const HOST_ID = 'local'
|
||||
|
||||
function record(
|
||||
ownerProcess: AgentSessionProcessIdentity | null,
|
||||
processlessAt?: number | null
|
||||
lease: {
|
||||
processlessAt?: number | null
|
||||
reservedSpawnToken?: string | null
|
||||
claimStatus?: AgentSessionClaimStatus
|
||||
runtimeFence?: number
|
||||
} = {}
|
||||
): AgentSessionRecord {
|
||||
return { sessionId: 'session-1', lease: { ownerProcess, processlessAt } } as AgentSessionRecord
|
||||
return {
|
||||
sessionId: 'session-1',
|
||||
providerHandleChain: [],
|
||||
lease: {
|
||||
ownerProcess,
|
||||
reservedSpawnToken: null,
|
||||
claimStatus: 'released',
|
||||
runtimeFence: 3,
|
||||
...lease
|
||||
}
|
||||
} as unknown as AgentSessionRecord
|
||||
}
|
||||
|
||||
const OWNER: AgentSessionProcessIdentity = {
|
||||
@@ -21,9 +37,11 @@ const OWNER: AgentSessionProcessIdentity = {
|
||||
spawnToken: 'token-1'
|
||||
}
|
||||
|
||||
const deadProbe = () => vi.fn(async () => ({ outcome: 'pid-absent' }) as const)
|
||||
|
||||
describe('structured agent-session owner probe', () => {
|
||||
it('probes an owner this host spawned', async () => {
|
||||
const probe = vi.fn(async () => ({ outcome: 'pid-absent' }) as const)
|
||||
const probe = deadProbe()
|
||||
const result = await createStructuredAgentSessionOwnerProbe(HOST_ID, probe)(record(OWNER))
|
||||
|
||||
expect(probe).toHaveBeenCalledWith({
|
||||
@@ -34,7 +52,7 @@ describe('structured agent-session owner probe', () => {
|
||||
})
|
||||
|
||||
it('refuses to probe an owner on another host, whose pid means nothing here', async () => {
|
||||
const probe = vi.fn(async () => ({ outcome: 'pid-absent' }) as const)
|
||||
const probe = deadProbe()
|
||||
const result = await createStructuredAgentSessionOwnerProbe(
|
||||
HOST_ID,
|
||||
probe
|
||||
@@ -44,22 +62,64 @@ describe('structured agent-session owner probe', () => {
|
||||
expect(result.outcome).toBe('indeterminate')
|
||||
})
|
||||
|
||||
it('leaves a reservation that never named a process for manual recovery', async () => {
|
||||
const probe = vi.fn(async () => ({ outcome: 'pid-absent' }) as const)
|
||||
const result = await createStructuredAgentSessionOwnerProbe(HOST_ID, probe)(record(null))
|
||||
it('leaves a reservation whose spawn token is still live on this host latched', async () => {
|
||||
const probe = deadProbe()
|
||||
const result = await createStructuredAgentSessionOwnerProbe(HOST_ID, probe, async () => [9001])(
|
||||
record(null, { claimStatus: 'reserved', reservedSpawnToken: 'token-1' })
|
||||
)
|
||||
|
||||
// Evicting here would put a second writer on a live Codex thread.
|
||||
expect(result.outcome).toBe('indeterminate')
|
||||
})
|
||||
|
||||
it('leaves a reservation latched on a host that cannot enumerate spawn tokens', async () => {
|
||||
const result = await createStructuredAgentSessionOwnerProbe(
|
||||
HOST_ID,
|
||||
deadProbe(),
|
||||
async () => null
|
||||
)(record(null, { claimStatus: 'reserved', reservedSpawnToken: 'token-1' }))
|
||||
|
||||
expect(result.outcome).toBe('indeterminate')
|
||||
})
|
||||
|
||||
it('frees a reservation once the host proves no process carries its token', async () => {
|
||||
const result = await createStructuredAgentSessionOwnerProbe(
|
||||
HOST_ID,
|
||||
deadProbe(),
|
||||
async () => []
|
||||
)(record(null, { claimStatus: 'reserved', reservedSpawnToken: 'token-1' }))
|
||||
|
||||
expect(result).toEqual({ outcome: 'reservation-unused' })
|
||||
})
|
||||
|
||||
it('frees a lease that names neither an owner nor a spawn token', async () => {
|
||||
const probe = deadProbe()
|
||||
const scan = vi.fn(async () => [] as number[])
|
||||
// Nothing was ever minted that a child could be carrying, so no scan is even needed;
|
||||
// answering `indeterminate` here is what latches every released record into recovery.
|
||||
const result = await createStructuredAgentSessionOwnerProbe(HOST_ID, probe, scan)(record(null))
|
||||
|
||||
expect(probe).not.toHaveBeenCalled()
|
||||
// Evicting here would need proof nothing spawned under the token; guessing
|
||||
// would put a second writer on a live Codex thread.
|
||||
expect(scan).not.toHaveBeenCalled()
|
||||
expect(result).toEqual({ outcome: 'reservation-unused' })
|
||||
})
|
||||
|
||||
it('still refuses a reservation that recorded no token to scan for', async () => {
|
||||
const result = await createStructuredAgentSessionOwnerProbe(
|
||||
HOST_ID,
|
||||
deadProbe(),
|
||||
async () => []
|
||||
)(record(null, { claimStatus: 'reserved' }))
|
||||
|
||||
expect(result.outcome).toBe('indeterminate')
|
||||
})
|
||||
|
||||
it('releases only a reservation carrying durable pre-spawn proof', async () => {
|
||||
const probe = vi.fn(async () => ({ outcome: 'pid-absent' }) as const)
|
||||
const probe = deadProbe()
|
||||
const result = await createStructuredAgentSessionOwnerProbe(
|
||||
HOST_ID,
|
||||
probe
|
||||
)(record(null, 1_800_000_000_000))
|
||||
)(record(null, { processlessAt: 1_800_000_000_000, claimStatus: 'reserved' }))
|
||||
|
||||
expect(probe).not.toHaveBeenCalled()
|
||||
expect(result).toEqual({ outcome: 'reservation-unused' })
|
||||
|
||||
@@ -19,7 +19,11 @@ import { StructuredAgentSessionHost } from '../native-chat/agent-session-wire/st
|
||||
import type { StructuredAgentSessionHandoffTransport } from '../native-chat/agent-session-wire/structured-agent-session-handoff-types'
|
||||
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 {
|
||||
probeAgentSessionProcessIdentity,
|
||||
probeAgentSessionReservation
|
||||
} from './agent-session-process-identity-probe'
|
||||
import { findAgentSessionSpawnTokenProcesses } from './agent-session-spawn-token-process-scan'
|
||||
import { readEchoedAgentSessionSpawnToken } from './agent-session-spawn-token-readback'
|
||||
import { agentSessionPtyWriteGate } from './agent-session-pty-write-gate'
|
||||
import { resolveLoginShellEnvironment } from '../startup/login-shell-environment'
|
||||
@@ -130,7 +134,8 @@ async function install(deps: StructuredAgentSessionRuntimeDeps): Promise<Install
|
||||
*/
|
||||
export function createStructuredAgentSessionOwnerProbe(
|
||||
hostId: string,
|
||||
probe = probeAgentSessionProcessIdentity
|
||||
probe = probeAgentSessionProcessIdentity,
|
||||
findSpawnTokenProcesses = findAgentSessionSpawnTokenProcesses
|
||||
): (record: AgentSessionRecord) => Promise<AgentSessionOwnerProbe> {
|
||||
return async (record) => {
|
||||
const owner = record.lease.ownerProcess
|
||||
@@ -138,12 +143,27 @@ export function createStructuredAgentSessionOwnerProbe(
|
||||
if (record.lease.processlessAt !== undefined && record.lease.processlessAt !== null) {
|
||||
return { outcome: 'reservation-unused' }
|
||||
}
|
||||
// Freeing a reservation needs positive proof that nothing spawned under
|
||||
// its token, and this host has no spawn-token process scan yet.
|
||||
return {
|
||||
outcome: 'indeterminate',
|
||||
reason: 'reservation named no process, and this host cannot scan for its spawn token'
|
||||
const spawnToken = record.lease.reservedSpawnToken
|
||||
if (spawnToken === null) {
|
||||
if (record.lease.claimStatus === 'reserved') {
|
||||
return {
|
||||
outcome: 'indeterminate',
|
||||
reason: 'reservation recorded no spawn token to scan for'
|
||||
}
|
||||
}
|
||||
// The token is minted before the child and is the only thing a child could be carrying.
|
||||
// No owner and no token means nothing on any host can be holding this lease — answering
|
||||
// `indeterminate` here is what latches an already-free record into recovery forever.
|
||||
return { outcome: 'reservation-unused' }
|
||||
}
|
||||
// Freeing a reservation needs positive proof that nothing spawned under its token. The scan
|
||||
// answers null where the platform cannot read another process's environment.
|
||||
return probeAgentSessionReservation({
|
||||
spawnToken,
|
||||
findProcessesWithSpawnToken: (token) => findSpawnTokenProcesses(token),
|
||||
hasProviderActivitySinceReservation: async () =>
|
||||
agentSessionReservationTouchedProvider(record)
|
||||
})
|
||||
}
|
||||
if (owner.hostId !== hostId) {
|
||||
// Checking a remote host's pid against this machine's process table is
|
||||
@@ -161,3 +181,14 @@ export function createStructuredAgentSessionOwnerProbe(
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* The only provider-side trace a reservation can leave in its own record: a handle link minted at
|
||||
* this fence. `proveAgentSessionOwner` refuses to append one before an identity is committed, so a
|
||||
* link at the reservation's fence means a child got far enough to resume the provider thread. It
|
||||
* cannot see activity the child produced without proving a handle, which is why it is paired with
|
||||
* the token scan rather than trusted alone.
|
||||
*/
|
||||
function agentSessionReservationTouchedProvider(record: AgentSessionRecord): boolean {
|
||||
return record.providerHandleChain.at(-1)?.mintedAtFence === record.lease.runtimeFence
|
||||
}
|
||||
|
||||
@@ -228,14 +228,57 @@ describe('restart reconciliation', () => {
|
||||
).toEqual({ disposition: 'recovering', stage: 'recovering', reason: 'no answer' })
|
||||
})
|
||||
|
||||
it('keeps a pre-restart conflict conflicted', () => {
|
||||
it('keeps a pre-restart conflict conflicted while its owner cannot be proven gone', () => {
|
||||
expect(
|
||||
adjudicateAgentSessionRestart({
|
||||
lease: lease({ claimStatus: 'conflicted' }),
|
||||
probe: INDETERMINATE,
|
||||
observedAt: 9_000
|
||||
})
|
||||
).toEqual({ disposition: 'conflicted', reason: 'claim conflicted before restart' })
|
||||
})
|
||||
|
||||
it('keeps a conflict conflicted when it names no process to prove anything about', () => {
|
||||
expect(
|
||||
adjudicateAgentSessionRestart({
|
||||
lease: lease({ claimStatus: 'conflicted', ownerProcess: null }),
|
||||
probe: { outcome: 'pid-absent' },
|
||||
observedAt: 9_000
|
||||
})
|
||||
).toEqual({ disposition: 'conflicted', reason: 'claim conflicted before restart' })
|
||||
})
|
||||
|
||||
it('frees a conflict whose named owner is proven gone', () => {
|
||||
// Why: the conflict protects one specific process. Once that process is proven gone there is
|
||||
// no claimant left, and a conflict with no exit is a session nobody can ever open again.
|
||||
expect(
|
||||
adjudicateAgentSessionRestart({
|
||||
lease: lease({ claimStatus: 'conflicted' }),
|
||||
probe: { outcome: 'pid-absent' },
|
||||
observedAt: 9_000
|
||||
})
|
||||
).toEqual({ disposition: 'conflicted', reason: 'claim conflicted before restart' })
|
||||
).toEqual({
|
||||
disposition: 'evicted',
|
||||
nextFence: 8,
|
||||
evidence: { kind: 'pid-absent', detail: 'recorded pid absent on host', observedAt: 9_000 }
|
||||
})
|
||||
})
|
||||
|
||||
it('frees a lease that names neither an owner nor a reservation, without moving the fence', () => {
|
||||
// Why: an evicted lease has no owner and no token, so a restart has nothing to probe.
|
||||
// Calling that an unproven reservation re-latched every released record on every boot.
|
||||
expect(
|
||||
adjudicateAgentSessionRestart({
|
||||
lease: lease({
|
||||
ownerProcess: null,
|
||||
reservedSpawnToken: null,
|
||||
claimStatus: 'released',
|
||||
handoffStage: 'recovering'
|
||||
}),
|
||||
probe: INDETERMINATE,
|
||||
observedAt: 9_000
|
||||
})
|
||||
).toEqual({ disposition: 'free', reason: 'lease has no owner and no reservation' })
|
||||
})
|
||||
|
||||
it('frees an abandoned native reservation even without a spawn-token scan', () => {
|
||||
|
||||
@@ -46,6 +46,8 @@ export type AgentSessionAcquisitionDecision =
|
||||
|
||||
export type AgentSessionRestartAdjudication =
|
||||
| { disposition: 'readopt' }
|
||||
/** Nothing is outstanding — no owner, no reservation. Clear any latched stage; the fence stays. */
|
||||
| { disposition: 'free'; reason: string }
|
||||
| { disposition: 'evicted'; nextFence: number; evidence: AgentSessionDeathEvidence }
|
||||
| { disposition: 'recovering'; stage: AgentSessionHandoffStage; reason: string }
|
||||
| { disposition: 'conflicted'; reason: string }
|
||||
@@ -180,11 +182,27 @@ export function adjudicateAgentSessionRestart(args: {
|
||||
}): AgentSessionRestartAdjudication {
|
||||
const { lease, probe, observedAt } = args
|
||||
if (lease.claimStatus === 'conflicted') {
|
||||
// Why: the conflict outlives the process that observed it; resolving to free would hand the
|
||||
// provider session to whichever side restarted first.
|
||||
const conflictedOwnerDeath =
|
||||
lease.ownerProcess === null ? null : deathEvidenceFor(probe, observedAt)
|
||||
if (conflictedOwnerDeath) {
|
||||
// Why: the conflict names one specific process. Present-time proof that THAT process is gone
|
||||
// leaves no claimant to protect, and a conflict with no exit is a session the user can never
|
||||
// open again. Without such proof the conflict still outlives the process that observed it.
|
||||
return {
|
||||
disposition: 'evicted',
|
||||
nextFence: nextAgentSessionFence(lease),
|
||||
evidence: conflictedOwnerDeath
|
||||
}
|
||||
}
|
||||
return { disposition: 'conflicted', reason: 'claim conflicted before restart' }
|
||||
}
|
||||
if (lease.ownerProcess === null) {
|
||||
if (lease.reservedSpawnToken === null && lease.claimStatus !== 'reserved') {
|
||||
// Why: the spawn token is minted before the child and is the only thing a child could be
|
||||
// carrying. With no owner and no token nothing can hold this lease, so it is already free —
|
||||
// treating it as an unproven reservation is what re-latches every released record on restart.
|
||||
return { disposition: 'free', reason: 'lease has no owner and no reservation' }
|
||||
}
|
||||
if (probe.outcome === 'reservation-unused') {
|
||||
return {
|
||||
disposition: 'evicted',
|
||||
|
||||
Reference in New Issue
Block a user