diff --git a/src/main/claude/claude-structured-owner-identity.test.ts b/src/main/claude/claude-structured-owner-identity.test.ts deleted file mode 100644 index a254ce4a94c..00000000000 --- a/src/main/claude/claude-structured-owner-identity.test.ts +++ /dev/null @@ -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>() - .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) - }) -}) diff --git a/src/main/claude/claude-structured-owner-identity.ts b/src/main/claude/claude-structured-owner-identity.ts index e833e38413e..1d13e6ec7c2 100644 --- a/src/main/claude/claude-structured-owner-identity.ts +++ b/src/main/claude/claude-structured-owner-identity.ts @@ -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 = readProcessStartTimeMs -): Promise { - 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 diff --git a/src/main/claude/claude-transcript-branch-proof.ts b/src/main/claude/claude-transcript-branch-proof.ts index b3f26bf8640..d7065caa275 100644 --- a/src/main/claude/claude-transcript-branch-proof.ts +++ b/src/main/claude/claude-transcript-branch-proof.ts @@ -27,38 +27,6 @@ export class ClaudeTranscriptTailIncompleteError extends Error { } } -export async function retryClaudeTranscriptTailRead( - read: () => Promise, - options: { - timeoutMs?: number - retryDelayMs?: number - now?: () => number - wait?: (delayMs: number) => Promise - } = {} -): Promise { - 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 diff --git a/src/main/codex/codex-app-server-connection.ts b/src/main/codex/codex-app-server-connection.ts index ef690e210ed..a73958d91c6 100644 --- a/src/main/codex/codex-app-server-connection.ts +++ b/src/main/codex/codex-app-server-connection.ts @@ -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 { + return terminateCodexAppServerProcessTree(child, spawnToken, { + dedicatedProcessGroup: hasDedicatedProcessGroup + }) + } const pending = new Map() 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 diff --git a/src/main/codex/codex-app-server-process-teardown.test.ts b/src/main/codex/codex-app-server-process-teardown.test.ts index aca0b9f5a62..013ee183d12 100644 --- a/src/main/codex/codex-app-server-process-teardown.test.ts +++ b/src/main/codex/codex-app-server-process-teardown.test.ts @@ -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) + }) }) diff --git a/src/main/codex/codex-app-server-process-teardown.ts b/src/main/codex/codex-app-server-process-teardown.ts index 4ac147cc102..4e3bf537c22 100644 --- a/src/main/codex/codex-app-server-process-teardown.ts +++ b/src/main/codex/codex-app-server-process-teardown.ts @@ -12,6 +12,7 @@ type TeardownChild = Pick export type CodexAppServerProcessTeardownDeps = { platform?: NodeJS.Platform + dedicatedProcessGroup?: boolean /** Diagnostic/recovery injection only; never used by the primary teardown. */ findSpawnTokenProcesses?: (spawnToken: string) => Promise captureDescendants?: (rootPid: number) => Promise @@ -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) } diff --git a/src/main/ipc/runtime-subscribe-lifecycle.test.ts b/src/main/ipc/runtime-subscribe-lifecycle.test.ts new file mode 100644 index 00000000000..689073c0264 --- /dev/null +++ b/src/main/ipc/runtime-subscribe-lifecycle.test.ts @@ -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 unknown>(), + listeners: new Map 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 { + const record: StreamRecord = { + emit, + settled: false, + signal: options.signal, + subscriptionId: request.id + } + streams.push(record) + return new Promise((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 + } +} + +function createSender(id: number): SenderHarness { + const senderListeners = new Map 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) + }) +}) diff --git a/src/main/ipc/runtime.ts b/src/main/ipc/runtime.ts index f36bf76970e..82bc2473486 100644 --- a/src/main/ipc/runtime.ts +++ b/src/main/ipc/runtime.ts @@ -25,7 +25,18 @@ function boundTerminalFitRestore(pending: Promise): Promise { export function registerRuntimeHandlers(runtime: OrcaRuntimeService): void { const pendingTerminalFitRestores = new Map>() - const localSubscriptions = new Map() + const subscriptionsBySender = new Map>() + + 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') diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-acquisition-options.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-acquisition-options.test.ts index ed349f22d37..6340e12a265 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-acquisition-options.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-acquisition-options.test.ts @@ -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' } + }) + } + }) + }) }) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-attach-flow.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-attach-flow.ts index 3e8c6ed82ad..b71017a931a 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-attach-flow.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-attach-flow.ts @@ -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 /** Settles writes already captured by the superseded journal before opening another. */ beforeJournalOpen?: () => Promise | 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 { + 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) } } diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-attach-orchestration.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-attach-orchestration.ts index ce570d9d6f1..74cb78d78b9 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-attach-orchestration.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-attach-orchestration.ts @@ -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 diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-attach.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-attach.ts index dc682433deb..cfdbf786e14 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-attach.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-attach.ts @@ -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 diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-handoff.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-handoff.test.ts index 6542c0c08c8..4c9f1e69609 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-handoff.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-handoff.test.ts @@ -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' + }) + }) }) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-host-runtime-state.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-host-runtime-state.ts index e0656ec66f7..d1db05df872 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-host-runtime-state.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-host-runtime-state.ts @@ -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 { - 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( diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-host.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-host.test.ts index 15de7bcc5a1..c32958a1abc 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-host.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-host.test.ts @@ -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() - .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() .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 } + ).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') diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-host.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-host.ts index b5230834c58..7248e22c8f4 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-host.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-host.ts @@ -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() } diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-legacy-restart.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-legacy-restart.test.ts deleted file mode 100644 index 13edd1f9028..00000000000 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-legacy-restart.test.ts +++ /dev/null @@ -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 (_sessionId: string, task: () => Promise) => 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) - }) -}) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-processless-reservation.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-processless-reservation.test.ts index 9c85243d882..a1b6b39f5e0 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-processless-reservation.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-processless-reservation.test.ts @@ -4,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() .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' } }) }) }) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-readable-restorer.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-readable-restorer.ts index 7f929f7e5a6..69dde97b697 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-readable-restorer.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-readable-restorer.ts @@ -18,23 +18,18 @@ export class StructuredAgentSessionReadableRestorer { hasSession: (sessionId: string) => boolean onReadable: (sessionId: string, restored: RestoredStructuredAgentSessionRead) => void restoreHandoff: (sessionId: string) => Promise - /** Reaps children carrying a token no lease claims, before recovery grants a new owner. */ - reapOrphanChildren?: () => Promise } ) {} restore(): Promise { - 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 { - // 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 { await restoreStructuredAgentSessionsOnRestart({ ...this.input, records: this.input.store.listRecords().filter(this.input.supportsRecord) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-recovery-exits.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-recovery-exits.test.ts index 3b18361518c..a5927dc0c14 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-recovery-exits.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-recovery-exits.test.ts @@ -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' } }) }) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-recovery-resolution.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-recovery-resolution.test.ts index 1dc0b7bbddb..4f83a1855f3 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-recovery-resolution.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-recovery-resolution.test.ts @@ -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' }) }) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-replay-outcome.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-replay-outcome.ts index f1a155485d4..a15cd9bf8be 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-replay-outcome.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-replay-outcome.ts @@ -30,6 +30,9 @@ export function resolveAgentSessionReplayOutcome(input: { } } if (outcome.status === 'unknown') { + if (input.rerunWhenReplayMissing) { + return { decision: 'rerun' } + } return { decision: 'refuse', refusal: { diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-settled-attach-retry.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-settled-attach-retry.test.ts new file mode 100644 index 00000000000..24dc81f4684 --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-settled-attach-retry.test.ts @@ -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() + 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 +let releaseAcquisition: Mock> +let dispatch: Mock + +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): 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>() + .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' + } + }) + }) +}) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-wedged-profile-migration.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-wedged-profile-migration.test.ts index 3370d4a2581..4af342227c2 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-wedged-profile-migration.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-wedged-profile-migration.test.ts @@ -176,9 +176,8 @@ function isAcquirable(lease: NonNullable>['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', diff --git a/src/main/runtime/agent-session-acquisition-failure-settlement.ts b/src/main/runtime/agent-session-acquisition-failure-settlement.ts new file mode 100644 index 00000000000..7ad20397813 --- /dev/null +++ b/src/main/runtime/agent-session-acquisition-failure-settlement.ts @@ -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 + 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 + } + }) +} diff --git a/src/main/runtime/agent-session-launch-env-backfill.test.ts b/src/main/runtime/agent-session-launch-env-admission.test.ts similarity index 88% rename from src/main/runtime/agent-session-launch-env-backfill.test.ts rename to src/main/runtime/agent-session-launch-env-admission.test.ts index 97e69f7c422..97b0f95f2b6 100644 --- a/src/main/runtime/agent-session-launch-env-backfill.test.ts +++ b/src/main/runtime/agent-session-launch-env-admission.test.ts @@ -18,8 +18,8 @@ function request(overrides: Partial = {}): 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']) diff --git a/src/main/runtime/agent-session-orphan-child-reaper.test.ts b/src/main/runtime/agent-session-orphan-child-reaper.test.ts index 06a5b7a1954..eb837ac4a22 100644 --- a/src/main/runtime/agent-session-orphan-child-reaper.test.ts +++ b/src/main/runtime/agent-session-orphan-child-reaper.test.ts @@ -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) + }) }) diff --git a/src/main/runtime/agent-session-orphan-child-reaper.ts b/src/main/runtime/agent-session-orphan-child-reaper.ts index fd13bd897a0..68d9c899024 100644 --- a/src/main/runtime/agent-session-orphan-child-reaper.ts +++ b/src/main/runtime/agent-session-orphan-child-reaper.ts @@ -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 + } } } diff --git a/src/main/runtime/agent-session-process-identity-probe.ts b/src/main/runtime/agent-session-process-identity-probe.ts index c19c40bc671..51577048d31 100644 --- a/src/main/runtime/agent-session-process-identity-probe.ts +++ b/src/main/runtime/agent-session-process-identity-probe.ts @@ -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' diff --git a/src/main/runtime/agent-session-record-store-file.ts b/src/main/runtime/agent-session-record-store-file.ts index 9068ae6a603..8b8ef6cc4e4 100644 --- a/src/main/runtime/agent-session-record-store-file.ts +++ b/src/main/runtime/agent-session-record-store-file.ts @@ -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 diff --git a/src/main/runtime/agent-session-record-store-legacy.test.ts b/src/main/runtime/agent-session-record-store-legacy.test.ts deleted file mode 100644 index feb4eaf293f..00000000000 --- a/src/main/runtime/agent-session-record-store-legacy.test.ts +++ /dev/null @@ -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 { - return AgentSessionRecordStore.open({ directory, hostId: 'local' }) -} - -async function writeLegacyStore(record: Record): Promise { - 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: {} }) - }) -}) diff --git a/src/main/runtime/agent-session-record-store-security.test.ts b/src/main/runtime/agent-session-record-store-security.test.ts index b4f2a01ccee..f70145cf5a2 100644 --- a/src/main/runtime/agent-session-record-store-security.test.ts +++ b/src/main/runtime/agent-session-record-store-security.test.ts @@ -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 - 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 () => { diff --git a/src/main/runtime/agent-session-record-store-security.ts b/src/main/runtime/agent-session-record-store-security.ts index 74acb28b340..cc9b302ebfa 100644 --- a/src/main/runtime/agent-session-record-store-security.ts +++ b/src/main/runtime/agent-session-record-store-security.ts @@ -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 - 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 - 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 { try { await chmod(path, mode) @@ -89,56 +29,11 @@ export async function hardenAgentSessionStorePermissions(filePath: string): Prom ]) } -async function prepareScrub(filePath: string): Promise { - 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 { - 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 { return withAgentSessionStoreTransactionLock(filePath, async () => { - await hardenAgentSessionStorePermissions(filePath) - await scrubAgentSessionStoreFiles(filePath) await hardenAgentSessionStorePermissions(filePath) return loadAgentSessionStore(filePath, hostId) }) diff --git a/src/main/runtime/agent-session-record-store.ts b/src/main/runtime/agent-session-record-store.ts index bc341c912ad..6ab4a81aca8 100644 --- a/src/main/runtime/agent-session-record-store.ts +++ b/src/main/runtime/agent-session-record-store.ts @@ -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 { const [renewed] = await this.renewLeases([args]) return renewed diff --git a/src/main/runtime/agent-session-record-unsupported-schema.test.ts b/src/main/runtime/agent-session-record-unsupported-schema.test.ts new file mode 100644 index 00000000000..1f807870682 --- /dev/null +++ b/src/main/runtime/agent-session-record-unsupported-schema.test.ts @@ -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) + }) +}) diff --git a/src/main/runtime/agent-session-recovery-publish-fault.test.ts b/src/main/runtime/agent-session-recovery-publish-fault.test.ts index 10785bbad29..37ad172bdfa 100644 --- a/src/main/runtime/agent-session-recovery-publish-fault.test.ts +++ b/src/main/runtime/agent-session-recovery-publish-fault.test.ts @@ -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' ) diff --git a/src/main/runtime/agent-session-reservation-admission.ts b/src/main/runtime/agent-session-reservation-admission.ts index 85c0123ccc5..1735d67e62c 100644 --- a/src/main/runtime/agent-session-reservation-admission.ts +++ b/src/main/runtime/agent-session-reservation-admission.ts @@ -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, diff --git a/src/main/runtime/agent-session-restart-handoff-adjudication.test.ts b/src/main/runtime/agent-session-restart-handoff-adjudication.test.ts index 1f913216e73..0a4e3c39a54 100644 --- a/src/main/runtime/agent-session-restart-handoff-adjudication.test.ts +++ b/src/main/runtime/agent-session-restart-handoff-adjudication.test.ts @@ -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' + } + ) + }) }) diff --git a/src/main/runtime/agent-session-restart-handoff-adjudication.ts b/src/main/runtime/agent-session-restart-handoff-adjudication.ts index 148f2ced896..99849248ace 100644 --- a/src/main/runtime/agent-session-restart-handoff-adjudication.ts +++ b/src/main/runtime/agent-session-restart-handoff-adjudication.ts @@ -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, diff --git a/src/main/runtime/agent-session-store-transaction-queue.ts b/src/main/runtime/agent-session-store-transaction-queue.ts index 77e2fe5058a..43266c0f113 100644 --- a/src/main/runtime/agent-session-store-transaction-queue.ts +++ b/src/main/runtime/agent-session-store-transaction-queue.ts @@ -138,7 +138,7 @@ export class AgentSessionStoreTransactionQueue { return run } - persistLoadedMigration(): Promise { + persistLoadedRewrite(): Promise { return this.transact(() => undefined) } diff --git a/src/main/runtime/structured-agent-session-runtime.test.ts b/src/main/runtime/structured-agent-session-runtime.test.ts index ca0120f343d..eebc6acd499 100644 --- a/src/main/runtime/structured-agent-session-runtime.test.ts +++ b/src/main/runtime/structured-agent-session-runtime.test.ts @@ -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 + ) + ) + }) +}) diff --git a/src/main/runtime/structured-agent-session-runtime.ts b/src/main/runtime/structured-agent-session-runtime.ts index 6e6911846eb..7bd7af34007 100644 --- a/src/main/runtime/structured-agent-session-runtime.ts +++ b/src/main/runtime/structured-agent-session-runtime.ts @@ -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 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 - if (!Object.hasOwn(record, 'launchEnv')) { - return { value, changed: false } - } - const { launchEnv: _discarded, ...rest } = record - return { value: rest, changed: true } -} diff --git a/src/shared/agent-session-lease-adjudication.test.ts b/src/shared/agent-session-lease-adjudication.test.ts index f479d239f5f..dd8b5433efd 100644 --- a/src/shared/agent-session-lease-adjudication.test.ts +++ b/src/shared/agent-session-lease-adjudication.test.ts @@ -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' }) }) }) diff --git a/src/shared/agent-session-lease-adjudication.ts b/src/shared/agent-session-lease-adjudication.ts index aa9d87ca60c..6d9deb54818 100644 --- a/src/shared/agent-session-lease-adjudication.ts +++ b/src/shared/agent-session-lease-adjudication.ts @@ -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: diff --git a/src/shared/agent-session-record-legacy.test-fixture.ts b/src/shared/agent-session-record-legacy.test-fixture.ts deleted file mode 100644 index 82edd659c16..00000000000 --- a/src/shared/agent-session-record-legacy.test-fixture.ts +++ /dev/null @@ -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 { - 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 { - const record = legacyAgentSessionRecordV1() - const link = (record.providerHandleChain as Record[])[0]! - return { - ...record, - providerHandleChain: [ - { - ...link, - linkId: `codex-1-${LEGACY_AGENT_SESSION_ID}`, - handle: { provider: 'codex', threadId: LEGACY_AGENT_SESSION_ID } - } - ] - } -} - -export function legacyMissingAuthorityRecordV1(): Record { - const record = legacyAgentSessionRecordV1() - const lease = { ...(record.lease as Record) } - for (const field of [ - 'claimStatus', - 'ownerProcess', - 'reservedSpawnToken', - 'handoffStage', - 'journalCheckpoint', - 'deathEvidence' - ]) { - delete lease[field] - } - const { accountHome: _accountHome, ...withoutAccountHome } = record - return { ...withoutAccountHome, lease } -} diff --git a/src/shared/agent-session-record-load.ts b/src/shared/agent-session-record-load.ts deleted file mode 100644 index 8f4b11715ce..00000000000 --- a/src/shared/agent-session-record-load.ts +++ /dev/null @@ -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), - 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 - } - } - } -} diff --git a/src/shared/agent-session-record.ts b/src/shared/agent-session-record.ts index edeadb36102..f81e1461218 100644 --- a/src/shared/agent-session-record.ts +++ b/src/shared/agent-session-record.ts @@ -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 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) && diff --git a/src/shared/inline-env-command-prefix.test.ts b/src/shared/inline-env-command-prefix.test.ts deleted file mode 100644 index 8f5b3bd6f7b..00000000000 --- a/src/shared/inline-env-command-prefix.test.ts +++ /dev/null @@ -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'`) - }) -}) diff --git a/src/shared/inline-env-command-prefix.ts b/src/shared/inline-env-command-prefix.ts deleted file mode 100644 index e471d77a1cb..00000000000 --- a/src/shared/inline-env-command-prefix.ts +++ /dev/null @@ -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 - 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 - platform: NodeJS.Platform - shell?: AgentStartupShell -}): string { - return `${buildInlineEnvCommandPrefix(args)}${args.command}` -} diff --git a/src/shared/structured-agent-session-mutation.test.ts b/src/shared/structured-agent-session-mutation.test.ts index d0ec288f7c0..ea92d8f0fd6 100644 --- a/src/shared/structured-agent-session-mutation.test.ts +++ b/src/shared/structured-agent-session-mutation.test.ts @@ -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) + ) + }) }) diff --git a/src/shared/structured-agent-session-mutation.ts b/src/shared/structured-agent-session-mutation.ts index c31c87b3ee0..ccc80475c93 100644 --- a/src/shared/structured-agent-session-mutation.ts +++ b/src/shared/structured-agent-session-mutation.ts @@ -9,7 +9,7 @@ function canonicalize(value: unknown): string { } const entries = Object.entries(value as Record) .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') - ) -} diff --git a/tests/e2e/cross-version-wire/versioned-agent-session-wire.ts b/tests/e2e/cross-version-wire/versioned-agent-session-wire.ts index 241585e141b..65eeaea7d26 100644 --- a/tests/e2e/cross-version-wire/versioned-agent-session-wire.ts +++ b/tests/e2e/cross-version-wire/versioned-agent-session-wire.ts @@ -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 { // 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> { - return import(/* @vite-ignore */ specifier) as Promise> + return import(/* @vite-ignore */ pathToFileURL(specifier).href) as Promise< + Record + > } async function loadReleaseBuild(checkout: ReleaseCheckout): Promise {