diff --git a/config/scripts/vitest-sqlite-runtime-files.mjs b/config/scripts/vitest-sqlite-runtime-files.mjs index d839482b65c9..d4d7279dd147 100644 --- a/config/scripts/vitest-sqlite-runtime-files.mjs +++ b/config/scripts/vitest-sqlite-runtime-files.mjs @@ -137,12 +137,16 @@ export const SQLITE_RUNTIME_INCLUDE = [ 'src/main/native-chat/agent-session-wire/structured-agent-session-repeat-press.test.ts', 'src/main/native-chat/agent-session-wire/structured-agent-session-repeated-stop.test.ts', 'src/main/native-chat/agent-session-wire/structured-agent-session-resend-answer.test.ts', + 'src/main/native-chat/agent-session-wire/structured-agent-session-replaced-server-turn.test.ts', 'src/main/native-chat/agent-session-wire/structured-agent-session-resend-ledger-refusal.test.ts', 'src/main/native-chat/agent-session-wire/structured-agent-session-resolved-append.test.ts', 'src/main/native-chat/agent-session-wire/structured-agent-session-restart-offer-newer-orca.test.ts', 'src/main/native-chat/agent-session-wire/structured-agent-session-restart-offer-withdrawal.test.ts', + 'src/main/native-chat/agent-session-wire/structured-agent-session-restart-provider-output.test.ts', + 'src/main/native-chat/agent-session-wire/structured-agent-session-restart-recovery-output.test.ts', 'src/main/native-chat/agent-session-wire/structured-agent-session-restart-restore.test.ts', 'src/main/native-chat/agent-session-wire/structured-agent-session-restart-status-publication.test.ts', + 'src/main/native-chat/agent-session-wire/structured-agent-session-restart-stream-duration.test.ts', 'src/main/native-chat/agent-session-wire/structured-agent-session-rewind-at-rest.test.ts', 'src/main/native-chat/agent-session-wire/structured-agent-session-rewind-live-send.test.ts', 'src/main/native-chat/agent-session-wire/structured-agent-session-rewind.test.ts', diff --git a/src/main/codex/codex-background-terminals.ts b/src/main/codex/codex-background-terminals.ts index a70205183eaa..f8dcd1f98fe0 100644 --- a/src/main/codex/codex-background-terminals.ts +++ b/src/main/codex/codex-background-terminals.ts @@ -225,13 +225,17 @@ export async function stopCodexBackgroundCommands( return { cancelled: false } } const { acquisitionGeneration } = session - const { stopped, survived } = await terminateCodexBackgroundTerminals(session.connection, terminals, { - timeoutMs, - isCurrent: () => - sessions.get(input.sessionId) === session && - !session.ended && - session.fence === input.fence && - session.acquisitionGeneration === acquisitionGeneration - }) + const { stopped, survived } = await terminateCodexBackgroundTerminals( + session.connection, + terminals, + { + timeoutMs, + isCurrent: () => + sessions.get(input.sessionId) === session && + !session.ended && + session.fence === input.fence && + session.acquisitionGeneration === acquisitionGeneration + } + ) return { cancelled: stopped > 0, ...(survived ? { stillRunning: true as const } : {}) } } diff --git a/src/main/native-chat/agent-session-journal/journal-open.ts b/src/main/native-chat/agent-session-journal/journal-open.ts index 2c066601ea7a..577efeccedd1 100644 --- a/src/main/native-chat/agent-session-journal/journal-open.ts +++ b/src/main/native-chat/agent-session-journal/journal-open.ts @@ -57,7 +57,7 @@ export function replayJournal(db: Database.Database, sessionId: string): Journal /** The same fold, fed a row at a time, for a caller that yields between batches of rows. */ export function startJournalRowFold(input: { sessionId: string; epoch: string }): { /** False once the fold has stopped: the rest of the rows are not read. */ - add: (entry: { seq: number; rowJson: string }) => boolean + add: (entry: { seq: number; rowJson: string; ts?: number }) => boolean finish: () => JournalLoad } { const state = createJournalReducerState(input.sessionId, input.epoch) @@ -67,7 +67,7 @@ export function startJournalRowFold(input: { sessionId: string; epoch: string }) let newer: { sequence: number } | null = null let empty = true - const add = (entry: { seq: number; rowJson: string }): boolean => { + const add = (entry: { seq: number; rowJson: string; ts?: number }): boolean => { empty = false const parsed = parseJournalRow(entry.rowJson) if (!parsed.ok && parsed.unreadable) { @@ -99,7 +99,7 @@ export function startJournalRowFold(input: { sessionId: string; epoch: string }) return true } anchored = true - applyJournalRow(state, row) + applyJournalRow(state, row, entry.ts) return true } const finish = (): JournalLoad => { diff --git a/src/main/native-chat/agent-session-journal/journal-provider-activity.ts b/src/main/native-chat/agent-session-journal/journal-provider-activity.ts new file mode 100644 index 000000000000..09a965b624ef --- /dev/null +++ b/src/main/native-chat/agent-session-journal/journal-provider-activity.ts @@ -0,0 +1,46 @@ +import { parseAgentJournalItemKey } from '../../../shared/agent-session-journal-item-key' +import type { AgentJournalItemBody } from '../../../shared/agent-session-journal-types' +import { readAgentJournalTurn } from '../../../shared/agent-session-turn-record' +import type { JournalReducerState } from './journal-reducer' +import type { JournalRow } from './journal-row-schema' + +/** Saved output extends a turn; client actions and recovery writes do not. */ +export function observeJournalProviderActivity( + state: Pick, + row: JournalRow, + itemId: string, + body: AgentJournalItemBody, + savedAt: number +): void { + if (row.recovered || !isProviderActivity(body)) { + return + } + const identity = parseAgentJournalItemKey(itemId) + // Codex detail rows use the Orca namespace; fallback rows already name their provider. + if ( + !identity || + (identity.provider === 'orca' && + !identity.clientMessageId.startsWith('codex-item:') && + !(body.kind === 'status' && body.providerFrame !== undefined)) + ) { + return + } + state.providerActivityAt.set( + row.fence, + Math.max(state.providerActivityAt.get(row.fence) ?? 0, savedAt) + ) +} + +function isProviderActivity(body: AgentJournalItemBody): boolean { + if (body.kind === 'message') { + return body.role !== 'user' + } + if (body.kind === 'tool-call') { + return body.endedAs === undefined + } + if (body.kind === 'approval' || body.kind === 'question') { + return body.resolution.state === 'pending' + } + const turn = readAgentJournalTurn(body) + return turn ? turn.state === 'running' : true +} 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 3890b9a7f28f..41da4e5f459b 100644 --- a/src/main/native-chat/agent-session-journal/journal-reducer.ts +++ b/src/main/native-chat/agent-session-journal/journal-reducer.ts @@ -26,6 +26,7 @@ import { agentSessionSendBodyFingerprint } from '../../../shared/structured-agen import { JournalDerivedTurnScope } from './journal-derived-turn-scope' import { removeJournalItem, statedOrDerivedTurnScope, upsertJournalItem } from './journal-item-fold' import { journalItemRevisionIsStale } from './journal-item-revision' +import { observeJournalProviderActivity } from './journal-provider-activity' import { isJournalStopOrResumeRow, type JournalRow } from './journal-row-schema' import { acceptSubmissionFromProviderItem, applyJournalSubmission } from './journal-submission-fold' import { applyJournalDispatchRow } from './journal-dispatch-reducer' @@ -52,6 +53,8 @@ export type JournalReducerState = { items: Map /** Fence of the writer that created each item: the generation a running turn belongs to. */ itemFences: Map + /** Latest saved output per owner, rebuilt during the existing row fold. */ + providerActivityAt: Map /** Revision of a removed item, so a late lower revision cannot resurrect it. */ tombstones: Map submissions: Map @@ -78,6 +81,7 @@ export function createJournalReducerState(sessionId: string, epoch: string): Jou highestFence: 0, items: new Map(), itemFences: new Map(), + providerActivityAt: new Map(), tombstones: new Map(), submissions: new Map(), receipts: new Map(), @@ -89,7 +93,11 @@ export function createJournalReducerState(sessionId: string, epoch: string): Jou } } -export function applyJournalRow(state: JournalReducerState, row: JournalRow): void { +export function applyJournalRow( + state: JournalReducerState, + row: JournalRow, + savedAt = row.ts +): void { state.lastSequence = Math.max(state.lastSequence, row.seq) state.highestFence = Math.max(state.highestFence, row.fence) if (row.kind === 'epoch') { @@ -101,6 +109,7 @@ export function applyJournalRow(state: JournalReducerState, row: JournalRow): vo return } const itemId = resolveJournalItemId(state, row.itemId, row.body) + observeJournalProviderActivity(state, row, row.itemId, row.body, savedAt) acceptSubmissionFromProviderItem(state, row.itemId, itemId, row) upsertJournalItem( state, @@ -132,6 +141,7 @@ export function applyJournalRow(state: JournalReducerState, row: JournalRow): vo continue } const { revision, body } = mutation + observeJournalProviderActivity(state, row, mutation.itemId, body, savedAt) const itemId = resolveJournalItemId(state, mutation.itemId, body) acceptSubmissionFromProviderItem(state, mutation.itemId, itemId, row) const producer = journalBatchMutationProducer(row, mutation) diff --git a/src/main/native-chat/agent-session-journal/journal-row-table.ts b/src/main/native-chat/agent-session-journal/journal-row-table.ts index 9074ae0019e6..7ff185857cf6 100644 --- a/src/main/native-chat/agent-session-journal/journal-row-table.ts +++ b/src/main/native-chat/agent-session-journal/journal-row-table.ts @@ -39,7 +39,8 @@ export function publishJournalSessionEpoch( export function insertJournalRow( db: Database.Database, sessionId: string, - row: JournalRow + row: JournalRow, + savedAt = row.ts ): number { const rowJson = serializeJournalRow(row) // A row the reader rejects would fail the chat's next load, so it is never written: the throw @@ -51,7 +52,8 @@ export function insertJournalRow( `the ${row.kind} row for ${sessionId} would not read back, so it was not written` ) } - db.prepare(INSERT_ROW).run(sessionId, row.epoch, row.seq, row.ts, rowJson) + // The table timestamp records the save; the JSON timestamp can remain pinned to item start. + db.prepare(INSERT_ROW).run(sessionId, row.epoch, row.seq, savedAt, rowJson) return Buffer.byteLength(rowJson, 'utf8') } diff --git a/src/main/native-chat/agent-session-journal/journal-row-writer.test.ts b/src/main/native-chat/agent-session-journal/journal-row-writer.test.ts index 72d884a01d95..24dc191bee99 100644 --- a/src/main/native-chat/agent-session-journal/journal-row-writer.test.ts +++ b/src/main/native-chat/agent-session-journal/journal-row-writer.test.ts @@ -13,6 +13,7 @@ import type { JournalHostDatabase } from './journal-host-database' import type { JournalRow } from './journal-row-schema' import { JournalRowWriter } from './journal-row-writer' import { JournalWriteQueue } from './journal-write-queue' +import { readJournalRowsAfter } from './journal-row-table' import { openTestJournalHostDatabase, readTestJournalRows, @@ -106,4 +107,21 @@ describe('journal row writer', () => { [2] ) }) + + it.each(['single', 'batch'] as const)( + 'keeps the save time separate from a pinned JSON timestamp in a %s write', + async (kind) => { + const { writer, committedRows } = writerHarness() + const pinned = (seq: number) => row(seq, 0) + await (kind === 'single' + ? writer.enqueue(pinned) + : writer.enqueueRows(() => [pinned, pinned])) + + const saved = readJournalRowsAfter(database.db, SESSION_ID, EPOCH, 0) + expect(saved).toHaveLength(kind === 'single' ? 1 : 2) + expect(saved.map((entry) => entry.ts)).toEqual(saved.map(() => 1)) + expect(saved.map((entry) => JSON.parse(entry.rowJson))).toEqual(committedRows) + expect(committedRows.map((entry) => entry.ts)).toEqual(saved.map(() => 0)) + } + ) }) diff --git a/src/main/native-chat/agent-session-journal/journal-row-writer.ts b/src/main/native-chat/agent-session-journal/journal-row-writer.ts index 3bbd9ef09ae0..4818ef20e183 100644 --- a/src/main/native-chat/agent-session-journal/journal-row-writer.ts +++ b/src/main/native-chat/agent-session-journal/journal-row-writer.ts @@ -27,7 +27,7 @@ export type JournalRowWriterDeps = { readOnly: () => boolean highestFence: () => number nextSequence: () => number - commit: (rows: readonly JournalRow[]) => void + commit: (rows: readonly JournalRow[], savedAt: number) => void /** Standing hook run for EVERY appended row — the queued-draft returned * transition rides here so no rejection path can bypass it. Bookkeeping: it * runs in its own savepoint, so its failure is reported and never vetoes the row. */ @@ -48,12 +48,13 @@ export class JournalRowWriter { ): Promise { return this.deps.serialize(() => { assertJournalWritable(this.deps.readOnly(), this.deps.sessionId) - const row = build(this.deps.nextSequence(), this.deps.now()) + const savedAt = this.deps.now() + const row = build(this.deps.nextSequence(), savedAt) assertJournalFence(row.fence, this.deps.highestFence()) try { // One INSERT: the chat's epoch pointer moves only when the epoch does. this.deps.database().transaction((db) => { - insertJournalRow(db, this.deps.sessionId, row) + insertJournalRow(db, this.deps.sessionId, row, savedAt) hook?.(db, row) receipt?.write(db) this.runBookkeeping(db, row) @@ -66,7 +67,7 @@ export class JournalRowWriter { // fail. Rejecting here instead would leave the next append reusing a // sequence the table already holds. The ledger first: it cannot throw, the fold can. receipt?.committed() - this.deps.commit([row]) + this.deps.commit([row], savedAt) return row }) } @@ -98,7 +99,7 @@ export class JournalRowWriter { try { this.deps.database().transaction((db) => { for (const row of rows) { - insertJournalRow(db, this.deps.sessionId, row) + insertJournalRow(db, this.deps.sessionId, row, ts) this.runBookkeeping(db, row) } receipt?.write(db) @@ -108,7 +109,7 @@ export class JournalRowWriter { throw error } receipt?.committed() - this.deps.commit(rows) + this.deps.commit(rows, ts) return rows } diff --git a/src/main/native-chat/agent-session-journal/journal-store-collaborators.ts b/src/main/native-chat/agent-session-journal/journal-store-collaborators.ts index 76a7dff8a303..f5bc7b062a51 100644 --- a/src/main/native-chat/agent-session-journal/journal-store-collaborators.ts +++ b/src/main/native-chat/agent-session-journal/journal-store-collaborators.ts @@ -40,7 +40,7 @@ export type JournalStoreHost = { readOnly: () => boolean cursor: () => AgentJournalCursor adopt: (loaded: JournalLoad) => void - commit: (rows: readonly JournalRow[]) => void + commit: (rows: readonly JournalRow[], savedAt: number) => void journal: () => AgentSessionJournal enqueue: (build: (seq: number, ts: number) => JournalRow) => Promise } diff --git a/src/main/native-chat/agent-session-journal/journal-store-restore.ts b/src/main/native-chat/agent-session-journal/journal-store-restore.ts index 0f5208dea9bb..e8cb28e6d321 100644 --- a/src/main/native-chat/agent-session-journal/journal-store-restore.ts +++ b/src/main/native-chat/agent-session-journal/journal-store-restore.ts @@ -33,9 +33,13 @@ export async function restoreJournalStore( replay: () => replayJournal(database.db, host.identity.sessionId), start: () => collaborators.epochController.start('session_created', 0), adopt: host.adopt, - // Roster notices are about the conversation, not any turn in it. + // Journal-open revisions are host recovery, not new provider output. appendItem: (identity, body, fence) => - host.journal().appendItem(identity, body, { fence, turnScope: AGENT_JOURNAL_THREAD_SCOPE }), + host.journal().appendItem(identity, body, { + fence, + turnScope: AGENT_JOURNAL_THREAD_SCOPE, + recovered: true + }), highestFence: () => host.state().highestFence }) } 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 25e3a7e8d865..9e274658b088 100644 --- a/src/main/native-chat/agent-session-journal/journal-store.ts +++ b/src/main/native-chat/agent-session-journal/journal-store.ts @@ -123,9 +123,9 @@ export class AgentSessionJournal { this.adoptLoadedJournal(loaded) this.onCommitted?.() }, - commit: (rows) => { + commit: (rows, savedAt) => { for (const row of rows) { - applyJournalRow(this.state, row) + applyJournalRow(this.state, row, savedAt) } this.onCommitted?.() }, @@ -249,6 +249,10 @@ export class AgentSessionJournal { /** Fence of the writer that created the item, while it is in the timeline. */ itemFence = (itemId: string): number | undefined => this.state.itemFences.get(itemId) + /** Latest saved output at this fence, excluding client actions and recovery bookkeeping. */ + lastProviderActivityAt = (fence: number): number | undefined => + this.state.providerActivityAt.get(fence) + submissions = (): AgentJournalSubmission[] => [...this.state.submissions.values()] submission = (clientMessageId: string) => this.state.submissions.get(clientMessageId) diff --git a/src/main/native-chat/agent-session-journal/journal-subagent-liveness.test.ts b/src/main/native-chat/agent-session-journal/journal-subagent-liveness.test.ts index 89805d366081..d9362e6008ad 100644 --- a/src/main/native-chat/agent-session-journal/journal-subagent-liveness.test.ts +++ b/src/main/native-chat/agent-session-journal/journal-subagent-liveness.test.ts @@ -306,10 +306,12 @@ describe('journal reopen after the writing host is gone', () => { // The host dies without ever settling them — no `ended`, so no session sweep. await live.close() + clock += 3_600_000 const reopened = await open() const afterRestart = reopened.snapshot().items.at(-1)! expect(afterRestart.itemId).toBe(beforeRestart.itemId) + expect(afterRestart.recovered).toBe(true) expect(rosterOf(afterRestart.body)).toMatchObject([ { id: 'a', state: 'unverifiable' }, { id: 'b', state: 'unverifiable' } diff --git a/src/main/native-chat/agent-session-timeline/provider-timeline-decision.ts b/src/main/native-chat/agent-session-timeline/provider-timeline-decision.ts index 61fdfe4cd496..57558fc73f2d 100644 --- a/src/main/native-chat/agent-session-timeline/provider-timeline-decision.ts +++ b/src/main/native-chat/agent-session-timeline/provider-timeline-decision.ts @@ -76,7 +76,11 @@ export type ProviderTimelineDecision = { /** The event's change to the state, made when the sink admits it. */ commit?: (state: ProviderTimelineState) => void /** The settlement the event owes, read from the journal when it runs. */ - settle?: { what: string; resolve: (journal: Journal) => readonly JournalLifecycleMutationInput[] } + settle?: { + what: string + recovered?: true + resolve: (journal: Journal) => readonly JournalLifecycleMutationInput[] + } writes?: readonly ProviderTimelineItemWrite[] /** The streamed item whose text this event's full snapshot replaces. */ closes?: string diff --git a/src/main/native-chat/agent-session-timeline/provider-timeline-transition-layout.ts b/src/main/native-chat/agent-session-timeline/provider-timeline-transition-layout.ts index f2c221ed8e81..663d8aefd140 100644 --- a/src/main/native-chat/agent-session-timeline/provider-timeline-transition-layout.ts +++ b/src/main/native-chat/agent-session-timeline/provider-timeline-transition-layout.ts @@ -79,6 +79,7 @@ export function planProviderTimelineWrites( plan.settlement({ settlementId: providerTimelineSettlementId(context, serial(), settle.what), reservedBytes: SETTLEMENT_RESERVED_BYTES, + recovered: settle.recovered, resolve: settle.resolve }) } diff --git a/src/main/native-chat/agent-session-timeline/provider-timeline-turn-decisions.ts b/src/main/native-chat/agent-session-timeline/provider-timeline-turn-decisions.ts index e754884ab3ce..973de114862f 100644 --- a/src/main/native-chat/agent-session-timeline/provider-timeline-turn-decisions.ts +++ b/src/main/native-chat/agent-session-timeline/provider-timeline-turn-decisions.ts @@ -260,6 +260,7 @@ export function decideSessionEnd( return { settle: { what: 'session-end', + recovered: true, resolve: (journal) => providerTimelineSettlement(journal, 'session', { turns: runningProviderTimelineTurns(journal), diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-attach-orchestration.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-attach-orchestration.ts index cd419133389c..2690d4b80a79 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-attach-orchestration.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-attach-orchestration.ts @@ -232,6 +232,7 @@ async function runAttachUnderAbort( fence, acquisitionGeneration, deathEvidence: priorDeathEvidence, + replaced: context.deps.store.replacedRuntime(sessionId), failureTextContext: structuredAgentSessionFailureWordsContext(priorRecord) }) } 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 ec7f478501f7..85738139603c 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 @@ -38,7 +38,7 @@ export type OpenedStructuredAgentSessionConversation = { } export type StructuredAgentSessionConversationOpenDeps = { - store: Pick + store: Pick journalDatabase: JournalHostDatabase logger: StructuredAgentSessionHostDeps['logger'] } @@ -79,10 +79,23 @@ export async function openStructuredAgentSessionConversation( return opened.session } -/** The open itself, indexed by nobody yet: the caller adopts the result. */ +/** The open itself, indexed by nobody yet: the caller adopts the result. Any open that settles + * needs the store, which names the runtimes this one replaced. */ +export async function openStructuredAgentSessionConversationJournal( + deps: StructuredAgentSessionConversationOpenDeps, + record: AgentSessionRecord, + options?: StructuredAgentSessionConversationOpenOptions +): Promise +/** An acquisition's open settles nothing itself, so it needs no store. */ export async function openStructuredAgentSessionConversationJournal( deps: Omit, record: AgentSessionRecord, + options: { acquisition: true } +): Promise +export async function openStructuredAgentSessionConversationJournal( + deps: Omit & + Partial>, + record: AgentSessionRecord, options: StructuredAgentSessionConversationOpenOptions = {} ): Promise { const { sessionId } = record @@ -148,7 +161,8 @@ export async function resettleOpenStructuredAgentSessionConversation( } async function settleGoneGeneration( - deps: Pick, + deps: Pick & + Partial>, record: AgentSessionRecord, journal: AgentSessionJournal ): Promise { @@ -159,6 +173,7 @@ async function settleGoneGeneration( fence: record.lease.runtimeFence, acquisitionGeneration: null, deathEvidence: record.lease.deathEvidence ?? null, + replaced: deps.store?.replacedRuntime(record.sessionId), failureTextContext: structuredAgentSessionFailureWordsContext(record) }) } catch (error) { 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 index a36518c5e614..cd9ecc7e7418 100644 --- 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 @@ -32,6 +32,8 @@ import { openTestJournalHostDatabase } from '../agent-session-journal/journal-host-database-test-support' import { openAgentSessionJournal } from '../agent-session-journal/journal-store-factory' +import { AgentSessionJournal } from '../agent-session-journal/journal-store' +import { STALE_SESSION_ROW_PREFIX } from '../../../shared/agent-session-stop-row-identity' import { AgentSessionAcquisitionExitUnprovenError, type StructuredAgentSessionAdapter @@ -181,6 +183,9 @@ function openHost(overrides: Partial): void { }) } +/** The turn ended where its owner was last proven alive, whatever proof ended it. */ +const INTERRUPTED_TURN = { turnId: 'turn-1', state: 'interrupted', completedAt: LAST_RENEWED_AT } + /** What a client reads over the wire. */ async function settledTurn() { return (await host.journalSnapshot(SESSION)).items @@ -195,6 +200,7 @@ beforeEach(async () => { }) afterEach(async () => { + vi.restoreAllMocks() await host?.flushAllStreamedEvents() closeTestJournalHostDatabases() await rm(root, { recursive: true, force: true }) @@ -302,10 +308,11 @@ 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. +// On desktop the chat on screen at relaunch opens before the startup reconcile has probed its owner. +// The Orca that held the owner is gone (this one replaced it), so the open itself ends the turn; the +// probe's later proof has nothing left to revise. 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 () => { + it('reads interrupted at the last renewal from the first open, and a subscriber is sent it once', async () => { openHost({ probeOwner: async () => ({ outcome: 'pid-absent' }) }) const events: AgentSessionSubscribeEvent[] = [] const unsubscribe = await host.subscribe({ @@ -313,18 +320,12 @@ describe('a turn a read reached before the reconcile proved its owner dead', () sessionId: SESSION, emit: (event) => events.push(event) }) - expect(await settledTurn()).toEqual(UNVERIFIABLE_TURN) + expect(await settledTurn()).toMatchObject(INTERRUPTED_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 } - ]) - ) + expect(turnStatesSent(events)).toMatchObject([INTERRUPTED_TURN]) const { items } = await host.journalSnapshot(SESSION) const [timing] = selectStructuredAgentTurnTimings(items).values() expect(completedStructuredAgentTurnSeconds(timing)).toBe(27) @@ -332,7 +333,7 @@ describe('a turn a read reached before the reconcile proved its owner dead', () unsubscribe() }) - it('reports the revision to the status feed as an interruption, which the chat folds as worked', async () => { + it('reports the open to the status feed as an interruption, which the chat folds as worked', async () => { const published: AgentSessionStatusSummary[] = [] openHost({ probeOwner: async () => ({ outcome: 'pid-absent' }), @@ -343,13 +344,13 @@ describe('a turn a read reached before the reconcile proved its owner dead', () published .filter((summary) => summary.sessionId === SESSION && summary.turnOutcome) .map((summary) => summary.turnOutcome) - expect(outcomes().at(-1)).toBe('unconfirmed') + // The sidebar's red Failed until seen; the turn folds as "Worked for 27s" beside its notice row. + expect(outcomes()).toEqual(['interruption']) await host.reconcileRestartLeases() await drainSession() - // The sidebar's red Failed until seen; the turn folds as "Worked for 27s" beside its notice row. - await vi.waitFor(() => expect(outcomes().at(-1)).toBe('interruption')) + expect(outcomes().at(-1)).toBe('interruption') const [timing] = selectStructuredAgentTurnTimings( (await host.journalSnapshot(SESSION)).items ).values() @@ -362,7 +363,7 @@ describe('a turn a read reached before the reconcile proved its owner dead', () ).toEqual({ key: 'workedFor', duration: '27s' }) }) - it('explains the turn its proof revised, so the cut reads once and offers Continue', async () => { + it('explains the turn it cut, so the cut reads once and offers Continue', async () => { // The Orca that crashed recorded that it started, and never that it ended. const crashed = crashedClaudeRecord() const ownerProcess = crashed.lease.ownerProcess @@ -380,7 +381,7 @@ describe('a turn a read reached before the reconcile proved its owner dead', () beginAgentSessionRuntimeRecord(root, 'runtime-crashed', TOOL_STARTED_AT - 60_000) store = await openTestAgentSessionRecordStore(root) openHost({ probeOwner: async () => ({ outcome: 'pid-absent' }) }) - expect(await settledTurn()).toEqual(UNVERIFIABLE_TURN) + expect(await settledTurn()).toMatchObject(INTERRUPTED_TURN) await host.reconcileRestartLeases() await drainSession() @@ -418,7 +419,7 @@ describe('a turn a read reached before the reconcile proved its owner dead', () expect(await host.journalSnapshot(SESSION)).toEqual(settled) }) - it('revises it when recovery stops a child that outlived Orca, with no send', async () => { + it('ends it at the last renewal although recovery stops a child that outlived Orca later', async () => { let alive = true const probeOwner = async (): Promise => alive @@ -441,7 +442,7 @@ describe('a turn a read reached before the reconcile proved its owner dead', () stopOwnerProcess }) await host.history({ sessionId: SESSION, direction: 'tail' }) - expect(await settledTurn()).toEqual(UNVERIFIABLE_TURN) + expect(await settledTurn()).toMatchObject(INTERRUPTED_TURN) // The reconcile only parks the live orphan in recovery; recovery's stop is what proves it gone. await host.restoreReadableSessions() @@ -510,31 +511,32 @@ describe('a turn a read reached before the reconcile proved its owner dead', () ]) }) - it('stays unverifiable when the revision cannot be written, and a later open revises it', async () => { + it('stays running when the open cannot write its settle, and a later open settles it', async () => { let now = RELAUNCHED_AT const log = recordingStructuredAgentSessionLogger() - openHost({ - probeOwner: async () => ({ outcome: 'pid-absent' }), - now: () => now, - logger: log.logger + const append = AgentSessionJournal.prototype.appendLifecycleBatch + let refused = false + vi.spyOn(AgentSessionJournal.prototype, 'appendLifecycleBatch').mockImplementation(function ( + this: AgentSessionJournal, + input + ) { + if (!refused && input.settlementId.startsWith(STALE_SESSION_ROW_PREFIX)) { + refused = true + return Promise.reject(new Error('disk full')) + } + return append.call(this, input) }) - await host.history({ sessionId: SESSION, direction: 'tail' }) - const { journal } = host.collaboratorsForTests().sessions.get(SESSION)! - vi.spyOn(journal, 'appendLifecycleBatch').mockRejectedValueOnce(new Error('disk full')) + openHost({ now: () => now, logger: log.logger }) - await host.reconcileRestartLeases() - await drainSession() + await host.history({ sessionId: SESSION, direction: 'tail' }) expect(log.scopes()).toEqual(['open-dead-generation']) - expect(await settledTurn()).toEqual(UNVERIFIABLE_TURN) - // The proof is durable on the record, so the next open converges. + expect(await settledTurn()).toMatchObject({ turnId: 'turn-1', state: 'running' }) + // Nothing is owed in between: the next open derives the same end again. 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 - }) + expect(await settledTurn()).toMatchObject(INTERRUPTED_TURN) }) }) @@ -583,15 +585,9 @@ async function hostWithFailingFirstStart(failure: Error) { } } -/** 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. +// before the turn is settled; what the record holds afterwards is about the start's own child. The +// turn is judged by the runtime this one replaced, at its owner's last renewal, never by that death. 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( @@ -613,7 +609,7 @@ describe('a turn a newer start could not settle before it failed', () => { }) await expect(attach(16)).resolves.toMatchObject({ ok: true }) - expect(await settledTurn()).toEqual(UNVERIFIABLE_TURN) + expect(await settledTurn()).toMatchObject(INTERRUPTED_TURN) }) it('is not judged by the watched exit of a start that failed', async () => { @@ -629,6 +625,6 @@ describe('a turn a newer start could not settle before it failed', () => { 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) + expect(await settledTurn()).toMatchObject(INTERRUPTED_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 c9742e18881d..0a32aa14d6fd 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 @@ -36,8 +36,9 @@ import { provenUnverifiableTurnRevisions, provenUnverifiedToolCallRevisions, runningTurnLifecycleRevisions, - stopFoundTurnLiveAt, + lastProvenTurnLiveAt, turnVerdictFromDeathEvidence, + turnVerdictFromReplacedRuntime, watchedExitRevisions, type StructuredAgentSessionTurnVerdict, type StructuredAgentSessionWatchedExit @@ -46,13 +47,18 @@ import { exitedRootTurnScope, settledRootTurnScope } from './structured-agent-session-exit-turn-scope' -import { orcaStopRowBody } from './structured-agent-session-orca-stop-row' import { hasUnfinishedStructuredAgentSessionWork, isInProgressStructuredAgentSessionItem, type DeadGenerationJournal } from './structured-agent-session-unfinished-work' import { codexUnopenedSendResolutions } from './structured-agent-session-unopened-send-withdrawal' +import type { AgentSessionReplacedRuntime } from '../../runtime/agent-session-replaced-runtime' +import { + isUnverifiableTurn, + staleStopRow, + type StaleCut +} from './structured-agent-session-stale-stop-row' /** Bounds the exit reason the lease keeps as log evidence; a provider diagnostic is held to the * same cap. */ @@ -177,11 +183,11 @@ 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. Proven death ends the turn interrupted, - * and a proof written after an earlier settle revises what that settle left `unverifiable`, the - * turn and the calls it closed alike. Must run before a new child's buffered events land, or a - * live turn would be judged. + * is acquired, or when a chat is reopened for reading. Derived from the journal, the lease's death + * evidence and the runtimes this one replaced each time, so nothing is owed in between. Either proof + * ends the turn interrupted, and a proof written after an earlier settle revises what that settle + * left `unverifiable`, the turn and the calls it closed alike. Must run before a new child's + * buffered events land, or a live turn would be judged. */ export async function settleStaleStructuredAgentSessionState(input: { journal: AgentSessionJournal @@ -189,23 +195,35 @@ export async function settleStaleStructuredAgentSessionState(input: { fence: number acquisitionGeneration: string | null deathEvidence: AgentSessionDeathEvidence | null + /** The runtimes this one replaced for this chat: every owner they held lost its pipes with them. */ + replaced?: AgentSessionReplacedRuntime /** Who the exit row names. */ failureTextContext?: AgentSessionFailureWordsContext }): Promise { - const { journal } = input + const { journal, replaced } = input const items = journal.snapshot().items // Each turn is judged by the evidence only if it names that turn's owner. - const verdictFor = (item: AgentJournalRenderItem) => + const deathVerdictFor = (item: AgentJournalRenderItem) => turnVerdictFromDeathEvidence( input.deathEvidence, journal.itemFence(item.itemId), - stopFoundTurnLiveAt(journal, item) + lastProvenTurnLiveAt(journal, item) + ) + const replacedVerdictFor = (item: AgentJournalRenderItem) => + turnVerdictFromReplacedRuntime( + replaced, + journal.itemFence(item.itemId), + lastProvenTurnLiveAt(journal, item) ) + const verdictFor = (item: AgentJournalRenderItem) => { + const byDeath = deathVerdictFor(item) + return byDeath.state === 'interrupted' ? byDeath : replacedVerdictFor(item) + } // A successful settlement leaves terminal items; a failed transaction leaves the same prefix. const generation = input.acquisitionGeneration ?? `seq-${journal.cursor().sequence}` const settlementId = `${STALE_SESSION_ROW_PREFIX}${input.sessionId}:${input.fence}:${generation}` // Calls an earlier settle closed with no proof, revised once a proof names their owner. - const mutations = provenUnverifiedToolCallRevisions(items, input.deathEvidence, journal) + const mutations = provenUnverifiedToolCallRevisions(items, input.deathEvidence, journal, replaced) for (const item of items) { // A turn already settled (a person's Stop) ends its calls as it ended; only a turn still running // leaves them to the evidence. @@ -220,34 +238,32 @@ export async function settleStaleStructuredAgentSessionState(input: { }) } } - const proven = provenUnverifiableTurnRevisions(items, input.deathEvidence, journal) + const proven = provenUnverifiableTurnRevisions(items, input.deathEvidence, journal, replaced) const turnEnds = [ ...items.flatMap((item) => runningTurnLifecycleRevisions([item], verdictFor(item))), ...proven ] mutations.push(...turnEnds) - const evidence = input.deathEvidence - if ( - evidence && - (proven.length > 0 || - items.some( - (item) => - isInProgressStructuredAgentSessionItem(item) && verdictFor(item).state === 'interrupted' - )) && - !endedByPersonsStop(journal, turnEnds) - ) { - mutations.unshift({ - kind: 'item', - // Named by the death it explains, so a retry after a partly written settle adds no second row. - identity: { - provider: 'orca', - clientMessageId: `${STALE_SESSION_ROW_PREFIX}${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, and how Orca ended when the provider died with it. - body: orcaStopRowBody(input.failureTextContext, evidence.runtimeEnd), - turnScope: settledRootTurnScope(items, turnEnds) - }) + const stopRow = staleStopRow(input, items, (item): StaleCut => { + const unverifiable = isUnverifiableTurn(item) + if (!unverifiable && !isInProgressStructuredAgentSessionItem(item)) { + return null + } + const fence = journal.itemFence(item.itemId) + // An earlier settle's `unverifiable` turn is revised by a death naming its owner or by its replaced runtime. + if ( + unverifiable + ? fence !== undefined && fence === input.deathEvidence?.ownerFence + : deathVerdictFor(item).state === 'interrupted' + ) { + return { by: 'death' } + } + return fence !== undefined && replacedVerdictFor(item).state === 'interrupted' + ? { by: 'replaced', fence } + : null + }) + if (stopRow && !endedByPersonsStop(journal, turnEnds)) { + mutations.unshift({ ...stopRow, turnScope: settledRootTurnScope(items, turnEnds) }) } if (mutations.length > 0) { await journal.appendLifecycleBatch({ diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-readable-restorer.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-readable-restorer.test.ts index b6c8444a28bc..87c03c071ea9 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-readable-restorer.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-readable-restorer.test.ts @@ -35,7 +35,11 @@ describe('StructuredAgentSessionReadableRestorer', () => { ) const restorer = new StructuredAgentSessionReadableRestorer({ openDeps: { - store: { getRecord: () => null, listRecords: () => records }, + store: { + getRecord: () => null, + listRecords: () => records, + replacedRuntime: () => undefined + }, journalDatabase: openTestJournalHostDatabase(stateDirectory), logger: recordingStructuredAgentSessionLogger().logger }, diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-replaced-server-turn.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-replaced-server-turn.test.ts new file mode 100644 index 000000000000..822fb635b966 --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-replaced-server-turn.test.ts @@ -0,0 +1,241 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import { agentJournalItemKey } from '../../../shared/agent-session-journal-item-key' +import { + AGENT_JOURNAL_THREAD_SCOPE, + type AgentJournalRenderItem +} from '../../../shared/agent-session-journal-types' +import type { AgentSessionOwnerProbe } from '../../../shared/agent-session-lease-adjudication' +import type { AgentSessionRecord } from '../../../shared/agent-session-record' +import { readAgentJournalTurn } from '../../../shared/agent-session-turn-record' +import { + completedStructuredAgentTurnSeconds, + selectStructuredAgentTurnTimings +} from '../../../shared/structured-agent-session-turn-timing' +import { withNativeChatCutTurnNotices } from '../../../shared/native-chat-cut-turn-notice' +import { beginAgentSessionRuntimeIncarnationForTest } from '../../runtime/agent-session-runtime-attribution' +import { openTestAgentSessionRecordStore } from '../../runtime/agent-session-record-store-test-harness' +import { + closeTestJournalHostDatabases, + openTestJournalHostDatabase +} from '../agent-session-journal/journal-host-database-test-support' +import { + openStructuredAgentSessionConversationJournal, + type StructuredAgentSessionConversationOpenDeps +} from './structured-agent-session-conversation-open' +import { StructuredAgentSessionHost } from './structured-agent-session-host' +import { HOST_TEST_NOW, hostTestAttachParams } from './structured-agent-session-host-test-data' +import { + createRestTestRig, + REST_TEST_CALLER, + restTestSend, + REST_TEST_SESSION as SESSION, + REST_TEST_THREAD as THREAD, + type RestTestRig +} from './structured-agent-session-rest-test-rig' + +const LAST_CHUNK_AT = HOST_TEST_NOW + 5_500 +const RESTARTED_AT = HOST_TEST_NOW + 3_600_000 +const CUT_REPLY = 'chunks 1-11' +const TURN = { provider: 'codex' as const, threadId: THREAD, turnId: 'cut-turn', ordinal: 0 } +const REPLY = { ...TURN, ordinal: 1 } +const NEXT_TURN = { provider: 'codex' as const, threadId: THREAD, turnId: 'next-turn', ordinal: 2 } +const NEXT_REPLY = { ...NEXT_TURN, ordinal: 3 } +// The provider's identity cannot be proven either way: the recovery releases it with no proof. +const UNPROVABLE: AgentSessionOwnerProbe = { outcome: 'indeterminate', reason: 'probe unavailable' } + +let rig: RestTestRig + +beforeEach(async () => { + rig = await createRestTestRig({ idleSweep: { intervalMs: 3_600_000 } }) + vi.spyOn(Date, 'now').mockImplementation(() => rig.clock.now) +}) + +afterEach(async () => { + vi.restoreAllMocks() + await rig.dispose() + closeTestJournalHostDatabases() +}) + +async function streamThenServerKilled(): Promise { + const attached = await rig.host.attach(REST_TEST_CALLER, hostTestAttachParams(null)) + if (!attached.ok) { + throw new Error(`attach refused: ${attached.refusal.code}`) + } + const sent = await rig.host.send(REST_TEST_CALLER, restTestSend('long stream', attached.fence)) + if (!sent.ok) { + throw new Error(`send refused: ${sent.refusal.code}`) + } + await vi.waitFor(() => expect(rig.adapter.dispatch).toHaveBeenCalledOnce()) + const events = rig.adapter.acquire.mock.calls[0]?.[0].events + if (!events) { + throw new Error('the provider has no event sink') + } + events.appendItem( + TURN, + { kind: 'turn', turnId: TURN.turnId, state: 'running', startedAt: HOST_TEST_NOW }, + { turnScope: AGENT_JOURNAL_THREAD_SCOPE, observedAt: HOST_TEST_NOW } + ) + for (let chunk = 1; chunk <= 11; chunk += 1) { + rig.clock.now = HOST_TEST_NOW + chunk * 500 + events.appendItem( + REPLY, + { + kind: 'message', + role: 'assistant', + state: 'running', + blocks: [{ type: 'text', text: `chunks 1-${chunk}` }] + }, + { + turnScope: { kind: 'turn', turnItemId: agentJournalItemKey(TURN) }, + observedAt: HOST_TEST_NOW + chunk * 500 + } + ) + await rig.host.flushStreamedEvents(SESSION) + } +} + +/** The server dies with no exit seen and no settle; a server process starts on the same host. */ +async function startServer(options: { newProcess: boolean }): Promise { + const old = rig.host + old.stopDelivery() + const { lifetime, runtimeState, sessions } = old.collaboratorsForTests() + lifetime.idleSweep.dispose() + await runtimeState.stopLeaseRenewal() + runtimeState.currentEventSink(SESSION)?.close() + await sessions.get(SESSION)?.journal.close() + closeTestJournalHostDatabases() + if (options.newProcess) { + beginAgentSessionRuntimeIncarnationForTest() + } + rig.clock.now = RESTARTED_AT + rig.store = await openTestAgentSessionRecordStore(rig.root) + rig.host = new StructuredAgentSessionHost({ + ...old.deps, + store: rig.store, + journalDatabase: openTestJournalHostDatabase(rig.root), + probeOwner: async () => UNPROVABLE, + now: () => rig.clock.now + }) +} + +function turnsOf(items: readonly AgentJournalRenderItem[]) { + return items.flatMap((item) => readAgentJournalTurn(item.body) ?? []) +} + +function repliesOf(items: readonly AgentJournalRenderItem[]): string[] { + return items.flatMap((item) => + item.body.kind === 'message' && item.body.role === 'assistant' + ? item.body.blocks.flatMap((block) => (block.type === 'text' ? [block.text] : [])) + : [] + ) +} + +function stopNotices(items: readonly AgentJournalRenderItem[]): AgentJournalRenderItem[] { + return withNativeChatCutTurnNotices(items, { agentName: 'Codex' }).filter( + (item) => item.body.kind === 'status' + ) +} + +async function expectCutTurnSettledOnce(): Promise { + const { items } = await rig.host.journalSnapshot(SESSION) + const cut = turnsOf(items).find((turn) => turn.turnId === TURN.turnId) + expect(cut).toMatchObject({ state: 'interrupted', completedAt: LAST_CHUNK_AT }) + const [timing] = selectStructuredAgentTurnTimings(items).values() + expect(completedStructuredAgentTurnSeconds(timing)).toBe(5) + expect(repliesOf(items).filter((text) => text === CUT_REPLY)).toHaveLength(1) + expect(stopNotices(items)).toHaveLength(1) +} + +describe('a server replaced mid-turn on the same execution host', () => { + it('interrupts the cut turn at its last saved output with no provider identity left', async () => { + await streamThenServerKilled() + await startServer({ newProcess: true }) + await rig.host.restoreReadableSessions() + await expectCutTurnSettledOnce() + + await rig.host.reconcileRestartLeases() + await rig.host.collaboratorsForTests().serialize(SESSION, async () => {}) + // Nothing proved the provider gone: the lease keeps no death and no identity. + const lease = rig.store.getRecord(SESSION)?.lease + expect(lease).toMatchObject({ claimStatus: 'released', deathEvidence: null }) + expect(lease?.ownerProcess).toBeNull() + await expectCutTurnSettledOnce() + + const attached = await rig.host.attach( + REST_TEST_CALLER, + hostTestAttachParams(lease?.runtimeFence ?? null) + ) + if (!attached.ok) { + throw new Error(`re-attach refused: ${attached.refusal.code}`) + } + const sent = await rig.host.send( + REST_TEST_CALLER, + restTestSend('after restart', attached.fence) + ) + expect(sent.ok).toBe(true) + await vi.waitFor(() => expect(rig.adapter.dispatch).toHaveBeenCalledTimes(2)) + const events = rig.adapter.acquire.mock.calls[1]?.[0].events + if (!events) { + throw new Error('the new provider has no event sink') + } + events.appendItem( + NEXT_TURN, + { kind: 'turn', turnId: NEXT_TURN.turnId, state: 'running', startedAt: RESTARTED_AT }, + { turnScope: AGENT_JOURNAL_THREAD_SCOPE, observedAt: RESTARTED_AT } + ) + events.appendItem( + NEXT_REPLY, + { kind: 'message', role: 'assistant', blocks: [{ type: 'text', text: 'fresh reply' }] }, + { turnScope: { kind: 'turn', turnItemId: agentJournalItemKey(NEXT_TURN) } } + ) + events.appendItem( + NEXT_TURN, + { + kind: 'turn', + turnId: NEXT_TURN.turnId, + state: 'completed', + startedAt: RESTARTED_AT, + completedAt: RESTARTED_AT + 1_000 + }, + { turnScope: AGENT_JOURNAL_THREAD_SCOPE } + ) + await rig.host.flushStreamedEvents(SESSION) + + await expectCutTurnSettledOnce() + const { items } = await rig.host.journalSnapshot(SESSION) + expect(turnsOf(items).map((turn) => turn.state)).toEqual(['interrupted', 'completed']) + expect(repliesOf(items)).toEqual([CUT_REPLY, 'fresh reply']) + }) + + it('revises a turn an earlier start left unverifiable once a new server replaces its owner', async () => { + await streamThenServerKilled() + // The same process reopening its store replaced no runtime: nothing proves the turn over. + await startServer({ newProcess: false }) + await rig.host.restoreReadableSessions() + await rig.host.reconcileRestartLeases() + await rig.host.collaboratorsForTests().serialize(SESSION, async () => {}) + const left = turnsOf((await rig.host.journalSnapshot(SESSION)).items) + expect(left).toEqual([expect.objectContaining({ state: 'unverifiable' })]) + + await startServer({ newProcess: true }) + await rig.host.restoreReadableSessions() + await expectCutTurnSettledOnce() + + const settled = await rig.host.journalSnapshot(SESSION) + await startServer({ newProcess: true }) + await rig.host.restoreReadableSessions() + expect((await rig.host.journalSnapshot(SESSION)).items).toEqual(settled.items) + }) + + it('only lets an acquisition open a journal without the store', () => { + const open = ( + deps: Omit, + record: AgentSessionRecord + ) => { + void openStructuredAgentSessionConversationJournal(deps, record, { acquisition: true }) + // @ts-expect-error a reader open settles, so without the store its turns would stay unverifiable + void openStructuredAgentSessionConversationJournal(deps, record) + } + expect(open).toBeTypeOf('function') + }) +}) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-restart-provider-output.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-restart-provider-output.test.ts new file mode 100644 index 000000000000..694443515dd9 --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-restart-provider-output.test.ts @@ -0,0 +1,361 @@ +import { createCodexJournalTranslator } from '../../codex/codex-structured-journal-translation' +import { codexStreamingJournalItem } from '../../codex/codex-structured-item-translation' +import { createClaudeJournalTranslator } from '../../claude/claude-structured-journal-translation' +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import { agentJournalItemKey } from '../../../shared/agent-session-journal-item-key' +import { AGENT_JOURNAL_THREAD_SCOPE } from '../../../shared/agent-session-journal-types' +import { readAgentJournalTurn } from '../../../shared/agent-session-turn-record' +import { + completedStructuredAgentTurnSeconds, + selectStructuredAgentTurnTimings +} from '../../../shared/structured-agent-session-turn-timing' +import { openTestAgentSessionRecordStore } from '../../runtime/agent-session-record-store-test-harness' +import { + closeTestJournalHostDatabases, + openTestJournalHostDatabase +} from '../agent-session-journal/journal-host-database-test-support' +import { StructuredAgentSessionHost } from './structured-agent-session-host' +import { HOST_TEST_NOW, hostTestAttachParams } from './structured-agent-session-host-test-data' +import { unhandledProviderFrameJournalItem } from './unhandled-provider-frame' +import { + createRestTestRig, + REST_TEST_CALLER, + restTestSend, + REST_TEST_SESSION as SESSION, + REST_TEST_THREAD as THREAD, + type RestTestRig +} from './structured-agent-session-rest-test-rig' + +const LAST_CHUNK_AT = HOST_TEST_NOW + 5_500 +const SAVED_AT = HOST_TEST_NOW + 60_000 +const RESTARTED_AT = HOST_TEST_NOW + 3_600_000 +const TURN = { provider: 'codex' as const, threadId: THREAD, turnId: 'stream-turn', ordinal: 0 } +const REPLY = { ...TURN, ordinal: 1 } +const TURN_SCOPE = { kind: 'turn' as const, turnItemId: agentJournalItemKey(TURN) } + +let rig: RestTestRig + +beforeEach(async () => { + rig = await createRestTestRig({ idleSweep: { intervalMs: 3_600_000 } }) + vi.spyOn(Date, 'now').mockImplementation(() => rig.clock.now) +}) + +afterEach(async () => { + vi.restoreAllMocks() + await rig.dispose() + closeTestJournalHostDatabases() +}) + +/** Drops the old host without an exit callback or graceful turn settlement. */ +async function restartAfterCrash(): Promise { + const old = rig.host + old.stopDelivery() + const { lifetime, runtimeState, sessions } = old.collaboratorsForTests() + lifetime.idleSweep.dispose() + await runtimeState.stopLeaseRenewal() + runtimeState.currentEventSink(SESSION)?.close() + await sessions.get(SESSION)?.journal.close() + closeTestJournalHostDatabases() + rig.clock.now = RESTARTED_AT + rig.store = await openTestAgentSessionRecordStore(rig.root) + rig.host = new StructuredAgentSessionHost({ + ...old.deps, + store: rig.store, + journalDatabase: openTestJournalHostDatabase(rig.root), + now: () => RESTARTED_AT + }) +} + +async function beginTurn() { + const attached = await rig.host.attach(REST_TEST_CALLER, hostTestAttachParams(null)) + if (!attached.ok) { + throw new Error('attach refused') + } + const sent = await rig.host.send(REST_TEST_CALLER, restTestSend('review', attached.fence)) + if (!sent.ok) { + throw new Error('send refused') + } + await vi.waitFor(() => expect(rig.adapter.dispatch).toHaveBeenCalledOnce()) + const events = rig.adapter.acquire.mock.calls[0]?.[0].events + if (!events) { + throw new Error('event sink absent') + } + events.appendItem( + TURN, + { kind: 'turn', turnId: TURN.turnId, state: 'running', startedAt: HOST_TEST_NOW }, + { turnScope: AGENT_JOURNAL_THREAD_SCOPE, observedAt: HOST_TEST_NOW } + ) + return events +} + +async function expectInterruptedOutput(completedAt = LAST_CHUNK_AT): Promise { + await restartAfterCrash() + await rig.host.restoreReadableSessions() + const snapshot = await rig.host.journalSnapshot(SESSION) + const turn = snapshot.items.map((item) => readAgentJournalTurn(item.body)).find(Boolean) + expect(turn).toMatchObject({ state: 'interrupted', completedAt }) +} + +describe('provider output survives host restart', () => { + it.each([ + ['agentMessage', 'item/agentMessage/delta'], + ['fileChange', 'item/fileChange/outputDelta'], + ['plan', 'item/plan/delta'], + ['reasoning', 'item/reasoning/textDelta'], + ['commandExecution', 'item/commandExecution/outputDelta'] + ] as const)('times saved Codex %s output rather than item start', async (type, method) => { + const events = await beginTurn() + let receivedAt = HOST_TEST_NOW + const translator = createCodexJournalTranslator({ + sink: events, + primaryThreadId: () => THREAD, + now: () => receivedAt, + schedule: () => () => {} + }) + translator.handle({ + type: 'notification', + sessionId: SESSION, + threadId: THREAD, + method: 'turn/started', + params: { turn: { id: TURN.turnId } }, + observedAt: receivedAt + }) + translator.handle({ + type: 'notification', + sessionId: SESSION, + threadId: THREAD, + method: 'item/started', + params: { + turnId: TURN.turnId, + item: { + id: 'actual-reply', + type, + text: '', + command: 'echo stream', + status: 'inProgress', + changes: [{ path: 'file.ts' }] + } + }, + observedAt: receivedAt + }) + for (let chunk = 1; chunk <= 11; chunk += 1) { + receivedAt = HOST_TEST_NOW + chunk * 500 + rig.clock.now = receivedAt + expect( + translator.handle({ + type: 'notification', + sessionId: SESSION, + threadId: THREAD, + method, + params: { + turnId: TURN.turnId, + itemId: 'actual-reply', + delta: `actual chunk ${chunk} `.repeat(4) + }, + observedAt: receivedAt + }) + ).toMatchObject({ accepted: true }) + translator.flush() + await rig.host.flushStreamedEvents(SESSION) + } + translator.dispose() + const before = await rig.host.journalSnapshot(SESSION) + const output = before.items.find((item) => + JSON.stringify(item.body).includes('actual chunk 11') + ) + expect(output?.observedAt).toBe(HOST_TEST_NOW) + expect( + rig.host.collaboratorsForTests().sessions.get(SESSION)?.journal.lastProviderActivityAt(1) + ).toBe(LAST_CHUNK_AT) + await restartAfterCrash() + await rig.host.restoreReadableSessions() + const snapshot = await rig.host.journalSnapshot(SESSION) + const turn = snapshot.items.map((item) => readAgentJournalTurn(item.body)).find(Boolean) + expect(turn).toMatchObject({ state: 'interrupted', completedAt: LAST_CHUNK_AT }) + }) + + it('counts streamed file-change output as later proof of provider life', async () => { + const events = await beginTurn() + for (let chunk = 1; chunk <= 11; chunk += 1) { + rig.clock.now = HOST_TEST_NOW + chunk * 500 + const translated = codexStreamingJournalItem( + { id: 'file-change-1', type: 'fileChange', changes: [{ path: 'file.ts' }] }, + `patch chunk ${chunk}` + ) + if (!translated.body) { + throw new Error('file change was not translated') + } + events.appendItem(REPLY, translated.body, { + turnScope: TURN_SCOPE, + observedAt: HOST_TEST_NOW + chunk * 500 + }) + await rig.host.flushStreamedEvents(SESSION) + } + expect(rig.store.getRecord(SESSION)?.lease.lastRenewedAt).toBe(HOST_TEST_NOW) + await restartAfterCrash() + await rig.host.restoreReadableSessions() + const snapshot = await rig.host.journalSnapshot(SESSION) + const turn = snapshot.items.map((item) => readAgentJournalTurn(item.body)).find(Boolean) + expect(turn).toMatchObject({ state: 'interrupted', completedAt: LAST_CHUNK_AT }) + }) + + it('uses a delayed first save while keeping the bubble at item start', async () => { + const events = await beginTurn() + const translator = createCodexJournalTranslator({ + sink: events, + primaryThreadId: () => THREAD, + schedule: () => () => {} + }) + translator.handle({ + type: 'notification', + sessionId: SESSION, + threadId: THREAD, + method: 'turn/started', + params: { turn: { id: TURN.turnId } }, + observedAt: HOST_TEST_NOW + }) + translator.handle({ + type: 'notification', + sessionId: SESSION, + threadId: THREAD, + method: 'item/started', + params: { turnId: TURN.turnId, item: { id: 'late-first', type: 'agentMessage', text: '' } }, + observedAt: HOST_TEST_NOW + }) + translator.handle({ + type: 'notification', + sessionId: SESSION, + threadId: THREAD, + method: 'item/agentMessage/delta', + params: { turnId: TURN.turnId, itemId: 'late-first', delta: 'first durable output' }, + observedAt: LAST_CHUNK_AT + }) + rig.clock.now = SAVED_AT + translator.flush() + await rig.host.flushStreamedEvents(SESSION) + translator.dispose() + const before = await rig.host.journalSnapshot(SESSION) + expect( + before.items.find((item) => JSON.stringify(item.body).includes('first durable output')) + ?.observedAt + ).toBe(HOST_TEST_NOW) + await expectInterruptedOutput(SAVED_AT) + }) + + it.each(['text_delta', 'thinking_delta'] as const)( + 'times delayed Claude %s checkpoints', + async (type) => { + const events = await beginTurn() + const translator = createClaudeJournalTranslator({ sink: events, schedule: () => () => {} }) + for (let chunk = 0; chunk <= 11; chunk += 1) { + translator.handle({ + type: 'message', + sessionId: SESSION, + observedAt: HOST_TEST_NOW + chunk * 500, + message: { + type: 'stream_event', + uuid: 'stream-frame', + session_id: 'claude-provider', + parent_tool_use_id: null, + event: { + type: 'content_block_delta', + index: 0, + delta: { type, text: `text chunk ${chunk} `, thinking: `thought chunk ${chunk} ` } + } + } + }) + } + rig.clock.now = SAVED_AT + translator.flush() + await rig.host.flushStreamedEvents(SESSION) + translator.dispose() + const before = await rig.host.journalSnapshot(SESSION) + const output = before.items.find((item) => JSON.stringify(item.body).includes('chunk 11')) + expect(output).toBeDefined() + if (type === 'thinking_delta') { + expect(output?.observedAt).toBe(HOST_TEST_NOW) + } + await expectInterruptedOutput(SAVED_AT) + } + ) + + it('counts a saved provider fallback but excludes a later host notice', async () => { + const events = await beginTurn() + rig.clock.now = LAST_CHUNK_AT + const fallback = unhandledProviderFrameJournalItem('codex', 'test-output', { + message: 'provider output' + }) + if (!fallback) { + throw new Error('provider output was not translated') + } + events.appendItem( + { provider: 'orca', clientMessageId: 'provider-frame:codex:acquisition:1' }, + fallback.body, + { turnScope: TURN_SCOPE } + ) + await rig.host.flushStreamedEvents(SESSION) + const journal = rig.host.collaboratorsForTests().sessions.get(SESSION)?.journal + if (!journal) { + throw new Error('journal absent') + } + rig.clock.now = SAVED_AT + await journal.appendItem( + { provider: 'orca', clientMessageId: 'host-notice' }, + { kind: 'status', text: 'host notice' }, + { fence: 1, turnScope: TURN_SCOPE } + ) + await expectInterruptedOutput() + }) + + it('counts a pending approval with no assistant output before the crash', async () => { + const events = await beginTurn() + rig.clock.now = LAST_CHUNK_AT + events.appendItem( + REPLY, + { + kind: 'approval', + title: 'Run?', + detail: null, + options: [], + resolution: { state: 'pending', selectedOptionId: null, resolvedBy: null, resolvedAt: null } + }, + { turnScope: TURN_SCOPE, observedAt: LAST_CHUNK_AT } + ) + await rig.host.flushStreamedEvents(SESSION) + await restartAfterCrash() + await rig.host.restoreReadableSessions() + const snapshot = await rig.host.journalSnapshot(SESSION) + const turn = snapshot.items.map((item) => readAgentJournalTurn(item.body)).find(Boolean) + expect(turn).toMatchObject({ state: 'interrupted', completedAt: LAST_CHUNK_AT }) + expect(snapshot.items.find((item) => item.body.kind === 'approval')?.body).toMatchObject({ + resolution: { state: 'cancelled' } + }) + const [timing] = selectStructuredAgentTurnTimings(snapshot.items).values() + expect(completedStructuredAgentTurnSeconds(timing)).toBe(5) + }) + + it('cannot record interruption from a surviving owner before exit proof', async () => { + const events = await beginTurn() + rig.clock.now = LAST_CHUNK_AT + events.appendItem( + REPLY, + { kind: 'status', text: 'working' }, + { turnScope: TURN_SCOPE, observedAt: LAST_CHUNK_AT } + ) + await rig.host.flushStreamedEvents(SESSION) + await restartAfterCrash() + rig.host.deps.probeOwner = async () => ({ + outcome: 'identity-matched', + matchedOn: ['process-start-time'] + }) + const stopOwnerProcess = vi.fn() + rig.host.deps.stopOwnerProcess = stopOwnerProcess + await rig.host.reconcileRestartLeases() + const snapshot = await rig.host.journalSnapshot(SESSION) + const turn = snapshot.items.map((item) => readAgentJournalTurn(item.body)).find(Boolean) + expect(turn?.state).toBe('unverifiable') + expect(turn?.completedAt).toBeUndefined() + expect(stopOwnerProcess).not.toHaveBeenCalled() + expect(rig.store.getRecord(SESSION)?.lease.deathEvidence).toBeFalsy() + }) +}) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-restart-recovery-output.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-restart-recovery-output.test.ts new file mode 100644 index 000000000000..1c78db236f3e --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-restart-recovery-output.test.ts @@ -0,0 +1,213 @@ +import { StructuredAgentRegistry } from './structured-agent-registry' +import { acpLaunchSpecFor } from '../../acp/acp-launch-specs' +import { acpStructuredAgentDefinition } from '../../acp/acp-structured-agent-definitions' +import { AcpBackgroundTaskTimeline } from '../../acp/acp-background-task-timeline' +import { createProviderTimelineAssembler } from '../agent-session-timeline/provider-timeline-assembler' +import { providerTimelineSink } from '../agent-session-timeline/provider-timeline-plan' +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import { readAgentJournalTurn } from '../../../shared/agent-session-turn-record' +import { + completedStructuredAgentTurnSeconds, + selectStructuredAgentTurnTimings +} from '../../../shared/structured-agent-session-turn-timing' +import { openTestAgentSessionRecordStore } from '../../runtime/agent-session-record-store-test-harness' +import { + closeTestJournalHostDatabases, + openTestJournalHostDatabase +} from '../agent-session-journal/journal-host-database-test-support' +import { StructuredAgentSessionHost } from './structured-agent-session-host' +import { + HOST_TEST_LOCATION, + HOST_TEST_NOW, + hostTestAttachParams +} from './structured-agent-session-host-test-data' +import { + createRestTestRig, + REST_TEST_CALLER, + restTestSend, + REST_TEST_SESSION as SESSION, + type RestTestRig +} from './structured-agent-session-rest-test-rig' + +const LAST_CHUNK_AT = HOST_TEST_NOW + 5_500 +const CONNECTION_CLOSED_AT = HOST_TEST_NOW + 60_000 +const RESTARTED_AT = HOST_TEST_NOW + 3_600_000 + +let rig: RestTestRig + +beforeEach(async () => { + rig = await createRestTestRig({ idleSweep: { intervalMs: 3_600_000 } }) + vi.spyOn(Date, 'now').mockImplementation(() => rig.clock.now) + const spec = acpLaunchSpecFor('grok') + if (!spec) { + throw new Error('ACP agent definition absent') + } + const definition = acpStructuredAgentDefinition(spec) + // The rig adapter has no compact; this recovery never compacts. + rig.host.deps.agents = new StructuredAgentRegistry([ + { + definition: { ...definition, capabilities: { ...definition.capabilities, compact: false } }, + adapter: rig.host.deps.adapter + } + ]) + rig.adapter.acquire.mockImplementation(async ({ fence, spawnToken }) => ({ + acquisitionGeneration: 'acp-generation', + process: { hostId: 'local', pid: 4242, processStartTimeMs: 1_700_000_000_000, spawnToken }, + link: { + linkId: `link-${fence}`, + handle: { transport: 'acp', agent: 'grok', nativeId: 'acp-session' }, + origin: 'created', + mintedAtFence: fence, + observedAt: rig.clock.now + } + })) + rig.adapter.dispatch.mockImplementation(async ({ clientMessageId }) => ({ + state: 'accepted', + providerIdentity: { provider: 'orca', clientMessageId } + })) +}) + +afterEach(async () => { + vi.restoreAllMocks() + await rig.dispose() + closeTestJournalHostDatabases() +}) + +/** Drops the old host without an exit callback or graceful turn settlement. */ +async function restartAfterCrash(): Promise { + const old = rig.host + old.stopDelivery() + const { lifetime, runtimeState, sessions } = old.collaboratorsForTests() + lifetime.idleSweep.dispose() + await runtimeState.stopLeaseRenewal() + runtimeState.currentEventSink(SESSION)?.close() + await sessions.get(SESSION)?.journal.close() + closeTestJournalHostDatabases() + rig.clock.now = RESTARTED_AT + rig.store = await openTestAgentSessionRecordStore(rig.root) + rig.host = new StructuredAgentSessionHost({ + ...old.deps, + store: rig.store, + journalDatabase: openTestJournalHostDatabase(rig.root), + now: () => RESTARTED_AT + }) +} + +async function beginTurn(workspaceKind: 'git-worktree' | 'folder') { + const attached = await rig.host.attach( + REST_TEST_CALLER, + hostTestAttachParams(null, { + provider: 'grok', + agent: 'grok', + providerHandle: undefined, + location: { ...HOST_TEST_LOCATION, workspaceKind }, + accountHome: { variable: 'GROK_HOME', path: rig.root } + }) + ) + if (!attached.ok) { + throw new Error('attach refused') + } + const sent = await rig.host.send( + REST_TEST_CALLER, + restTestSend('background work', attached.fence) + ) + if (!sent.ok) { + throw new Error('send refused') + } + await vi.waitFor(() => expect(rig.adapter.dispatch).toHaveBeenCalledOnce()) + const events = rig.adapter.acquire.mock.calls[0]?.[0].events + if (!events) { + throw new Error('event sink absent') + } + expect(rig.store.getRecord(SESSION)?.provider).toBe('grok') + return events +} + +describe.each([ + ['journal-open', false], + ['session-end', false], + ['session-end', true] +] as const)('%s recovery with buffered output: %s', (recovery, bufferedOutput) => { + it.each(['git-worktree', 'folder'] as const)( + 'excludes ACP background-task recovery in a %s workspace', + async (workspaceKind) => { + const events = await beginTurn(workspaceKind) + const sink = providerTimelineSink(events) + if (!sink) { + throw new Error('timeline sink absent') + } + const assembler = createProviderTimelineAssembler({ + sink, + sessionId: SESSION, + agent: 'grok', + generation: 'acp-generation', + namespace: 'acp-session', + schedule: () => () => {} + }) + assembler.apply({ type: 'turn.open', turn: 'acp-turn', at: HOST_TEST_NOW }) + rig.clock.now = LAST_CHUNK_AT + const tasks = new AcpBackgroundTaskTimeline(() => undefined) + for (const event of tasks.translate( + [ + { + taskId: 'task-1', + kind: 'command', + label: 'sleep 20', + state: 'working', + startedAt: HOST_TEST_NOW + } + ], + { turn: 'acp-turn' } + )) { + expect(assembler.apply(event).admission).toMatchObject({ accepted: true }) + } + assembler.flush() + await rig.host.flushStreamedEvents(SESSION) + if (bufferedOutput) { + expect( + assembler.apply({ + type: 'text.delta', + item: { stream: 'tail' }, + channel: 'assistant', + text: 'buffered provider output' + }).admission + ).toMatchObject({ accepted: true }) + } + if (recovery === 'session-end') { + rig.clock.now = CONNECTION_CLOSED_AT + expect( + assembler.apply({ type: 'session.ended', verdict: { state: 'unverifiable' } }).admission + ).toMatchObject({ accepted: true }) + await rig.host.flushStreamedEvents(SESSION) + } + assembler.dispose() + const journal = rig.host.collaboratorsForTests().sessions.get(SESSION)?.journal + if (!journal) { + throw new Error('journal absent') + } + const completedAt = bufferedOutput ? CONNECTION_CLOSED_AT : LAST_CHUNK_AT + expect(journal.lastProviderActivityAt(1)).toBe(completedAt) + await restartAfterCrash() + await rig.host.restoreReadableSessions() + const snapshot = await rig.host.journalSnapshot(SESSION) + const background = snapshot.items.find( + (item) => + item.body.kind === 'message' && + item.body.blocks.some((block) => block.type === 'background-task') + ) + expect(background?.body).toMatchObject({ + blocks: expect.arrayContaining([expect.objectContaining({ state: 'unverifiable' })]) + }) + const turn = snapshot.items.map((item) => readAgentJournalTurn(item.body)).find(Boolean) + expect(turn).toMatchObject({ state: 'interrupted', completedAt }) + const [timing] = selectStructuredAgentTurnTimings(snapshot.items).values() + expect(completedStructuredAgentTurnSeconds(timing)).toBe(bufferedOutput ? 60 : 5) + expect( + rig.host.collaboratorsForTests().sessions.get(SESSION)?.journal.lastProviderActivityAt(1) + ).toBe(completedAt) + await restartAfterCrash() + await rig.host.restoreReadableSessions() + expect((await rig.host.journalSnapshot(SESSION)).items).toEqual(snapshot.items) + } + ) +}) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-restart-restore.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-restart-restore.test.ts index 7df838d04e82..1b74a23911f3 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-restart-restore.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-restart-restore.test.ts @@ -28,7 +28,7 @@ afterAll(() => { }) const NO_OPEN_DEPS = { - store: { getRecord: () => null, listRecords: () => [] }, + store: { getRecord: () => null, listRecords: () => [], replacedRuntime: () => undefined }, journalDatabase: openTestJournalHostDatabase(stateDirectory), logger: recordingStructuredAgentSessionLogger().logger } diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-restart-stream-duration.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-restart-stream-duration.test.ts new file mode 100644 index 000000000000..5deb2e482db9 --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-restart-stream-duration.test.ts @@ -0,0 +1,269 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import { agentJournalItemKey } from '../../../shared/agent-session-journal-item-key' +import { AGENT_JOURNAL_THREAD_SCOPE } from '../../../shared/agent-session-journal-types' +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 { withNativeChatCutTurnNotices } from '../../../shared/native-chat-cut-turn-notice' +import { openTestAgentSessionRecordStore } from '../../runtime/agent-session-record-store-test-harness' +import { + closeTestJournalHostDatabases, + openTestJournalHostDatabase +} from '../agent-session-journal/journal-host-database-test-support' +import { StructuredAgentSessionHost } from './structured-agent-session-host' +import { + HOST_TEST_LOCATION, + HOST_TEST_NOW, + hostTestAttachParams +} from './structured-agent-session-host-test-data' +import { + createRestTestRig, + REST_TEST_CALLER, + restTestSend, + REST_TEST_SESSION as SESSION, + REST_TEST_THREAD as THREAD, + type RestTestRig +} from './structured-agent-session-rest-test-rig' + +const LAST_CHUNK_AT = HOST_TEST_NOW + 5_500 +const RESTARTED_AT = HOST_TEST_NOW + 3_600_000 +const TURN = { provider: 'codex' as const, threadId: THREAD, turnId: 'stream-turn', ordinal: 0 } +const REPLY = { ...TURN, ordinal: 1 } +const TURN_SCOPE = { kind: 'turn' as const, turnItemId: agentJournalItemKey(TURN) } + +let rig: RestTestRig + +beforeEach(async () => { + rig = await createRestTestRig({ idleSweep: { intervalMs: 3_600_000 } }) + vi.spyOn(Date, 'now').mockImplementation(() => rig.clock.now) +}) + +afterEach(async () => { + vi.restoreAllMocks() + await rig.dispose() + closeTestJournalHostDatabases() +}) + +async function streamBeforeCrash( + workspaceKind: 'folder' | 'git-worktree' = 'git-worktree' +): Promise { + const attached = await rig.host.attach( + REST_TEST_CALLER, + hostTestAttachParams(null, { location: { ...HOST_TEST_LOCATION, workspaceKind } }) + ) + if (!attached.ok) { + throw new Error(`attach refused: ${attached.refusal.code}`) + } + const sent = await rig.host.send(REST_TEST_CALLER, restTestSend('long stream', attached.fence)) + if (!sent.ok) { + throw new Error(`send refused: ${sent.refusal.code}`) + } + await vi.waitFor(() => expect(rig.adapter.dispatch).toHaveBeenCalledOnce()) + const events = rig.adapter.acquire.mock.calls[0]?.[0].events + if (!events) { + throw new Error('the provider has no event sink') + } + events.appendItem( + TURN, + { kind: 'turn', turnId: TURN.turnId, state: 'running', startedAt: HOST_TEST_NOW }, + { turnScope: AGENT_JOURNAL_THREAD_SCOPE, observedAt: HOST_TEST_NOW } + ) + for (let chunk = 1; chunk <= 11; chunk += 1) { + rig.clock.now = HOST_TEST_NOW + chunk * 500 + events.appendItem( + REPLY, + { + kind: 'message', + role: 'assistant', + state: 'running', + blocks: [{ type: 'text', text: `chunks 1-${chunk}` }] + }, + { turnScope: TURN_SCOPE, observedAt: HOST_TEST_NOW + chunk * 500 } + ) + await rig.host.flushStreamedEvents(SESSION) + } + expect(rig.store.getRecord(SESSION)?.lease.lastRenewedAt).toBe(HOST_TEST_NOW) +} + +/** Drops the old host without an exit callback or graceful turn settlement. */ +async function restartAfterCrash(): Promise { + const old = rig.host + old.stopDelivery() + const { lifetime, runtimeState, sessions } = old.collaboratorsForTests() + lifetime.idleSweep.dispose() + await runtimeState.stopLeaseRenewal() + runtimeState.currentEventSink(SESSION)?.close() + await sessions.get(SESSION)?.journal.close() + closeTestJournalHostDatabases() + rig.clock.now = RESTARTED_AT + rig.store = await openTestAgentSessionRecordStore(rig.root) + rig.host = new StructuredAgentSessionHost({ + ...old.deps, + store: rig.store, + journalDatabase: openTestJournalHostDatabase(rig.root), + now: () => RESTARTED_AT + }) +} + +async function expectSettledStream(): Promise { + const snapshot = await rig.host.journalSnapshot(SESSION) + expect( + snapshot.items.find((item) => item.itemId === agentJournalItemKey(REPLY))?.body + ).toMatchObject({ + kind: 'message', + blocks: [{ type: 'text', text: 'chunks 1-11' }] + }) + const turn = snapshot.items.map((item) => readAgentJournalTurn(item.body)).find(Boolean) + expect(turn).toMatchObject({ state: 'interrupted', completedAt: LAST_CHUNK_AT }) + const [timing] = selectStructuredAgentTurnTimings(snapshot.items).values() + expect(completedStructuredAgentTurnSeconds(timing)).toBe(5) + expect( + withNativeChatCutTurnNotices(snapshot.items, { agentName: 'Codex' }).filter( + (item) => item.body.kind === 'status' && item.body.text !== 'host notice' + ) + ).toHaveLength(1) + expect(rig.adapter.acquire).toHaveBeenCalledOnce() +} + +describe('a provider stream cut short before its next lease renewal', () => { + it.each(['git-worktree', 'folder'] as const)( + 'settles and replays the duration exactly once in a %s workspace', + async (workspaceKind) => { + await streamBeforeCrash(workspaceKind) + const cursor = (await rig.host.journalSnapshot(SESSION)).cursor + await restartAfterCrash() + await rig.host.restoreReadableSessions() + + await expectSettledStream() + const events: AgentSessionSubscribeEvent[] = [] + const unsubscribe = await rig.host.subscribe({ + id: 'reconnecting-client', + sessionId: SESSION, + cursor, + emit: (event) => events.push(event) + }) + const replayedTurns = events.flatMap((event) => { + const items = + event.type === 'snapshot' + ? event.page.items + : event.type === 'batch' + ? event.batch.items + : [] + return items.flatMap((item) => readAgentJournalTurn(item.body) ?? []) + }) + expect(replayedTurns).toEqual([ + expect.objectContaining({ state: 'interrupted', completedAt: LAST_CHUNK_AT }) + ]) + unsubscribe() + const settled = await rig.host.journalSnapshot(SESSION) + await rig.host.restoreReadableSessions() + expect(await rig.host.journalSnapshot(SESSION)).toEqual(settled) + + await restartAfterCrash() + await rig.host.restoreReadableSessions() + await expectSettledStream() + // A reopen can mark its queue, but never revises the settled turn again. + expect((await rig.host.journalSnapshot(SESSION)).items).toEqual(settled.items) + } + ) + + it('retains the output bound when a client opens the journal before owner reconciliation', async () => { + await streamBeforeCrash() + await restartAfterCrash() + const beforeProof = await rig.host.journalSnapshot(SESSION) + expect( + beforeProof.items.some((item) => readAgentJournalTurn(item.body)?.state === 'unverifiable') + ).toBe(true) + + await rig.host.reconcileRestartLeases() + await rig.host.collaboratorsForTests().serialize(SESSION, async () => {}) + + await expectSettledStream() + }) + + it('also preserves the interrupted turn and duration after a local desktop quit', async () => { + await streamBeforeCrash() + rig.clock.now = LAST_CHUNK_AT + await rig.host.flushAllStreamedEvents({ trigger: 'quit' }) + await restartAfterCrash() + await rig.host.restoreReadableSessions() + + await expectSettledStream() + expect(rig.adapter.closeSession).toHaveBeenCalledOnce() + }) + + // A new server replacing the owner's runtime ends it anyway (structured-agent-session-replaced-server-turn). + it('leaves the end unverifiable when its own runtime reopens and cannot prove owner death', async () => { + await streamBeforeCrash() + await restartAfterCrash() + rig.host.deps.probeOwner = async () => ({ + outcome: 'indeterminate', + reason: 'probe unavailable' + }) + await rig.host.restoreReadableSessions() + + const snapshot = await rig.host.journalSnapshot(SESSION) + const turn = snapshot.items.map((item) => readAgentJournalTurn(item.body)).find(Boolean) + expect(turn?.state).toBe('unverifiable') + expect(turn?.completedAt).toBeUndefined() + expect(rig.adapter.acquire).toHaveBeenCalledOnce() + }) + + it('ignores later client actions, recovery revisions, and another owner generation', async () => { + await streamBeforeCrash() + const journal = rig.host.collaboratorsForTests().sessions.get(SESSION)?.journal + if (!journal) { + throw new Error('the streaming journal is not open') + } + const later = RESTARTED_AT - 1_000 + rig.clock.now = later + await journal.appendItem( + { provider: 'orca', clientMessageId: 'late-client-send' }, + { kind: 'message', role: 'user', blocks: [{ type: 'text', text: 'next task' }] }, + { fence: 1, turnScope: TURN_SCOPE, observedAt: later } + ) + await journal.appendItem( + { ...REPLY, ordinal: 2 }, + { + kind: 'approval', + title: 'Run?', + detail: null, + options: [], + resolution: { + state: 'resolved', + selectedOptionId: 'allow', + resolvedBy: 'client', + resolvedAt: later + } + }, + { fence: 1, turnScope: TURN_SCOPE, observedAt: later } + ) + const originalReply = journal.itemBody(agentJournalItemKey(REPLY)) + if (!originalReply) { + throw new Error('the streamed reply is missing') + } + await journal.appendLifecycleBatch({ + settlementId: 'earlier-read-recovery', + fence: 1, + recovered: true, + mutations: [{ kind: 'item', identity: REPLY, body: originalReply, turnScope: TURN_SCOPE }] + }) + await journal.appendItem( + { provider: 'orca', clientMessageId: 'late-host-notice' }, + { kind: 'status', text: 'host notice' }, + { fence: 1, turnScope: TURN_SCOPE } + ) + await journal.appendItem( + { ...REPLY, ordinal: 3 }, + { kind: 'message', role: 'assistant', blocks: [{ type: 'text', text: 'another owner' }] }, + { fence: 2, turnScope: TURN_SCOPE, observedAt: later } + ) + await restartAfterCrash() + await rig.host.restoreReadableSessions() + + await expectSettledStream() + }) +}) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-stale-stop-row.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-stale-stop-row.ts new file mode 100644 index 000000000000..ad39aa2975b0 --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-stale-stop-row.ts @@ -0,0 +1,68 @@ +// The row a stale settle writes about why the turns it cut stopped: a death that names their owner, +// or the replacement of the Orca runtime that held them. One row per proof, named by it, so a retry +// after a partly written settle adds no second row. + +import { STALE_SESSION_ROW_PREFIX } from '../../../shared/agent-session-stop-row-identity' +import type { AgentSessionFailureWordsContext } from '../../../shared/agent-session-failure-words' +import type { + AgentJournalRenderItem, + AgentJournalStatusItem +} from '../../../shared/agent-session-journal-types' +import type { AgentSessionDeathEvidence } from '../../../shared/agent-session-record' +import { readAgentJournalTurn } from '../../../shared/agent-session-turn-record' +import type { AgentSessionReplacedRuntime } from '../../runtime/agent-session-replaced-runtime' +import { orcaStopRowBody } from './structured-agent-session-orca-stop-row' + +/** Which proof ends an item the settle found running or left `unverifiable`, if any does. */ +export type StaleCut = { by: 'death' } | { by: 'replaced'; fence: number } | null + +export function isUnverifiableTurn(item: AgentJournalRenderItem): boolean { + return readAgentJournalTurn(item.body)?.state === 'unverifiable' +} + +export function staleStopRow( + input: { + sessionId: string + deathEvidence: AgentSessionDeathEvidence | null + replaced?: AgentSessionReplacedRuntime + failureTextContext?: AgentSessionFailureWordsContext + }, + items: readonly AgentJournalRenderItem[], + cutBy: (item: AgentJournalRenderItem) => StaleCut +): { + kind: 'item' + identity: { provider: 'orca'; clientMessageId: string } + body: AgentJournalStatusItem +} | null { + const cuts = items.map(cutBy) + const evidence = input.deathEvidence + // The death evidence is Orca's log text, never a sentence for a person: the row says only that + // the provider stopped, and how Orca ended when the provider died with it. + if (evidence && cuts.some((cut) => cut?.by === 'death')) { + return { + kind: 'item', + identity: { + provider: 'orca', + clientMessageId: `${STALE_SESSION_ROW_PREFIX}${input.sessionId}:death-${evidence.ownerFence ?? 'unowned'}-${evidence.observedAt}` + }, + body: orcaStopRowBody(input.failureTextContext, evidence.runtimeEnd) + } + } + const fences = cuts.flatMap((cut) => (cut?.by === 'replaced' ? [cut.fence] : [])) + if (!input.replaced || fences.length === 0) { + return null + } + const fence = Math.max(...fences) + return { + kind: 'item', + // Named by the owner it explains, which every later runtime finds the same. + identity: { + provider: 'orca', + clientMessageId: `${STALE_SESSION_ROW_PREFIX}${input.sessionId}:replaced-${fence}` + }, + body: orcaStopRowBody( + input.failureTextContext, + fence === input.replaced.fence ? input.replaced.runtimeEnd : undefined + ) + } +} 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 c0652dff2d59..9efb8202f880 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 @@ -311,6 +311,7 @@ describe('stale session state on a cold acquire', () => { const journal = { snapshot: () => ({ items }), itemFence: () => 1, + lastProviderActivityAt: () => undefined, stopMarks: { latest: () => null }, cursor: () => ({ epoch: 'epoch-1', sequence: 8 }), appendLifecycleBatch @@ -441,7 +442,7 @@ 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 () => { + it('uses provider output after a renewal, excluding a later client message', async () => { const root = await mkdtemp(join(tmpdir(), 'orca-stale-session-')) const journals = createTrackedJournalOpener() let now = 100 @@ -467,13 +468,19 @@ describe('stale session state on a cold acquire', () => { ) now = 200 await journal.appendItem(command, { ...shell, state: 'running' }, { fence: 1, turnScope }) - // Rows can outlast the last renewal; only the renewal is proof of life. + // Streaming revisions prove the provider worked after its last lease renewal. now = 700 await journal.appendItem( command, { ...shell, input: { command: 'pnpm test', streamed: 'ok' }, state: 'running' }, { fence: 1, turnScope } ) + now = 800 + await journal.appendItem( + { provider: 'orca', clientMessageId: 'late-send' }, + { kind: 'message', role: 'user', blocks: [{ type: 'text', text: 'next task' }] }, + { fence: 1, turnScope } + ) now = 9_000 await settleStaleStructuredAgentSessionState({ @@ -493,7 +500,7 @@ describe('stale session state on a cold acquire', () => { const items = journal.snapshot().items expect(items.map((item) => readAgentJournalTurn(item.body)).find(Boolean)).toMatchObject({ state: 'interrupted', - completedAt: 650 + completedAt: 700 }) // The probe's detail is Orca's, so the row carries none. expect( 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 ec4d76dcbdc0..aaebb2ce2f0d 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,9 +1,10 @@ // What the host may durably say about a turn whose provider child is gone. // -// `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` +// `interrupted` requires proof that the turn is over: death evidence naming that turn's owner by +// fence (a watched exit, or a local probe that found the recorded pid gone or reused), or the +// replacement of the Orca runtime that held the owner's pipes, which this host proves by holding its +// lock. A release nothing proved within one runtime (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 { @@ -20,6 +21,7 @@ import { readAgentJournalTurn } from '../../../shared/agent-session-turn-record' import type { AgentSessionDeathEvidence } from '../../../shared/agent-session-record' +import type { AgentSessionReplacedRuntime } from '../../runtime/agent-session-replaced-runtime' import type { JournalLifecycleMutationInput } from '../agent-session-journal/journal-row-builders' import type { AgentSessionJournal } from '../agent-session-journal/journal-store' @@ -35,7 +37,7 @@ export function turnVerdictFromDeathEvidence( evidence: AgentSessionDeathEvidence | null | undefined, /** Fence of the owner that wrote the turn. */ turnFence: number | undefined, - /** When a Stop event found the turn running (`stopFoundTurnLiveAt`): a later proof of life. */ + /** Saved provider output or a Stop that found the turn running: a later proof of life. */ liveAt?: number ): StructuredAgentSessionTurnVerdict { if (!evidence) { @@ -54,8 +56,8 @@ export function turnVerdictFromDeathEvidence( return interruptedTurnVerdict(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. A - // Stop that found the turn running is a later renewal, so the end reads after that Stop. + // counts the time Orca was down. Saved provider output and a Stop that found the turn running can + // prove life after that renewal; a client's later send or a recovery write cannot. const lastAlive = Math.max(evidence.lastProvenAliveAt ?? evidence.observedAt, liveAt ?? 0) return interruptedTurnVerdict(Math.min(lastAlive, evidence.observedAt)) } @@ -66,14 +68,31 @@ function interruptedTurnVerdict(completedAt: number): StructuredAgentSessionTurn : { state: 'interrupted' } } -/** When the latest Stop event found `item`'s turn running: E1 writes one only for a live turn. */ -export function stopFoundTurnLiveAt( - journal: Pick, +/** A turn whose owner a replaced runtime held is over: nothing it did after reaches the chat. It + * ends at the owner's last proof of life, which that runtime wrote before this one took over. */ +export function turnVerdictFromReplacedRuntime( + replaced: AgentSessionReplacedRuntime | undefined, + turnFence: number | undefined, + liveAt?: number +): StructuredAgentSessionTurnVerdict { + if (!replaced || turnFence === undefined || turnFence > replaced.fence) { + return UNVERIFIABLE_TURN_VERDICT + } + const renewedAt = turnFence === replaced.fence ? (replaced.lastProvenAliveAt ?? 0) : 0 + return interruptedTurnVerdict(Math.max(renewedAt, liveAt ?? 0)) +} + +/** The latest saved output from this turn's owner, including a Stop that found it running. */ +export function lastProvenTurnLiveAt( + journal: Pick, item: AgentJournalRenderItem ): number | undefined { const stop = journal.stopMarks.latest() const turnId = readAgentJournalTurn(item.body)?.turnId - return stop && turnId !== undefined && stop.event.turnId === turnId ? stop.event.at : undefined + const stoppedAt = stop && turnId !== undefined && stop.event.turnId === turnId ? stop.event.at : 0 + const fence = journal.itemFence(item.itemId) + const providerAt = fence === undefined ? 0 : (journal.lastProviderActivityAt(fence) ?? 0) + return Math.max(stoppedAt, providerAt) || undefined } /** Every turn this settle interrupts is a person's Stop's to end (`turnEndAfterStop`), so it reads @@ -111,21 +130,22 @@ export function runningTurnLifecycleRevisions( export function provenUnverifiableTurnRevisions( items: readonly AgentJournalRenderItem[], evidence: AgentSessionDeathEvidence | null | undefined, - journal: Pick + journal: Pick, + /** A runtime replacement proves it for every owner that runtime held, a death only for its own. */ + replaced?: AgentSessionReplacedRuntime ): JournalLifecycleMutationInput[] { - const ownerFence = evidence?.ownerFence - if (ownerFence === undefined) { - return [] - } return items.flatMap((item) => { const turn = readAgentJournalTurn(item.body) - return turn?.state === 'unverifiable' && journal.itemFence(item.itemId) === ownerFence - ? turnLifecycleRevision( - item, - turn, - turnVerdictFromDeathEvidence(evidence, ownerFence, stopFoundTurnLiveAt(journal, item)) - ) - : [] + if (turn?.state !== 'unverifiable') { + return [] + } + const fence = journal.itemFence(item.itemId) + const liveAt = lastProvenTurnLiveAt(journal, item) + const verdict = + evidence?.ownerFence !== undefined && fence === evidence.ownerFence + ? turnVerdictFromDeathEvidence(evidence, fence, liveAt) + : turnVerdictFromReplacedRuntime(replaced, fence, liveAt) + return verdict.state === 'interrupted' ? turnLifecycleRevision(item, turn, verdict) : [] }) } @@ -160,16 +180,17 @@ export function watchedExitRevisions( export function provenUnverifiedToolCallRevisions( items: readonly AgentJournalRenderItem[], evidence: AgentSessionDeathEvidence | null | undefined, - journal: Pick + journal: Pick, + replaced?: AgentSessionReplacedRuntime ): JournalLifecycleMutationInput[] { const ownerFence = evidence?.ownerFence - if (ownerFence === undefined) { - return [] - } + const proven = (fence: number | undefined) => + fence !== undefined && + (fence === ownerFence || (replaced !== undefined && fence <= replaced.fence)) return items.flatMap((item): JournalLifecycleMutationInput[] => { return item.body.kind === 'tool-call' && isUnverifiedEndAgentJournalToolCall(item.body) && - journal.itemFence(item.itemId) === ownerFence + proven(journal.itemFence(item.itemId)) ? [ { kind: 'item', diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-transition.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-transition.ts index 18805b0b7404..d1cfdec43356 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-transition.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-transition.ts @@ -45,6 +45,7 @@ export type StructuredAgentSessionTransitionStep = kind: 'settlement' /** Unique per settlement: the journal applies one id once. */ settlementId: string + recovered?: true /** Paces the queue only; the mutations are the journal's to choose. */ reservedBytes: number /** Read at execution; none writes nothing. */ @@ -118,6 +119,7 @@ function transitionAppend( batch: { settlementId: step.settlementId, fence, + recovered: step.recovered, resolve: () => step.resolve(journal) } } 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 ab73a2a786aa..b1fdcb5d8605 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 @@ -99,6 +99,7 @@ describe('provider-exit settlement', () => { items: [lifecycleItem('turn-1', 1, { state: 'running', startedAt: 1_000 })] }), itemFence: () => 7, + lastProviderActivityAt: () => undefined, stopMarks: { latest: () => null, personStopDecides: () => false }, appendLifecycleBatch } 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 436d7075643e..6727b0144c03 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 @@ -552,9 +552,10 @@ describe('already-wedged profiles become usable on load', () => { } ) - it('marks a running turn left behind by a released lease unverifiable on a cold acquire', async () => { + it('interrupts a running turn left behind by a released lease on a cold acquire', async () => { // No settlement latch: the record was released cleanly, but the journal still says a turn is - // running. The child that wrote it is gone and nothing observed its exit. + // running. Nothing observed the child's exit, but the runtime that held it was replaced, and + // nothing after its start proves it alive. await seedStore(wedgedRecord({ claimStatus: 'released', handoffStage: null })) await seedRunningTurn() openHost() @@ -564,8 +565,9 @@ describe('already-wedged profiles become usable on load', () => { expect(acquire).toHaveBeenCalledOnce() expect(turnLifecycle('turn-1')).toEqual({ turnId: 'turn-1', - state: 'unverifiable', + state: 'interrupted', startedAt: NOW - 5_000, + completedAt: NOW - 5_000, recovered: true }) expect( diff --git a/src/main/runtime/agent-session-record-store.ts b/src/main/runtime/agent-session-record-store.ts index 9cca024aca1b..b96728fb7c88 100644 --- a/src/main/runtime/agent-session-record-store.ts +++ b/src/main/runtime/agent-session-record-store.ts @@ -104,7 +104,7 @@ export class AgentSessionRecordStore { hostId: string }): AgentSessionRecordStore { const rows = loadAgentSessionStoreRows(args.journalDatabase.db) - const transactions = new AgentSessionStoreTransactions(args.journalDatabase, rows) + const transactions = new AgentSessionStoreTransactions(args.journalDatabase, rows, args.hostId) return new AgentSessionRecordStore(transactions, args.hostId) } @@ -122,6 +122,9 @@ export class AgentSessionRecordStore { listRecords = (): AgentSessionRecord[] => [...this.state.records.values()] + /** The runtimes this one replaced for a chat, as found at load; absent for a chat it began. */ + replacedRuntime = (sessionId: string) => this.transactions.replacedRuntimes.get(sessionId) + /** Every chat this host holds a row for, readable or not. */ listHeldSessionIds = (): string[] => heldAgentSessionIds(this.state) @@ -183,9 +186,7 @@ export class AgentSessionRecordStore { ) /** A record this build cannot validate: readable as present, never grantable as a writer. */ - isSessionUnreadable(sessionId: string): boolean { - return this.state.unreadableRecords.has(sessionId) - } + isSessionUnreadable = (sessionId: string): boolean => this.state.unreadableRecords.has(sessionId) listOperationRows = (): AgentSessionOperationRow[] => [...this.state.operations.values()] diff --git a/src/main/runtime/agent-session-replaced-runtime.test.ts b/src/main/runtime/agent-session-replaced-runtime.test.ts new file mode 100644 index 000000000000..afccc5b4b4be --- /dev/null +++ b/src/main/runtime/agent-session-replaced-runtime.test.ts @@ -0,0 +1,54 @@ +import { describe, expect, it } from 'vitest' +import type { AgentSessionLease } from '../../shared/agent-session-record' +import { + agentSessionLeaseFixture, + agentSessionRecordFixture +} from '../../shared/agent-session-record.test-fixture' +import { agentSessionReplacedRuntimes } from './agent-session-replaced-runtime' +import type { AgentSessionStoreState } from './agent-session-store-state' + +const SESSION = 'session-alpha-1' +const OWNER = { hostId: 'local', pid: 4242, processStartTimeMs: 1, spawnToken: 'spawn-owner' } + +function loaded(lease: Partial, runtimeEnd?: 'crash'): AgentSessionStoreState { + const record = agentSessionRecordFixture( + agentSessionLeaseFixture({ unreconciled: true, ...lease }) + ) + return { + records: new Map([[SESSION, record]]), + operations: new Map(), + retiredClaimKeys: [], + unreadableRecords: new Map(), + sessionTabs: null, + runtimeEnds: runtimeEnd ? new Map([['earlier', runtimeEnd]]) : new Map() + } +} + +function replacedFor(state: AgentSessionStoreState) { + return agentSessionReplacedRuntimes(state, { + hostId: 'local', + incarnation: 'this-runtime' + }).get(SESSION) +} + +describe('the runtimes a new server replaced, as it loads their records', () => { + it('names every fence an earlier runtime granted, its owner renewal, and how it ended', () => { + expect( + replacedFor(loaded({ ownerProcess: { ...OWNER, runtime: 'earlier' } }, 'crash')) + ).toEqual({ fence: 7, lastProvenAliveAt: 30_000, runtimeEnd: 'crash' }) + }) + + it('covers a chat an earlier runtime released, which no owner holds now', () => { + expect( + replacedFor(loaded({ ownerProcess: null, reservedSpawnToken: null, claimStatus: 'released' })) + ).toEqual({ fence: 7 }) + }) + + it.each([ + ['an owner this runtime recorded', { ownerProcess: { ...OWNER, runtime: 'this-runtime' } }], + ['an owner on another host', { ownerProcess: { ...OWNER, hostId: 'remote' } }], + ['a terminal agent an older build recorded', { claimStatus: 'conflicted' as const }] + ])('replaces nothing for %s', (_label, lease) => { + expect(replacedFor(loaded(lease))).toBeUndefined() + }) +}) diff --git a/src/main/runtime/agent-session-replaced-runtime.ts b/src/main/runtime/agent-session-replaced-runtime.ts new file mode 100644 index 000000000000..d58eec7a13f2 --- /dev/null +++ b/src/main/runtime/agent-session-replaced-runtime.ts @@ -0,0 +1,43 @@ +// What this runtime took over from the Orca runtimes that ran chats on this state directory before +// it. It loaded their records holding the profile's instance lock, so none of them is still running: +// every provider they started lost its pipes with them, and nothing it did after can reach the chat. +// That is the execution host's own proof that a turn those runtimes left running is over, whether or +// not the provider process itself can still be proven gone. + +import type { AgentSessionOrcaStopCause } from '../../shared/agent-session-orca-stop' +import type { AgentSessionStoreState } from './agent-session-store-state' + +export type AgentSessionReplacedRuntime = { + /** The highest fence a replaced runtime granted; every lower fence was granted before it. */ + fence: number + /** The last renewal of the owner loaded at `fence`, its last proof of life. */ + lastProvenAliveAt?: number + /** How the runtime holding the owner loaded at `fence` ended, when its record says. */ + runtimeEnd?: AgentSessionOrcaStopCause +} + +/** One entry per chat whose fences were all granted by runtimes this one replaced. */ +export function agentSessionReplacedRuntimes( + state: AgentSessionStoreState, + args: { hostId: string; incarnation: string } +): ReadonlyMap { + const replaced = new Map() + for (const [sessionId, { lease }] of state.records) { + const owner = lease.ownerProcess + // A terminal's own agent never wrote through Orca's pipes; another host's runtime is not one this + // replaced; and an owner this runtime recorded (a host reinstalled in one process) may still be held. + if ( + lease.claimStatus === 'conflicted' || + (owner && (owner.hostId !== args.hostId || owner.runtime === args.incarnation)) + ) { + continue + } + const runtimeEnd = owner?.runtime ? state.runtimeEnds?.get(owner.runtime) : undefined + replaced.set(sessionId, { + fence: lease.runtimeFence, + ...(owner ? { lastProvenAliveAt: lease.lastRenewedAt } : {}), + ...(runtimeEnd ? { runtimeEnd } : {}) + }) + } + return replaced +} diff --git a/src/main/runtime/agent-session-store-transactions.ts b/src/main/runtime/agent-session-store-transactions.ts index 3877da33af42..e70875904c1b 100644 --- a/src/main/runtime/agent-session-store-transactions.ts +++ b/src/main/runtime/agent-session-store-transactions.ts @@ -21,6 +21,10 @@ import { beginAgentSessionRuntimeRecord, readAgentSessionRuntimeEnds } from './agent-session-runtime-end-record' +import { + agentSessionReplacedRuntimes, + type AgentSessionReplacedRuntime +} from './agent-session-replaced-runtime' import { agentSessionStoreDraftRowWrites, draftAgentSessionStoreState, @@ -81,16 +85,23 @@ type StagedStoreTransaction = { export class AgentSessionStoreTransactions { private queue: Promise = Promise.resolve() private published: AgentSessionStoreState + /** What this runtime took over from the runtimes before it, as loaded; never changes after. */ + readonly replacedRuntimes: ReadonlyMap constructor( private readonly journalDatabase: JournalHostDatabase, - loaded: AgentSessionStoreState + loaded: AgentSessionStoreState, + hostId: string ) { freezeRows(loaded, null) this.published = { ...loaded, runtimeEnds: readAgentSessionRuntimeEnds(journalDatabase.stateDirectory) } + this.replacedRuntimes = agentSessionReplacedRuntimes(this.published, { + hostId, + incarnation: agentSessionRuntimeIncarnation() + }) // Before any owner this runtime records: a crash is concluded only from a recorded start. beginAgentSessionRuntimeRecord( journalDatabase.stateDirectory, diff --git a/src/renderer/src/components/editor/editor-cmd-save-target.ts b/src/renderer/src/components/editor/editor-cmd-save-target.ts index 75559af2a1d8..2be47558f7dc 100644 --- a/src/renderer/src/components/editor/editor-cmd-save-target.ts +++ b/src/renderer/src/components/editor/editor-cmd-save-target.ts @@ -23,7 +23,5 @@ export function getEditorCmdSaveFileId( : null } const activeTab = state.getActiveTab(FLOATING_TERMINAL_WORKTREE_ID) - return activeTab && isEditorTabContentType(activeTab.contentType) - ? activeTab.entityId - : null + return activeTab && isEditorTabContentType(activeTab.contentType) ? activeTab.entityId : null } diff --git a/src/renderer/src/components/native-chat/NativeChatInterruptedContinue.test.tsx b/src/renderer/src/components/native-chat/NativeChatInterruptedContinue.test.tsx index 9ece96a3dc0c..6c35294a4ffe 100644 --- a/src/renderer/src/components/native-chat/NativeChatInterruptedContinue.test.tsx +++ b/src/renderer/src/components/native-chat/NativeChatInterruptedContinue.test.tsx @@ -89,6 +89,7 @@ function Harness(props: Props): React.JSX.Element { {continuation.offeredTurnItemId ?? 'none'} {String(continuation.view.continueAvailable)} + {String(continuation.view.remoteHost)} {composerError ?? continuation.continueError?.text ?? 'none'}