fix(native-chat): settle a reply cut off by a server restart

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',

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 } : {}) }
 }

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 => {

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<JournalReducerState, 'providerActivityAt'>,
+  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
+}

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<string, AgentJournalRenderItem>
   /** Fence of the writer that created each item: the generation a running turn belongs to. */
   itemFences: Map<string, number>
+  /** Latest saved output per owner, rebuilt during the existing row fold. */
+  providerActivityAt: Map<number, number>
   /** Revision of a removed item, so a late lower revision cannot resurrect it. */
   tombstones: Map<string, number>
   submissions: Map<string, AgentJournalSubmission>
@@ -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)

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')
 }
 

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))
+    }
+  )
 })

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<JournalRow> {
     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
   }
 

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<JournalRow>
 }

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
   })
 }

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)

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' }

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

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
     })
   }

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),

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)
           })
         }