diff --git a/src/main/native-chat/agent-session-journal/journal-reducer.ts b/src/main/native-chat/agent-session-journal/journal-reducer.ts index 7ce1f4ef1c1d..d7e22575702d 100644 --- a/src/main/native-chat/agent-session-journal/journal-reducer.ts +++ b/src/main/native-chat/agent-session-journal/journal-reducer.ts @@ -42,6 +42,8 @@ export type JournalReducerState = { oldestSequence: number highestFence: number items: Map + /** Fence of the writer that created each item: the generation a running turn belongs to. */ + itemFences: Map /** Revision of a removed item, so a late lower revision cannot resurrect it. */ tombstones: Map submissions: Map @@ -61,6 +63,7 @@ export function createJournalReducerState(sessionId: string, epoch: string): Jou oldestSequence: 1, highestFence: 0, items: new Map(), + itemFences: new Map(), tombstones: new Map(), submissions: new Map(), receipts: new Map(), @@ -82,7 +85,13 @@ export function applyJournalRow(state: JournalReducerState, row: JournalRow): vo } const itemId = resolveJournalItemId(state, row.itemId, row.body) acceptSubmissionFromProviderItem(state, row.itemId, itemId, row) - upsertItem(state, itemId, row.revision, journalRenderItem(itemId, row.revision, row.body, row)) + upsertItem( + state, + itemId, + row.revision, + journalRenderItem(itemId, row.revision, row.body, row), + row.fence + ) return } if (row.kind === 'tombstone') { @@ -103,7 +112,7 @@ export function applyJournalRow(state: JournalReducerState, row: JournalRow): vo acceptSubmissionFromProviderItem(state, mutation.itemId, itemId, row) const producer = journalBatchMutationProducer(row, mutation) const item = journalRenderItem(itemId, revision, body, row, producer, sequenceIndex) - upsertItem(state, itemId, revision, item) + upsertItem(state, itemId, revision, item, row.fence) } else { removeItem(state, resolveItemId(state, mutation.itemId), mutation.revision) } @@ -187,7 +196,8 @@ function upsertItem( state: JournalReducerState, itemId: string, revision: number, - next: AgentJournalRenderItem + next: AgentJournalRenderItem, + fence: number ): void { const tombstoned = state.tombstones.get(itemId) if (tombstoned !== undefined && revision <= tombstoned) { @@ -199,6 +209,7 @@ function upsertItem( } if (!existing) { state.items.set(itemId, next) + state.itemFences.set(itemId, fence) state.tombstones.delete(itemId) return } @@ -237,6 +248,7 @@ function removeItem(state: JournalReducerState, itemId: string, revision: number } state.tombstones.set(itemId, revision) state.items.delete(itemId) + state.itemFences.delete(itemId) } function applySubmission( @@ -255,7 +267,7 @@ function applySubmission( ...(row.handoverRecorded ? { handoverRecorded: true, acceptedSequence: row.seq } : {}) }) const itemId = agentJournalSubmissionKey(row.clientMessageId) - upsertItem(state, itemId, 0, journalRenderItem(itemId, 0, row.body, row)) + upsertItem(state, itemId, 0, journalRenderItem(itemId, 0, row.body, row), row.fence) } function acceptSubmissionFromProviderItem( diff --git a/src/main/native-chat/agent-session-journal/journal-store.ts b/src/main/native-chat/agent-session-journal/journal-store.ts index 45c42b45f8ad..3be3aebbc514 100644 --- a/src/main/native-chat/agent-session-journal/journal-store.ts +++ b/src/main/native-chat/agent-session-journal/journal-store.ts @@ -234,6 +234,9 @@ export class AgentSessionJournal { /** Includes revisions and completion tombstones, whose timestamps disappear from render items. */ lastActivityAt = (): number => this.state.lastActivityAt + /** Fence of the writer that created the item, while it is in the timeline. */ + itemFence = (itemId: string): number | undefined => this.state.itemFences.get(itemId) + submissions = (): AgentJournalSubmission[] => [...this.state.submissions.values()] pendingSubmissions = (): AgentJournalSubmission[] => diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-conversation-open.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-conversation-open.ts index bce79ca71f87..dd8853954e3f 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-conversation-open.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-conversation-open.ts @@ -10,6 +10,7 @@ import type { AgentJournalResetReason } from '../../../shared/agent-session-journal-types' import { journalDirectoryFor } from '../agent-session-journal/journal-paths' +import type { AgentSessionJournal } from '../agent-session-journal/journal-store' import { openAgentSessionJournalWithRecovery } from './agent-session-journal-recovery' import { computeAgentSessionPayloadFingerprint } from '../../../shared/agent-session-mutation-envelope' import type { AgentSessionRecord } from '../../../shared/agent-session-record' @@ -103,23 +104,11 @@ export async function openStructuredAgentSessionConversationJournal( } catch (error) { deps.onEventSinkError?.({ sessionId, error }) } - try { - // No child in this process writes to a journal nobody had open, so whatever it shows running - // belongs to a generation that is gone, whatever the lease still claims. Settled before any - // reader or child sees it. - if (!options.acquisition) { - await settleStaleStructuredAgentSessionState({ - journal: opened.journal, - sessionId, - fence, - acquisitionGeneration: null, - deathEvidence: record.lease.deathEvidence ?? null, - failureTextContext: structuredAgentSessionFailureWordsContext(record) - }) - } - } catch (error) { - // Best effort: the next acquire re-derives it. - deps.onEventSinkError?.({ sessionId, error }) + // No child in this process writes to a journal nobody had open, so whatever it shows running + // belongs to a generation that is gone, whatever the lease still claims. Settled before any + // reader or child sees it. + if (!options.acquisition) { + await settleGoneGeneration(deps, record, opened.journal) } return { session: { journal: opened.journal, params, child: null }, @@ -127,6 +116,42 @@ export async function openStructuredAgentSessionConversationJournal( } } +/** + * The open's settle again, for a conversation already open: a proof of death written since it + * opened (the startup reconcile, a recovery) revises what the open could only call `unverifiable`. + * A record holds a proof only while released, so no child here is writing. A no-op once revised. + */ +export async function resettleOpenStructuredAgentSessionConversation( + deps: StructuredAgentSessionConversationOpenDeps, + sessionId: string, + session: StructuredAgentSessionHostSession | undefined +): Promise { + const record = deps.store.getRecord(sessionId) + if (session && record?.lease.deathEvidence) { + await settleGoneGeneration(deps, record, session.journal) + } +} + +async function settleGoneGeneration( + deps: Pick, + record: AgentSessionRecord, + journal: AgentSessionJournal +): Promise { + try { + await settleStaleStructuredAgentSessionState({ + journal, + sessionId: record.sessionId, + fence: record.lease.runtimeFence, + acquisitionGeneration: null, + deathEvidence: record.lease.deathEvidence ?? null, + failureTextContext: structuredAgentSessionFailureWordsContext(record) + }) + } catch (error) { + // Best effort: the next open or acquire re-derives it. + deps.onEventSinkError?.({ sessionId: record.sessionId, error }) + } +} + export function attachParamsForRecord( record: AgentSessionRecord, input: { diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-crash-turn-end.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-crash-turn-end.test.ts new file mode 100644 index 000000000000..0d83491ee21f --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-crash-turn-end.test.ts @@ -0,0 +1,528 @@ +// Where a turn a crash cut short ends, when the provider wrote nothing while it worked. +// +// Claude reports a Bash call once when it starts and again only when it finishes, so a command +// that runs for half a minute leaves one journal row at its start. The lease renewal the host +// wrote every ten seconds is what saw the child working after that row. + +import { mkdir, mkdtemp, rm, writeFile } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import type { AgentSessionOwnerProbe } from '../../../shared/agent-session-lease-adjudication' +import type { AgentSessionRecord } from '../../../shared/agent-session-record' +import type { AgentSessionSubscribeEvent } from '../../../shared/agent-session-wire' +import { readAgentJournalTurn } from '../../../shared/agent-session-turn-record' +import { + completedStructuredAgentTurnSeconds, + selectStructuredAgentTurnTimings +} from '../../../shared/structured-agent-session-turn-timing' +import { AgentSessionRecordStore } from '../../runtime/agent-session-record-store' +import { AGENT_SESSION_STORE_FILE_NAME } from '../../runtime/agent-session-record-store-file' +import { journalDirectoryFor } from '../agent-session-journal/journal-paths' +import { openAgentSessionJournal } from '../agent-session-journal/journal-store-factory' +import { + AgentSessionAcquisitionExitUnprovenError, + type StructuredAgentSessionAdapter +} from './structured-agent-session-adapter' +import { resettleOpenStructuredAgentSessionConversation } from './structured-agent-session-conversation-open' +import { StructuredAgentSessionHost } from './structured-agent-session-host' +import { STRUCTURED_AGENT_SESSION_IDLE_MS } from './structured-agent-session-idle-sweep' +import type { StructuredAgentSessionHostDeps } from './structured-agent-session-host-types' +import { + hostTestAttachParams, + HOST_TEST_LOCATION as LOCATION, + HOST_TEST_SESSION as SESSION +} from './structured-agent-session-host-test-data' + +const PROVIDER_SESSION = 'provider-session-alpha-1' +/** The tool call's row: the last thing the provider wrote before the crash. */ +const TOOL_STARTED_AT = 1_800_000_000_000 +/** The last renewal before the crash, while the command was still running. */ +const LAST_RENEWED_AT = TOOL_STARTED_AT + 25_000 +/** Orca comes back an hour later. */ +const RELAUNCHED_AT = TOOL_STARTED_AT + 60 * 60 * 1000 + +let root: string +let store: AgentSessionRecordStore +let host: StructuredAgentSessionHost + +function crashedClaudeRecord(): AgentSessionRecord { + const linkId = 'claude-13-link' + return { + schemaVersion: 2, + sessionId: SESSION, + location: LOCATION, + provider: 'claude', + providerHandleChain: [ + { + linkId, + handle: { provider: 'claude', sessionId: PROVIDER_SESSION, leafUuid: null }, + origin: 'created', + mintedAtFence: 13, + observedAt: TOOL_STARTED_AT - 60_000 + } + ], + accountHome: { variable: 'CLAUDE_CONFIG_DIR', path: '/home/dev/.claude' }, + createdAt: TOOL_STARTED_AT - 60_000, + updatedAt: LAST_RENEWED_AT, + lease: { + sessionId: SESSION, + runtimeKind: 'native', + runtimeFence: 13, + handoffStage: null, + provenHandleLinkId: linkId, + ownerProcess: { + hostId: 'local', + pid: 12_546, + processStartTimeMs: TOOL_STARTED_AT - 60_000, + spawnToken: 'spawn-crashed' + }, + reservedSpawnToken: 'spawn-crashed', + leaseDeadlineAt: LAST_RENEWED_AT + 30_000, + lastRenewedAt: LAST_RENEWED_AT, + handoffOperationId: null, + journalCheckpoint: null, + claimKeyId: 'key-1', + claimStatus: 'live', + unreconciled: false, + deathEvidence: null + } + } +} + +async function seedCrashedStore(): Promise { + const directory = join(root, 'store') + await mkdir(directory, { recursive: true }) + await writeFile( + join(directory, AGENT_SESSION_STORE_FILE_NAME), + JSON.stringify({ + schemaVersion: 2, + hostId: 'local', + records: { [SESSION]: crashedClaudeRecord() }, + operations: {}, + retiredClaimKeys: [], + unusableRecords: {} + }), + 'utf-8' + ) + store = await AgentSessionRecordStore.open({ directory, hostId: 'local' }) +} + +/** A running turn whose only row after its start is a Bash call that never reported back, for a + * send the provider never acknowledged: the reopen settles it at the fence after the crash. */ +async function seedClaudeToolTurn(): Promise { + let now = TOOL_STARTED_AT - 2_000 + const journal = await openAgentSessionJournal({ + identity: { + sessionId: SESSION, + workspaceId: LOCATION.workspaceId, + hostId: LOCATION.executionHostId, + agent: 'claude', + providerHandle: { kind: 'claude', sessionId: PROVIDER_SESSION, leafUuid: null } + }, + journalDir: journalDirectoryFor(root, { + workspaceId: LOCATION.workspaceId, + sessionId: SESSION + }), + now: () => now + }) + await journal.appendSubmission({ + clientMessageId: 'send-1', + payloadFingerprint: '0'.repeat(64), + body: { kind: 'message', role: 'user', blocks: [{ type: 'text', text: 'run the loop' }] }, + fence: 13 + }) + await journal.appendItem( + { provider: 'claude', sessionId: PROVIDER_SESSION, uuid: 'uuid-turn' }, + { kind: 'turn', turnId: 'turn-1', state: 'running', startedAt: now }, + { fence: 13 } + ) + now = TOOL_STARTED_AT + await journal.appendItem( + { provider: 'claude', sessionId: PROVIDER_SESSION, uuid: 'uuid-bash' }, + { + kind: 'tool-call', + name: 'Bash', + input: { command: 'for i in $(seq 90); do sleep 1; done' }, + state: 'running' + }, + { fence: 13 } + ) + await journal.close() +} + +function openHost(overrides: Partial): void { + host = new StructuredAgentSessionHost({ + store, + adapter: { + acquire: vi.fn(), + dispatch: vi.fn(), + cancelTurn: vi.fn(), + answerPrompt: vi.fn(), + setOption: vi.fn(), + supportsCreate: () => true + }, + journalRoot: root, + claimKeyId: 'key-1', + mintSpawnToken: () => 'spawn-new', + now: () => RELAUNCHED_AT, + ...overrides + }) +} + +/** What a client reads over the wire. */ +async function settledTurn() { + return (await host.journalSnapshot(SESSION)).items + .map((item) => readAgentJournalTurn(item.body)) + .find(Boolean) +} + +beforeEach(async () => { + root = await mkdtemp(join(tmpdir(), 'orca-crash-turn-end-')) + await seedCrashedStore() + await seedClaudeToolTurn() +}) + +afterEach(async () => { + await host?.flushAllStreamedEvents() + await rm(root, { recursive: true, force: true }) +}) + +describe('a turn a crash cut short mid-tool', () => { + it('ends at the last renewal, not at the tool call the provider last reported', async () => { + openHost({ probeOwner: async () => ({ outcome: 'pid-absent' }) }) + + await host.restoreReadableSessions() + + expect(store.getRecord(SESSION)?.lease.deathEvidence).toMatchObject({ + kind: 'pid-absent', + observedAt: RELAUNCHED_AT, + lastProvenAliveAt: LAST_RENEWED_AT + }) + expect(await settledTurn()).toMatchObject({ + state: 'interrupted', + completedAt: LAST_RENEWED_AT + }) + // "Worked for 27s", where the tool call's row alone reads 2s. + const [timing] = selectStructuredAgentTurnTimings( + (await host.journalSnapshot(SESSION)).items + ).values() + expect(completedStructuredAgentTurnSeconds(timing)).toBe(27) + }) + + it('ends at the pre-crash renewal when the child outlived Orca and recovery stopped it', async () => { + // The orphan is alive at relaunch, but its output went nowhere: none of that is work shown. + let alive = true + const probeOwner = async (): Promise => + alive + ? { outcome: 'identity-matched', matchedOn: ['spawn-token'] } + : { outcome: 'pid-absent' } + const stopOwnerProcess = vi.fn(() => { + alive = false + }) + openHost({ probeOwner, stopOwnerProcess }) + + await host.restoreReadableSessions() + + expect(stopOwnerProcess).toHaveBeenCalledOnce() + expect(store.getRecord(SESSION)?.lease.deathEvidence).toMatchObject({ + kind: 'pid-absent', + lastProvenAliveAt: LAST_RENEWED_AT + }) + expect(await settledTurn()).toMatchObject({ + state: 'interrupted', + completedAt: LAST_RENEWED_AT + }) + }) +}) + +/** Every turn state a subscriber was sent, in order: snapshots, then live batches. */ +function turnStatesSent(events: readonly AgentSessionSubscribeEvent[]) { + return events + .flatMap((event) => + event.type === 'snapshot' ? event.page.items : event.type === 'batch' ? event.batch.items : [] + ) + .flatMap((item) => readAgentJournalTurn(item.body) ?? []) +} + +function attach(fence: number) { + return host.attach( + { callerKey: 'client-1' }, + hostTestAttachParams(fence, { + provider: 'claude', + agent: 'claude', + accountHome: { variable: 'CLAUDE_CONFIG_DIR', path: '/home/dev/.claude' }, + providerHandle: { kind: 'claude', sessionId: PROVIDER_SESSION, leafUuid: null } + }) + ) +} + +/** Runs whatever this session's serialize already has queued. */ +function drainSession(): Promise { + return host.collaboratorsForTests().serialize(SESSION, async () => {}) +} + +// On desktop the chat on screen at relaunch opens before the startup reconcile has probed its owner, +// so the open can only call the turn unverifiable; the proof, whoever writes it, then revises it. +describe('a turn a read reached before the reconcile proved its owner dead', () => { + it('reads unverifiable, then interrupted at the last renewal, and a subscriber is sent both', async () => { + openHost({ probeOwner: async () => ({ outcome: 'pid-absent' }) }) + const events: AgentSessionSubscribeEvent[] = [] + const unsubscribe = await host.subscribe({ + id: 'reader-1', + sessionId: SESSION, + emit: (event) => events.push(event) + }) + expect(await settledTurn()).toEqual(UNVERIFIABLE_TURN) + + await host.reconcileRestartLeases() + await drainSession() + + // A paired or phone client learns of the revision through the ordinary journal publish. + await vi.waitFor(() => + expect(turnStatesSent(events)).toMatchObject([ + { state: 'unverifiable' }, + { state: 'interrupted', completedAt: LAST_RENEWED_AT } + ]) + ) + const { items } = await host.journalSnapshot(SESSION) + const [timing] = selectStructuredAgentTurnTimings(items).values() + expect(completedStructuredAgentTurnSeconds(timing)).toBe(27) + expect(items.filter((item) => item.body.kind === 'status')).toHaveLength(1) + unsubscribe() + }) + + it('revises nothing twice, whoever re-runs the settle', async () => { + openHost({ probeOwner: async () => ({ outcome: 'pid-absent' }) }) + await host.history({ sessionId: SESSION, direction: 'tail' }) + await host.reconcileRestartLeases() + await drainSession() + const settled = await host.journalSnapshot(SESSION) + + await host + .collaboratorsForTests() + .serialize(SESSION, () => + resettleOpenStructuredAgentSessionConversation( + host.deps, + SESSION, + host.collaboratorsForTests().sessions.get(SESSION) + ) + ) + await host.restoreReadableSessions() + + expect(await host.journalSnapshot(SESSION)).toEqual(settled) + }) + + it('revises it when recovery stops a child that outlived Orca, with no send', async () => { + let alive = true + const probeOwner = async (): Promise => + alive + ? { outcome: 'identity-matched', matchedOn: ['spawn-token'] } + : { outcome: 'pid-absent' } + const stopOwnerProcess = vi.fn(() => { + alive = false + }) + const acquire = vi.fn() + openHost({ + adapter: { + acquire, + dispatch: vi.fn(), + cancelTurn: vi.fn(), + answerPrompt: vi.fn(), + setOption: vi.fn(), + supportsCreate: () => true + }, + probeOwner, + stopOwnerProcess + }) + await host.history({ sessionId: SESSION, direction: 'tail' }) + expect(await settledTurn()).toEqual(UNVERIFIABLE_TURN) + + // The reconcile only parks the live orphan in recovery; recovery's stop is what proves it gone. + await host.restoreReadableSessions() + await drainSession() + + expect(stopOwnerProcess).toHaveBeenCalledOnce() + expect(acquire).not.toHaveBeenCalled() + expect(await settledTurn()).toMatchObject({ + state: 'interrupted', + completedAt: LAST_RENEWED_AT + }) + }) + + it('never touches the turn a start after the crash is writing', async () => { + const acquire = vi.fn( + async ({ fence, spawnToken, onSpawned, events }) => { + const process = { + hostId: 'local', + pid: 8_000, + processStartTimeMs: RELAUNCHED_AT, + spawnToken + } + await onSpawned?.(process) + // The new child is already working when its start lands, ahead of the queued revision. + events?.appendItem( + { provider: 'claude', sessionId: PROVIDER_SESSION, uuid: 'uuid-turn-2' }, + { kind: 'turn', turnId: 'turn-2', state: 'running', startedAt: RELAUNCHED_AT } + ) + return { + process, + link: { + linkId: `claude-${fence}-link`, + handle: { provider: 'claude', sessionId: PROVIDER_SESSION, leafUuid: null }, + origin: 'resumed', + mintedAtFence: fence, + observedAt: RELAUNCHED_AT + } + } + } + ) + openHost({ + adapter: { + acquire, + dispatch: vi.fn(), + cancelTurn: vi.fn(), + answerPrompt: vi.fn(), + setOption: vi.fn(), + supportsCreate: () => true + }, + probeOwner: async () => ({ outcome: 'pid-absent' }) + }) + await host.history({ sessionId: SESSION, direction: 'tail' }) + + // The start runs the reconcile itself, so the revision it queues waits behind the start. + await expect(attach(14)).resolves.toMatchObject({ ok: true }) + await host.flushStreamedEvents(SESSION) + await drainSession() + + const turns = (await host.journalSnapshot(SESSION)).items.flatMap( + (item) => readAgentJournalTurn(item.body) ?? [] + ) + expect(turns).toMatchObject([ + { turnId: 'turn-1', state: 'interrupted', completedAt: LAST_RENEWED_AT }, + { turnId: 'turn-2', state: 'running' } + ]) + }) + + it('stays unverifiable when the revision cannot be written, and a later open revises it', async () => { + let now = RELAUNCHED_AT + const onEventSinkError = vi.fn() + openHost({ + probeOwner: async () => ({ outcome: 'pid-absent' }), + now: () => now, + onEventSinkError + }) + await host.history({ sessionId: SESSION, direction: 'tail' }) + const { journal } = host.collaboratorsForTests().sessions.get(SESSION)! + vi.spyOn(journal, 'appendLifecycleBatch').mockRejectedValueOnce(new Error('disk full')) + + await host.reconcileRestartLeases() + await drainSession() + + expect(onEventSinkError).toHaveBeenCalledOnce() + expect(await settledTurn()).toEqual(UNVERIFIABLE_TURN) + // The proof is durable on the record, so the next open converges. + now += STRUCTURED_AGENT_SESSION_IDLE_MS + 1 + await host.collaboratorsForTests().lifetime.idleSweep.tick() + expect(host.hasSession(SESSION)).toBe(false) + expect(await settledTurn()).toMatchObject({ + state: 'interrupted', + completedAt: LAST_RENEWED_AT + }) + }) +}) + +/** A relaunched host whose first start after the crash, at fence 15, fails with `failure`; every + * later start succeeds. The crashed owner, and any child left behind, probe gone. */ +async function hostWithFailingFirstStart(failure: Error) { + let now = RELAUNCHED_AT + const acquire = vi.fn( + async ({ fence, spawnToken, onSpawned }) => { + const process = { hostId: 'local', pid: 7_000 + fence, processStartTimeMs: now, spawnToken } + await onSpawned?.(process) + if (fence === 15) { + throw failure + } + return { + process, + link: { + linkId: `claude-${fence}-link`, + handle: { provider: 'claude', sessionId: PROVIDER_SESSION, leafUuid: null }, + origin: 'resumed', + mintedAtFence: fence, + observedAt: now + } + } + } + ) + openHost({ + adapter: { + acquire, + releaseAcquisition: async () => true, + dispatch: vi.fn(), + cancelTurn: vi.fn(), + answerPrompt: vi.fn(), + setOption: vi.fn(), + supportsCreate: () => true + }, + probeOwner: async () => ({ outcome: 'pid-absent' }), + now: () => now + }) + await host.reconcileRestartLeases() + return { + attach, + advance: (ms: number) => { + now += ms + } + } +} + +/** The turn as the crashed owner left it, with no end: nothing proves when that owner stopped. */ +const UNVERIFIABLE_TURN = { + turnId: 'turn-1', + state: 'unverifiable', + startedAt: TOOL_STARTED_AT - 2_000 +} + +// The relaunch proved the fence-13 owner gone, but a start reserving fence 15 clears that proof +// before the turn is settled; what the record holds afterwards is about the start's own child. +describe('a turn a newer start could not settle before it failed', () => { + it('is not judged by the death of the child it left for recovery', async () => { + const { attach, advance } = await hostWithFailingFirstStart( + new AgentSessionAcquisitionExitUnprovenError(new Error('hung')) + ) + + await attach(14).catch(() => undefined) + expect(store.getRecord(SESSION)?.lease).toMatchObject({ + runtimeFence: 15, + handoffStage: 'recovering', + lastRenewedAt: RELAUNCHED_AT + }) + advance(60_000) + // Recovery records that child's death, then the client retries at the fence it was told. + await expect(attach(15)).resolves.toMatchObject({ refusal: { currentFence: 16 } }) + expect(store.getRecord(SESSION)?.lease.deathEvidence).toMatchObject({ + ownerFence: 15, + lastProvenAliveAt: RELAUNCHED_AT + }) + await expect(attach(16)).resolves.toMatchObject({ ok: true }) + + expect(await settledTurn()).toEqual(UNVERIFIABLE_TURN) + }) + + it('is not judged by the watched exit of a start that failed', async () => { + const { attach, advance } = await hostWithFailingFirstStart(new Error('claude exited (code 1)')) + + await expect(attach(14)).resolves.toMatchObject({ ok: false }) + expect(store.getRecord(SESSION)?.lease.deathEvidence).toMatchObject({ + kind: 'exit-observed', + ownerFence: 15, + observedAt: RELAUNCHED_AT + }) + advance(60_000) + await expect(attach(16)).resolves.toMatchObject({ ok: true }) + + // Main ended it at the failed start, an hour after the crash, with the start's reason. + expect(await settledTurn()).toEqual(UNVERIFIABLE_TURN) + }) +}) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-dead-generation-settlement.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-dead-generation-settlement.ts index f6a166c7c02d..8ee3430c5c91 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-dead-generation-settlement.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-dead-generation-settlement.ts @@ -25,6 +25,7 @@ import { } from './structured-agent-session-start-failure-row' import type { AgentSessionDeathEvidence } from '../../../shared/agent-session-record' import { + provenUnverifiableTurnRevisions, runningTurnLifecycleRevisions, turnVerdictFromDeathEvidence, type StructuredAgentSessionTurnVerdict @@ -185,9 +186,9 @@ export async function settleStructuredAgentSessionDeadGeneration(input: { /** * Settles whatever a generation with no child in this process left running: found when a new child * is acquired, or when a chat is reopened for reading. Derived from the journal and the lease's - * death evidence each time, so nothing is owed in between. Only an observed exit earns an end time - * and the exit copy. Must run before a new child's buffered events land, or a live turn would be - * judged. + * death evidence each time, so nothing is owed in between. Proven death ends the turn interrupted, + * and a proof written after an earlier settle revises what that settle left `unverifiable`. Must + * run before a new child's buffered events land, or a live turn would be judged. */ export async function settleStaleStructuredAgentSessionState(input: { journal: AgentSessionJournal @@ -200,7 +201,10 @@ export async function settleStaleStructuredAgentSessionState(input: { }): Promise { const { journal } = input const items = journal.snapshot().items - const verdict = turnVerdictFromDeathEvidence(input.deathEvidence) + // Each turn is judged by the evidence only if it names that turn's owner. + const verdictFor = (item: AgentJournalRenderItem) => + turnVerdictFromDeathEvidence(input.deathEvidence, journal.itemFence(item.itemId)) + // Per attempt: a retry re-partitions only what is left, and a reused chunk id would skip it. const generation = input.acquisitionGeneration ?? `seq-${journal.cursor().sequence}` const settlementId = `stale-session:${input.sessionId}:${input.fence}:${generation}` const mutations: JournalLifecycleMutationInput[] = [] @@ -211,11 +215,24 @@ export async function settleStaleStructuredAgentSessionState(input: { mutations.push({ kind: 'item', identity, body }) } } - mutations.push(...runningTurnLifecycleRevisions(items, verdict)) - if (verdict.state === 'interrupted' && items.some(isInProgressItem)) { + const proven = provenUnverifiableTurnRevisions(items, input.deathEvidence, journal) + mutations.push( + ...items.flatMap((item) => runningTurnLifecycleRevisions([item], verdictFor(item))), + ...proven + ) + const evidence = input.deathEvidence + if ( + evidence && + (proven.length > 0 || + items.some((item) => isInProgressItem(item) && verdictFor(item).state === 'interrupted')) + ) { mutations.unshift({ kind: 'item', - identity: { provider: 'orca', clientMessageId: settlementId }, + // Named by the death it explains, so a retry after a partly written settle adds no second row. + identity: { + provider: 'orca', + clientMessageId: `stale-session:${input.sessionId}:death-${evidence.ownerFence ?? 'unowned'}-${evidence.observedAt}` + }, // The death evidence is Orca's log text, never a sentence for a person: the row says only // that the provider stopped. body: { diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-host-delivery.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-host-delivery.ts index dace93edc621..d85d1dfee0f2 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-host-delivery.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-host-delivery.ts @@ -7,6 +7,7 @@ import type { AgentJournalResetReason } from '../../../shared/agent-session-jour import type { AgentSessionJournal } from '../agent-session-journal/journal-store' import { openStructuredAgentSessionConversation, + resettleOpenStructuredAgentSessionConversation, type OpenedStructuredAgentSessionConversation, type StructuredAgentSessionConversationOpenOptions } from './structured-agent-session-conversation-open' @@ -28,6 +29,8 @@ export type StructuredAgentSessionConversationDelivery = { sessionId: string, options?: StructuredAgentSessionConversationOpenOptions ) => Promise + /** Stops the loop and the resettle on a proof of death; quit's first step. */ + dispose: () => void /** Indexes a conversation some other open produced, as `open` would have. */ adoptOpened: ( sessionId: string, @@ -76,9 +79,26 @@ export function createStructuredAgentSessionConversationDelivery(input: { loop.wake(sessionId) } } + // A chat open before its owner's death was proven revises what its open settled. Queued, never + // awaited: the writer can hold this session's serialize (an attach recovering its lease). + const stopResettling = deps.store.onDeathEvidence((sessionId) => { + if (sessions.has(sessionId)) { + void input + .trackStart( + input.serialize(sessionId, () => + resettleOpenStructuredAgentSessionConversation(deps, sessionId, sessions.get(sessionId)) + ) + ) + .catch((error: unknown) => deps.onEventSinkError?.({ sessionId, error })) + } + }) return { loop, adoptOpened, + dispose: () => { + loop.dispose() + stopResettling() + }, open: (sessionId, options) => openStructuredAgentSessionConversation({ deps, sessions, adoptOpened }, sessionId, options) } diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-host-test-abandon.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-host-test-abandon.ts index de928eb32e7f..a4069875bade 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-host-test-abandon.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-host-test-abandon.ts @@ -16,7 +16,7 @@ export async function abandonStructuredAgentSessionHost( host: StructuredAgentSessionHost ): Promise { // First, so the drain below cannot race the loop into enqueueing another step. - host['conversationDelivery'].loop.dispose() + host['conversationDelivery'].dispose() host['lifetime'].dispose() await host['runtimeState'].stopLeaseRenewal() await host['tasks'].drainAttaches() 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 eba441430749..26697d3f6819 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 @@ -247,7 +247,7 @@ export class StructuredAgentSessionHost { // Trigger inlined rather than imported: `AgentSessionResumeTrigger` in shared is the canonical // type, and this file has no line budget left for the import. async flushAllStreamedEvents(options?: { trigger?: 'quit' | 'update' }): Promise { - this.conversationDelivery.loop.dispose() + this.conversationDelivery.dispose() await flushStructuredAgentSessionHost({ ...this.lifetimeContext(), idleSweep: this.lifetime, diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-send-open-stale-turn.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-send-open-stale-turn.test.ts index a1f196d8b612..0a14f4361197 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-send-open-stale-turn.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-send-open-stale-turn.test.ts @@ -86,7 +86,6 @@ async function relaunchAfterCrashMidTurn( return { host, acquire } } -/** Neither proof of the old owner's death is an observed exit, so the turn ends `unverifiable`. */ async function turnStates(host: StructuredAgentSessionHost) { return (await host.journalSnapshot(SESSION)).items .flatMap((item) => readAgentJournalTurn(item.body) ?? []) @@ -94,18 +93,20 @@ async function turnStates(host: StructuredAgentSessionHost) { } // A host that cannot prove the old owner gone leaves its lease in recovery, still claimed, until -// the next acquire resolves it: the open is not that acquire, and the turn is no less gone. +// the next acquire resolves it: the open is not that acquire, and the turn is no less gone. Only +// the proof decides whether it reads interrupted. const PROBES = [ - ['the old owner is proven gone', { outcome: 'pid-absent' }], + ['the old owner is proven gone', { outcome: 'pid-absent' }, 'interrupted'], [ 'nothing proves the old owner gone', - { outcome: 'indeterminate', reason: 'This host cannot probe structured session owners.' } + { outcome: 'indeterminate', reason: 'This host cannot probe structured session owners.' }, + 'unverifiable' ] -] as const satisfies readonly (readonly [string, AgentSessionOwnerProbe])[] +] as const satisfies readonly (readonly [string, AgentSessionOwnerProbe, string])[] it.each(PROBES)( 'settles a turn a dead generation left running when a send opens the chat and its start fails, when %s', - async (_when, probe) => { + async (_when, probe, settled) => { const { host, acquire } = await relaunchAfterCrashMidTurn(probe) expect(host.hasSession(SESSION)).toBe(false) @@ -123,32 +124,34 @@ it.each(PROBES)( ) expect(acquire).toHaveBeenCalledOnce() - expect(await turnStates(host)).toEqual(['unverifiable']) + expect(await turnStates(host)).toEqual([settled]) await host.flushAllStreamedEvents() } ) it.each(PROBES)( 'settles the same turn when a reader opens the chat, when %s', - async (_when, probe) => { + async (_when, probe, settled) => { const { host } = await relaunchAfterCrashMidTurn(probe) await host.revealSession(SESSION) - expect(await turnStates(host)).toEqual(['unverifiable']) + expect(await turnStates(host)).toEqual([settled]) await host.flushAllStreamedEvents() } ) -// On desktop the chat on screen at relaunch reads before startup reconciles the leases. -it('settles it when a read reaches the chat before the startup reconcile', async () => { +// On desktop the chat on screen at relaunch reads before startup reconciles the leases: nothing +// has proved its owner gone yet, and the reconcile's proof then revises what the open settled. +it('settles it when a read reaches the chat before the startup reconcile, then revises it', async () => { const { host } = await relaunchAfterCrashMidTurn({ outcome: 'pid-absent' }, { reconcile: false }) expect(hostTestState().store.getRecord(SESSION)?.lease.claimStatus).toBe('live') await host.history({ sessionId: SESSION, direction: 'tail' }) + expect(await turnStates(host)).toEqual(['unverifiable']) await host.reconcileRestartLeases() await host.restoreReadableSessions([SESSION]) - expect(await turnStates(host)).toEqual(['unverifiable']) + expect(await turnStates(host)).toEqual(['interrupted']) await host.flushAllStreamedEvents() }) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-stale-turn-open.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-stale-turn-open.test.ts index 5598cd3c4f72..83959233f6f3 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-stale-turn-open.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-stale-turn-open.test.ts @@ -28,8 +28,9 @@ afterEach(async () => { await rig.dispose() }) -/** A turn in flight whose agent exits, and whose exit settlement the journal refuses. */ -async function exitWithUnwrittenSettlement(): Promise { +/** A turn in flight whose agent exits, and whose exit settlement the journal refuses: by default + * also the retry that recording the exit queues, so only an open is left to settle it. */ +async function exitWithUnwrittenSettlement(refusedWrites = 2): Promise { await foundRestTestChat(rig) const open = rig.host.collaboratorsForTests().sessions.get(SESSION)! const running = open.child! @@ -40,7 +41,10 @@ async function exitWithUnwrittenSettlement(): Promise { { kind: 'turn', turnId: 'working', state: 'running' } ) await rig.host.flushStreamedEvents(SESSION) - vi.spyOn(open.journal, 'appendLifecycleBatch').mockRejectedValueOnce(new Error('disk full')) + const append = vi.spyOn(open.journal, 'appendLifecycleBatch') + for (let refused = 0; refused < refusedWrites; refused += 1) { + append.mockRejectedValueOnce(new Error('disk full')) + } await rig.host.handleAdapterEvent({ type: 'ended', sessionId: SESSION, @@ -55,6 +59,7 @@ async function exitWithUnwrittenSettlement(): Promise { deathEvidence: { kind: 'exit-observed' } }) ) + await rig.host.collaboratorsForTests().serialize(SESSION, async () => {}) } async function workingTurnState(): Promise { @@ -65,6 +70,18 @@ async function workingTurnState(): Promise { } describe('a turn its gone agent left running', () => { + it('is settled in place by the retry recording the exit queues, with no reopen', async () => { + await exitWithUnwrittenSettlement(1) + + expect(rig.host.collaboratorsForTests().sessions.has(SESSION)).toBe(true) + expect(await workingTurnState()).toBe('interrupted') + const { items } = await rig.host.journalSnapshot(SESSION) + expect(JSON.stringify(items)).toContain( + 'Codex stopped while this response was in progress. You can continue in this conversation.' + ) + expect(rig.adapter.acquire).toHaveBeenCalledOnce() + }) + it('is settled when the idle-closed conversation is opened again, and its reader sees it', async () => { await exitWithUnwrittenSettlement() expect(await workingTurnState()).toBe('running') diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-stale-turn-verdict.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-stale-turn-verdict.test.ts index 4ab7b3ad010c..2c7a91e76378 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-stale-turn-verdict.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-stale-turn-verdict.test.ts @@ -4,6 +4,8 @@ import { join } from 'node:path' import { describe, expect, it, vi } from 'vitest' import { agentJournalItemKey } from '../../../shared/agent-session-journal-item-key' import type { AgentJournalRenderItem } from '../../../shared/agent-session-journal-types' +import type { AgentSessionDeathEvidence } from '../../../shared/agent-session-record' +import { readAgentJournalTurn } from '../../../shared/agent-session-turn-record' import type { AgentSessionJournal } from '../agent-session-journal/journal-store' import { createTrackedJournalOpener } from '../agent-session-journal/journal-store-test-open' import { settleStaleStructuredAgentSessionState } from './structured-agent-session-dead-generation-settlement' @@ -75,17 +77,77 @@ function promptItem(state: 'pending' | 'resolved', sequence: number): AgentJourn } describe('turn verdict from death evidence', () => { - it('earns an end time only from an observed exit', () => { + /** Judges a turn the fence-1 owner wrote, by evidence naming that owner. */ + const verdictForTurn = (evidence: AgentSessionDeathEvidence | null | undefined) => + turnVerdictFromDeathEvidence(evidence && { ownerFence: 1, ...evidence }, 1) + + it('ends a watched exit at the exit', () => { + expect(verdictForTurn({ kind: 'exit-observed', detail: 'exit', observedAt: 500 })).toEqual({ + state: 'interrupted', + completedAt: 500 + }) + }) + + it.each(['pid-absent', 'identity-mismatch'] as const)( + 'ends a %s proof at the last proof of life, never after the probe', + (kind) => { + const proof = (lastProvenAliveAt?: number) => ({ + kind, + detail: 'gone', + observedAt: 9_000, + ...(lastProvenAliveAt === undefined ? {} : { lastProvenAliveAt }) + }) + // Probed at 9000, long after the crash: the downtime is never counted as work. + expect(verdictForTurn(proof(8_000))).toEqual({ state: 'interrupted', completedAt: 8_000 }) + expect(verdictForTurn(proof(9_500))).toEqual({ state: 'interrupted', completedAt: 9_000 }) + // A proof that recorded no proof of life has only the probe. + expect(verdictForTurn(proof())).toEqual({ state: 'interrupted', completedAt: 9_000 }) + } + ) + + it('ends a watched exit at the exit even when a renewal is recorded', () => { expect( - turnVerdictFromDeathEvidence({ kind: 'exit-observed', detail: 'exit', observedAt: 500 }) + verdictForTurn({ + kind: 'exit-observed', + detail: 'exit', + observedAt: 500, + lastProvenAliveAt: 400 + }) ).toEqual({ state: 'interrupted', completedAt: 500 }) - expect( - turnVerdictFromDeathEvidence({ kind: 'pid-absent', detail: 'gone', observedAt: 500 }) - ).toEqual({ state: 'unverifiable' }) - expect( - turnVerdictFromDeathEvidence({ kind: 'identity-mismatch', detail: 'pid', observedAt: 500 }) - ).toEqual({ state: 'unverifiable' }) - expect(turnVerdictFromDeathEvidence(null)).toEqual({ state: 'unverifiable' }) + }) + + it('leaves a release nothing proved unverifiable', () => { + expect(verdictForTurn(null)).toEqual({ state: 'unverifiable' }) + expect(verdictForTurn(undefined)).toEqual({ state: 'unverifiable' }) + }) + it('keeps the rule an older build applied to evidence that names no owner', () => { + // Only a watched exit was proof then; a probe's proof stays unverifiable. + const legacy = { detail: 'gone', observedAt: 9_000, lastProvenAliveAt: 8_000 } + expect(turnVerdictFromDeathEvidence({ ...legacy, kind: 'exit-observed' }, 1)).toEqual({ + state: 'interrupted', + completedAt: 9_000 + }) + expect(turnVerdictFromDeathEvidence({ ...legacy, kind: 'pid-absent' }, 1)).toEqual({ + state: 'unverifiable' + }) + }) + + it('gives a turn no end from evidence about any other owner', () => { + const proof = { + kind: 'pid-absent' as const, + detail: 'gone', + observedAt: 9_000, + lastProvenAliveAt: 8_000 + } + // A newer start's death, and a turn whose writer the timeline no longer knows: neither proves + // the turn's own owner gone. + for (const [evidence, turnFence] of [ + [{ ...proof, ownerFence: 3 }, 1], + [{ ...proof, kind: 'exit-observed' as const, ownerFence: 3 }, 1], + [{ ...proof, ownerFence: 1 }, undefined] + ] as const) { + expect(turnVerdictFromDeathEvidence(evidence, turnFence)).toEqual({ state: 'unverifiable' }) + } }) }) @@ -162,6 +224,13 @@ describe('running turn lifecycle revisions', () => { }) }) + it('never ends a turn before it began, when the last proof of life predates it', () => { + const item = lifecycleItem('turn-2', 'running', 2, { startedAt: 500 }) + expect( + runningTurnLifecycleRevisions([item], { state: 'interrupted', completedAt: 300 }) + ).toMatchObject([{ body: { kind: 'turn', state: 'interrupted', completedAt: 500 } }]) + }) + it('revises a legacy status-form running row from an older host into a typed turn', () => { expect( runningTurnLifecycleRevisions([legacyLifecycleItem('turn-2', 30)], { state: 'unverifiable' }) @@ -180,11 +249,22 @@ describe('running turn lifecycle revisions', () => { }) }) +/** A probe that found the fence-1 owner gone at 9000, last proven alive at 100. */ +const PROVEN: AgentSessionDeathEvidence = { + kind: 'pid-absent', + detail: 'recorded pid absent on host', + observedAt: 9_000, + ownerFence: 1, + lastProvenAliveAt: 100 +} + describe('stale session state on a cold acquire', () => { function journalWith(items: AgentJournalRenderItem[]) { const appendLifecycleBatch = vi.fn(async () => ({ epoch: 'epoch-1', sequence: 9 })) + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: the settle reads only these journal members. const journal = { snapshot: () => ({ items }), + itemFence: () => 1, cursor: () => ({ epoch: 'epoch-1', sequence: 8 }), appendLifecycleBatch } as unknown as AgentSessionJournal @@ -310,6 +390,255 @@ describe('stale session state on a cold acquire', () => { } }) + it('ends a probe-proven turn at its last proof of life, not at a row written after it', async () => { + const root = await mkdtemp(join(tmpdir(), 'orca-stale-session-')) + const journals = createTrackedJournalOpener() + let now = 100 + try { + const journal = await journals.open({ + identity: { + sessionId: 'session-1', + workspaceId: 'workspace-1', + hostId: 'local', + agent: 'codex', + providerHandle: { kind: 'codex', threadId: THREAD } + }, + journalDir: root, + now: () => now + }) + const command = { provider: 'codex' as const, threadId: THREAD, turnId: 'turn-1', ordinal: 1 } + const shell = { kind: 'tool-call' as const, name: 'shell', input: { command: 'pnpm test' } } + await journal.appendItem( + { ...command, ordinal: 0 }, + { kind: 'turn', turnId: 'turn-1', state: 'running', startedAt: 100 }, + { fence: 1 } + ) + now = 200 + await journal.appendItem(command, { ...shell, state: 'running' }, { fence: 1 }) + // Rows can outlast the last renewal; only the renewal is proof of life. + now = 700 + await journal.appendItem( + command, + { ...shell, input: { command: 'pnpm test', streamed: 'ok' }, state: 'running' }, + { fence: 1 } + ) + now = 9_000 + + await settleStaleStructuredAgentSessionState({ + journal, + sessionId: 'session-1', + fence: 2, + acquisitionGeneration: 'generation-2', + deathEvidence: { + kind: 'pid-absent', + detail: 'recorded pid absent on host', + observedAt: 9_000, + ownerFence: 1, + lastProvenAliveAt: 650 + } + }) + + const items = journal.snapshot().items + expect(items.map((item) => readAgentJournalTurn(item.body)).find(Boolean)).toMatchObject({ + state: 'interrupted', + completedAt: 650 + }) + // The probe's detail is Orca's, so the row carries none. + expect( + items.flatMap((item) => (item.body.kind === 'status' ? [item.body.text] : [])) + ).toEqual([ + 'The agent stopped while this response was in progress. You can continue in this conversation.' + ]) + } finally { + await journals.closeAll() + await rm(root, { recursive: true, force: true }) + } + }) + + it('ends a crashed turn at its last proof of life, not at a send accepted before the proof', async () => { + const root = await mkdtemp(join(tmpdir(), 'orca-stale-session-')) + const journals = createTrackedJournalOpener() + let now = 100 + try { + const journal = await journals.open({ + identity: { + sessionId: 'session-1', + workspaceId: 'workspace-1', + hostId: 'local', + agent: 'codex', + providerHandle: { kind: 'codex', threadId: THREAD } + }, + journalDir: root, + now: () => now + }) + await journal.appendItem( + RUNNING_IDENTITY, + { kind: 'turn', turnId: 'turn-2', state: 'running', startedAt: 100 }, + { fence: 1 } + ) + now = 200 + await journal.appendItem( + { ...RUNNING_IDENTITY, ordinal: 1 }, + { kind: 'tool-call', name: 'shell', input: { command: 'pnpm test' }, state: 'running' }, + { fence: 1 } + ) + const settle = (deathEvidence: AgentSessionDeathEvidence | null) => + settleStaleStructuredAgentSessionState({ + journal, + sessionId: 'session-1', + fence: 1, + acquisitionGeneration: null, + deathEvidence + }) + // An hour later the relaunch opens the chat before anything proved the owner gone. + now = 3_600_000 + await settle(null) + const turn = () => journal.snapshot().items.map((item) => readAgentJournalTurn(item.body))[0] + expect(turn()).toMatchObject({ state: 'unverifiable' }) + now = 3_605_000 + await journal.appendSubmission({ + clientMessageId: 'send-1', + payloadFingerprint: 'fingerprint-1', + body: { kind: 'message', role: 'user', blocks: [{ type: 'text', text: 'still there?' }] }, + fence: 1, + handoverRecorded: true + }) + await journal.resolveDispatch({ clientMessageId: 'send-1', state: 'pending', fence: 1 }) + now = 3_606_000 + + await settle({ + kind: 'pid-absent', + detail: 'gone', + observedAt: 3_606_000, + ownerFence: 1, + lastProvenAliveAt: 210 + }) + + expect(turn()).toMatchObject({ state: 'interrupted', completedAt: 210 }) + } finally { + await journals.closeAll() + await rm(root, { recursive: true, force: true }) + } + }) + + it('revises an unverifiable turn only from a proof naming its own owner, and only once', async () => { + const root = await mkdtemp(join(tmpdir(), 'orca-stale-session-')) + const journals = createTrackedJournalOpener() + let now = 100 + try { + const journal = await journals.open({ + identity: { + sessionId: 'session-1', + workspaceId: 'workspace-1', + hostId: 'local', + agent: 'codex', + providerHandle: { kind: 'codex', threadId: THREAD } + }, + journalDir: root, + now: () => now + }) + await journal.appendItem( + RUNNING_IDENTITY, + { kind: 'turn', turnId: 'turn-2', state: 'running', startedAt: 100 }, + { fence: 1 } + ) + now = 400 + // The open, before anything proved the fence-1 owner gone. + const settle = (deathEvidence: AgentSessionDeathEvidence | null) => + settleStaleStructuredAgentSessionState({ + journal, + sessionId: 'session-1', + fence: 2, + acquisitionGeneration: null, + deathEvidence + }) + await settle(null) + now = 9_000 + const turn = () => journal.snapshot().items.map((item) => readAgentJournalTurn(item.body))[0] + expect(turn()).toMatchObject({ state: 'unverifiable' }) + + const unrevised = journal.cursor() + await expect(settle({ ...PROVEN, ownerFence: 2 })).resolves.toBe(0) + const { ownerFence: _ownerFence, ...olderBuildProof } = PROVEN + await expect(settle({ ...olderBuildProof, kind: 'exit-observed' })).resolves.toBe(0) + expect(journal.cursor()).toEqual(unrevised) + expect(turn()).toMatchObject({ state: 'unverifiable' }) + + await expect(settle(PROVEN)).resolves.toBe(2) + expect(turn()).toEqual({ + turnId: 'turn-2', + state: 'interrupted', + startedAt: 100, + completedAt: 100 + }) + const revised = journal.cursor() + await expect(settle(PROVEN)).resolves.toBe(0) + expect(journal.cursor()).toEqual(revised) + expect(journal.snapshot().items.filter((item) => item.body.kind === 'status')).toHaveLength(1) + } finally { + await journals.closeAll() + await rm(root, { recursive: true, force: true }) + } + }) + + it('adds one row for a death however many attempts its settle takes to write', async () => { + const root = await mkdtemp(join(tmpdir(), 'orca-stale-session-')) + const journals = createTrackedJournalOpener() + let now = 100 + try { + const journal = await journals.open({ + identity: { + sessionId: 'session-1', + workspaceId: 'workspace-1', + hostId: 'local', + agent: 'codex', + providerHandle: { kind: 'codex', threadId: THREAD } + }, + journalDir: root, + now: () => now + }) + // Enough running turns that the settle writes two batches, and its retry two again. + for (let index = 0; index < 399; index++) { + const turnId = `turn-${index}` + await journal.appendItem( + { ...RUNNING_IDENTITY, turnId }, + { kind: 'turn', turnId, state: 'running', startedAt: 100 }, + { fence: 1 } + ) + } + now = 9_000 + const settle = () => + settleStaleStructuredAgentSessionState({ + journal, + sessionId: 'session-1', + fence: 2, + acquisitionGeneration: null, + deathEvidence: PROVEN + }) + const turns = () => + journal.snapshot().items.flatMap((item) => readAgentJournalTurn(item.body) ?? []) + const statusRows = () => + journal.snapshot().items.filter((item) => item.body.kind === 'status') + const append = journal.appendLifecycleBatch.bind(journal) + const secondBatchFails = vi + .spyOn(journal, 'appendLifecycleBatch') + .mockImplementationOnce(append) + .mockRejectedValueOnce(new Error('disk full')) + + await expect(settle()).rejects.toThrow('disk full') + secondBatchFails.mockRestore() + expect(turns().filter((turn) => turn.state === 'running')).toHaveLength(200) + expect(statusRows()).toHaveLength(1) + + await settle() + expect(turns().filter((turn) => turn.state !== 'interrupted')).toEqual([]) + expect(statusRows()).toHaveLength(1) + } finally { + await journals.closeAll() + await rm(root, { recursive: true, force: true }) + } + }) + it('writes nothing when no turn is running and keys on the journal position without a generation', async () => { const idle = journalWith([ lifecycleItem('turn-1', 'completed', 1, { startedAt: 10, completedAt: 20 }) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-stale-turn-verdict.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-stale-turn-verdict.ts index b6e9b9bde5ea..b23018aeba83 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-stale-turn-verdict.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-stale-turn-verdict.ts @@ -1,8 +1,10 @@ // What the host may durably say about a turn whose provider child is gone. // -// `interrupted` requires the host to have seen the child exit; that receipt is the only end time it -// is allowed to record. Everything weaker — a pid probe, an identity mismatch, a journal found -// running with no observed exit on the record — is `unverifiable` and carries no end at all. +// `interrupted` requires proof that the child which wrote the turn is gone: death evidence naming +// that turn's owner by fence — a watched exit, or a local probe that found the recorded pid gone or +// reused. A release nothing proved — lost contact, an unverifiable identity, a stop that outlived the +// ladder — carries none, and neither does a later owner's death; the turn is then `unverifiable` +// with no end at all, until a proof naming its owner is written and revises it. import { parseAgentJournalItemKey } from '../../../shared/agent-session-journal-item-key' import type { @@ -15,6 +17,7 @@ import { } from '../../../shared/agent-session-turn-record' import type { AgentSessionDeathEvidence } from '../../../shared/agent-session-record' import type { JournalLifecycleMutationInput } from '../agent-session-journal/journal-row-builders' +import type { AgentSessionJournal } from '../agent-session-journal/journal-store' export type StructuredAgentSessionTurnVerdict = | { state: 'interrupted'; completedAt: number } @@ -25,11 +28,31 @@ export const UNVERIFIABLE_TURN_VERDICT: StructuredAgentSessionTurnVerdict = { } export function turnVerdictFromDeathEvidence( - evidence: AgentSessionDeathEvidence | null | undefined + evidence: AgentSessionDeathEvidence | null | undefined, + /** Fence of the owner that wrote the turn. */ + turnFence: number | undefined ): StructuredAgentSessionTurnVerdict { - return evidence?.kind === 'exit-observed' - ? { state: 'interrupted', completedAt: evidence.observedAt } - : UNVERIFIABLE_TURN_VERDICT + if (!evidence) { + return UNVERIFIABLE_TURN_VERDICT + } + if (evidence.ownerFence === undefined) { + // Evidence an older build wrote names no owner; it keeps the rule that build applied. + return evidence.kind === 'exit-observed' + ? { state: 'interrupted', completedAt: evidence.observedAt } + : UNVERIFIABLE_TURN_VERDICT + } + if (evidence.ownerFence !== turnFence) { + return UNVERIFIABLE_TURN_VERDICT + } + if (evidence.kind === 'exit-observed') { + return { state: 'interrupted', completedAt: evidence.observedAt } + } + // A probe finds a dead child long after it died; its last renewal bounds the end, so the turn never + // counts the time Orca was down. Timeline rows don't: a send can land there after the death. + return { + state: 'interrupted', + completedAt: Math.min(evidence.lastProvenAliveAt ?? evidence.observedAt, evidence.observedAt) + } } /** Revises every still-running lifecycle item in place, keeping its identity and start. */ @@ -37,23 +60,43 @@ export function runningTurnLifecycleRevisions( items: readonly AgentJournalRenderItem[], verdict: StructuredAgentSessionTurnVerdict ): JournalLifecycleMutationInput[] { - const revisions: JournalLifecycleMutationInput[] = [] - for (const item of items) { + return items.flatMap((item) => { const turn = readAgentJournalTurn(item.body) - if (turn?.state !== 'running') { - continue - } - const identity = parseAgentJournalItemKey(item.itemId) - if (!identity) { - continue - } - revisions.push({ - kind: 'item', - identity, - body: agentJournalTurnBody(settledLifecycle(turn, verdict)) - }) + return turn?.state === 'running' ? turnLifecycleRevision(item, turn, verdict) : [] + }) +} + +/** + * A turn an earlier settle could only call `unverifiable`, because the proof had not been written + * yet, revised once a proof names the owner that wrote it. Only ever upward, and never from an + * older build's proof, which names no owner. + */ +export function provenUnverifiableTurnRevisions( + items: readonly AgentJournalRenderItem[], + evidence: AgentSessionDeathEvidence | null | undefined, + journal: Pick +): JournalLifecycleMutationInput[] { + const ownerFence = evidence?.ownerFence + if (ownerFence === undefined) { + return [] } - return revisions + return items.flatMap((item) => { + const turn = readAgentJournalTurn(item.body) + return turn?.state === 'unverifiable' && journal.itemFence(item.itemId) === ownerFence + ? turnLifecycleRevision(item, turn, turnVerdictFromDeathEvidence(evidence, ownerFence)) + : [] + }) +} + +function turnLifecycleRevision( + item: AgentJournalRenderItem, + turn: AgentJournalTurnLifecycle, + verdict: StructuredAgentSessionTurnVerdict +): JournalLifecycleMutationInput[] { + const identity = parseAgentJournalItemKey(item.itemId) + return identity + ? [{ kind: 'item', identity, body: agentJournalTurnBody(settledLifecycle(turn, verdict)) }] + : [] } /** The verdict owns the turn's end and nothing else; every other field the row @@ -69,7 +112,10 @@ function settledLifecycle( durationMs: _durationMs, ...kept } = lifecycle - return verdict.state === 'interrupted' - ? { ...kept, state: verdict.state, completedAt: verdict.completedAt } - : { ...kept, state: verdict.state } + if (verdict.state !== 'interrupted') { + return { ...kept, state: verdict.state } + } + // A renewal can predate the turn, which started with its owner alive; it never ends before that. + const began = Math.max(lifecycle.requestedAt ?? 0, lifecycle.startedAt ?? 0) + return { ...kept, state: verdict.state, completedAt: Math.max(verdict.completedAt, began) } } diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-surface-lifetime.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-surface-lifetime.test.ts index 98e50783bdf3..b04a9b3d820a 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-surface-lifetime.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-surface-lifetime.test.ts @@ -767,7 +767,9 @@ describe('an unexpected provider exit', () => { runtimeFence: exitedFence + 1, deathEvidence: { kind: 'exit-observed', detail: 'provider exited', observedAt: NOW } }) - // Nothing retries the settlement; the journal writes again, and the next acquire re-derives it. + // The retry recording the exit queues is refused too; once the journal writes again, the next + // acquire re-derives it. + await host.collaboratorsForTests().serialize(SESSION, async () => {}) refusing.mockRestore() dispatch.mockResolvedValueOnce({ diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-unexpected-exit.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-unexpected-exit.test.ts index 0685af605499..451629f7b45f 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-unexpected-exit.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-unexpected-exit.test.ts @@ -96,6 +96,7 @@ describe('provider-exit settlement', () => { snapshot: () => ({ items: [lifecycleItem('turn-1', 1, { state: 'running', startedAt: 1_000 })] }), + itemFence: () => 7, appendLifecycleBatch, markPendingSubmissionsUnknown: vi.fn(async () => []) } @@ -123,7 +124,7 @@ describe('provider-exit settlement', () => { observedAt } ) - // Released, not latched: the next acquire or read restore settles what this write left. + // Released, not latched: a later settle from this evidence finishes what this write left. expect(record.lease).toMatchObject({ claimStatus: 'released', handoffStage: null, diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-unexpected-exit.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-unexpected-exit.ts index 3a47d1240f24..5025c5a6f82f 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-unexpected-exit.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-unexpected-exit.ts @@ -131,8 +131,8 @@ export async function settleUnexpectedStructuredAgentSessionExit< acquisitionGeneration: child.generation, now: context.now(), exitObservedAt: observedAt, - // Bare cause: whatever this settlement could not write is settled from it later, by the - // next acquire or read restore, and `exit-observed` already says the rest. + // Bare cause: whatever this settlement could not write is settled from it later (the + // settle recording it queues, or the next open or acquire); `exit-observed` says the rest. exitReason: unexpectedEvent.reason.slice(0, MAX_UNEXPECTED_EXIT_REASON_CHARS) }) } catch (error) { 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 98b212a8606c..a492b0b65ce1 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 @@ -191,6 +191,9 @@ function isAcquirable(lease: NonNullable>['le ) } +/** After the owner's last renewal, so a turn it proves dead ends at its start, never before. */ +const SEEDED_TURN_STARTED_AT = NOW - 5_000 + async function seedRunningTurn(provider: 'codex' | 'claude' = 'codex'): Promise { const journal = await openAgentSessionJournal({ identity: { @@ -209,7 +212,7 @@ async function seedRunningTurn(provider: 'codex' | 'claude' = 'codex'): Promise< provider === 'codex' ? { provider: 'codex', threadId: THREAD, turnId: 'turn-1', ordinal: 0 } : { provider: 'claude', sessionId: 'provider-session-alpha-1', uuid: 'uuid-running' }, - { kind: 'turn', turnId: 'turn-1', state: 'running', startedAt: NOW - 5_000 }, + { kind: 'turn', turnId: 'turn-1', state: 'running', startedAt: SEEDED_TURN_STARTED_AT }, { fence: 13 } ) await journal.close() @@ -306,6 +309,16 @@ describe('already-wedged profiles become usable on load', () => { 'a quit that left the owner for a probe to prove gone', wedgedRecord({ claimStatus: 'live', handoffStage: null, ownerProcess: DEAD_OWNER }), null, + { state: 'interrupted', completedAt: SEEDED_TURN_STARTED_AT } + ], + [ + 'a quit that left an owner on a host this one cannot probe', + wedgedRecord({ + claimStatus: 'live', + handoffStage: null, + ownerProcess: { ...DEAD_OWNER, hostId: 'remote-host' } + }), + null, { state: 'unverifiable' } ] ] as const)( @@ -314,8 +327,18 @@ describe('already-wedged profiles become usable on load', () => { await seedStore({ ...seeded, lease: { ...seeded.lease, deathEvidence } }) await seedRunningTurn() const published: AgentSessionStatusSummary[] = [] + const remote = seeded.lease.ownerProcess?.hostId === 'remote-host' openHost({ - statusSink: { publish: (summary) => published.push(summary), forget: () => {} } + statusSink: { publish: (summary) => published.push(summary), forget: () => {} }, + // Loss of contact is never proof of death: a remote owner only ever probes indeterminate. + ...(remote + ? { + probeOwner: async () => ({ + outcome: 'indeterminate' as const, + reason: 'owner runs on remote-host, which this host cannot probe' + }) + } + : {}) }) await host.restoreReadableSessions() @@ -328,9 +351,25 @@ describe('already-wedged profiles become usable on load', () => { ...verdict }) expect(activeStructuredAgentSessionTurnId(restoredJournal().snapshot().items)).toBe(null) + // The row never carries the proof's detail, which is Orca's log text. + const statusRows = restoredJournal() + .snapshot() + .items.flatMap((item) => (item.body.kind === 'status' ? [item.body.text] : [])) + expect(statusRows).toEqual( + remote + ? [] + : [ + 'Codex stopped while this response was in progress. You can continue in this conversation.' + ] + ) // What the sidebar reads: every status this restart published says the chat is not working. expect(published.filter((summary) => summary.sessionId === SESSION)).not.toEqual([]) expect(published.map((summary) => summary.status)).not.toContain('working') + // A crash is not something the user did: no outcome is claimed, so no reader files it as a + // cancellation the user already knows about. + expect(published.map((summary) => summary.turnOutcome)).toEqual( + published.map(() => undefined) + ) } ) @@ -350,11 +389,12 @@ describe('already-wedged profiles become usable on load', () => { expect(acquire).toHaveBeenCalledOnce() expect(activeStructuredAgentSessionTurnId(restoredJournal().snapshot().items)).toBe(null) - // A pid probe proved the owner gone; nobody saw it exit, so the turn has no end. + // A pid probe proved the owner gone, so the turn was cut short when it was last proven alive. expect(turnLifecycle('turn-1')).toEqual({ turnId: 'turn-1', - state: 'unverifiable', + state: 'interrupted', startedAt: NOW - 5_000, + completedAt: SEEDED_TURN_STARTED_AT, recovered: true }) expect(store.getRecord(SESSION)?.lease).toMatchObject({ @@ -443,11 +483,12 @@ describe('already-wedged profiles become usable on load', () => { expect(acquire).toHaveBeenCalledOnce() expect(store.getRecord(SESSION)?.lease).toMatchObject({ claimStatus: 'live' }) - // A pid probe proved the owner gone; nobody saw it exit, so the turn has no end. + // A pid probe proved the owner gone, so the turn was cut short when it was last proven alive. expect(turnLifecycle('turn-1')).toEqual({ turnId: 'turn-1', - state: 'unverifiable', + state: 'interrupted', startedAt: NOW - 5_000, + completedAt: SEEDED_TURN_STARTED_AT, recovered: true }) } diff --git a/src/main/runtime/agent-session-acquisition-failure-settlement.ts b/src/main/runtime/agent-session-acquisition-failure-settlement.ts index a3e114081186..f6efaa890c37 100644 --- a/src/main/runtime/agent-session-acquisition-failure-settlement.ts +++ b/src/main/runtime/agent-session-acquisition-failure-settlement.ts @@ -86,12 +86,14 @@ export function settleFailedAgentSessionPostAcquisitionAttachment( } const next = args.exitProof === 'unproven' - ? withLease(record, { - ...record.lease, - handoffStage: 'recovering', - handoffOperationId: null, - lastRenewedAt: args.now - }) + ? { + ...withLease(record, { + ...record.lease, + handoffStage: 'recovering', + handoffOperationId: null + }), + updatedAt: args.now + } : withLease(record, { ...record.lease, runtimeFence: nextAgentSessionFence(record.lease), @@ -101,18 +103,15 @@ export function settleFailedAgentSessionPostAcquisitionAttachment( claimStatus: 'released', lastRenewedAt: args.now, handoffOperationId: null, - deathEvidence: - args.exitProof === 'root-exit-observed' - ? { - kind: 'exit-observed', - detail: 'the provider process exited; its descendants were not proven gone', - observedAt: args.now - } - : { - kind: 'exit-observed', - detail: 'post-acquisition cleanup proved no provider child remains', - observedAt: args.now - } + deathEvidence: { + kind: 'exit-observed', + detail: + args.exitProof === 'root-exit-observed' + ? 'the provider process exited; its descendants were not proven gone' + : 'post-acquisition cleanup proved no provider child remains', + observedAt: args.now, + ownerFence: record.lease.runtimeFence + } }) state.records.set(args.sessionId, next) state.operations = settleAgentSessionOperation(state.operations, args) @@ -133,14 +132,17 @@ function settleFailedLease( throw new Error('agent_session_ownership_unknown') } if (args.exitProof === 'unproven' && record.lease.ownerProcess) { - return withLease(record, { - ...record.lease, - handoffStage: 'recovering', - // 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 - }) + // Parking in recovery proves nothing alive, so `lastRenewedAt` keeps the spawn's proof. + return { + ...withLease(record, { + ...record.lease, + handoffStage: 'recovering', + // 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 + }), + updatedAt: args.now + } } return withLease(record, { ...record.lease, @@ -151,7 +153,7 @@ function settleFailedLease( claimStatus: 'released', lastRenewedAt: args.now, handoffOperationId: null, - deathEvidence: acquisitionDeathEvidence(args.exitProof, args.now) + deathEvidence: acquisitionDeathEvidence(args.exitProof, args.now, record.lease.runtimeFence) }) } @@ -159,25 +161,27 @@ function settleFailedLease( * when nothing was. */ function acquisitionDeathEvidence( exitProof: AgentSessionAcquisitionExitProof, - observedAt: number + observedAt: number, + ownerFence: number ): AgentSessionDeathEvidence | null { if (exitProof === 'unproven') { return null } + const proof = { observedAt, ownerFence } if (exitProof === 'processless') { - return { kind: 'pid-absent', detail: 'reservation failed before spawn', observedAt } + return { kind: 'pid-absent', detail: 'reservation failed before spawn', ...proof } } if (exitProof === 'root-exit-observed') { return { kind: 'exit-observed', detail: 'the provider process exited; its descendants were not proven gone', - observedAt + ...proof } } // Cleanup proved no child of this attempt remains; it may never have spawned. return { kind: 'exit-observed', detail: 'acquisition cleanup proved no provider child remains', - observedAt + ...proof } } diff --git a/src/main/runtime/agent-session-death-evidence-persistence.test.ts b/src/main/runtime/agent-session-death-evidence-persistence.test.ts new file mode 100644 index 000000000000..6e0bff8098cd --- /dev/null +++ b/src/main/runtime/agent-session-death-evidence-persistence.test.ts @@ -0,0 +1,228 @@ +import { mkdtemp, rm, writeFile } 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 { AgentSessionDeathEvidence } from '../../shared/agent-session-record' +import { + agentSessionLeaseFixture, + agentSessionRecordFixture +} from '../../shared/agent-session-record.test-fixture' +import type { AgentSessionFailedAcquisitionSettlement } from './agent-session-acquisition-failure-settlement' +import { AgentSessionRecordStore } from './agent-session-record-store' +import { agentSessionStorePath } from './agent-session-record-store-file' + +const SESSION = 'session-alpha-1' +/** The fixture lease's last renewal. */ +const LAST_RENEWED_AT = 30_000 + +let directory: string + +async function seed(deathEvidence: AgentSessionDeathEvidence | null): Promise { + const lease = + deathEvidence === null + ? agentSessionLeaseFixture() + : agentSessionLeaseFixture({ + ownerProcess: null, + reservedSpawnToken: null, + claimStatus: 'released', + deathEvidence + }) + await writeFile( + agentSessionStorePath(directory), + JSON.stringify({ + schemaVersion: 2, + hostId: 'local', + records: { [SESSION]: agentSessionRecordFixture(lease) }, + operations: {}, + retiredClaimKeys: [], + unusableRecords: {} + }) + ) +} + +function open(): Promise { + return AgentSessionRecordStore.open({ directory, hostId: 'local' }) +} + +beforeEach(async () => { + directory = await mkdtemp(join(tmpdir(), 'orca-death-evidence-')) +}) + +afterEach(async () => { + await rm(directory, { recursive: true, force: true }) +}) + +describe('death evidence on disk', () => { + it('records the last renewal before the crash and reads it back after another restart', async () => { + await seed(null) + const crashed = await open() + await crashed.reconcileOnRestart({ + probe: async () => ({ outcome: 'pid-absent' }), + now: 90_000 + }) + const evidence = { + kind: 'pid-absent', + detail: 'recorded pid absent on host', + observedAt: 90_000, + ownerFence: 7, + lastProvenAliveAt: LAST_RENEWED_AT + } + expect(crashed.getRecord(SESSION)?.lease.deathEvidence).toEqual(evidence) + expect((await open()).getRecord(SESSION)?.lease.deathEvidence).toEqual(evidence) + }) + + it('loads evidence an older build wrote without a proven-alive time', async () => { + const olderBuild = { kind: 'pid-absent' as const, detail: 'gone', observedAt: 90_000 } + await seed(olderBuild) + const store = await open() + expect(store.isSessionUnreadable(SESSION)).toBe(false) + expect(store.getRecord(SESSION)?.lease.deathEvidence).toEqual(olderBuild) + }) + + it.each([ + ['proves the owner alive after the probe found it gone', { lastProvenAliveAt: 90_001 }], + ['names no possible owner', { ownerFence: -1 }] + ])('quarantines evidence that %s', async (_why, field) => { + await seed({ kind: 'pid-absent', detail: 'gone', observedAt: 90_000, ...field }) + expect((await open()).isSessionUnreadable(SESSION)).toBe(true) + }) +}) + +describe('who is told a proof of death landed', () => { + it('tells a listener once the proof is committed, and never for a write that proves nothing', async () => { + await seed(null) + const store = await open() + const told: unknown[] = [] + store.onDeathEvidence((sessionId) => told.push(store.getRecord(sessionId)?.lease.deathEvidence)) + const unsubscribed = vi.fn() + store.onDeathEvidence(unsubscribed)() + + await expect( + store.evictProvenDeadOwner({ + sessionId: SESSION, + expectedFence: 6, + probe: { outcome: 'pid-absent' }, + now: 80_000 + }) + ).rejects.toThrow() + await store.reconcileOnRestart({ probe: async () => ({ outcome: 'pid-absent' }), now: 90_000 }) + // The proof already on the record is not news to a later write. + await store.setSessionTabVisibility(SESSION, true) + + expect(told).toEqual([expect.objectContaining({ kind: 'pid-absent', ownerFence: 7 })]) + expect(unsubscribed).not.toHaveBeenCalled() + }) +}) + +describe('a failed acquisition', () => { + const NOW = 1_800_000_000_000 + const OPERATION_ID = `${NOW}-${'1'.padStart(32, '0')}` + + /** Reserve and observe the spawn at NOW, as an attach does before it can fail. */ + async function spawnedOwner(store: AgentSessionRecordStore): Promise { + const reserved = await store.reserveOwner({ + sessionId: SESSION, + location: agentSessionRecordFixture().location, + provider: 'claude', + accountHome: { variable: 'CLAUDE_CONFIG_DIR', path: '/home/dev/.claude' }, + expectedFence: null, + spawnToken: 'spawn-a', + claimKeyId: 'key-1', + handoffOperationId: OPERATION_ID, + probe: { outcome: 'indeterminate', reason: 'no answer' }, + operation: { callerKey: 'client-1', operationId: OPERATION_ID, fingerprint: 'fp-1' }, + now: NOW + }) + const fence = reserved.record.lease.runtimeFence + await store.commitProcessIdentity({ + sessionId: SESSION, + fence, + process: { hostId: 'local', pid: 4242, processStartTimeMs: NOW, spawnToken: 'spawn-a' }, + now: NOW + }) + return fence + } + + function unproven(fence: number, now: number): AgentSessionFailedAcquisitionSettlement { + return { + sessionId: SESSION, + fence, + spawnToken: 'spawn-a', + callerKey: 'client-1', + operationId: OPERATION_ID, + outcome: { status: 'failed', code: 'agent_session_operation_invalid', message: 'failed' }, + exitProof: 'unproven', + now + } + } + + it.each([ + ['before the owner proved its handle', false], + ['after the owner proved its handle', true] + ] as const)( + 'parks in recovery keeping the last proof of life, not the failure, %s', + async (_when, proved) => { + const store = await open() + const fence = await spawnedOwner(store) + if (proved) { + await store.proveOwner({ + sessionId: SESSION, + fence, + link: { + linkId: 'link-1', + handle: { provider: 'claude', sessionId: 'provider-session-1', leafUuid: null }, + origin: 'created', + mintedAtFence: fence, + observedAt: NOW + }, + now: NOW + }) + await store.settleFailedPostAcquisitionAttachment(unproven(fence, NOW + 60_000)) + } else { + await store.settleFailedAcquisition(unproven(fence, NOW + 60_000)) + } + expect(store.getRecord(SESSION)?.lease).toMatchObject({ + handoffStage: 'recovering', + lastRenewedAt: NOW + }) + + const evicted = await store.evictProvenDeadOwner({ + sessionId: SESSION, + expectedFence: fence, + probe: { outcome: 'pid-absent' }, + now: NOW + 120_000 + }) + expect(evicted.lease.deathEvidence).toMatchObject({ lastProvenAliveAt: NOW }) + } + ) + + it.each(['exit-proven', 'root-exit-observed', 'processless'] as const)( + 'names its own fence in the proof when cleanup settles it %s', + async (exitProof) => { + const store = await open() + const fence = await spawnedOwner(store) + const settled = + exitProof === 'processless' + ? await store.settleFailedAcquisition({ ...unproven(fence, NOW), exitProof }) + : await (async () => { + await store.proveOwner({ + sessionId: SESSION, + fence, + link: { + linkId: 'link-1', + handle: { provider: 'claude', sessionId: 'provider-session-1', leafUuid: null }, + origin: 'created', + mintedAtFence: fence, + observedAt: NOW + }, + now: NOW + }) + return store.settleFailedPostAcquisitionAttachment({ + ...unproven(fence, NOW), + exitProof + }) + })() + expect(settled.lease.deathEvidence).toMatchObject({ ownerFence: fence }) + } + ) +}) diff --git a/src/main/runtime/agent-session-record-store.ts b/src/main/runtime/agent-session-record-store.ts index 314323a32380..925fa38d8d18 100644 --- a/src/main/runtime/agent-session-record-store.ts +++ b/src/main/runtime/agent-session-record-store.ts @@ -77,6 +77,8 @@ export const AGENT_SESSION_LEASE_TTL_MS = 30_000, AGENT_SESSION_LEASE_RENEW_INTERVAL_MS = 10_000 export class AgentSessionRecordStore { + private readonly deathEvidenceListeners = new Set<(sessionId: string) => void>() + private constructor(private readonly transactions: AgentSessionStoreTransactionQueue) {} static async open(args: { directory: string; hostId: string }): Promise { @@ -328,6 +330,32 @@ export class AgentSessionRecordStore { }) } + /** Told, once committed, of each session a transaction wrote a proof of death for — whichever + * transition wrote it, since every one lands here. Must not throw. */ + onDeathEvidence(listener: (sessionId: string) => void): () => void { + this.deathEvidenceListeners.add(listener) + return () => this.deathEvidenceListeners.delete(listener) + } + /** Serialize every mutation against the latest committed disk state. */ - private transact = (apply: () => T): Promise => this.transactions.transact(apply) + private transact = async (apply: () => T): Promise => { + let proven: string[] = [] + const result = await this.transactions.transact(() => { + if (this.deathEvidenceListeners.size === 0) { + return apply() + } + const before = new Map( + [...this.state.records].map(([id, record]) => [id, record.lease.deathEvidence]) + ) + const applied = apply() + proven = [...this.state.records] + .filter(([id, { lease }]) => lease.deathEvidence && lease.deathEvidence !== before.get(id)) + .map(([id]) => id) + return applied + }) + for (const sessionId of proven) { + this.deathEvidenceListeners.forEach((listener) => listener(sessionId)) + } + return result + } } diff --git a/src/main/runtime/agent-session-restart-lease-transitions.ts b/src/main/runtime/agent-session-restart-lease-transitions.ts index 70104588a584..b4269d21c677 100644 --- a/src/main/runtime/agent-session-restart-lease-transitions.ts +++ b/src/main/runtime/agent-session-restart-lease-transitions.ts @@ -52,10 +52,13 @@ export function applyAgentSessionRestartAdjudication(args: { deathEvidence: adjudication.evidence }) } - return withLease(record, { - ...record.lease, - handoffStage: adjudication.stage, - unreconciled: false, - lastRenewedAt: args.now - }) + // Parking in recovery proves nothing alive, so `lastRenewedAt` keeps the pre-crash proof. + return { + ...withLease(record, { + ...record.lease, + handoffStage: adjudication.stage, + unreconciled: false + }), + updatedAt: args.now + } } diff --git a/src/main/runtime/agent-session-surface-release-transition.test.ts b/src/main/runtime/agent-session-surface-release-transition.test.ts index 5e671acc77f5..0ea30d901976 100644 --- a/src/main/runtime/agent-session-surface-release-transition.test.ts +++ b/src/main/runtime/agent-session-surface-release-transition.test.ts @@ -18,5 +18,7 @@ describe('agent session surface release transition', () => { }) expect(released.lease.runtimeFence).toBe(9) + // The proof names the owner it released, not the fence the floor moved to. + expect(released.lease.deathEvidence).toMatchObject({ kind: 'exit-observed', ownerFence: 7 }) }) }) diff --git a/src/main/runtime/agent-session-surface-release-transition.ts b/src/main/runtime/agent-session-surface-release-transition.ts index 7da1e7633eff..fe98a8705248 100644 --- a/src/main/runtime/agent-session-surface-release-transition.ts +++ b/src/main/runtime/agent-session-surface-release-transition.ts @@ -51,7 +51,8 @@ export function releaseAgentSessionOwnerAfterSurfaceClose(args: { deathEvidence: { kind: 'exit-observed', detail: args.exitReason ?? 'the last surface holding this session released it', - observedAt: args.exitObservedAt ?? args.now + observedAt: args.exitObservedAt ?? args.now, + ownerFence: record.lease.runtimeFence } }) } diff --git a/src/main/runtime/rpc/methods/structured-agent-session-at-rest.test.ts b/src/main/runtime/rpc/methods/structured-agent-session-at-rest.test.ts index df00957215f0..1cc5050d8031 100644 --- a/src/main/runtime/rpc/methods/structured-agent-session-at-rest.test.ts +++ b/src/main/runtime/rpc/methods/structured-agent-session-at-rest.test.ts @@ -424,7 +424,10 @@ describe('an agent exit', () => { { kind: 'turn', turnId: 'working', state: 'running' } ) await rig.host.flushStreamedEvents(SESSION) - vi.spyOn(open.journal, 'appendLifecycleBatch').mockRejectedValueOnce(new Error('disk full')) + // The exit's own write and the retry recording the exit queues are both refused. + vi.spyOn(open.journal, 'appendLifecycleBatch') + .mockRejectedValueOnce(new Error('disk full')) + .mockRejectedValueOnce(new Error('disk full')) await rig.host.handleAdapterEvent({ type: 'ended', sessionId: SESSION, @@ -440,6 +443,7 @@ describe('an agent exit', () => { deathEvidence: { kind: 'exit-observed' } }) ) + await rig.host.collaboratorsForTests().serialize(SESSION, async () => {}) const sent = await rig.host.send( CALLER, diff --git a/src/shared/agent-session-lease-adjudication.test.ts b/src/shared/agent-session-lease-adjudication.test.ts index 7256f940aa43..35e794efd0d2 100644 --- a/src/shared/agent-session-lease-adjudication.test.ts +++ b/src/shared/agent-session-lease-adjudication.test.ts @@ -222,14 +222,28 @@ describe('restart reconciliation', () => { expect(result).toEqual({ disposition: 'evicted', nextFence: 8, + // The death happened between the last renewal and the probe. evidence: { kind: 'identity-mismatch', detail: 'mismatched process-start-time', - observedAt: 9_000 + observedAt: 9_000, + ownerFence: 7, + lastProvenAliveAt: 500 } }) }) + it('never records the owner alive after the probe that found it gone', () => { + // A clock stepped back across the restart would otherwise write an interval load rejects. + expect( + adjudicateAgentSessionRestart({ + lease: lease({ lastRenewedAt: 9_500 }), + probe: { outcome: 'pid-absent' }, + observedAt: 9_000 + }) + ).toMatchObject({ evidence: { observedAt: 9_000, lastProvenAliveAt: 9_000 } }) + }) + it('hands an unverifiable owner to recovery resolution instead of evicting it', () => { expect( adjudicateAgentSessionRestart({ @@ -269,7 +283,13 @@ describe('restart reconciliation', () => { ).toEqual({ disposition: 'evicted', nextFence: 8, - evidence: { kind: 'pid-absent', detail: 'recorded pid absent on host', observedAt: 9_000 } + evidence: { + kind: 'pid-absent', + detail: 'recorded pid absent on host', + observedAt: 9_000, + ownerFence: 7, + lastProvenAliveAt: 500 + } }) }) @@ -308,7 +328,13 @@ describe('restart reconciliation', () => { ).toEqual({ disposition: 'evicted', nextFence: 8, - evidence: { kind: 'pid-absent', detail: 'reservation never spawned', observedAt: 9_000 } + // The proof is about this reservation's own fence. + evidence: { + kind: 'pid-absent', + detail: 'reservation never spawned', + observedAt: 9_000, + ownerFence: 7 + } }) } ) diff --git a/src/shared/agent-session-lease-adjudication.ts b/src/shared/agent-session-lease-adjudication.ts index b0b806ddad72..271b136456f0 100644 --- a/src/shared/agent-session-lease-adjudication.ts +++ b/src/shared/agent-session-lease-adjudication.ts @@ -80,16 +80,23 @@ export function isProvenAliveProbe(probe: AgentSessionOwnerProbe): boolean { function deathEvidenceFor( probe: AgentSessionOwnerProbe, - observedAt: number + observedAt: number, + lease: AgentSessionLease ): AgentSessionDeathEvidence | null { + // Capped so a clock that stepped back across the restart still writes a valid interval. + const interval = { + observedAt, + ownerFence: lease.runtimeFence, + lastProvenAliveAt: Math.min(lease.lastRenewedAt, observedAt) + } if (probe.outcome === 'exit-observed') { - return { kind: 'exit-observed', detail: 'observed process exit', observedAt } + return { kind: 'exit-observed', detail: 'observed process exit', ...interval } } if (probe.outcome === 'pid-absent') { - return { kind: 'pid-absent', detail: 'recorded pid absent on host', observedAt } + return { kind: 'pid-absent', detail: 'recorded pid absent on host', ...interval } } if (probe.outcome === 'identity-mismatch') { - return { kind: 'identity-mismatch', detail: `mismatched ${probe.field}`, observedAt } + return { kind: 'identity-mismatch', detail: `mismatched ${probe.field}`, ...interval } } return null } @@ -240,7 +247,12 @@ export function adjudicateAgentSessionRestart(args: { nextFence: nextAgentSessionFence(lease), evidence: probe.outcome === 'reservation-unused' - ? { kind: 'pid-absent', detail: 'reservation never spawned', observedAt } + ? { + kind: 'pid-absent', + detail: 'reservation never spawned', + observedAt, + ownerFence: lease.runtimeFence + } : null } } @@ -253,7 +265,7 @@ export function adjudicateAgentSessionRestart(args: { reason: 'owner outlived the runtime that held its transport' } } - const evidence = deathEvidenceFor(probe, observedAt) + const evidence = deathEvidenceFor(probe, observedAt, lease) if (evidence) { return { disposition: 'evicted', nextFence: nextAgentSessionFence(lease), evidence } } diff --git a/src/shared/agent-session-record.ts b/src/shared/agent-session-record.ts index 2b85695a8459..12a26e64ac77 100644 --- a/src/shared/agent-session-record.ts +++ b/src/shared/agent-session-record.ts @@ -87,6 +87,13 @@ export type AgentSessionDeathEvidence = { kind: 'exit-observed' | 'pid-absent' | 'identity-mismatch' detail: string observedAt: number + /** Fence of the owner (or reservation) this death is about; a fence names exactly one. Absent on + * evidence older builds wrote, which then speaks for no turn. */ + ownerFence?: number + /** The death interval's lower bound: the last time the runtime holding the owner's transport + * proved it alive. Only a probe's proof records it: absent on a surface-release exit, a failed + * start, and evidence older builds wrote. */ + lastProvenAliveAt?: number } export type AgentSessionLease = { @@ -102,6 +109,8 @@ export type AgentSessionLease = { /** Reserved before any process exists, then matched against the child's environment. */ reservedSpawnToken: string | null leaseDeadlineAt: number + /** While `ownerProcess` is set, the last time its transport holder proved it alive; parking in + * `recovering` proves nothing, so it leaves this alone. */ lastRenewedAt: number handoffOperationId: string | null journalCheckpoint: AgentSessionJournalCheckpoint | null @@ -276,13 +285,20 @@ function isAgentSessionDeathEvidence(value: unknown): value is AgentSessionDeath return false } const evidence = value as Partial + const { observedAt, lastProvenAliveAt, ownerFence } = evidence return ( (evidence.kind === 'exit-observed' || evidence.kind === 'pid-absent' || evidence.kind === 'identity-mismatch') && isBoundedString(evidence.detail, MAX_ID_LENGTH) && - Number.isSafeInteger(evidence.observedAt) && - (evidence.observedAt as number) >= 0 + typeof observedAt === 'number' && + Number.isSafeInteger(observedAt) && + observedAt >= 0 && + (ownerFence === undefined || (Number.isSafeInteger(ownerFence) && ownerFence >= 0)) && + (lastProvenAliveAt === undefined || + (Number.isSafeInteger(lastProvenAliveAt) && + lastProvenAliveAt >= 0 && + lastProvenAliveAt <= observedAt)) ) } diff --git a/tests/e2e/cross-version-wire/agent-session-death-evidence-downgrade.unit.test.ts b/tests/e2e/cross-version-wire/agent-session-death-evidence-downgrade.unit.test.ts new file mode 100644 index 000000000000..96780a185176 --- /dev/null +++ b/tests/e2e/cross-version-wire/agent-session-death-evidence-downgrade.unit.test.ts @@ -0,0 +1,61 @@ +import { mkdtempSync, rmSync, writeFileSync } from 'node:fs' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { expect, test } from 'vitest' +import { + agentSessionLeaseFixture, + agentSessionRecordFixture +} from '../../../src/shared/agent-session-record.test-fixture' +import { AgentSessionRecordStore } from '../../../src/main/runtime/agent-session-record-store' +import { agentSessionStorePath } from '../../../src/main/runtime/agent-session-record-store-file' +import { importReleaseCheckoutModule, materializeReleaseCheckout } from './release-checkout' + +// The last release before a proof of death named its owner and its last proof of life. +const BASELINE_REF = 'v1.4.211' +const SESSION = 'session-alpha-1' + +test('an older build loads a proof of death that names its owner and its last proof of life', async () => { + const directory = mkdtempSync(join(tmpdir(), 'orca-death-evidence-downgrade-')) + try { + // A live owner at fence 7, last renewed at 30000, found gone when this build relaunches. + writeFileSync( + agentSessionStorePath(directory), + JSON.stringify({ + schemaVersion: 2, + hostId: 'local', + records: { [SESSION]: agentSessionRecordFixture(agentSessionLeaseFixture()) }, + operations: {}, + retiredClaimKeys: [], + unusableRecords: {} + }) + ) + const store = await AgentSessionRecordStore.open({ directory, hostId: 'local' }) + await store.reconcileOnRestart({ probe: async () => ({ outcome: 'pid-absent' }), now: 90_000 }) + const evidence = store.getRecord(SESSION)?.lease.deathEvidence + expect(evidence).toEqual({ + kind: 'pid-absent', + detail: 'recorded pid absent on host', + observedAt: 90_000, + ownerFence: 7, + lastProvenAliveAt: 30_000 + }) + + const checkout = await materializeReleaseCheckout(BASELINE_REF) + const baseline = await importReleaseCheckoutModule( + checkout, + 'src/main/runtime/agent-session-record-store.ts' + ) + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: the pinned release exports this class with the open/read members called below; a missing one fails the test. + const OldStore = baseline.AgentSessionRecordStore as { + open: (args: { directory: string; hostId: string }) => Promise + } + const old = await OldStore.open({ directory, hostId: 'local' }) + expect(old.isSessionUnreadable(SESSION)).toBe(false) + expect(old.getRecord(SESSION)?.lease).toMatchObject({ + claimStatus: 'released', + deathEvidence: evidence + }) + } finally { + rmSync(directory, { recursive: true, force: true }) + } +})