fix(agent-session): harden recovery ownership

This commit is contained in:
Brennan Benson
2026-08-27 18:52:00 -07:00
parent 551672c28e
commit 23a8696b75
53 changed files with 1929 additions and 1179 deletions
@@ -1,48 +0,0 @@
import { describe, expect, it, vi } from 'vitest'
import { claudeProcessIdentity } from './claude-structured-owner-identity'
const IDENTITY = {
sessionId: 'session-identity',
workspaceId: 'workspace-1',
hostId: 'local',
agent: 'claude' as const,
providerHandle: { kind: 'claude' as const, sessionId: 'provider-1', leafUuid: null }
}
describe('claude process identity', () => {
it('records the observed start time alongside the spawn token', async () => {
await expect(
claudeProcessIdentity(
{ 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(
claudeProcessIdentity({ 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', async () => {
const readStartTime = vi.fn(async () => null)
await expect(
claudeProcessIdentity({ identity: IDENTITY, spawnToken: 'spawn-a', pid: 4242 }, readStartTime)
).rejects.toThrow('start time')
expect(readStartTime).toHaveBeenCalledTimes(3)
})
})
@@ -1,42 +1,4 @@
import type { AgentSessionJournalIdentity } from '../../shared/agent-session-journal-types'
import type { AgentSessionProviderHandleLink } from '../../shared/agent-session-provider-handle'
import type { AgentSessionProcessIdentity } from '../../shared/agent-session-record'
import { readProcessStartTimeMs } from '../runtime/agent-session-process-identity-probe'
export const CLAUDE_SPAWN_TOKEN_ENV = 'ORCA_AGENT_SESSION_SPAWN_TOKEN'
const START_TIME_READ_ATTEMPTS = 3
export async function claudeProcessIdentity(
input: {
identity: AgentSessionJournalIdentity
spawnToken: string
pid: number | undefined
},
readStartTime: (pid: number) => Promise<number | null> = readProcessStartTimeMs
): Promise<AgentSessionProcessIdentity> {
if (input.pid === undefined) {
throw new Error('claude stream-json 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) {
// Recording null makes every later owner probe indeterminate, so refuse before the lease
// can publish a writer that cannot be re-proved after a restart.
throw new Error(`claude stream-json start time for pid ${input.pid} could not be read`)
}
return {
hostId: input.identity.hostId,
pid: input.pid,
processStartTimeMs,
spawnToken: input.spawnToken
}
}
export function claudeProviderHandleLink(input: {
sessionId: string
@@ -27,38 +27,6 @@ export class ClaudeTranscriptTailIncompleteError extends Error {
}
}
export async function retryClaudeTranscriptTailRead<T>(
read: () => Promise<T>,
options: {
timeoutMs?: number
retryDelayMs?: number
now?: () => number
wait?: (delayMs: number) => Promise<void>
} = {}
): Promise<T> {
const timeoutMs = options.timeoutMs ?? 15_000
const retryDelayMs = options.retryDelayMs ?? 100
const now = options.now ?? Date.now
const wait =
options.wait ?? ((delayMs: number) => new Promise((resolve) => setTimeout(resolve, delayMs)))
const deadline = now() + timeoutMs
let incompleteTail: ClaudeTranscriptTailIncompleteError | null = null
do {
try {
return await read()
} catch (error) {
if (!(error instanceof ClaudeTranscriptTailIncompleteError)) {
throw error
}
incompleteTail = error
}
if (now() < deadline) {
await wait(retryDelayMs)
}
} while (now() < deadline)
throw incompleteTail
}
export function proveClaudeTranscriptBranchFromJsonl(input: {
contents: string
providerSessionId: string
+13 -5
View File
@@ -70,8 +70,16 @@ export async function openCodexAppServerConnection(
for (const key of launch.envToDelete ?? []) {
delete childEnv[key]
}
const child = spawnImpl(createProviderSpawnSpec(launch, childEnv, process.platform))
const spawnSpec = createProviderSpawnSpec(launch, childEnv, process.platform)
const child = spawnImpl(spawnSpec)
const spawnToken = launch.env?.[CODEX_SPAWN_TOKEN_ENV]
const hasDedicatedProcessGroup = spawnImpl === spawnProcess && spawnSpec.detached
function terminateProcessTree(): Promise<boolean> {
return terminateCodexAppServerProcessTree(child, spawnToken, {
dedicatedProcessGroup: hasDedicatedProcessGroup
})
}
const pending = new Map<number, PendingRequest>()
let stderrTail = ''
@@ -145,7 +153,7 @@ export async function openCodexAppServerConnection(
failPending(error)
return
}
void terminateCodexAppServerProcessTree(child, spawnToken)
void terminateProcessTree()
handleUnexpectedEnd(error)
})
@@ -199,7 +207,7 @@ export async function openCodexAppServerConnection(
stdoutBuffer += chunk
if (Buffer.byteLength(stdoutBuffer) > STDOUT_LINE_MAX_BYTES) {
child.stdout.destroy()
void terminateCodexAppServerProcessTree(child, spawnToken)
void terminateProcessTree()
handleUnexpectedEnd(new Error('codex app-server emitted an oversized JSONL line'))
return
}
@@ -219,7 +227,7 @@ export async function openCodexAppServerConnection(
dispatchMessage(parsed)
} catch (error) {
child.stdout.destroy()
void terminateCodexAppServerProcessTree(child, spawnToken)
void terminateProcessTree()
handleUnexpectedEnd(error instanceof Error ? error : new Error(String(error)))
return
}
@@ -300,7 +308,7 @@ export async function openCodexAppServerConnection(
if (!exited) {
await waitForProcessExitUntil(exitPromise, GRACEFUL_EXIT_MS)
if (!exited) {
const treeExited = await terminateCodexAppServerProcessTree(child, spawnToken)
const treeExited = await terminateProcessTree()
if (!treeExited) {
failPending(new Error('codex app-server process-tree exit was not proven'))
return false
@@ -84,4 +84,65 @@ describe('terminateCodexAppServerProcessTree', () => {
expect(target.kill).toHaveBeenLastCalledWith('SIGKILL')
})
it('signals a proven dedicated POSIX process group without scanning descendants', async () => {
const target = child()
const captureDescendants = vi.fn()
const signalProcessGroup = vi.fn()
await expect(
terminateCodexAppServerProcessTree(target, undefined, {
platform: 'darwin',
dedicatedProcessGroup: true,
captureDescendants,
signalProcessGroup
})
).resolves.toBe(true)
expect(signalProcessGroup).toHaveBeenCalledWith(1234, 'SIGKILL')
expect(captureDescendants).not.toHaveBeenCalled()
expect(target.kill).not.toHaveBeenCalled()
})
it('keeps the dedicated-group wrapper reachable when signalling is unproven', async () => {
const target = child()
await expect(
terminateCodexAppServerProcessTree(target, undefined, {
platform: 'linux',
dedicatedProcessGroup: true,
signalProcessGroup: () => {
throw Object.assign(new Error('denied'), { code: 'EPERM' })
}
})
).resolves.toBe(false)
expect(target.kill).not.toHaveBeenCalled()
})
it('tears down 40 dedicated groups without process-table scans or cross-group fanout', async () => {
const killMocks = Array.from({ length: 40 }, () => vi.fn(() => true))
const targets = killMocks.map((kill, index) => ({
pid: 10_000 + index,
kill: kill as ChildProcess['kill']
}))
const captureDescendants = vi.fn()
const signalProcessGroup = vi.fn()
const results = await Promise.all(
targets.map((target) =>
terminateCodexAppServerProcessTree(target, undefined, {
platform: 'linux',
dedicatedProcessGroup: true,
captureDescendants,
signalProcessGroup
})
)
)
expect(results).toEqual(Array.from({ length: targets.length }, () => true))
expect(signalProcessGroup.mock.calls).toEqual(targets.map((target) => [target.pid, 'SIGKILL']))
expect(captureDescendants).not.toHaveBeenCalled()
expect(killMocks.every((kill) => kill.mock.calls.length === 0)).toBe(true)
})
})
@@ -12,6 +12,7 @@ type TeardownChild = Pick<ChildProcess, 'pid' | 'kill'>
export type CodexAppServerProcessTeardownDeps = {
platform?: NodeJS.Platform
dedicatedProcessGroup?: boolean
/** Diagnostic/recovery injection only; never used by the primary teardown. */
findSpawnTokenProcesses?: (spawnToken: string) => Promise<number[] | null>
captureDescendants?: (rootPid: number) => Promise<DescendantSnapshot | null>
@@ -24,6 +25,21 @@ export type CodexAppServerProcessTeardownDeps = {
now?: () => number
}
function terminateDedicatedPosixGroup(
rootPid: number,
deps: CodexAppServerProcessTeardownDeps
): boolean {
const signalGroup =
deps.signalProcessGroup ??
((pgid: number, signal: NodeJS.Signals) => process.kill(-pgid, signal))
try {
signalGroup(rootPid, 'SIGKILL')
return true
} catch (error) {
return (error as NodeJS.ErrnoException).code === 'ESRCH'
}
}
function sendSignal(pid: number, signal: NodeJS.Signals): void {
try {
process.kill(pid, signal)
@@ -140,6 +156,9 @@ async function terminateOnce(
child.kill('SIGKILL')
return true
}
if (deps.dedicatedProcessGroup) {
return terminateDedicatedPosixGroup(rootPid, deps)
}
return terminatePosixTree(child, rootPid, spawnToken, deps)
}
@@ -0,0 +1,267 @@
import { beforeEach, describe, expect, it, vi } from 'vitest'
type StreamRecord = {
emit: (response: string) => void
settled: boolean
signal: AbortSignal
subscriptionId: string
}
const { handlers, listeners, streams } = vi.hoisted(() => ({
handlers: new Map<string, (_event: unknown, args?: unknown) => unknown>(),
listeners: new Map<string, (_event: unknown, args?: unknown) => unknown>(),
streams: [] as StreamRecord[]
}))
vi.mock('electron', () => ({
BrowserWindow: { fromWebContents: vi.fn() },
ipcMain: {
handle: vi.fn((channel: string, handler: (_event: unknown, args?: unknown) => unknown) => {
handlers.set(channel, handler)
}),
on: vi.fn((channel: string, handler: (_event: unknown, args?: unknown) => unknown) => {
listeners.set(channel, handler)
}),
removeAllListeners: vi.fn(),
removeHandler: vi.fn()
}
}))
vi.mock('../runtime/rpc/dispatcher', () => ({
RpcDispatcher: class {
dispatchStreaming(
request: { id: string },
emit: (response: string) => void,
options: { signal: AbortSignal }
): Promise<void> {
const record: StreamRecord = {
emit,
settled: false,
signal: options.signal,
subscriptionId: request.id
}
streams.push(record)
return new Promise<void>((resolve) => {
options.signal.addEventListener('abort', () => {
record.settled = true
resolve()
})
})
}
}
}))
import { registerRuntimeHandlers } from './runtime'
type SenderHarness = {
destroy: () => void
emitDidNavigate: () => void
emitRenderProcessGone: () => void
listenerCount: (eventName: string) => number
sender: {
id: number
isDestroyed: () => boolean
on: (eventName: string, callback: () => void) => void
once: (eventName: string, callback: () => void) => void
send: ReturnType<typeof vi.fn>
}
}
function createSender(id: number): SenderHarness {
const senderListeners = new Map<string, (() => void)[]>()
let destroyed = false
const add = (eventName: string, callback: () => void): void => {
const callbacks = senderListeners.get(eventName) ?? []
callbacks.push(callback)
senderListeners.set(eventName, callbacks)
}
const fire = (eventName: string): void => {
for (const callback of senderListeners.get(eventName) ?? []) {
callback()
}
}
return {
destroy: () => {
destroyed = true
fire('destroyed')
},
emitDidNavigate: () => fire('did-navigate'),
emitRenderProcessGone: () => fire('render-process-gone'),
listenerCount: (eventName) => (senderListeners.get(eventName) ?? []).length,
sender: {
id,
isDestroyed: () => destroyed,
on: add,
once: (eventName, callback) => {
const wrapped = (): void => {
const callbacks = senderListeners.get(eventName) ?? []
senderListeners.set(
eventName,
callbacks.filter((candidate) => candidate !== wrapped)
)
callback()
}
add(eventName, wrapped)
},
send: vi.fn()
}
}
}
function subscribe(sender: SenderHarness['sender'], subscriptionId: string): void {
const handler = handlers.get('runtime:subscribe')
if (!handler) {
throw new Error('runtime:subscribe handler not registered')
}
handler({ sender }, { subscriptionId, method: 'agentSession.watch' })
}
function streamFor(subscriptionId: string): StreamRecord {
const record = streams.findLast((entry) => entry.subscriptionId === subscriptionId)
if (!record) {
throw new Error(`no stream dispatched for ${subscriptionId}`)
}
return record
}
const FRAME = JSON.stringify({ ok: true, result: { seq: 1 } })
describe('runtime:subscribe renderer lifecycle cleanup', () => {
beforeEach(() => {
handlers.clear()
listeners.clear()
streams.length = 0
registerRuntimeHandlers({} as never)
})
it('aborts a live stream once its sender commits a navigation', () => {
const harness = createSender(1)
subscribe(harness.sender, 'sub-reload')
const stream = streamFor('sub-reload')
stream.emit(FRAME)
expect(harness.sender.send).toHaveBeenCalledTimes(1)
expect(harness.sender.send).toHaveBeenCalledWith('runtime:subscription:sub-reload', {
ok: true,
result: { seq: 1 }
})
harness.emitDidNavigate()
expect(stream.signal.aborted).toBe(true)
expect(stream.settled).toBe(true)
stream.emit(FRAME)
expect(harness.sender.send).toHaveBeenCalledTimes(1)
})
it('aborts a live stream when the renderer process dies without destruction', () => {
const harness = createSender(2)
subscribe(harness.sender, 'sub-crash')
const stream = streamFor('sub-crash')
harness.emitRenderProcessGone()
expect(stream.signal.aborted).toBe(true)
stream.emit(FRAME)
expect(harness.sender.send).not.toHaveBeenCalled()
})
it('still aborts on sender destruction', () => {
const harness = createSender(3)
subscribe(harness.sender, 'sub-destroyed')
const stream = streamFor('sub-destroyed')
harness.destroy()
expect(stream.signal.aborted).toBe(true)
stream.emit(FRAME)
expect(harness.sender.send).not.toHaveBeenCalled()
})
it('scopes navigation cleanup to the navigating sender', () => {
const navigating = createSender(4)
const surviving = createSender(5)
subscribe(navigating.sender, 'sub-navigating')
subscribe(surviving.sender, 'sub-surviving')
navigating.emitDidNavigate()
expect(streamFor('sub-navigating').signal.aborted).toBe(true)
expect(streamFor('sub-surviving').signal.aborted).toBe(false)
streamFor('sub-surviving').emit(FRAME)
expect(surviving.sender.send).toHaveBeenCalledTimes(1)
})
it('streams to a fresh post-reload subscription while the orphan stays dead', async () => {
const harness = createSender(6)
subscribe(harness.sender, 'sub-old')
const orphan = streamFor('sub-old')
harness.emitDidNavigate()
expect(orphan.signal.aborted).toBe(true)
// The settled stream's cleanup runs off the dispatch promise.
await Promise.resolve()
subscribe(harness.sender, 'sub-new')
const fresh = streamFor('sub-new')
expect(fresh.signal.aborted).toBe(false)
fresh.emit(FRAME)
orphan.emit(FRAME)
expect(harness.sender.send).toHaveBeenCalledTimes(1)
expect(harness.sender.send).toHaveBeenCalledWith('runtime:subscription:sub-new', {
ok: true,
result: { seq: 1 }
})
// Lifecycle listeners are registered once per sender, not once per subscription.
expect(harness.listenerCount('did-navigate')).toBe(1)
expect(harness.listenerCount('render-process-gone')).toBe(1)
})
it('aborts every stream of a sender with several live subscriptions', () => {
const harness = createSender(7)
subscribe(harness.sender, 'sub-a')
subscribe(harness.sender, 'sub-b')
harness.emitDidNavigate()
expect(streamFor('sub-a').signal.aborted).toBe(true)
expect(streamFor('sub-b').signal.aborted).toBe(true)
})
it('keeps explicit unsubscribe working', () => {
const harness = createSender(8)
subscribe(harness.sender, 'sub-explicit')
const stream = streamFor('sub-explicit')
const unsubscribe = listeners.get('runtime:unsubscribe')
if (!unsubscribe) {
throw new Error('runtime:unsubscribe listener not registered')
}
unsubscribe({ sender: harness.sender }, { subscriptionId: 'sub-explicit' })
expect(stream.signal.aborted).toBe(true)
})
it('scopes colliding subscription ids and unsubscribe to their sender', () => {
const firstHarness = createSender(9)
const secondHarness = createSender(10)
subscribe(firstHarness.sender, 'sub-collision')
const first = streams.at(-1)!
subscribe(secondHarness.sender, 'sub-collision')
const second = streams.at(-1)!
expect(first.signal.aborted).toBe(false)
expect(second.signal.aborted).toBe(false)
const unsubscribe = listeners.get('runtime:unsubscribe')
if (!unsubscribe) {
throw new Error('runtime:unsubscribe listener not registered')
}
unsubscribe({ sender: firstHarness.sender }, { subscriptionId: 'sub-collision' })
expect(first.signal.aborted).toBe(true)
expect(second.signal.aborted).toBe(false)
})
})
+34 -11
View File
@@ -25,7 +25,18 @@ function boundTerminalFitRestore(pending: Promise<boolean>): Promise<boolean> {
export function registerRuntimeHandlers(runtime: OrcaRuntimeService): void {
const pendingTerminalFitRestores = new Map<string, Promise<boolean>>()
const localSubscriptions = new Map<string, AbortController>()
const subscriptionsBySender = new Map<number, Map<string, AbortController>>()
const abortSenderSubscriptions = (senderId: number): void => {
const subscriptions = subscriptionsBySender.get(senderId)
if (!subscriptions) {
return
}
for (const controller of subscriptions.values()) {
controller.abort()
}
subscriptions.clear()
}
ipcMain.removeHandler('runtime:syncWindowGraph')
ipcMain.removeHandler('runtime:getStatus')
ipcMain.removeHandler('runtime:call')
@@ -83,19 +94,30 @@ export function registerRuntimeHandlers(runtime: OrcaRuntimeService): void {
event,
args: { subscriptionId: string; method: string; params?: unknown }
): { subscribed: boolean } => {
const previous = localSubscriptions.get(args.subscriptionId)
const senderId = event.sender.id
let senderSubscriptions = subscriptionsBySender.get(senderId)
if (!senderSubscriptions) {
senderSubscriptions = new Map()
subscriptionsBySender.set(senderId, senderSubscriptions)
// Why: reload reuses WebContents, so destruction alone cannot stop the old document's streams.
const abortAll = (): void => abortSenderSubscriptions(senderId)
event.sender.on('did-navigate', abortAll)
event.sender.on('render-process-gone', abortAll)
event.sender.once('destroyed', () => {
abortAll()
subscriptionsBySender.delete(senderId)
})
}
const previous = senderSubscriptions.get(args.subscriptionId)
previous?.abort()
const controller = new AbortController()
localSubscriptions.set(args.subscriptionId, controller)
senderSubscriptions.set(args.subscriptionId, controller)
const channel = `runtime:subscription:${args.subscriptionId}`
const onSenderDestroyed = (): void => controller.abort()
const stop = (): void => {
event.sender.removeListener('destroyed', onSenderDestroyed)
if (localSubscriptions.get(args.subscriptionId) === controller) {
localSubscriptions.delete(args.subscriptionId)
if (senderSubscriptions.get(args.subscriptionId) === controller) {
senderSubscriptions.delete(args.subscriptionId)
}
}
event.sender.once('destroyed', onSenderDestroyed)
void new RpcDispatcher({ runtime, methods: ALL_RPC_METHODS })
.dispatchStreaming(
{
@@ -121,9 +143,10 @@ export function registerRuntimeHandlers(runtime: OrcaRuntimeService): void {
}
)
ipcMain.on('runtime:unsubscribe', (_event, args: { subscriptionId: string }) => {
localSubscriptions.get(args.subscriptionId)?.abort()
localSubscriptions.delete(args.subscriptionId)
ipcMain.on('runtime:unsubscribe', (event, args: { subscriptionId: string }) => {
const senderSubscriptions = subscriptionsBySender.get(event.sender.id)
senderSubscriptions?.get(args.subscriptionId)?.abort()
senderSubscriptions?.delete(args.subscriptionId)
})
ipcMain.removeHandler('runtime:getTerminalFitOverrides')
@@ -5,6 +5,7 @@ import { afterEach, describe, expect, it, vi } from 'vitest'
import { computeAgentSessionPayloadFingerprint } from '../../../shared/agent-session-mutation-envelope'
import type { AgentSessionOptionsResult } from '../../../shared/agent-session-wire'
import { AgentSessionRecordStore } from '../../runtime/agent-session-record-store'
import type { AgentSessionRecord } from '../../../shared/agent-session-record'
import type { StructuredAgentSessionAdapter } from './structured-agent-session-adapter'
import {
attachFingerprintFields,
@@ -91,6 +92,14 @@ function adapter(input: {
}
}
function expectSettledAttachLease(record: AgentSessionRecord | null): void {
expect(record).not.toBeNull()
const lease = record!.lease
const durableState = lease.handoffStage ?? lease.claimStatus
expect(['live', 'released', 'recovering', 'manual-recovery']).toContain(durableState)
expect(lease.handoffStage).not.toBe('new-owner-proving')
}
describe('structured session acquisition options', () => {
it('persists provider options before proving a resumed legacy record', async () => {
root = await mkdtemp(join(tmpdir(), 'orca-acquisition-options-'))
@@ -197,4 +206,143 @@ describe('structured session acquisition options', () => {
expect(releaseAcquisition).toHaveBeenCalledOnce()
expect(store.getRecord(SESSION)?.lease.ownerProcess).toBeNull()
})
describe.each([
['adapter acquire', 'acquire'],
['options read', 'options'],
['identity commit', 'identity'],
['owner proof', 'proof'],
['journal attach', 'journal']
] as const)('%s failure', (_label, failurePoint) => {
it.each([
['proven cleanup', true],
['unproven cleanup', false],
['cleanup error', 'throws']
] as const)('atomically settles the lease and operation after %s', async (_case, cleanup) => {
const exitProven = cleanup === true
root = await mkdtemp(join(tmpdir(), `orca-acquisition-${failurePoint}-`))
const storeDir = join(root, 'store')
const store = await AgentSessionRecordStore.open({ directory: storeDir, hostId: 'local' })
const base = adapter({
origin: 'created',
options: { current: { model: 'gpt-5.6-terra' }, models: [] }
})
const injected = new Error(`${failurePoint} failed`)
const acquire = vi.mocked(base.acquire)
const readOptions = vi.mocked(base.readOptions!)
if (failurePoint === 'acquire') {
acquire.mockRejectedValueOnce(injected)
} else if (failurePoint === 'options') {
readOptions.mockRejectedValueOnce(injected)
} else if (failurePoint === 'identity') {
vi.spyOn(store, 'commitProcessIdentity').mockRejectedValueOnce(injected)
} else if (failurePoint === 'proof') {
vi.spyOn(store, 'proveOwner').mockRejectedValueOnce(injected)
} else if (failurePoint === 'journal') {
acquire.mockImplementation(async ({ fence, spawnToken }) => ({
process: {
hostId: 'local',
pid: 4242,
processStartTimeMs: NOW,
spawnToken
},
link: {
linkId: `link-${fence}`,
handle: { provider: 'codex', threadId: 'legacy-thread' },
origin: store.getRecord(SESSION)?.providerHandleChain.length ? 'resumed' : 'created',
mintedAtFence: fence,
observedAt: NOW
}
}))
}
const releaseAcquisition = vi.fn(async () => {
if (cleanup === 'throws') {
throw new Error('cleanup failed')
}
return cleanup
})
const failingAdapter = {
...base,
acquire,
readOptions,
releaseAcquisition,
...(failurePoint === 'journal'
? { historyFilePath: vi.fn().mockRejectedValueOnce(injected).mockResolvedValue(null) }
: {})
}
const perform = (
target: AgentSessionRecordStore,
operationId: string,
fence: number | null
) =>
performAttach({
store: target,
adapter: failingAdapter,
journalRoot: root!,
authority: {
spawnToken: operationId === CREATE_OPERATION ? 'spawn-a' : 'spawn-b',
claimKeyId: 'key-1',
handoffOperationId: operationId,
probe: { outcome: 'reservation-unused' }
},
callerKey: 'client-1',
params: attachParams(operationId, fence),
now: () => NOW,
onAttached: () => {}
})
await expect(perform(store, CREATE_OPERATION, null)).rejects.toThrow(
exitProven ? injected.message : 'agent_session_acquisition_exit_unproven'
)
const reopened = await AgentSessionRecordStore.open({
directory: storeDir,
hostId: 'local'
})
const failedRecord = reopened.getRecord(SESSION)
expectSettledAttachLease(failedRecord)
expect(
reopened.listOperationRows().find((row) => row.operationId === CREATE_OPERATION)?.outcome
).toMatchObject({ status: 'failed' })
await reopened.reconcileOnRestart({
probe: async (record) =>
exitProven || record.lease.ownerProcess === null
? exitProven
? { outcome: 'reservation-unused' }
: { outcome: 'indeterminate', reason: 'owner identity was never committed' }
: { outcome: 'identity-matched', matchedOn: ['process-start-time'] },
now: NOW + 1
})
if (exitProven) {
expect(failedRecord?.lease).toMatchObject({
runtimeFence: 2,
claimStatus: 'released',
handoffStage: null,
handoffOperationId: null,
ownerProcess: null,
reservedSpawnToken: null
})
await expect(perform(reopened, RESUME_OPERATION, 2)).resolves.toMatchObject({ ok: true })
expectSettledAttachLease(reopened.getRecord(SESSION))
} else {
expect(failedRecord?.lease).toMatchObject({
runtimeFence: 1,
claimStatus: failurePoint === 'journal' ? 'live' : 'reserved',
handoffStage:
failurePoint === 'proof' || failurePoint === 'journal'
? 'recovering'
: 'manual-recovery',
// The settled operation must not stay named by the lease as an in-flight transfer.
handoffOperationId: null,
reservedSpawnToken: 'spawn-a'
})
await expect(perform(reopened, RESUME_OPERATION, 1)).resolves.toMatchObject({
ok: false,
refusal: { code: 'agent_session_ownership_unknown' }
})
}
})
})
})
@@ -25,7 +25,9 @@ import {
import type { AgentSessionRecordStore } from '../../runtime/agent-session-record-store'
import type { StructuredAgentSessionAdapter } from './structured-agent-session-adapter'
import {
AgentSessionAcquisitionExitUnprovenError,
AgentSessionAcquisitionRefusal,
AgentSessionPreSpawnError,
isAgentSessionPreSpawnError,
rethrowAfterAgentSessionAcquisitionCleanup
} from './structured-agent-session-adapter'
@@ -52,6 +54,8 @@ export type AttachFlowInput = {
onAcquiring?: () => Promise<void> | void
/** Settles writes already captured by the superseded journal before opening another. */
beforeJournalOpen?: () => Promise<void> | void
/** Removes any partial host publication after journal attachment fails. */
onAttachFailed?: () => void
}
export async function performAttach(
@@ -79,7 +83,6 @@ export async function performAttach(
})
)
record = reserved.record
reservedRecord = record
replayed = reserved.disposition === 'replayed'
if (
replayed &&
@@ -95,32 +98,59 @@ export async function performAttach(
return { ok: false, refusal: replay.refusal }
}
}
reservedRecord = record
if (!agentSessionLeaseAdmitsWriter(record.lease)) {
record = await acquireOwner(input, record)
}
} catch (error) {
if (error instanceof AgentSessionAcquisitionRefusal) {
const refusal = { code: error.code, message: error.message }
await store.recordOperationOutcome({
callerKey: input.callerKey,
operationId: params.envelope.clientOperationId,
outcome: { status: 'failed', code: refusal.code, message: refusal.message }
})
return { ok: false, refusal }
}
if (reservedRecord && isAgentSessionPreSpawnError(error)) {
const spawnToken = reservedRecord.lease.reservedSpawnToken
if (spawnToken) {
const processlessAt = input.now()
await store.setReservationProcesslessProof({
const spawnToken = reservedRecord?.lease.reservedSpawnToken
if (reservedRecord && spawnToken) {
// A pre-spawn failure is its own processless proof; the settlement records the
// evidence and the failed operation in one durable transaction.
const exitProof = isAgentSessionPreSpawnError(error)
? 'processless'
: error instanceof AgentSessionAcquisitionExitUnprovenError
? 'unproven'
: 'exit-proven'
const outcome =
error instanceof AgentSessionAcquisitionExitUnprovenError
? {
status: 'failed' as const,
code: 'agent_session_ownership_unknown',
message: error.message
}
: error instanceof AgentSessionAcquisitionRefusal
? {
status: 'failed' as const,
code: error.code,
message: error.message
}
: {
status: 'failed' as const,
code: 'agent_session_operation_invalid',
message: error instanceof Error ? error.message : String(error)
}
try {
await store.settleFailedAcquisition({
sessionId,
fence: reservedRecord.lease.runtimeFence,
spawnToken,
processlessAt,
now: processlessAt
callerKey: input.callerKey,
operationId: params.envelope.clientOperationId,
outcome,
exitProof,
now: input.now()
})
} catch (settlementError) {
throw new AggregateError(
[error, settlementError],
'agent session acquisition failure settlement failed'
)
}
}
if (error instanceof AgentSessionAcquisitionRefusal) {
return { ok: false, refusal: { code: error.code, message: error.message } }
}
return {
ok: false,
refusal: classifyStoreFailure(
@@ -131,19 +161,24 @@ export async function performAttach(
}
}
await input.beforeJournalOpen?.()
const attached = await attachJournal({
record,
params,
journalRoot: input.journalRoot,
adapter: input.adapter
})
input.onAttached(attached)
await store.recordOperationOutcome({
callerKey: input.callerKey,
operationId: params.envelope.clientOperationId,
outcome: { status: 'succeeded', sessionId }
})
let attached: AttachedJournal
try {
await input.beforeJournalOpen?.()
attached = await attachJournal({
record,
params,
journalRoot: input.journalRoot,
adapter: input.adapter
})
input.onAttached(attached)
await store.recordOperationOutcome({
callerKey: input.callerKey,
operationId: params.envelope.clientOperationId,
outcome: { status: 'succeeded', sessionId }
})
} catch (error) {
return settlePostAcquisitionAttachFailure(input, record, error)
}
const fence = record.lease.runtimeFence
return {
@@ -160,6 +195,45 @@ export async function performAttach(
}
}
async function settlePostAcquisitionAttachFailure(
input: AttachFlowInput,
record: AgentSessionRecord,
cause: unknown
): Promise<never> {
let cleanupError: unknown = cause
let exitProof: 'exit-proven' | 'unproven' = 'unproven'
try {
await rethrowAfterAgentSessionAcquisitionCleanup(input.adapter, record.sessionId, cause)
} catch (error) {
cleanupError = error
exitProof =
error instanceof AgentSessionAcquisitionExitUnprovenError ? 'unproven' : 'exit-proven'
}
input.onAttachFailed?.()
try {
await input.store.settleFailedPostAcquisitionAttachment({
sessionId: record.sessionId,
fence: record.lease.runtimeFence,
spawnToken: record.lease.reservedSpawnToken ?? '',
callerKey: input.callerKey,
operationId: input.params.envelope.clientOperationId,
outcome: {
status: 'failed',
code: 'agent_session_operation_invalid',
message: cause instanceof Error ? cause.message : String(cause)
},
exitProof,
now: input.now()
})
} catch (settlementError) {
throw new AggregateError(
[cleanupError, settlementError],
'agent session post-acquisition attachment failure settlement failed'
)
}
throw cleanupError
}
/** A reservation with no process behind it is only a promise to spawn; the
* adapter makes it real and the store then grants the writer. */
async function acquireOwner(
@@ -172,23 +246,27 @@ async function acquireOwner(
throw new Error('agent_session_ownership_unknown')
}
// Pre-spawn proof is single-use: this retry may create a child after the durable clear.
record = await input.store.setReservationProcesslessProof({
sessionId: record.sessionId,
fence,
spawnToken,
processlessAt: null,
now: input.now()
})
await input.onAcquiring?.()
const acquired = await input.adapter.acquire({
identity: journalIdentityFor(record, input.params),
fence,
// Retries must recover the original reservation, not mint a second child.
spawnToken,
...(record.options ? { options: record.options } : {}),
...(input.eventSink ? { events: input.eventSink } : {})
})
try {
try {
record = await input.store.setReservationProcesslessProof({
sessionId: record.sessionId,
fence,
spawnToken,
processlessAt: null,
now: input.now()
})
await input.onAcquiring?.()
} catch (error) {
throw new AgentSessionPreSpawnError(error)
}
const acquired = await input.adapter.acquire({
identity: journalIdentityFor(record, input.params),
fence,
// Retries must recover the original reservation, not mint a second child.
spawnToken,
...(record.options ? { options: record.options } : {}),
...(input.eventSink ? { events: input.eventSink } : {})
})
const options = await readNativeSessionOptions({
adapter: input.adapter,
sessionId: record.sessionId,
@@ -213,6 +291,9 @@ async function acquireOwner(
...(options ? { options } : {})
})
} catch (error) {
if (isAgentSessionPreSpawnError(error)) {
throw error
}
return rethrowAfterAgentSessionAcquisitionCleanup(input.adapter, record.sessionId, error)
}
}
@@ -43,7 +43,7 @@ export function attachStructuredAgentSession(
await eventSink.drained()
},
authority: {
spawnToken: context.deps.mintSpawnToken?.() ?? randomUUID(),
spawnToken: () => context.deps.mintSpawnToken?.() ?? randomUUID(),
claimKeyId: context.deps.claimKeyId,
handoffOperationId: params.envelope.clientOperationId,
probe: await context.runtimeState.probeOwner(sessionId),
@@ -53,6 +53,11 @@ export function attachStructuredAgentSession(
callerKey,
params,
now: () => context.now(),
onAttachFailed: () => {
context.sessions.delete(sessionId)
eventSink.close()
context.runtimeState.discardEventSink(sessionId)
},
onAttached: (attached) => {
const fence = context.deps.store.getRecord(sessionId)?.lease.runtimeFence ?? 0
const previousFence = context.sessions.get(sessionId)?.fence
@@ -60,7 +60,7 @@ export type AgentSessionAttachParams = {
/** Host-supplied half of the reservation. */
export type AgentSessionAttachAuthority = {
spawnToken: string
spawnToken: string | (() => string)
claimKeyId: string
handoffOperationId: string | null
probe: AgentSessionOwnerProbe
@@ -18,6 +18,7 @@ import type {
const NOW = 1_800_000_000_000
const SESSION = 'session-handoff'
const PLAIN_RESIDUE = 'session-plain-residue'
const THREAD = '019fd532-7c11-7a90-b6de-4e1a2c3d5f60'
let root: string
@@ -288,4 +289,96 @@ describe('structured session ownership recovery on restore', () => {
store.getRecord(SESSION)?.lease.runtimeFence
)
})
it('continues only the persisted TUI handoff after a store restart', async () => {
const plainOperation = operationId()
await store.reserveOwner({
sessionId: PLAIN_RESIDUE,
location: {
executionHostId: 'local',
wslDistro: null,
workspaceId: 'workspace-1',
workspaceKind: 'git-worktree'
},
provider: 'codex',
accountHome: { variable: 'CODEX_HOME', path: join(root, 'codex-home') },
runtimeKind: 'native',
expectedFence: null,
spawnToken: 'plain-residue-token',
claimKeyId: 'key-1',
handoffOperationId: plainOperation,
probe: { outcome: 'reservation-unused' },
operation: { callerKey: 'test', operationId: plainOperation, fingerprint: 'plain-attach' },
now: NOW
})
const handoffOperation = operationId()
let interrupted = await setStoredAgentSessionHandoffStage(store, {
sessionId: SESSION,
fence: 1,
stage: 'preparing',
handoffOperationId: handoffOperation,
now: NOW
})
interrupted = await stopStoredAgentSessionOwnerForHandoff(store, {
sessionId: SESSION,
expectedFence: interrupted.lease.runtimeFence,
operationId: handoffOperation,
now: NOW
})
const interruptedFence = interrupted.lease.runtimeFence
store = await AgentSessionRecordStore.open({ directory: join(root, 'store'), hostId: 'local' })
await store.reconcileOnRestart({
probe: async (record) =>
record.sessionId === PLAIN_RESIDUE
? { outcome: 'indeterminate', reason: 'plain reservation cannot be attributed' }
: { outcome: 'pid-absent' },
now: NOW + 1_000
})
journal = await openAgentSessionJournal({
identity: {
sessionId: SESSION,
workspaceId: 'workspace-1',
hostId: 'local',
agent: 'codex',
providerHandle: { kind: 'codex', threadId: THREAD }
},
journalDir: join(root, 'journal')
})
launchTui = vi.fn(async ({ fence, spawnToken }) => makeTuiOwner(fence, spawnToken))
coordinator = createCoordinator()
await coordinator.restore(PLAIN_RESIDUE)
expect(launchTui).not.toHaveBeenCalled()
expect(acquireNativeCalls).toBe(0)
expect(store.getRecord(PLAIN_RESIDUE)?.lease).toMatchObject({
runtimeKind: 'native',
runtimeFence: 1,
handoffStage: 'manual-recovery',
claimStatus: 'reserved',
reservedSpawnToken: 'plain-residue-token'
})
expect(
store.listOperationRows().find((row) => row.operationId === plainOperation)?.outcome
).toMatchObject({ status: 'failed', code: 'agent_session_ownership_unknown' })
await coordinator.restore(SESSION)
expect(launchTui).toHaveBeenCalledOnce()
expect(acquireNativeCalls).toBe(1)
expect(launchTui.mock.calls[0]?.[0]).toMatchObject({
record: {
sessionId: SESSION,
lease: { handoffOperationId: handoffOperation }
},
fence: interruptedFence + 3
})
expect(store.getRecord(SESSION)?.lease).toMatchObject({
runtimeKind: 'tui',
runtimeFence: interruptedFence + 3,
handoffStage: null,
handoffOperationId: null,
claimStatus: 'live'
})
})
})
@@ -60,15 +60,6 @@ export class StructuredAgentSessionHostRuntimeState {
await Promise.all([...this.eventSinks.values()].map((sink) => sink.drained()))
}
/**
* Process ownership is enforced at spawn time by the OS supervisor. Token
* enumeration is retained as Linux-only recovery evidence and is never the
* orphan reaper of record (macOS/Windows are explicitly unverifiable).
*/
reapOrphanChildren(): Promise<number[]> {
return Promise.resolve([])
}
/** Exit from a latched recovery stage when present-time evidence permits one. */
resolveRecovery(sessionId: string): Promise<'resolved' | 'unresolved' | 'not-applicable'> {
return resolveStructuredSessionRecovery(
@@ -208,49 +208,7 @@ describe('attach', () => {
expect(retry).toMatchObject({ ok: true, replayed: true })
})
it('reuses the persisted spawn token when acquisition is retried', async () => {
const spawnTokens: string[] = []
const acquire = vi
.fn<StructuredAgentSessionAdapter['acquire']>()
.mockImplementationOnce(async ({ spawnToken }) => {
spawnTokens.push(spawnToken)
throw new Error('reply lost')
})
.mockImplementation(async ({ fence, spawnToken }) => {
spawnTokens.push(spawnToken)
return {
process: {
hostId: 'local',
pid: 4242,
processStartTimeMs: 1_700_000_000_000,
spawnToken
},
link: {
linkId: `link-${fence}`,
handle: { provider: 'codex', threadId: THREAD },
origin: 'created',
mintedAtFence: fence,
observedAt: NOW
}
}
})
let token = 0
host = new StructuredAgentSessionHost({
store,
adapter: { ...adapter(), acquire },
journalRoot: root,
claimKeyId: 'key-1',
mintSpawnToken: () => `spawn-${++token}`,
now: () => NOW
})
const params = attachParams()
await expect(host.attach(CALLER, params)).rejects.toThrow('reply lost')
expect(await host.attach(CALLER, params)).toMatchObject({ ok: true, replayed: true })
expect(spawnTokens).toEqual(['spawn-1', 'spawn-1'])
})
it('finishes proof when a retry finds the process identity already committed', async () => {
it('retires a failed proved acquisition before admitting a fresh operation', async () => {
const acquire = vi
.fn<StructuredAgentSessionAdapter['acquire']>()
.mockImplementationOnce(async ({ fence, spawnToken }) => ({
@@ -296,7 +254,12 @@ describe('attach', () => {
await expect(host.attach(CALLER, params)).rejects.toThrow(
'agent_session_provider_handle_stale_fence'
)
expect(await host.attach(CALLER, params)).toMatchObject({ ok: true, replayed: true })
expect(await host.attach(CALLER, params)).toMatchObject({
ok: false,
refusal: { code: 'agent_session_operation_invalid' }
})
const releasedFence = store.getRecord(SESSION)?.lease.runtimeFence ?? 0
expect(await host.attach(CALLER, ensureParams(releasedFence))).toMatchObject({ ok: true })
expect(acquire).toHaveBeenCalledTimes(2)
expect(store.getRecord(SESSION)?.lease.claimStatus).toBe('live')
expect(releaseAcquisition).toHaveBeenCalledWith({ sessionId: SESSION })
@@ -409,6 +372,38 @@ describe('send', () => {
expect(state.ok && state.page.submissions).toHaveLength(1)
})
it('advances an explicit retry after a ledger-unknown send is reconciled in the journal', async () => {
await attach()
const journal = (
host as unknown as { sessions: Map<string, { journal: AgentSessionJournal }> }
).sessions.get(SESSION)!.journal
vi.spyOn(journal, 'resolveDispatch').mockRejectedValueOnce(new Error('journal resolve failed'))
const body = hostTestMessage('possibly delivered before persistence failed')
const params = { envelope: envelope('agentSession.send', { body }), body }
await expect(host.send(CALLER, params)).rejects.toThrow('journal resolve failed')
expect(
store.listOperationRows().find((row) => row.operationId === params.envelope.clientOperationId)
?.outcome
).toEqual({ status: 'unknown' })
expect(dispatch).toHaveBeenCalledTimes(1)
await journal.markPendingSubmissionsUnknown(store.getRecord(SESSION)?.lease.runtimeFence ?? 1)
await expect(host.send(CALLER, params)).resolves.toMatchObject({
ok: false,
refusal: { code: 'agent_session_operation_unknown' }
})
expect(dispatch).toHaveBeenCalledTimes(1)
await expect(host.send(CALLER, { ...params, retryUnknown: true })).resolves.toMatchObject({
ok: true,
replayed: false,
value: { submission: { dispatchState: 'accepted' } }
})
expect(dispatch).toHaveBeenCalledTimes(2)
expect(journal.submissions()).toHaveLength(1)
})
it('refuses a stale fence and hands back the current one', async () => {
const record = await attach()
const body = hostTestMessage('add a retry')
@@ -105,8 +105,7 @@ export class StructuredAgentSessionHost {
serialize: (sessionId, task) => this.serialize(sessionId, task),
hasSession: (sessionId) => this.sessions.has(sessionId),
onReadable: (sessionId, restored) => this.sessions.set(sessionId, restored),
restoreHandoff: (sessionId) => this.handoffs.restore(sessionId),
reapOrphanChildren: () => this.runtimeState.reapOrphanChildren()
restoreHandoff: (sessionId) => this.handoffs.restore(sessionId)
})
this.runtimeState.startLeaseRenewal()
}
@@ -1,38 +0,0 @@
import { describe, expect, it, vi } from 'vitest'
import {
legacyAgentSessionRecordV1,
LEGACY_AGENT_SESSION_ID
} from '../../../shared/agent-session-record-legacy.test-fixture'
import { loadAgentSessionRecord } from '../../../shared/agent-session-record-load'
import { isResumableStructuredAgentSessionRecord } from './structured-agent-session-resume-eligibility'
import { restoreStructuredAgentSessionsOnRestart } from './structured-agent-session-restart-restore'
describe('legacy structured session restart', () => {
it('does not auto-attach an upgraded v1 record', async () => {
const loaded = loadAgentSessionRecord(legacyAgentSessionRecordV1())
if (loaded.status !== 'upgraded') {
throw new Error(`expected upgrade, got ${loaded.status}`)
}
const serialize = vi.fn(async <T>(_sessionId: string, task: () => Promise<T>) => task())
await restoreStructuredAgentSessionsOnRestart({
store: { getRecord: () => loaded.record } as never,
journalRoot: '/isolated/no-journal',
records: [loaded.record],
reconcile: async () => null,
resolveRecovery: async () => 'not-applicable',
serialize: serialize as never,
hasSession: () => false,
onReadable: vi.fn(),
restoreHandoff: vi.fn(async () => undefined)
})
expect(loaded.record).toMatchObject({
sessionId: LEGACY_AGENT_SESSION_ID,
lease: { claimStatus: 'conflicted', handoffStage: 'manual-recovery' }
})
// Restart no longer resumes ANYTHING, so the guard that matters now is the one a surface hits:
// a conflicted record under manual recovery is not a record a hold may acquire.
expect(isResumableStructuredAgentSessionRecord(loaded.record)).toBe(false)
})
})
@@ -4,7 +4,6 @@ import { join } from 'node:path'
import { afterEach, describe, expect, it, vi } from 'vitest'
import { computeAgentSessionPayloadFingerprint } from '../../../shared/agent-session-mutation-envelope'
import { AgentSessionRecordStore } from '../../runtime/agent-session-record-store'
import { createStructuredAgentSessionOwnerProbe } from '../../runtime/structured-agent-session-runtime'
import {
AgentSessionPreSpawnError,
type StructuredAgentSessionAdapter
@@ -18,6 +17,7 @@ import { performAttach } from './structured-agent-session-attach-flow'
const NOW = 1_800_000_000_000
const SESSION = 'session-alpha'
const OPERATION = `${NOW}-${'1'.padStart(32, '0')}`
const NEXT_OPERATION = `${NOW}-${'2'.padStart(32, '0')}`
let root: string | null = null
afterEach(async () => {
@@ -27,12 +27,15 @@ afterEach(async () => {
root = null
})
function attachParams(): AgentSessionAttachParams {
function attachParams(
operationId = OPERATION,
expectedRuntimeFence: number | null = null
): AgentSessionAttachParams {
const params: AgentSessionAttachParams = {
envelope: {
sessionId: SESSION,
clientOperationId: OPERATION,
expectedRuntimeFence: null,
clientOperationId: operationId,
expectedRuntimeFence,
payloadFingerprint: ''
},
location: {
@@ -61,7 +64,7 @@ function attachParams(): AgentSessionAttachParams {
}
describe('processless structured session reservation', () => {
it('persists pre-spawn proof so restart can release only the proven reservation', async () => {
it('settles a pre-spawn failure and its processless evidence in one durable transaction', async () => {
root = await mkdtemp(join(tmpdir(), 'orca-processless-reservation-'))
const storeDir = join(root, 'store')
const store = await AgentSessionRecordStore.open({ directory: storeDir, hostId: 'local' })
@@ -70,6 +73,8 @@ describe('processless structured session reservation', () => {
throw new AgentSessionPreSpawnError(new Error('workspace no longer exists'))
})
} as unknown as StructuredAgentSessionAdapter
const processlessProof = vi.spyOn(store, 'setReservationProcesslessProof')
const settlement = vi.spyOn(store, 'settleFailedAcquisition')
await expect(
performAttach({
@@ -88,14 +93,27 @@ describe('processless structured session reservation', () => {
onAttached: () => {}
})
).rejects.toThrow('workspace no longer exists')
expect(store.getRecord(SESSION)?.lease.processlessAt).toBe(NOW)
expect(settlement).toHaveBeenCalledExactlyOnceWith(
expect.objectContaining({ exitProof: 'processless', spawnToken: 'spawn-a' })
)
// No separate durable proof write: the only proof call is acquisition's single-use clear.
expect(processlessProof).toHaveBeenCalledExactlyOnceWith(
expect.objectContaining({ processlessAt: null })
)
expect(store.getRecord(SESSION)?.lease).toMatchObject({
claimStatus: 'released',
handoffStage: null,
handoffOperationId: null,
runtimeFence: 2,
processlessAt: null,
reservedSpawnToken: null,
deathEvidence: { kind: 'pid-absent', detail: 'reservation failed before spawn' }
})
expect(store.listOperationRows()[0]?.outcome).toMatchObject({ status: 'failed' })
const reopened = await AgentSessionRecordStore.open({ directory: storeDir, hostId: 'local' })
await reopened.reconcileOnRestart({
probe: async (record) =>
record.lease.processlessAt === NOW
? { outcome: 'reservation-unused' }
: { outcome: 'indeterminate', reason: 'missing pre-spawn proof' },
probe: async () => ({ outcome: 'indeterminate', reason: 'no owner to probe' }),
now: NOW + 1
})
expect(reopened.getRecord(SESSION)?.lease).toMatchObject({
@@ -105,7 +123,7 @@ describe('processless structured session reservation', () => {
})
})
it('consumes stale pre-spawn proof before a retry can spawn', async () => {
it('does not rerun a settled pre-spawn failure and admits a fresh operation', async () => {
root = await mkdtemp(join(tmpdir(), 'orca-processless-retry-'))
const storeDir = join(root, 'store')
const store = await AgentSessionRecordStore.open({ directory: storeDir, hostId: 'local' })
@@ -113,21 +131,21 @@ describe('processless structured session reservation', () => {
acquire: vi
.fn<StructuredAgentSessionAdapter['acquire']>()
.mockRejectedValueOnce(new AgentSessionPreSpawnError(new Error('launch not ready')))
.mockResolvedValueOnce({
.mockImplementationOnce(async ({ fence, spawnToken }) => ({
process: {
hostId: 'local',
pid: 4242,
processStartTimeMs: NOW,
spawnToken: 'spawn-a'
spawnToken
},
link: {
linkId: 'link-1',
handle: { provider: 'codex', threadId: 'thread-1' },
origin: 'created',
mintedAtFence: 1,
mintedAtFence: fence,
observedAt: NOW
}
}),
})),
releaseAcquisition: vi.fn(async () => true)
} as unknown as StructuredAgentSessionAdapter
const input = {
@@ -147,26 +165,28 @@ describe('processless structured session reservation', () => {
}
await expect(performAttach(input)).rejects.toThrow('launch not ready')
expect(store.getRecord(SESSION)?.lease.processlessAt).toBe(NOW)
vi.spyOn(store, 'commitProcessIdentity').mockRejectedValueOnce(new Error('simulated crash'))
await expect(performAttach(input)).rejects.toThrow('simulated crash')
expect(adapter.releaseAcquisition).toHaveBeenCalledWith({ sessionId: SESSION })
const reopened = await AgentSessionRecordStore.open({ directory: storeDir, hostId: 'local' })
expect(reopened.getRecord(SESSION)?.lease.processlessAt).toBeNull()
await reopened.reconcileOnRestart({
probe: createStructuredAgentSessionOwnerProbe('local'),
now: NOW + 1
await expect(performAttach(input)).resolves.toMatchObject({
ok: false,
refusal: { code: 'agent_session_operation_invalid' }
})
// 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: 'released',
expect(adapter.acquire).toHaveBeenCalledOnce()
await expect(
performAttach({
...input,
authority: {
...input.authority,
spawnToken: 'spawn-b',
handoffOperationId: NEXT_OPERATION
},
params: attachParams(NEXT_OPERATION, 2)
})
).resolves.toMatchObject({ ok: true })
expect(store.getRecord(SESSION)?.lease).toMatchObject({
claimStatus: 'live',
handoffStage: null,
runtimeFence: 2,
ownerProcess: null,
reservedSpawnToken: null
runtimeFence: 3,
ownerProcess: { spawnToken: 'spawn-b' }
})
})
})
@@ -18,23 +18,18 @@ 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>
}
) {}
restore(): Promise<void> {
this.restorePromise ??= this.reapThenRestore().catch((error: unknown) => {
this.restorePromise ??= this.restoreReadableSessions().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?.()
private async restoreReadableSessions(): Promise<void> {
await restoreStructuredAgentSessionsOnRestart({
...this.input,
records: this.input.store.listRecords().filter(this.input.supportsRecord)
@@ -118,30 +118,84 @@ afterEach(async () => {
})
describe('recovery exits', () => {
it('frees a reservation stranded by a crash between reserve and identity commit', async () => {
it('keeps an ownerless unproven acquisition in manual recovery across restart', async () => {
acquire.mockRejectedValueOnce(new Error('simulated crash before identity commit'))
await expect(host.attach(CALLER, hostTestAttachParams(null))).rejects.toThrow('simulated crash')
await expect(host.attach(CALLER, hostTestAttachParams(null))).rejects.toThrow(
'agent_session_acquisition_exit_unproven'
)
expect(store.getRecord(SESSION)?.lease).toMatchObject({
claimStatus: 'reserved',
ownerProcess: null
handoffStage: 'manual-recovery',
handoffOperationId: null,
ownerProcess: null,
runtimeFence: 1,
reservedSpawnToken: 'spawn-a'
})
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({
const refused = await host.attach(CALLER, hostTestAttachParams(1))
expect(refused).toMatchObject({
ok: false,
refusal: { code: 'agent_session_checkpoint_stale', currentFence: 2 }
refusal: { code: 'agent_session_ownership_unknown' }
})
const retried = await host.attach(CALLER, hostTestAttachParams(2))
expect(retried).toMatchObject({ ok: true })
expect(acquire).toHaveBeenCalledOnce()
})
it('releases an unproven acquisition whose owner later dies, without replaying it as a handoff', async () => {
// A real session first, so restart restore has a journal to read and runs the
// handoff restorer over the residue instead of skipping the record.
expect((await host.attach(CALLER, hostTestAttachParams(null))).ok).toBe(true)
await reopenStore()
// The resume fails at owner proof and cleanup cannot prove exit: the settlement
// keeps the reservation latched at `recovering` with its committed owner identity.
vi.spyOn(store, 'proveOwner').mockRejectedValueOnce(new Error('handle proof lost'))
openHost({
mintSpawnToken: () => 'spawn-b',
probeOwner: async () => ({ outcome: 'pid-absent' })
})
await expect(host.attach(CALLER, hostTestAttachParams(2))).rejects.toThrow(
'agent_session_acquisition_exit_unproven'
)
expect(store.getRecord(SESSION)?.lease).toMatchObject({
claimStatus: 'reserved',
handoffStage: 'recovering',
handoffOperationId: null,
ownerProcess: { spawnToken: 'spawn-b' },
runtimeFence: 3,
reservedSpawnToken: 'spawn-b'
})
// The unproven owner dies before the next launch; reconciliation and handoff restore run.
await reopenStore()
openHost({
mintSpawnToken: () => 'spawn-c',
probeOwner: async () => ({ outcome: 'pid-absent' })
})
await host.restoreReadableSessions()
// Death proof releases the residue outright: no handoff continuation, no spawned child,
// no stage a user would have to clear by hand.
expect(acquire).toHaveBeenCalledTimes(2)
expect(store.getRecord(SESSION)?.lease).toMatchObject({
claimStatus: 'released',
handoffStage: null,
handoffOperationId: null,
ownerProcess: null,
reservedSpawnToken: null,
runtimeFence: 4
})
// The ordinary native recovery path remains: the first surface hold resumes it.
await host.hold(SESSION, 'surface-1')
expect(acquire).toHaveBeenCalledTimes(3)
expect(store.getRecord(SESSION)?.lease).toMatchObject({
claimStatus: 'live',
handoffStage: null,
ownerProcess: { pid: 4242 }
runtimeFence: 5,
ownerProcess: { spawnToken: 'spawn-c' }
})
})
@@ -106,7 +106,7 @@ function deps(
}
describe('structured session recovery resolution', () => {
it('releases a latched ownerless native reservation', async () => {
it('does not release an ownerless native reservation without processless proof', async () => {
const store = await openStore()
await reserve(store)
await latch(store, 'recovering')
@@ -116,12 +116,12 @@ describe('structured session recovery resolution', () => {
SESSION
)
expect(result).toBe('resolved')
expect(result).toBe('unresolved')
expect(store.getRecord(SESSION)?.lease).toMatchObject({
claimStatus: 'released',
handoffStage: null,
runtimeFence: 2,
reservedSpawnToken: null
claimStatus: 'reserved',
handoffStage: 'recovering',
runtimeFence: 1,
reservedSpawnToken: 'spawn-recovery'
})
})
@@ -30,6 +30,9 @@ export function resolveAgentSessionReplayOutcome<TValue>(input: {
}
}
if (outcome.status === 'unknown') {
if (input.rerunWhenReplayMissing) {
return { decision: 'rerun' }
}
return {
decision: 'refuse',
refusal: {
@@ -0,0 +1,392 @@
// What a client's retry sees around a failed attach settlement: a crash between
// reserve and settlement replays into the original reservation, and a settled
// failure refuses sends without ever re-dispatching on the user's behalf.
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 { computeAgentSessionPayloadFingerprint } from '../../../shared/agent-session-mutation-envelope'
import type { AgentSessionMutationEnvelope } from '../../../shared/agent-session-wire'
import type * as DurableFileWrite from '../../durable-file-write'
import { AgentSessionRecordStore } from '../../runtime/agent-session-record-store'
import type {
AgentSessionDispatchOutcome,
StructuredAgentSessionAdapter
} from './structured-agent-session-adapter'
import { StructuredAgentSessionHost } from './structured-agent-session-host'
import {
HOST_TEST_NOW as NOW,
HOST_TEST_SESSION as SESSION,
HOST_TEST_THREAD as THREAD,
hostTestAttachParams,
hostTestMessage,
hostTestOperationId,
resetHostTestOperationIds
} from './structured-agent-session-host-test-data'
const CALLER = { callerKey: 'client-1' }
const publishFault = vi.hoisted(() => ({ failOnPublish: 0, publishCount: 0 }))
vi.mock('../../durable-file-write', async (importOriginal) => {
const actual = await importOriginal<typeof DurableFileWrite>()
return {
...actual,
renameDurable: async (tmpPath: string, finalPath: string) => {
if (finalPath.endsWith('agent-sessions.json')) {
publishFault.publishCount += 1
}
if (
finalPath.endsWith('agent-sessions.json') &&
publishFault.publishCount === publishFault.failOnPublish
) {
throw new Error('simulated crash before failed-settlement publish')
}
return actual.renameDurable(tmpPath, finalPath)
}
}
})
let root: string
let store: AgentSessionRecordStore
let host: StructuredAgentSessionHost
let acquire: Mock<StructuredAgentSessionAdapter['acquire']>
let releaseAcquisition: Mock<NonNullable<StructuredAgentSessionAdapter['releaseAcquisition']>>
let dispatch: Mock<StructuredAgentSessionAdapter['dispatch']>
function accepted(): AgentSessionDispatchOutcome {
return {
state: 'accepted',
providerIdentity: { provider: 'codex', threadId: THREAD, turnId: 'turn-1', ordinal: 1 }
}
}
function adapter(): StructuredAgentSessionAdapter {
return {
acquire,
releaseAcquisition,
dispatch,
cancelTurn: vi.fn(async () => ({ cancelled: true })),
answerPrompt: vi.fn(async () => undefined),
setOption: vi.fn(async () => undefined)
}
}
function envelope(method: string, fields: Record<string, unknown>): AgentSessionMutationEnvelope {
return {
sessionId: SESSION,
clientOperationId: hostTestOperationId(),
expectedRuntimeFence: store.getRecord(SESSION)?.lease.runtimeFence ?? 1,
payloadFingerprint: computeAgentSessionPayloadFingerprint({
method,
sessionId: SESSION,
fields
})
}
}
beforeEach(async () => {
root = await mkdtemp(join(tmpdir(), 'orca-settled-attach-retry-'))
publishFault.failOnPublish = 0
publishFault.publishCount = 0
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
}
}))
releaseAcquisition = vi.fn(async () => true)
dispatch = vi.fn(async () => accepted())
store = await AgentSessionRecordStore.open({ directory: join(root, 'store'), hostId: 'local' })
host = new StructuredAgentSessionHost({
store,
adapter: adapter(),
journalRoot: root,
claimKeyId: 'key-1',
mintSpawnToken: () => 'spawn-a',
now: () => NOW
})
})
afterEach(async () => {
publishFault.failOnPublish = 0
await host.flushAllStreamedEvents()
await rm(root, { recursive: true, force: true })
})
describe('settled attach retry', () => {
it('settles a post-acquisition journal failure and retries without a restart', async () => {
const historyFilePath = vi
.fn<NonNullable<StructuredAgentSessionAdapter['historyFilePath']>>()
.mockRejectedValueOnce(new Error('journal path unavailable'))
.mockResolvedValue(null)
host = new StructuredAgentSessionHost({
store,
adapter: { ...adapter(), historyFilePath },
journalRoot: root,
claimKeyId: 'key-1',
mintSpawnToken: () => 'spawn-a',
now: () => NOW
})
const first = hostTestAttachParams(null)
await expect(host.attach(CALLER, first)).rejects.toThrow('journal path unavailable')
expect(releaseAcquisition).toHaveBeenCalledOnce()
expect(store.getRecord(SESSION)?.lease).toMatchObject({
claimStatus: 'released',
handoffStage: null,
ownerProcess: null,
reservedSpawnToken: null,
runtimeFence: 2
})
expect(
store.listOperationRows().find((row) => row.operationId === first.envelope.clientOperationId)
?.outcome
).toMatchObject({ status: 'failed' })
await expect(host.attach(CALLER, hostTestAttachParams(2))).resolves.toMatchObject({ ok: true })
expect(acquire).toHaveBeenCalledTimes(2)
expect(store.getRecord(SESSION)?.lease).toMatchObject({
claimStatus: 'live',
handoffStage: null,
runtimeFence: 3
})
})
it('continues a safe same-host pending replay with the reserved token', async () => {
const spawnTokens: string[] = []
acquire.mockImplementation(async ({ fence, spawnToken }) => {
spawnTokens.push(spawnToken)
if (spawnTokens.length === 1) {
throw new Error('reply lost')
}
return {
process: {
hostId: 'local',
pid: 4242,
processStartTimeMs: 1_700_000_000_000,
spawnToken
},
link: {
linkId: `link-${fence}`,
handle: { provider: 'codex', threadId: THREAD },
origin: 'created',
mintedAtFence: fence,
observedAt: NOW
}
}
})
const mintSpawnToken = vi.fn(() => 'spawn-safe')
host = new StructuredAgentSessionHost({
store,
adapter: adapter(),
journalRoot: root,
claimKeyId: 'key-1',
mintSpawnToken,
now: () => NOW
})
const params = hostTestAttachParams(null)
publishFault.failOnPublish = 2
await expect(host.attach(CALLER, params)).rejects.toThrow(
'agent session acquisition failure settlement failed'
)
expect(await host.attach(CALLER, params)).toMatchObject({ ok: true, replayed: true })
expect(mintSpawnToken).toHaveBeenCalledOnce()
expect(spawnTokens).toEqual(['spawn-safe', 'spawn-safe'])
})
it('fences a crash-interrupted reservation replay until positive recovery', async () => {
const spawnTokens: string[] = []
acquire.mockImplementation(async ({ fence, spawnToken }) => {
spawnTokens.push(spawnToken)
if (spawnTokens.length === 1) {
throw new Error('reply lost')
}
return {
process: {
hostId: 'local',
pid: 4242,
processStartTimeMs: 1_700_000_000_000,
spawnToken
},
link: {
linkId: `link-${fence}`,
handle: { provider: 'codex', threadId: THREAD },
origin: 'created',
mintedAtFence: fence,
observedAt: NOW
}
}
})
let token = 0
const mintSpawnToken = vi.fn(() => `spawn-${++token}`)
let reservationUnused = false
host = new StructuredAgentSessionHost({
store,
adapter: adapter(),
journalRoot: root,
claimKeyId: 'key-1',
mintSpawnToken,
probeOwner: async () =>
reservationUnused
? { outcome: 'reservation-unused' }
: { outcome: 'indeterminate', reason: 'spawn token scan unavailable' },
now: () => NOW
})
const params = hostTestAttachParams(null)
publishFault.failOnPublish = 2
await expect(host.attach(CALLER, params)).rejects.toThrow(
'agent session acquisition failure settlement failed'
)
expect(publishFault.publishCount).toBe(2)
expect(store.getRecord(SESSION)?.lease).toMatchObject({
claimStatus: 'reserved',
handoffStage: 'new-owner-proving',
runtimeFence: 1,
reservedSpawnToken: 'spawn-1',
ownerProcess: null
})
await host.flushAllStreamedEvents()
store = await AgentSessionRecordStore.open({ directory: join(root, 'store'), hostId: 'local' })
host = new StructuredAgentSessionHost({
store,
adapter: adapter(),
journalRoot: root,
claimKeyId: 'key-1',
mintSpawnToken,
probeOwner: async () =>
reservationUnused
? { outcome: 'reservation-unused' }
: { outcome: 'indeterminate', reason: 'spawn token scan unavailable' },
now: () => NOW
})
const refused = await host.attach(CALLER, params)
if (refused.ok) {
throw new Error('expected the replayed reservation to stay fenced')
}
expect(refused.refusal.code).toBe('agent_session_ownership_unknown')
expect(acquire).toHaveBeenCalledTimes(1)
expect(releaseAcquisition).toHaveBeenCalledTimes(1)
expect(mintSpawnToken).toHaveBeenCalledTimes(1)
expect(spawnTokens).toEqual(['spawn-1'])
expect(store.getRecord(SESSION)?.lease).toMatchObject({
claimStatus: 'reserved',
handoffStage: 'manual-recovery',
runtimeFence: 1,
reservedSpawnToken: 'spawn-1',
ownerProcess: null
})
expect(
store.listOperationRows().find((row) => row.operationId === params.envelope.clientOperationId)
?.outcome
).toEqual({ status: 'pending' })
reservationUnused = true
await host.hold(SESSION, 'desktop-chat:retry')
expect(mintSpawnToken).toHaveBeenCalledTimes(2)
expect(spawnTokens).toEqual(['spawn-1', 'spawn-2'])
expect(store.getRecord(SESSION)?.lease).toMatchObject({
claimStatus: 'live',
runtimeKind: 'native',
runtimeFence: 3,
handoffStage: null,
handoffOperationId: null,
ownerProcess: { spawnToken: 'spawn-2' }
})
})
it('restores an unknown submission without redispatch before a distinct send', async () => {
expect((await host.attach(CALLER, hostTestAttachParams(null))).ok).toBe(true)
dispatch.mockRejectedValueOnce(new Error('socket closed'))
const body = hostTestMessage('possibly delivered')
const unknownParams = {
envelope: envelope('agentSession.send', { body }),
body
}
const first = await host.send(CALLER, unknownParams)
expect(first).toMatchObject({ ok: true, value: { submission: { dispatchState: 'unknown' } } })
await host.flushAllStreamedEvents()
store = await AgentSessionRecordStore.open({ directory: join(root, 'store'), hostId: 'local' })
host = new StructuredAgentSessionHost({
store,
adapter: adapter(),
journalRoot: root,
claimKeyId: 'key-1',
mintSpawnToken: () => 'spawn-restarted',
probeOwner: async () => ({ outcome: 'pid-absent' }),
now: () => NOW
})
await host.restoreReadableSessions()
await host.hold(SESSION, 'desktop-chat:restart')
expect(store.getRecord(SESSION)?.lease).toMatchObject({
claimStatus: 'live',
handoffStage: null,
runtimeFence: 3
})
expect(dispatch).toHaveBeenCalledTimes(1)
const newBody = hostTestMessage('are you there?')
const sent = await host.send(CALLER, {
envelope: envelope('agentSession.send', { body: newBody }),
body: newBody
})
if (!sent.ok) {
throw new Error(`unexpected restored send refusal: ${sent.refusal.message}`)
}
expect(dispatch).toHaveBeenCalledTimes(2)
const restoredHistory = host.history({ sessionId: SESSION, direction: 'tail' })
if (!restoredHistory.ok) {
throw new Error(`unexpected restored history reset: ${restoredHistory.reset}`)
}
expect(
restoredHistory.page.submissions.find(
(submission) => submission.clientMessageId === unknownParams.envelope.clientOperationId
)?.dispatchState
).toBe('unknown')
const explicitRetry = await host.send(CALLER, {
...unknownParams,
envelope: {
...unknownParams.envelope,
expectedRuntimeFence: store.getRecord(SESSION)?.lease.runtimeFence ?? 3
},
retryUnknown: true
})
expect(explicitRetry).toMatchObject({
ok: true,
value: { submission: { dispatchState: 'accepted' } }
})
expect(dispatch).toHaveBeenCalledTimes(3)
})
it('records proven acquisition cleanup as durable death evidence', async () => {
acquire.mockRejectedValueOnce(new Error('resume rejected'))
await expect(host.attach(CALLER, hostTestAttachParams(null))).rejects.toThrow('resume rejected')
expect(releaseAcquisition).toHaveBeenCalledTimes(1)
expect(store.getRecord(SESSION)?.lease).toMatchObject({
claimStatus: 'released',
deathEvidence: {
kind: 'exit-observed',
detail: 'acquisition cleanup proved no provider child remains'
}
})
})
})
@@ -176,9 +176,8 @@ function isAcquirable(lease: NonNullable<ReturnType<typeof store.getRecord>>['le
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.
// A crash can leave a conflicted current-schema row in manual recovery; positive death proof
// must make it acquirable again without discarding the provider handle.
await seedStore(
wedgedRecord({
claimStatus: 'conflicted',
@@ -0,0 +1,144 @@
import {
agentSessionOperationKey,
settleAgentSessionOperation,
type AgentSessionOperationOutcome
} from '../../shared/agent-session-operation-ledger'
import { nextAgentSessionFence } from '../../shared/agent-session-next-fence'
import type { AgentSessionRecord } from '../../shared/agent-session-record'
import { assertFence, withLease } from './agent-session-lease-transitions'
import type { AgentSessionStoreState } from './agent-session-record-store-file'
export type AgentSessionFailedAcquisitionSettlement = {
sessionId: string
fence: number
spawnToken: string
callerKey: string
operationId: string
outcome: Extract<AgentSessionOperationOutcome, { status: 'failed' }>
exitProof: 'exit-proven' | 'processless' | 'unproven'
now: number
}
export type AgentSessionFailedPostAcquisitionAttachmentSettlement =
AgentSessionFailedAcquisitionSettlement
/** Liveness invariant: a settled attach never leaves its reservation in new-owner-proving. */
export function settleFailedAgentSessionAcquisition(
state: AgentSessionStoreState,
args: AgentSessionFailedAcquisitionSettlement
): AgentSessionRecord {
const operation = state.operations.get(agentSessionOperationKey(args.callerKey, args.operationId))
if (!operation || operation.outcome.status !== 'pending') {
throw new Error('agent_session_operation_conflict')
}
const record = state.records.get(args.sessionId)
if (!record) {
throw new Error('agent_session_identity_required')
}
const next = settleFailedLease(record, args)
state.records.set(args.sessionId, next)
state.operations = settleAgentSessionOperation(state.operations, args)
return next
}
/** A proved native owner still is not publishable until its journal attaches. */
export function settleFailedAgentSessionPostAcquisitionAttachment(
state: AgentSessionStoreState,
args: AgentSessionFailedPostAcquisitionAttachmentSettlement
): AgentSessionRecord {
const operation = state.operations.get(agentSessionOperationKey(args.callerKey, args.operationId))
if (!operation || operation.outcome.status !== 'pending') {
throw new Error('agent_session_operation_conflict')
}
const record = state.records.get(args.sessionId)
if (!record) {
throw new Error('agent_session_identity_required')
}
assertFence(record.lease, args.fence)
if (
record.lease.runtimeKind !== 'native' ||
record.lease.claimStatus !== 'live' ||
record.lease.handoffStage !== null ||
record.lease.ownerProcess?.spawnToken !== args.spawnToken ||
record.lease.reservedSpawnToken !== args.spawnToken ||
record.lease.provenHandleLinkId === null
) {
throw new Error('agent_session_ownership_unknown')
}
const next =
args.exitProof === 'unproven'
? withLease(record, {
...record.lease,
handoffStage: 'recovering',
handoffOperationId: null,
lastRenewedAt: args.now
})
: withLease(record, {
...record.lease,
runtimeFence: nextAgentSessionFence(record.lease),
handoffStage: null,
ownerProcess: null,
reservedSpawnToken: null,
processlessAt: null,
claimStatus: 'released',
lastRenewedAt: args.now,
handoffOperationId: null,
deathEvidence: {
kind: 'exit-observed',
detail: 'post-acquisition cleanup proved no provider child remains',
observedAt: args.now
}
})
state.records.set(args.sessionId, next)
state.operations = settleAgentSessionOperation(state.operations, args)
return next
}
function settleFailedLease(
record: AgentSessionRecord,
args: AgentSessionFailedAcquisitionSettlement
): AgentSessionRecord {
assertFence(record.lease, args.fence)
if (
record.lease.claimStatus !== 'reserved' ||
record.lease.handoffStage !== 'new-owner-proving' ||
record.lease.reservedSpawnToken !== args.spawnToken ||
record.lease.handoffOperationId !== args.operationId
) {
throw new Error('agent_session_ownership_unknown')
}
if (args.exitProof === 'unproven') {
return withLease(record, {
...record.lease,
handoffStage: record.lease.ownerProcess ? 'recovering' : 'manual-recovery',
// The operation is durably settled failed below; a lease still naming it would
// read as an in-flight transfer to every consumer that keys on the stage + id pair.
handoffOperationId: null,
lastRenewedAt: args.now
})
}
return withLease(record, {
...record.lease,
runtimeFence: nextAgentSessionFence(record.lease),
handoffStage: null,
ownerProcess: null,
reservedSpawnToken: null,
processlessAt: null,
claimStatus: 'released',
lastRenewedAt: args.now,
handoffOperationId: null,
deathEvidence:
args.exitProof === 'processless'
? {
kind: 'pid-absent',
detail: 'reservation failed before spawn',
observedAt: args.now
}
: {
// Cleanup proved no child of this attempt remains; it may never have spawned.
kind: 'exit-observed',
detail: 'acquisition cleanup proved no provider child remains',
observedAt: args.now
}
})
}
@@ -18,8 +18,8 @@ function request(overrides: Partial<AgentSessionReserveRequest> = {}): AgentSess
workspaceId: 'workspace-1',
workspaceKind: 'git-worktree'
},
provider: 'claude',
accountHome: { variable: 'CLAUDE_CONFIG_DIR', path: '/home/dev/.claude' },
provider: 'codex',
accountHome: { variable: 'CODEX_HOME', path: '/home/dev/.codex' },
runtimeKind: 'native',
expectedFence: null,
spawnToken: 'spawn-a',
@@ -44,7 +44,7 @@ afterEach(async () => {
await rm(directory, { recursive: true, force: true })
})
describe('legacy agent session launch environment', () => {
describe('agent session launch environment admission', () => {
it('does not persist ambient launch variables', async () => {
const store = await AgentSessionRecordStore.open({ directory, hostId: 'local' })
await store.reserveOwner(request())
@@ -54,7 +54,7 @@ describe('legacy agent session launch environment', () => {
spawnToken: 'spawn-b',
launchEnv: {
PATH: '/custom/bin:/usr/bin',
ANTHROPIC_AUTH_TOKEN: 'fixture-token'
OPENAI_API_KEY: 'fixture-token'
},
operation: {
callerKey: 'client-1',
@@ -65,13 +65,13 @@ describe('legacy agent session launch environment', () => {
)
const raw = await readFile(join(directory, 'agent-sessions.json'), 'utf-8')
expect(raw).not.toContain('ANTHROPIC_AUTH_TOKEN')
expect(raw).not.toContain('OPENAI_API_KEY')
expect(raw).not.toContain('"PATH"')
const reopened = await AgentSessionRecordStore.open({ directory, hostId: 'local' })
expect(reopened.getRecord(SESSION)).not.toHaveProperty('launchEnv')
})
it('rejects an environment that could not be reloaded before writing it', async () => {
it('rejects an environment that could not be validated before writing', async () => {
const store = await AgentSessionRecordStore.open({ directory, hostId: 'local' })
const launchEnv = Object.fromEntries(
Array.from({ length: 257 }, (_, index) => [`KEY_${index}`, 'value'])
@@ -1,7 +1,5 @@
// 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.
// A child spawned under a reservation whose record was lost is invisible to the lease. Reaping
// is bounded cleanup only; it never replaces the lease proof required to grant another writer.
import { describe, expect, it, vi } from 'vitest'
import type { AgentSessionRecordStore } from './agent-session-record-store'
@@ -53,4 +51,18 @@ describe('orphan agent-session child reaper', () => {
expect(stopped).toEqual([])
expect(stop).not.toHaveBeenCalled()
})
it('surfaces a failure to signal an observed orphan', async () => {
const failure = Object.assign(new Error('not permitted'), { code: 'EPERM' })
await expect(
stopOrphanAgentSessionChildren({
store: storeWithLeasedTokens([]),
scan: async () => new Map([['token-lost', [202]]]),
stop: () => {
throw failure
}
})
).rejects.toBe(failure)
})
})
@@ -1,11 +1,11 @@
/**
* Stops provider children that carry an Orca spawn token no lease claims.
* Best-effort stop for 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.
* Only a token match justifies signalling a process; neither age nor CPU is evidence, and a host
* that cannot enumerate tokens stops nothing. This never proves process exit or licenses a new
* owner; lease adjudication remains the single-writer boundary.
*/
import type { AgentSessionRecordStore } from './agent-session-record-store'
@@ -19,8 +19,10 @@ 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.
} catch (error) {
if ((error as NodeJS.ErrnoException).code !== 'ESRCH') {
throw error
}
}
}
@@ -4,8 +4,8 @@
* Pids are reused within minutes on a busy host, and reuse happens precisely in the recovery
* case, so a bare pid match is never proof. Every element of the identity tuple is unavailable
* somewhere — start time costs a CIM query on Windows and is missing in some containers, /proc
* does not exist on macOS — so an unanswerable probe reports `indeterminate` and the lease fails
* closed to manual recovery rather than minting a second writer.
* does not exist on macOS — so an exact but unanswerable identity stays fenced in `recovering`;
* an ownerless, unattributable reservation enters `manual-recovery`.
*/
import { readFile } from 'node:fs/promises'
@@ -15,9 +15,11 @@ import {
isAgentSessionOperationRow,
type AgentSessionOperationRow
} from '../../shared/agent-session-operation-ledger'
import { scrubAgentSessionRecordLaunchEnv } from '../../shared/agent-session-launch-env-scrub'
import type { AgentSessionRecord } from '../../shared/agent-session-record'
import { loadAgentSessionRecord } from '../../shared/agent-session-record-load'
import {
AGENT_SESSION_RECORD_SCHEMA_VERSION,
isAgentSessionRecord,
type AgentSessionRecord
} from '../../shared/agent-session-record'
import {
copyFileDurable,
durableWriteTempPath,
@@ -48,7 +50,7 @@ export type LoadedAgentSessionStore = {
readOnly: boolean
/** True when the primary file was unusable and the previous committed copy was used. */
recoveredFromBackup: boolean
/** True when a legacy row or store needs a durable current-schema rewrite. */
/** True when the normalized current-schema quarantine must be persisted. */
needsRewrite: boolean
}
@@ -108,8 +110,11 @@ function parseState(
return null
}
const schemaVersion = file.schemaVersion as number
if (schemaVersion < AGENT_SESSION_STORE_SCHEMA_VERSION) {
return null
}
if (
schemaVersion <= AGENT_SESSION_STORE_SCHEMA_VERSION &&
schemaVersion === AGENT_SESSION_STORE_SCHEMA_VERSION &&
(typeof file.records !== 'object' || file.records === null || Array.isArray(file.records))
) {
return null
@@ -127,59 +132,57 @@ function parseState(
return null
}
const state = emptyState(hostId)
state.schemaVersion =
schemaVersion <= AGENT_SESSION_STORE_SCHEMA_VERSION
? AGENT_SESSION_STORE_SCHEMA_VERSION
: schemaVersion
state.schemaVersion = schemaVersion
state.hostId = file.hostId
let needsRewrite = schemaVersion < AGENT_SESSION_STORE_SCHEMA_VERSION
let needsRewrite = false
if (typeof file.records === 'object' && file.records !== null) {
for (const [sessionId, value] of Object.entries(file.records)) {
const scrubbed = scrubAgentSessionRecordLaunchEnv(value)
const loaded = loadAgentSessionRecord(scrubbed.value)
needsRewrite ||= scrubbed.changed
if (loaded.status !== 'unusable' && loaded.record.sessionId === sessionId) {
state.records.set(sessionId, loaded.record)
needsRewrite ||= loaded.status === 'upgraded'
const record = isAgentSessionRecord(value) ? value : null
if (record?.sessionId === sessionId) {
state.records.set(sessionId, record)
} else {
state.unreadableRecords.set(sessionId, {
reason: loaded.status === 'unusable' ? loaded.reason : 'record_key_session_id_mismatch',
raw: scrubbed.value
})
needsRewrite ||= schemaVersion <= AGENT_SESSION_STORE_SCHEMA_VERSION
const valueSchemaVersion =
typeof value === 'object' &&
value !== null &&
(value as { schemaVersion?: unknown }).schemaVersion
const reason = record
? 'record_key_session_id_mismatch'
: valueSchemaVersion === AGENT_SESSION_RECORD_SCHEMA_VERSION
? 'current_shape_invalid'
: 'unsupported_schema'
state.unreadableRecords.set(sessionId, { reason, raw: value })
needsRewrite ||= schemaVersion === AGENT_SESSION_STORE_SCHEMA_VERSION
}
}
}
if (typeof file.unusableRecords === 'object' && file.unusableRecords !== null) {
for (const [sessionId, value] of Object.entries(file.unusableRecords)) {
if (typeof value !== 'object' || value === null) {
if (schemaVersion <= AGENT_SESSION_STORE_SCHEMA_VERSION) {
if (schemaVersion === AGENT_SESSION_STORE_SCHEMA_VERSION) {
return null
}
continue
}
const unusable = value as { reason?: unknown; raw?: unknown }
if (typeof unusable.reason !== 'string' || unusable.reason.length === 0) {
if (schemaVersion <= AGENT_SESSION_STORE_SCHEMA_VERSION) {
if (schemaVersion === AGENT_SESSION_STORE_SCHEMA_VERSION) {
return null
}
continue
}
const scrubbed = scrubAgentSessionRecordLaunchEnv(unusable.raw)
needsRewrite ||= scrubbed.changed
state.unreadableRecords.set(sessionId, { reason: unusable.reason, raw: scrubbed.value })
state.unreadableRecords.set(sessionId, { reason: unusable.reason, raw: unusable.raw })
}
}
if (typeof file.operations === 'object' && file.operations !== null) {
for (const [key, value] of Object.entries(file.operations)) {
if (!isAgentSessionOperationRow(value)) {
if (schemaVersion <= AGENT_SESSION_STORE_SCHEMA_VERSION) {
if (schemaVersion === AGENT_SESSION_STORE_SCHEMA_VERSION) {
return null
}
continue
}
if (key !== agentSessionOperationKey(value.callerKey, value.operationId)) {
if (schemaVersion <= AGENT_SESSION_STORE_SCHEMA_VERSION) {
if (schemaVersion === AGENT_SESSION_STORE_SCHEMA_VERSION) {
return null
}
continue
@@ -197,7 +200,7 @@ function parseState(
!Number.isSafeInteger(key.retiredAt) ||
(key.retiredAt as number) < 0
) {
if (schemaVersion <= AGENT_SESSION_STORE_SCHEMA_VERSION) {
if (schemaVersion === AGENT_SESSION_STORE_SCHEMA_VERSION) {
return null
}
continue
@@ -1,102 +0,0 @@
import { mkdtemp, readFile, rm, writeFile } from 'node:fs/promises'
import { tmpdir } from 'node:os'
import { join } from 'node:path'
import { afterEach, beforeEach, describe, expect, it } from 'vitest'
import {
LEGACY_AGENT_SESSION_ID,
legacyAgentSessionRecordV1,
legacyDesktopProviderHandleRecordV1,
legacyMissingAuthorityRecordV1
} from '../../shared/agent-session-record-legacy.test-fixture'
import { AgentSessionRecordStore } from './agent-session-record-store'
import { agentSessionStorePath } from './agent-session-record-store-file'
let directory: string
beforeEach(async () => {
directory = await mkdtemp(join(tmpdir(), 'orca-agent-session-legacy-'))
})
afterEach(async () => {
await rm(directory, { recursive: true, force: true })
})
async function open(): Promise<AgentSessionRecordStore> {
return AgentSessionRecordStore.open({ directory, hostId: 'local' })
}
async function writeLegacyStore(record: Record<string, unknown>): Promise<void> {
await writeFile(
agentSessionStorePath(directory),
JSON.stringify({
schemaVersion: 1,
hostId: 'local',
records: { [LEGACY_AGENT_SESSION_ID]: record },
operations: {},
retiredClaimKeys: []
})
)
}
describe('legacy agent session record store', () => {
it('upgrades a complete v1 identity into durable manual recovery', async () => {
await writeLegacyStore(legacyAgentSessionRecordV1())
const reopened = await open()
const migrated = reopened.getRecord(LEGACY_AGENT_SESSION_ID)
expect(migrated).toMatchObject({
schemaVersion: 2,
lease: {
claimStatus: 'conflicted',
handoffStage: 'manual-recovery',
unreconciled: true
}
})
expect(migrated).not.toHaveProperty('options')
expect(migrated).not.toHaveProperty('launchEnv')
const persisted = JSON.parse(await readFile(agentSessionStorePath(directory), 'utf-8'))
expect(persisted).toMatchObject({
schemaVersion: 2,
records: {
[LEGACY_AGENT_SESSION_ID]: {
schemaVersion: 2,
lease: { claimStatus: 'conflicted', handoffStage: 'manual-recovery' }
}
},
unusableRecords: {}
})
})
it.each([
[
'desktop id as provider handle',
legacyDesktopProviderHandleRecordV1,
'legacy_provider_handle_is_desktop_session'
],
['missing authority fields', legacyMissingAuthorityRecordV1, 'legacy_shape_invalid']
])('durably quarantines v1 rows with %s', async (_name, fixture, reason) => {
await writeLegacyStore(fixture())
const reopened = await open()
expect(reopened.getRecord(LEGACY_AGENT_SESSION_ID)).toBeNull()
expect(reopened.isSessionUnreadable(LEGACY_AGENT_SESSION_ID)).toBe(true)
const persisted = JSON.parse(await readFile(agentSessionStorePath(directory), 'utf-8'))
expect(persisted.records).not.toHaveProperty(LEGACY_AGENT_SESSION_ID)
expect(persisted.unusableRecords[LEGACY_AGENT_SESSION_ID]).toMatchObject({ reason })
})
it('migrates an older store schema while opening it', async () => {
const filePath = agentSessionStorePath(directory)
await writeFile(
filePath,
JSON.stringify({ schemaVersion: 0, hostId: 'local', records: {}, operations: {} })
)
await open()
const persisted = JSON.parse(await readFile(filePath, 'utf-8'))
expect(persisted).toMatchObject({ schemaVersion: 2, unusableRecords: {} })
})
})
@@ -1,9 +1,7 @@
import { chmod, mkdtemp, readFile, rm, stat, writeFile } from 'node:fs/promises'
import { chmod, mkdtemp, rm, stat } from 'node:fs/promises'
import { tmpdir } from 'node:os'
import { join } from 'node:path'
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
import type { AgentSessionRecord } from '../../shared/agent-session-record'
import { createCodexStructuredLaunchResolver } from '../codex/codex-structured-launch-resolution'
import { afterEach, beforeEach, describe, expect, it } from 'vitest'
import { AgentSessionRecordStore } from './agent-session-record-store'
import { agentSessionStorePath } from './agent-session-record-store-file'
import type { AgentSessionReserveRequest } from './agent-session-reservation-admission'
@@ -11,73 +9,6 @@ import type { AgentSessionReserveRequest } from './agent-session-reservation-adm
const NOW = 1_800_000_000_000
let directory: string
function resumableRecord(sessionId: string): AgentSessionRecord {
const threadId = `thread-${sessionId}`
return {
schemaVersion: 2,
sessionId,
location: {
executionHostId: 'local',
wslDistro: null,
workspaceId: `workspace-${sessionId}`,
workspaceKind: 'folder'
},
provider: 'codex',
providerHandleChain: [
{
linkId: `link-${sessionId}`,
handle: { provider: 'codex', threadId },
origin: 'created',
mintedAtFence: 1,
observedAt: NOW
}
],
accountHome: { variable: 'CODEX_HOME', path: `/accounts/${sessionId}` },
launchEnv: {
PATH: '/custom/bin:/usr/bin',
MY_COMPANY_CREDS: 'fixture-credentials',
GITHUB_TOKEN: 'fixture-token'
},
lease: {
sessionId,
runtimeKind: 'native',
runtimeFence: 1,
handoffStage: null,
provenHandleLinkId: null,
ownerProcess: null,
reservedSpawnToken: null,
leaseDeadlineAt: NOW,
lastRenewedAt: NOW,
handoffOperationId: null,
journalCheckpoint: null,
claimKeyId: 'key-1',
claimStatus: 'released',
unreconciled: false,
deathEvidence: null
},
createdAt: NOW,
updatedAt: NOW
}
}
function storePayload(): string {
const records = Object.fromEntries(
['session-live-a', 'session-live-b'].map((sessionId) => [
sessionId,
{ ...resumableRecord(sessionId), futureRecordField: { preserved: true } }
])
)
return JSON.stringify({
schemaVersion: 2,
hostId: 'local',
records,
operations: {},
retiredClaimKeys: [],
unusableRecords: {},
futureStoreField: { preserved: true }
})
}
function reserveRequest(): AgentSessionReserveRequest {
return {
sessionId: 'session-created',
@@ -89,7 +20,6 @@ function reserveRequest(): AgentSessionReserveRequest {
},
provider: 'codex',
accountHome: { variable: 'CODEX_HOME', path: '/accounts/created' },
launchEnv: { PATH: '/usr/bin' },
runtimeKind: 'native',
expectedFence: null,
spawnToken: 'spawn-created',
@@ -114,80 +44,6 @@ afterEach(async () => {
})
describe('agent session record store security', () => {
it('scrubs both committed generations without losing resumable sessions', async () => {
const filePath = agentSessionStorePath(directory)
await Promise.all([
writeFile(filePath, storePayload()),
writeFile(`${filePath}.bak`, storePayload())
])
if (process.platform !== 'win32') {
await chmod(directory, 0o755)
await Promise.all([chmod(filePath, 0o644), chmod(`${filePath}.bak`, 0o644)])
}
const store = await AgentSessionRecordStore.open({ directory, hostId: 'local' })
expect(store.readOnly).toBe(false)
expect(
store
.listRecords()
.map((record) => record.sessionId)
.sort()
).toEqual(['session-live-a', 'session-live-b'])
for (const committedPath of [filePath, `${filePath}.bak`]) {
const raw = await readFile(committedPath, 'utf-8')
expect(raw).not.toContain('MY_COMPANY_CREDS')
expect(raw).not.toContain('GITHUB_TOKEN')
const parsed = JSON.parse(raw) as {
records: Record<string, AgentSessionRecord & { futureRecordField?: unknown }>
futureStoreField?: unknown
}
const { records } = parsed
expect(Object.keys(records).sort()).toEqual(['session-live-a', 'session-live-b'])
expect(records['session-live-a']).not.toHaveProperty('launchEnv')
expect(parsed.futureStoreField).toEqual({ preserved: true })
expect(records['session-live-a']?.futureRecordField).toEqual({ preserved: true })
if (process.platform !== 'win32') {
expect((await stat(committedPath)).mode & 0o777).toBe(0o600)
}
}
if (process.platform !== 'win32') {
expect((await stat(directory)).mode & 0o777).toBe(0o700)
}
const resolveCommand = vi.fn(() => '/custom/bin/codex')
const resolveLaunch = createCodexStructuredLaunchResolver({
store,
resolveWorkspacePath: async (workspaceId) => `/workspaces/${workspaceId}`,
resolveCommand,
resolveEnvironment: async () => ({
PATH: '/fresh/bin:/usr/bin',
MY_COMPANY_CREDS: 'fresh-fixture-credentials'
})
})
for (const sessionId of ['session-live-a', 'session-live-b']) {
const launch = await resolveLaunch({
identity: {
sessionId,
workspaceId: `workspace-${sessionId}`,
hostId: 'local',
agent: 'codex',
providerHandle: { kind: 'codex', threadId: `thread-${sessionId}` }
}
})
expect(launch).toMatchObject({
command: '/custom/bin/codex',
codexHome: `/accounts/${sessionId}`,
resumeThreadId: `thread-${sessionId}`,
env: {
PATH: '/fresh/bin:/usr/bin',
MY_COMPANY_CREDS: 'fresh-fixture-credentials'
}
})
}
expect(resolveCommand).toHaveBeenCalledTimes(2)
expect(resolveCommand).toHaveBeenCalledWith({ pathEnv: '/fresh/bin:/usr/bin' })
})
it.skipIf(process.platform === 'win32')(
'creates the directory and store owner-only',
async () => {
@@ -1,7 +1,5 @@
import { chmod, mkdir, readFile, rm } from 'node:fs/promises'
import { chmod, mkdir } from 'node:fs/promises'
import { dirname } from 'node:path'
import { scrubAgentSessionRecordLaunchEnv } from '../../shared/agent-session-launch-env-scrub'
import { durableWriteTempPath, renameDurable, writeTempFileDurable } from '../durable-file-write'
import {
loadAgentSessionStore,
type LoadedAgentSessionStore
@@ -11,64 +9,6 @@ import { withAgentSessionStoreTransactionLock } from './agent-session-store-tran
const OWNER_DIRECTORY_MODE = 0o700
const OWNER_FILE_MODE = 0o600
type PreparedScrub = { filePath: string; tmpPath: string; published: boolean }
function scrubRecordMap(value: unknown): { value: unknown; changed: boolean } {
if (typeof value !== 'object' || value === null || Array.isArray(value)) {
return { value, changed: false }
}
let changed = false
const entries = Object.entries(value).map(([sessionId, record]) => {
const scrubbed = scrubAgentSessionRecordLaunchEnv(record)
changed ||= scrubbed.changed
return [sessionId, scrubbed.value]
})
return { value: changed ? Object.fromEntries(entries) : value, changed }
}
function scrubUnreadableRecordMap(value: unknown): { value: unknown; changed: boolean } {
if (typeof value !== 'object' || value === null || Array.isArray(value)) {
return { value, changed: false }
}
let changed = false
const entries = Object.entries(value).map(([sessionId, unusable]) => {
if (typeof unusable !== 'object' || unusable === null || Array.isArray(unusable)) {
return [sessionId, unusable]
}
const row = unusable as Record<string, unknown>
const scrubbed = scrubAgentSessionRecordLaunchEnv(row.raw)
changed ||= scrubbed.changed
return [sessionId, scrubbed.changed ? { ...row, raw: scrubbed.value } : row]
})
return { value: changed ? Object.fromEntries(entries) : value, changed }
}
function scrubStorePayload(raw: string): string | null {
let parsed: unknown
try {
parsed = JSON.parse(raw) as unknown
} catch {
return null
}
if (typeof parsed !== 'object' || parsed === null || Array.isArray(parsed)) {
return null
}
const store = parsed as Record<string, unknown>
if (!Number.isSafeInteger(store.schemaVersion) || (store.schemaVersion as number) < 0) {
return null
}
const records = scrubRecordMap(store.records)
const unreadableRecords = scrubUnreadableRecordMap(store.unusableRecords)
if (!records.changed && !unreadableRecords.changed) {
return null
}
return JSON.stringify({
...store,
...(records.changed ? { records: records.value } : {}),
...(unreadableRecords.changed ? { unusableRecords: unreadableRecords.value } : {})
})
}
async function chmodIfPresent(path: string, mode: number): Promise<void> {
try {
await chmod(path, mode)
@@ -89,56 +29,11 @@ export async function hardenAgentSessionStorePermissions(filePath: string): Prom
])
}
async function prepareScrub(filePath: string): Promise<PreparedScrub | null> {
let raw: string
try {
raw = await readFile(filePath, 'utf-8')
} catch (error) {
if ((error as NodeJS.ErrnoException).code === 'ENOENT') {
return null
}
throw error
}
const payload = scrubStorePayload(raw)
if (payload === null) {
return null
}
const tmpPath = durableWriteTempPath(filePath)
await writeTempFileDurable(tmpPath, payload, OWNER_FILE_MODE)
return { filePath, tmpPath, published: false }
}
/** Scrubs each committed generation independently so backup-only sessions survive. */
export async function scrubAgentSessionStoreFiles(filePath: string): Promise<void> {
const prepared: PreparedScrub[] = []
try {
for (const candidate of [`${filePath}.bak`, filePath]) {
const scrub = await prepareScrub(candidate)
if (scrub) {
prepared.push(scrub)
}
}
// Both replacements are durable first; publishing backup then live never removes the live path.
for (const scrub of prepared) {
await renameDurable(scrub.tmpPath, scrub.filePath)
scrub.published = true
}
} finally {
await Promise.all(
prepared
.filter((scrub) => !scrub.published)
.map((scrub) => rm(scrub.tmpPath, { force: true }).catch(() => {}))
)
}
}
export async function loadProtectedAgentSessionStore(
filePath: string,
hostId: string
): Promise<LoadedAgentSessionStore> {
return withAgentSessionStoreTransactionLock(filePath, async () => {
await hardenAgentSessionStorePermissions(filePath)
await scrubAgentSessionStoreFiles(filePath)
await hardenAgentSessionStorePermissions(filePath)
return loadAgentSessionStore(filePath, hostId)
})
+24 -15
View File
@@ -29,6 +29,12 @@ import {
setAgentSessionJournalCheckpoint,
type AgentSessionProcessIdentityCommit
} from './agent-session-lease-transitions'
import {
settleFailedAgentSessionAcquisition,
settleFailedAgentSessionPostAcquisitionAttachment,
type AgentSessionFailedAcquisitionSettlement,
type AgentSessionFailedPostAcquisitionAttachmentSettlement
} from './agent-session-acquisition-failure-settlement'
import {
renewAgentSessionLeases,
type AgentSessionLeaseRenewal
@@ -41,6 +47,7 @@ import {
type AgentSessionReservationProcesslessProof
} from './agent-session-processless-reservation'
import {
admitPendingAgentSessionReservationReplay,
applyAgentSessionReservation,
evaluateAgentSessionReserveOperation,
requireAgentSessionRecordForReplay,
@@ -80,7 +87,7 @@ export class AgentSessionRecordStore {
diskRevision
)
if (loaded.needsRewrite && !loaded.readOnly && !loaded.recoveredFromBackup) {
await transactions.persistLoadedMigration()
await transactions.persistLoadedRewrite()
}
return new AgentSessionRecordStore(transactions)
}
@@ -101,13 +108,10 @@ export class AgentSessionRecordStore {
return this.state.hostId
}
getRecord(sessionId: string): AgentSessionRecord | null {
return this.state.records.get(sessionId) ?? null
}
getRecord = (sessionId: string): AgentSessionRecord | null =>
this.state.records.get(sessionId) ?? null
listRecords(): AgentSessionRecord[] {
return [...this.state.records.values()]
}
listRecords = (): AgentSessionRecord[] => [...this.state.records.values()]
listByScope(location: AgentSessionExecutionLocation): AgentSessionRecord[] {
const scope = agentSessionScopeKey(location)
@@ -119,9 +123,7 @@ export class AgentSessionRecordStore {
return this.state.unreadableRecords.has(sessionId)
}
listOperationRows(): AgentSessionOperationRow[] {
return [...this.state.operations.values()]
}
listOperationRows = (): AgentSessionOperationRow[] => [...this.state.operations.values()]
isClaimKeyVerifiable(keyId: string, now: number): boolean {
const retired = this.state.retiredClaimKeys.find((entry) => entry.keyId === keyId)
@@ -147,11 +149,10 @@ export class AgentSessionRecordStore {
throw new Error(decision.code)
}
if (decision.decision === 'replay') {
const record = requireAgentSessionRecordForReplay(
this.state,
decision.row,
request.sessionId
)
let record = requireAgentSessionRecordForReplay(this.state, decision.row, request.sessionId)
if (decision.row.outcome.status === 'pending' && request.handoffOperationId !== null) {
record = admitPendingAgentSessionReservationReplay(record, request)
}
return { record, disposition: 'replayed' as const, operationRow: decision.row }
}
const result = applyAgentSessionReservation(this.state, request, AGENT_SESSION_LEASE_TTL_MS)
@@ -201,6 +202,14 @@ export class AgentSessionRecordStore {
})
}
/** Settle the failed attach and its reservation in one durable transaction. */
settleFailedAcquisition = (args: AgentSessionFailedAcquisitionSettlement) =>
this.transact(() => settleFailedAgentSessionAcquisition(this.state, args))
settleFailedPostAcquisitionAttachment = (
args: AgentSessionFailedPostAcquisitionAttachmentSettlement
) => this.transact(() => settleFailedAgentSessionPostAcquisitionAttachment(this.state, args))
async renewLease(args: AgentSessionLeaseRenewal): Promise<AgentSessionRecord> {
const [renewed] = await this.renewLeases([args])
return renewed
@@ -0,0 +1,97 @@
import { mkdtemp, readFile, rm, writeFile } from 'node:fs/promises'
import { tmpdir } from 'node:os'
import { join } from 'node:path'
import { afterEach, beforeEach, describe, expect, it } from 'vitest'
import { agentSessionRecordFixture } from '../../shared/agent-session-record.test-fixture'
import { AgentSessionRecordStore } from './agent-session-record-store'
import {
AGENT_SESSION_STORE_SCHEMA_VERSION,
agentSessionStorePath
} from './agent-session-record-store-file'
import type { AgentSessionReserveRequest } from './agent-session-reservation-admission'
const NOW = 1_800_000_000_000
const SESSION_ID = 'session-alpha-1'
let directory: string
function reserveRequest(): AgentSessionReserveRequest {
return {
sessionId: SESSION_ID,
location: {
executionHostId: 'local',
wslDistro: null,
workspaceId: 'workspace-1',
workspaceKind: 'git-worktree'
},
provider: 'codex',
accountHome: { variable: 'CODEX_HOME', path: '/home/user/.codex' },
runtimeKind: 'native',
expectedFence: null,
spawnToken: 'spawn-new',
claimKeyId: 'key-1',
handoffOperationId: null,
probe: { outcome: 'reservation-unused' },
operation: {
callerKey: 'client-1',
operationId: `${NOW}-00000000000000000000000000000001`,
fingerprint: 'fp-1'
},
now: NOW
}
}
beforeEach(async () => {
directory = await mkdtemp(join(tmpdir(), 'orca-agent-session-unsupported-'))
})
afterEach(async () => {
await rm(directory, { recursive: true, force: true })
})
describe('unsupported agent session record schema', () => {
it('quarantines without upgrading and keeps the session fail-closed', async () => {
const filePath = agentSessionStorePath(directory)
const unsupported = { ...agentSessionRecordFixture(), schemaVersion: 1 }
const payload = JSON.stringify({
schemaVersion: AGENT_SESSION_STORE_SCHEMA_VERSION,
hostId: 'local',
records: { [SESSION_ID]: unsupported },
operations: {},
retiredClaimKeys: [],
unusableRecords: {}
})
await Promise.all([writeFile(filePath, payload), writeFile(`${filePath}.bak`, payload)])
const store = await AgentSessionRecordStore.open({ directory, hostId: 'local' })
expect(store.getRecord(SESSION_ID)).toBeNull()
expect(store.isSessionUnreadable(SESSION_ID)).toBe(true)
await expect(store.reserveOwner(reserveRequest())).rejects.toThrow(
'execution_owner_reconciling'
)
const persisted = JSON.parse(await readFile(filePath, 'utf-8'))
expect(persisted.records).not.toHaveProperty(SESSION_ID)
expect(persisted.unusableRecords[SESSION_ID]).toMatchObject({
reason: 'unsupported_schema',
raw: { schemaVersion: 1 }
})
})
it('rejects an ad-hoc store schema without rewriting it', async () => {
const filePath = agentSessionStorePath(directory)
const payload = JSON.stringify({
schemaVersion: 1,
hostId: 'local',
records: {},
operations: {},
retiredClaimKeys: [],
unusableRecords: {}
})
await writeFile(filePath, payload)
await expect(AgentSessionRecordStore.open({ directory, hostId: 'local' })).rejects.toThrow(
'agent_session_store_corrupt'
)
await expect(readFile(filePath, 'utf-8')).resolves.toBe(payload)
})
})
@@ -71,7 +71,7 @@ describe('backup recovery publication', () => {
)
publishFault.armed = true
await expect(queue.persistLoadedMigration()).rejects.toThrow(
await expect(queue.persistLoadedRewrite()).rejects.toThrow(
'simulated death before primary publish'
)
@@ -12,7 +12,10 @@ import {
type AgentSessionOperationDecision,
type AgentSessionOperationRow
} from '../../shared/agent-session-operation-ledger'
import type { AgentSessionOwnerProbe } from '../../shared/agent-session-lease-adjudication'
import {
evaluateAgentSessionAcquisition,
type AgentSessionOwnerProbe
} from '../../shared/agent-session-lease-adjudication'
import {
AGENT_SESSION_RECORD_SCHEMA_VERSION,
agentSessionExecutionLocationsEqual,
@@ -36,13 +39,15 @@ export type AgentSessionReserveRequest = {
location: AgentSessionExecutionLocation
provider: AgentSessionHandleProvider
accountHome: AgentSessionAccountHome
/** Legacy internal input accepted for compatibility and deliberately discarded. */
/** Arguments pinned on first reservation so owner replacement repeats the same launch. */
launchArgs?: AgentSessionLaunchArgs
/** Current launch input validated here but never written to the durable record. */
launchEnv?: AgentSessionLaunchEnv
runtimeKind: AgentSessionReservation['runtimeKind']
/** Null when the session does not exist yet; otherwise the fence the caller last observed. */
expectedFence: number | null
spawnToken: string
/** A supplier is invoked only when this operation wins a new reservation. */
spawnToken: string | (() => string)
claimKeyId: string
handoffOperationId: string | null
probe: AgentSessionOwnerProbe
@@ -92,6 +97,26 @@ export function requireAgentSessionRecordForReplay(
return record
}
export function admitPendingAgentSessionReservationReplay(
record: AgentSessionRecord,
request: AgentSessionReserveRequest
): AgentSessionRecord {
const decision = evaluateAgentSessionAcquisition({
lease: record.lease,
expectedFence: record.lease.runtimeFence,
handoffOperationId: request.handoffOperationId,
probe: request.probe
})
if (decision.decision === 'refused') {
throw new Error(decision.code)
}
if (decision.decision !== 'retry-reservation') {
// A replay may continue only its still-present reservation; recovery requires a fresh intent.
throw new Error('agent_session_ownership_unknown')
}
return record
}
export function applyAgentSessionReservation(
state: AgentSessionStoreState,
request: AgentSessionReserveRequest,
@@ -108,7 +133,8 @@ export function applyAgentSessionReservation(
}
const reservation: AgentSessionReservation = {
runtimeKind: request.runtimeKind,
spawnToken: request.spawnToken,
spawnToken:
typeof request.spawnToken === 'function' ? request.spawnToken() : request.spawnToken,
claimKeyId: request.claimKeyId,
handoffOperationId: request.handoffOperationId,
leaseTtlMs: request.leaseTtlMs ?? leaseTtlMs,
@@ -65,9 +65,7 @@ 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.
it('routes an ownerless indeterminate reservation to manual recovery at the same fence', () => {
const abandoned = record('new-owner-proving')
abandoned.lease.runtimeKind = 'native'
abandoned.lease.ownerProcess = null
@@ -77,18 +75,61 @@ describe('restarted handoff adjudication', () => {
{ outcome: 'indeterminate', reason: 'no spawn-token scan' },
NOW + 1_000
).lease
).toMatchObject({
runtimeKind: 'native',
runtimeFence: 4,
handoffStage: 'manual-recovery',
handoffOperationId: 'handoff-op-1',
claimStatus: 'reserved',
ownerProcess: null,
reservedSpawnToken: 'spawn-restart'
})
})
it('releases a proving reservation only with durable processless proof', () => {
const processless = record('new-owner-proving')
processless.lease.runtimeKind = 'native'
processless.lease.ownerProcess = null
processless.lease.processlessAt = NOW
expect(
adjudicateRestartedAgentSessionHandoff(
processless,
{ outcome: 'reservation-unused' },
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', () => {
it.each([
['dead pid', { outcome: 'pid-absent' } as const],
[
'reused pid with a different start time',
{ outcome: 'identity-mismatch', field: 'process-start-time' } as const
]
])('releases a proving owner with exact %s proof', (_name, probe) => {
const proving = record('new-owner-proving')
proving.lease.runtimeKind = 'native'
expect(adjudicateRestartedAgentSessionHandoff(proving, probe, NOW + 1_000).lease).toMatchObject(
{
runtimeKind: 'native',
runtimeFence: 5,
handoffStage: null,
handoffOperationId: null,
claimStatus: 'released',
ownerProcess: null,
reservedSpawnToken: null
}
)
})
it('preserves the existing TUI handoff rollback after proving its target dead', () => {
expect(
adjudicateRestartedAgentSessionHandoff(
record('new-owner-proving'),
@@ -102,4 +143,30 @@ describe('restarted handoff adjudication', () => {
handoffOperationId: 'handoff-op-1'
})
})
it.each([
[
'live exact identity',
{ outcome: 'identity-matched', matchedOn: ['process-start-time'] } as const,
'recovering'
],
[
'indeterminate identity',
{ outcome: 'indeterminate', reason: 'start time unavailable' } as const,
'recovering'
]
] as const)('keeps the fence and token for a %s', (_name, probe, expectedStage) => {
const proving = record('new-owner-proving')
proving.lease.runtimeKind = 'native'
expect(adjudicateRestartedAgentSessionHandoff(proving, probe, NOW + 1_000).lease).toMatchObject(
{
runtimeFence: 4,
handoffStage: expectedStage,
handoffOperationId: 'handoff-op-1',
claimStatus: 'reserved',
ownerProcess: { pid: 4242, processStartTimeMs: NOW - 1_000 },
reservedSpawnToken: 'spawn-restart'
}
)
})
})
@@ -30,20 +30,16 @@ export function adjudicateRestartedAgentSessionHandoff(
if (adjudication.disposition !== 'evicted') {
return updateLease(record, {
...record.lease,
handoffStage: 'recovering',
handoffStage:
adjudication.disposition === 'conflicted' ? 'manual-recovery' : adjudication.stage,
claimStatus:
adjudication.disposition === 'conflicted' ? 'conflicted' : record.lease.claimStatus,
unreconciled: false,
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.
if (record.lease.handoffStage === 'new-owner-proving' && record.lease.runtimeKind === 'native') {
// The attempted new owner is proven absent; no writer remains to roll back or readopt.
return updateLease(record, {
...record.lease,
runtimeFence: adjudication.nextFence,
@@ -58,14 +54,11 @@ export function adjudicateRestartedAgentSessionHandoff(
deathEvidence: adjudication.evidence
})
}
const provingTarget = record.lease.handoffStage === 'new-owner-proving'
const provingTuiTarget =
record.lease.handoffStage === 'new-owner-proving' && record.lease.runtimeKind === 'tui'
return updateLease(record, {
...record.lease,
runtimeKind: provingTarget
? record.lease.runtimeKind === 'native'
? 'tui'
: 'native'
: record.lease.runtimeKind,
runtimeKind: provingTuiTarget ? 'native' : record.lease.runtimeKind,
runtimeFence: adjudication.nextFence,
handoffStage: 'old-owner-stopped',
ownerProcess: null,
@@ -138,7 +138,7 @@ export class AgentSessionStoreTransactionQueue {
return run
}
persistLoadedMigration(): Promise<void> {
persistLoadedRewrite(): Promise<void> {
return this.transact(() => undefined)
}
@@ -1,10 +1,17 @@
import { describe, expect, it, vi } from 'vitest'
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 {
AgentSessionClaimStatus,
AgentSessionProcessIdentity,
AgentSessionRecord
} from '../../shared/agent-session-record'
import { createStructuredAgentSessionOwnerProbe } from './structured-agent-session-runtime'
import {
createStructuredAgentSessionOwnerProbe,
ensureStructuredAgentSessionHost,
stopStructuredAgentSessionRuntime
} from './structured-agent-session-runtime'
const HOST_ID = 'local'
@@ -125,3 +132,71 @@ describe('structured agent-session owner probe', () => {
expect(result).toEqual({ outcome: 'reservation-unused' })
})
})
describe('structured agent-session runtime install', () => {
let stateDirectory: string | null = null
afterEach(async () => {
await stopStructuredAgentSessionRuntime()
if (stateDirectory) {
await rm(stateDirectory, { recursive: true, force: true })
stateDirectory = null
}
vi.restoreAllMocks()
})
it('starts orphan reaping and reports failures without failing installation', async () => {
stateDirectory = await mkdtemp(join(tmpdir(), 'orca-structured-runtime-'))
const failure = new Error('scan failed')
const reapOrphanChildren = vi.fn(async () => {
throw failure
})
const onError = vi.fn()
await expect(
ensureStructuredAgentSessionHost({
stateDirectory,
hostId: HOST_ID,
claimKeyId: 'key-1',
resolveWorkspacePath: async () => stateDirectory!,
resolveEnvironment: async () => ({}),
reapOrphanChildren,
onError
})
).resolves.toBeDefined()
await vi.waitFor(() =>
expect(onError).toHaveBeenCalledWith({
scope: 'agent-session-orphan-child-reaper',
error: failure
})
)
expect(reapOrphanChildren).toHaveBeenCalledWith({ store: expect.anything() })
})
it('logs an orphan-reaper failure when no reporter is configured', async () => {
stateDirectory = await mkdtemp(join(tmpdir(), 'orca-structured-runtime-'))
const failure = new Error('scan failed')
const consoleError = vi.spyOn(console, 'error').mockImplementation(() => {})
await expect(
ensureStructuredAgentSessionHost({
stateDirectory,
hostId: HOST_ID,
claimKeyId: 'key-1',
resolveWorkspacePath: async () => stateDirectory!,
resolveEnvironment: async () => ({}),
reapOrphanChildren: async () => {
throw failure
}
})
).resolves.toBeDefined()
await vi.waitFor(() =>
expect(consoleError).toHaveBeenCalledWith(
'[structured-agent-session] orphan reaper failed',
failure
)
)
})
})
@@ -19,6 +19,7 @@ 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 { stopOrphanAgentSessionChildren } from './agent-session-orphan-child-reaper'
import {
probeAgentSessionProcessIdentities,
probeAgentSessionProcessIdentity,
@@ -55,6 +56,7 @@ export type StructuredAgentSessionRuntimeDeps = {
resolveCodexOverrides?: () => NodeJS.ProcessEnv
onError?: (input: { scope: string; error: unknown }) => void
handoffTransport?: StructuredAgentSessionHandoffTransport
reapOrphanChildren?: typeof stopOrphanAgentSessionChildren
}
type InstalledRuntime = {
@@ -109,6 +111,21 @@ async function install(deps: StructuredAgentSessionRuntimeDeps): Promise<Install
hostId: deps.hostId
})
agentSessionPtyWriteGate.attachRecordLookup((sessionId) => store.getRecord(sessionId))
// Why: only the durable store can identify a provider child lost before record publication.
void (deps.reapOrphanChildren ?? stopOrphanAgentSessionChildren)({ store }).catch((error) => {
try {
if (deps.onError) {
deps.onError({ scope: 'agent-session-orphan-child-reaper', error })
} else {
console.error('[structured-agent-session] orphan reaper failed', error)
}
} catch (reportingError) {
console.error(
'[structured-agent-session] orphan reaper error reporting failed',
reportingError
)
}
})
try {
const codex = new CodexStructuredSessionAdapter({
resolveLaunch: createCodexStructuredLaunchResolver({
@@ -153,8 +170,8 @@ async function install(deps: StructuredAgentSessionRuntimeDeps): Promise<Install
/**
* The lease's only source of truth about a previous owner. Everything it cannot
* answer PID-reuse-safely reports `indeterminate`, which routes the session to
* manual recovery instead of minting a second writer on the same Codex thread.
* answer PID-reuse-safely reports `indeterminate`. An exact owner stays fenced in `recovering`;
* an ownerless, unattributable reservation enters `manual-recovery`.
*/
export function createStructuredAgentSessionOwnerProbe(
hostId: string,
@@ -1,14 +0,0 @@
export function scrubAgentSessionRecordLaunchEnv(value: unknown): {
value: unknown
changed: boolean
} {
if (typeof value !== 'object' || value === null || Array.isArray(value)) {
return { value, changed: false }
}
const record = value as Record<string, unknown>
if (!Object.hasOwn(record, 'launchEnv')) {
return { value, changed: false }
}
const { launchEnv: _discarded, ...rest } = record
return { value: rest, changed: true }
}
@@ -152,22 +152,25 @@ describe('acquisition compare-and-swap', () => {
})
})
it('refuses a different handoff operation and replays the matching one', () => {
const mid = lease({
handoffStage: 'old-owner-stopped',
handoffOperationId: 'op-1',
ownerProcess: null,
claimStatus: 'reserved'
})
expect(acquire(mid, { outcome: 'reservation-unused' }, 'op-2')).toEqual({
decision: 'refused',
code: 'agent_session_operation_conflict'
})
expect(acquire(mid, { outcome: 'reservation-unused' }, 'op-1')).toEqual({
decision: 'retry-reservation',
fence: 7
})
})
it.each(['old-owner-stopped', 'new-owner-proving'] as const)(
'refuses a different handoff operation and replays the matching one at %s',
(handoffStage) => {
const mid = lease({
handoffStage,
handoffOperationId: 'op-1',
ownerProcess: null,
claimStatus: 'reserved'
})
expect(acquire(mid, { outcome: 'reservation-unused' }, 'op-2')).toEqual({
decision: 'refused',
code: 'agent_session_operation_conflict'
})
expect(acquire(mid, { outcome: 'reservation-unused' }, 'op-1')).toEqual({
decision: 'retry-reservation',
fence: 7
})
}
)
it('refuses a reservation whose spawn may have won the race with the crash', () => {
const reserved = lease({ ownerProcess: null, claimStatus: 'reserved', handoffStage: null })
@@ -218,7 +221,9 @@ describe('restart reconciliation', () => {
})
})
it('sends an unverifiable owner to recovery instead of evicting it', () => {
it('keeps re-asking about an unverifiable owner instead of evicting it', () => {
// A recorded exact identity can still be probed later; manual recovery is reserved
// for leases that name no process at all.
expect(
adjudicateAgentSessionRestart({
lease: lease({ leaseDeadlineAt: 1 }),
@@ -281,13 +286,15 @@ describe('restart reconciliation', () => {
).toEqual({ disposition: 'free', reason: 'lease has no owner and no reservation' })
})
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.
it('does not infer an ownerless native reservation is unused from restart alone', () => {
const reserved = lease({ ownerProcess: null, claimStatus: 'reserved' })
expect(
adjudicateAgentSessionRestart({ lease: reserved, probe: INDETERMINATE, observedAt: 9_000 })
).toMatchObject({ disposition: 'evicted', nextFence: 8 })
).toEqual({
disposition: 'recovering',
stage: 'manual-recovery',
reason: 'reservation with no proven process'
})
})
it('frees a TUI reservation only when a probe proves nothing ever spawned', () => {
@@ -302,7 +309,7 @@ describe('restart reconciliation', () => {
).toMatchObject({ disposition: 'evicted', nextFence: 8 })
expect(
adjudicateAgentSessionRestart({ lease: reserved, probe: INDETERMINATE, observedAt: 9_000 })
).toMatchObject({ disposition: 'recovering' })
).toMatchObject({ disposition: 'recovering', stage: 'manual-recovery' })
})
})
+7 -19
View File
@@ -2,8 +2,9 @@
* Single-writer lease adjudication.
*
* Every decision here fails closed: expiry alone never grants a second owner, an unverifiable
* process counts as possibly alive, and a stage that cannot prove an owner ends in manual
* recovery rather than handing the session to the other runtime. This is the opposite polarity
* process counts as possibly alive, and a stage that cannot prove an owner keeps re-asking —
* or, when it names no process at all, ends in manual recovery — rather than handing the
* session to the other runtime. This is the opposite polarity
* from daemon adoption checks, which fail open on a missing start time — a wrong answer there
* refuses an adoption, a wrong answer here creates two writers on one provider session.
*/
@@ -142,7 +143,7 @@ export function evaluateAgentSessionAcquisition(args: {
}
if (
lease.ownerProcess === null &&
lease.handoffStage === 'old-owner-stopped' &&
STAGES_ADMITTING_NEW_OWNER.has(lease.handoffStage) &&
lease.claimStatus === 'reserved' &&
lease.reservedSpawnToken !== null
) {
@@ -210,24 +211,9 @@ 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: nextAgentSessionFence(lease),
evidence: {
kind: 'pid-absent',
detail: 'native reservation abandoned before an owner was proven',
observedAt
}
}
}
return {
disposition: 'recovering',
stage: 'recovering',
stage: 'manual-recovery',
reason: 'reservation with no proven process'
}
}
@@ -249,6 +235,8 @@ export function adjudicateAgentSessionRestart(args: {
return { disposition: 'evicted', nextFence: nextAgentSessionFence(lease), evidence }
}
return {
// Why: an exact recorded identity can still be probed later, so the system keeps
// re-asking; only a record naming nobody (above) needs the user to decide.
disposition: 'recovering',
stage: 'recovering',
reason:
@@ -1,79 +0,0 @@
/** Anonymized derivatives of schema-1 rows observed across ad-hoc structured-session builds. */
export const LEGACY_AGENT_SESSION_ID = 'desktop_mstny52oa2243469'
export const LEGACY_CODEX_THREAD_ID = '01a002e9-9a1c-7d42-a642-e481f64446f1'
export function legacyAgentSessionRecordV1(): Record<string, unknown> {
return {
schemaVersion: 1,
sessionId: LEGACY_AGENT_SESSION_ID,
location: {
executionHostId: 'local',
wslDistro: null,
workspaceId: 'repo-fixture::/isolated/legacy-worktree',
workspaceKind: 'git-worktree'
},
provider: 'codex',
providerHandleChain: [
{
linkId: `codex-1-${LEGACY_CODEX_THREAD_ID}`,
handle: { provider: 'codex', threadId: LEGACY_CODEX_THREAD_ID },
origin: 'created',
mintedAtFence: 1,
observedAt: 1_800_000_000_000
}
],
accountHome: { variable: 'CODEX_HOME', path: '/isolated/codex-home' },
lease: {
sessionId: LEGACY_AGENT_SESSION_ID,
runtimeKind: 'native',
runtimeFence: 2,
handoffStage: null,
provenHandleLinkId: null,
ownerProcess: null,
reservedSpawnToken: null,
leaseDeadlineAt: 1_800_000_030_000,
lastRenewedAt: 1_800_000_000_000,
handoffOperationId: null,
journalCheckpoint: null,
claimKeyId: 'legacy-key-1',
claimStatus: 'released',
unreconciled: false,
deathEvidence: null
},
createdAt: 1_800_000_000_000,
updatedAt: 1_800_000_000_000
}
}
export function legacyDesktopProviderHandleRecordV1(): Record<string, unknown> {
const record = legacyAgentSessionRecordV1()
const link = (record.providerHandleChain as Record<string, unknown>[])[0]!
return {
...record,
providerHandleChain: [
{
...link,
linkId: `codex-1-${LEGACY_AGENT_SESSION_ID}`,
handle: { provider: 'codex', threadId: LEGACY_AGENT_SESSION_ID }
}
]
}
}
export function legacyMissingAuthorityRecordV1(): Record<string, unknown> {
const record = legacyAgentSessionRecordV1()
const lease = { ...(record.lease as Record<string, unknown>) }
for (const field of [
'claimStatus',
'ownerProcess',
'reservedSpawnToken',
'handoffStage',
'journalCheckpoint',
'deathEvidence'
]) {
delete lease[field]
}
const { accountHome: _accountHome, ...withoutAccountHome } = record
return { ...withoutAccountHome, lease }
}
-78
View File
@@ -1,78 +0,0 @@
import {
AGENT_SESSION_RECORD_SCHEMA_VERSION,
isAgentSessionRecord,
type AgentSessionRecord
} from './agent-session-record'
export type AgentSessionRecordLoadResult =
| { status: 'current'; record: AgentSessionRecord }
| { status: 'upgraded'; record: AgentSessionRecord }
| {
status: 'unusable'
reason:
| 'current_shape_invalid'
| 'unsupported_schema'
| 'legacy_shape_invalid'
| 'legacy_provider_handle_missing'
| 'legacy_provider_handle_is_desktop_session'
| 'legacy_account_home_provider_mismatch'
}
function legacyHandleUsesDesktopSessionId(record: AgentSessionRecord): boolean {
return record.providerHandleChain.some((link) => {
const providerId =
link.handle.provider === 'claude' ? link.handle.sessionId : link.handle.threadId
return providerId === record.sessionId
})
}
/** V1 covered multiple ad-hoc shapes, so only structurally current, provider-safe rows migrate. */
export function loadAgentSessionRecord(value: unknown): AgentSessionRecordLoadResult {
if (isAgentSessionRecord(value)) {
return { status: 'current', record: value }
}
if (
typeof value === 'object' &&
value !== null &&
(value as { schemaVersion?: unknown }).schemaVersion === AGENT_SESSION_RECORD_SCHEMA_VERSION
) {
return { status: 'unusable', reason: 'current_shape_invalid' }
}
if (
typeof value !== 'object' ||
value === null ||
(value as { schemaVersion?: unknown }).schemaVersion !== 1
) {
return { status: 'unusable', reason: 'unsupported_schema' }
}
const candidate = {
...(value as Record<string, unknown>),
schemaVersion: AGENT_SESSION_RECORD_SCHEMA_VERSION
}
if (!isAgentSessionRecord(candidate)) {
return { status: 'unusable', reason: 'legacy_shape_invalid' }
}
if (candidate.providerHandleChain.length === 0) {
return { status: 'unusable', reason: 'legacy_provider_handle_missing' }
}
if (legacyHandleUsesDesktopSessionId(candidate)) {
return { status: 'unusable', reason: 'legacy_provider_handle_is_desktop_session' }
}
const expectedAccountVariable =
candidate.provider === 'codex' ? 'CODEX_HOME' : 'CLAUDE_CONFIG_DIR'
if (candidate.accountHome.variable !== expectedAccountVariable) {
return { status: 'unusable', reason: 'legacy_account_home_provider_mismatch' }
}
return {
status: 'upgraded',
record: {
...candidate,
lease: {
...candidate.lease,
claimStatus: 'conflicted',
handoffStage: 'manual-recovery',
unreconciled: true
}
}
}
}
+3 -5
View File
@@ -54,8 +54,8 @@ export type AgentSessionHandoffStage =
/**
* PID-reuse-safe process identity. `spawnToken` is the only element available on every platform:
* process start time costs a CIM query on Windows and is absent in some containers, so a runtime
* that cannot echo the token back qualifies for manual recovery only.
* process start time costs a CIM query on Windows and is absent in some containers. An exact
* identity stays in `recovering`; an ownerless, unattributable reservation uses `manual-recovery`.
*/
export type AgentSessionProcessIdentity = {
hostId: string
@@ -122,8 +122,6 @@ export type AgentSessionRecord = {
/** Provider options acknowledged for the next turn, restored across owner replacement. */
options?: Record<string, string>
launchArgs?: AgentSessionLaunchArgs
/** Legacy read-compatible field; current stores scrub and omit it. */
launchEnv?: AgentSessionLaunchEnv
lease: AgentSessionLease
createdAt: number
updatedAt: number
@@ -330,7 +328,7 @@ export function isAgentSessionRecord(value: unknown): value is AgentSessionRecor
isAgentSessionAccountHome(record.accountHome) &&
(record.options === undefined || isAgentSessionOptions(record.options)) &&
(record.launchArgs === undefined || isAgentSessionLaunchArgs(record.launchArgs)) &&
(record.launchEnv === undefined || isAgentSessionLaunchEnv(record.launchEnv)) &&
!Object.hasOwn(record, 'launchEnv') &&
isAgentSessionLease(record.lease) &&
record.lease.sessionId === record.sessionId &&
Number.isSafeInteger(record.createdAt) &&
@@ -1,73 +0,0 @@
import { describe, expect, it } from 'vitest'
import { withInlineEnvAssignments } from './inline-env-command-prefix'
const ENV = { ORCA_AGENT_SESSION_NONCE: 'nonce-1' }
describe('inline env assignments for a typed command', () => {
it('leaves the command untouched when there is nothing to assign', () => {
expect(
withInlineEnvAssignments({ command: 'codex resume x', env: {}, platform: 'darwin' })
).toBe('codex resume x')
})
// Why one POSIX form rather than a fish dialect: Orca cannot reliably detect the login shell of a
// remote or WSL host, so `AgentStartupShell` deliberately has no fish member. The single emitted
// form has to parse in every Unix shell instead, which is what this pins.
it('uses env(1) on POSIX so fish parses it too', () => {
expect(
withInlineEnvAssignments({
command: "codex resume 'thread-1'",
env: ENV,
platform: 'linux',
shell: 'posix'
})
// Why not `NAME=value cmd`: fish rejects that form outright.
).toBe("env ORCA_AGENT_SESSION_NONCE='nonce-1' codex resume 'thread-1'")
})
it('emits that same fish-safe form when no shell is given', () => {
expect(
withInlineEnvAssignments({ command: 'codex resume x', env: ENV, platform: 'darwin' })
).toBe("env ORCA_AGENT_SESSION_NONCE='nonce-1' codex resume x")
})
it('uses PowerShell and cmd assignment syntax on Windows', () => {
expect(
withInlineEnvAssignments({
command: 'codex resume thread-1',
env: ENV,
platform: 'win32',
shell: 'powershell'
})
).toBe("$env:ORCA_AGENT_SESSION_NONCE='nonce-1'; codex resume thread-1")
expect(
withInlineEnvAssignments({
command: 'codex resume thread-1',
env: ENV,
platform: 'win32',
shell: 'cmd'
})
).toBe('set "ORCA_AGENT_SESSION_NONCE=nonce-1" && codex resume thread-1')
})
it('quotes values for the target dialect', () => {
expect(
withInlineEnvAssignments({
command: 'codex',
env: { N: "it's" },
platform: 'darwin',
shell: 'posix'
})
// Why this idiom and not `'\''`: closing, adding a double-quoted quote, then reopening parses
// in fish too, which does not honour a backslash escape inside single quotes.
).toContain(`N='it'"'"'s'`)
expect(
withInlineEnvAssignments({
command: 'codex',
env: { N: "it's" },
platform: 'win32',
shell: 'powershell'
})
).toContain(`$env:N='it''s'`)
})
})
-46
View File
@@ -1,46 +0,0 @@
// Env assignments for a command Orca TYPES into a shell it already owns, rather
// than one it spawns with an env map. The adopted-Codex return path has no spawn
// to attach env to: the pane is a live shell and the resume command is keystrokes.
import {
quoteStartupArg,
resolveStartupShell,
type AgentStartupShell
} from './tui-agent-startup-shell'
export function buildInlineEnvCommandPrefix(args: {
env: Record<string, string>
platform: NodeJS.Platform
shell?: AgentStartupShell
}): string {
const entries = Object.entries(args.env)
if (entries.length === 0) {
return ''
}
const shell = resolveStartupShell(args.platform, args.shell)
const quote = (value: string): string => quoteStartupArg(value, shell)
if (shell === 'powershell') {
return `${entries.map(([name, value]) => `$env:${name}=${quote(value)}`).join('; ')}; `
}
if (shell === 'cmd') {
// Why: `set` here outlives the command, unlike the POSIX one-shot form. Safe
// because every caller overwrites the same names on its next invocation, so a
// leftover value can only ever fail a match — never satisfy a fresh one.
return `${entries.map(([name, value]) => `set ${quote(`${name}=${value}`)}`).join(' && ')} && `
}
// Why `env` rather than the bare `NAME=value cmd` prefix: fish rejects that form
// outright ("Unsupported use of '='"), and the pane's POSIX dialect is not
// recorded anywhere Orca can read at write time. `env` parses in sh, bash, zsh
// and fish alike, and matches how every other Orca launch path reaches the agent
// binary — through the executable, not a shell alias.
return `env ${entries.map(([name, value]) => `${name}=${quote(value)}`).join(' ')} `
}
/** Prefixes a typed command so the assignments apply to that invocation. */
export function withInlineEnvAssignments(args: {
command: string
env: Record<string, string>
platform: NodeJS.Platform
shell?: AgentStartupShell
}): string {
return `${buildInlineEnvCommandPrefix(args)}${args.command}`
}
@@ -1,41 +1,8 @@
import { describe, expect, it } from 'vitest'
import {
showStructuredAgentSessionChoice,
structuredAgentSessionPayloadFingerprint
} from './structured-agent-session-mutation'
import { structuredAgentSessionPayloadFingerprint } from './structured-agent-session-mutation'
import { computeAgentSessionPayloadFingerprint } from './agent-session-mutation-envelope'
describe('structured agent session client mutations', () => {
it('keeps the chat choice invisible without both capability and workspace support', () => {
expect(
showStructuredAgentSessionChoice({
hostCapability: false,
workspaceSupport: true,
agent: 'codex'
})
).toBe(false)
expect(
showStructuredAgentSessionChoice({
hostCapability: true,
workspaceSupport: false,
agent: 'codex'
})
).toBe(false)
expect(
showStructuredAgentSessionChoice({
hostCapability: true,
workspaceSupport: true,
agent: 'claude'
})
).toBe(true)
expect(
showStructuredAgentSessionChoice({
hostCapability: true,
workspaceSupport: true,
agent: 'codex'
})
).toBe(true)
})
it('canonicalizes payload fields before hashing', () => {
const first = structuredAgentSessionPayloadFingerprint({
method: 'agentSession.send',
@@ -51,4 +18,16 @@ describe('structured agent session client mutations', () => {
expect(first).toBe(second)
expect(first).toMatch(/^[a-f0-9]{64}$/)
})
it('matches host code-unit ordering for mixed-case and non-ASCII keys', () => {
const input = {
method: 'agentSession.send',
sessionId: 'session-1',
fields: { a: 1, A: 2, é: 3, 中: 4 }
}
expect(structuredAgentSessionPayloadFingerprint(input)).toBe(
computeAgentSessionPayloadFingerprint(input)
)
})
})
@@ -9,7 +9,7 @@ function canonicalize(value: unknown): string {
}
const entries = Object.entries(value as Record<string, unknown>)
.filter(([, entry]) => entry !== undefined)
.sort(([left], [right]) => left.localeCompare(right))
.sort(([left], [right]) => (left < right ? -1 : left > right ? 1 : 0))
return `{${entries.map(([key, entry]) => `${JSON.stringify(key)}:${canonicalize(entry)}`).join(',')}}`
}
@@ -37,29 +37,3 @@ export function createStructuredAgentSessionOperationId(
}
return `${timestamp}-${entropy}`
}
export function structuredAgentSessionCreateFingerprint(input: {
sessionId: string
worktree: string
agent?: StructuredAgent
}): string {
return structuredAgentSessionPayloadFingerprint({
method: 'agentSession.create',
sessionId: input.sessionId,
fields: { worktree: input.worktree, agent: input.agent ?? 'codex' }
})
}
export type StructuredAgent = 'claude' | 'codex'
export function showStructuredAgentSessionChoice(input: {
hostCapability: boolean
workspaceSupport: boolean
agent: string
}): boolean {
return (
input.hostCapability &&
input.workspaceSupport &&
(input.agent === 'codex' || input.agent === 'claude')
)
}
@@ -1,5 +1,6 @@
import { readFileSync, readdirSync } from 'node:fs'
import { join } from 'node:path'
import { pathToFileURL } from 'node:url'
import { materializeReleaseCheckout, REPO_ROOT, type ReleaseCheckout } from './release-checkout'
/**
@@ -120,7 +121,9 @@ async function loadWorkingTreeBuild(): Promise<AgentSessionWireBuild> {
// Why @vite-ignore: the checkout is created at run time, so Vite cannot glob it at
// transform time. Vite-node still resolves and transforms the target on demand.
function importFromCheckout(specifier: string): Promise<Record<string, unknown>> {
return import(/* @vite-ignore */ specifier) as Promise<Record<string, unknown>>
return import(/* @vite-ignore */ pathToFileURL(specifier).href) as Promise<
Record<string, unknown>
>
}
async function loadReleaseBuild(checkout: ReleaseCheckout): Promise<AgentSessionWireBuild> {