fix(native-chat): a turn a proven crash cut short reads interrupted

src/main/native-chat/agent-session-journal/journal-reducer.ts

@@ -42,6 +42,8 @@ export type JournalReducerState = {
   oldestSequence: number
   highestFence: number
   items: Map<string, AgentJournalRenderItem>
+  /** Fence of the writer that created each item: the generation a running turn belongs to. */
+  itemFences: Map<string, number>
   /** Revision of a removed item, so a late lower revision cannot resurrect it. */
   tombstones: Map<string, number>
   submissions: Map<string, AgentJournalSubmission>
@@ -61,6 +63,7 @@ export function createJournalReducerState(sessionId: string, epoch: string): Jou
     oldestSequence: 1,
     highestFence: 0,
     items: new Map(),
+    itemFences: new Map(),
     tombstones: new Map(),
     submissions: new Map(),
     receipts: new Map(),
@@ -82,7 +85,13 @@ export function applyJournalRow(state: JournalReducerState, row: JournalRow): vo
     }
     const itemId = resolveJournalItemId(state, row.itemId, row.body)
     acceptSubmissionFromProviderItem(state, row.itemId, itemId, row)
-    upsertItem(state, itemId, row.revision, journalRenderItem(itemId, row.revision, row.body, row))
+    upsertItem(
+      state,
+      itemId,
+      row.revision,
+      journalRenderItem(itemId, row.revision, row.body, row),
+      row.fence
+    )
     return
   }
   if (row.kind === 'tombstone') {
@@ -103,7 +112,7 @@ export function applyJournalRow(state: JournalReducerState, row: JournalRow): vo
         acceptSubmissionFromProviderItem(state, mutation.itemId, itemId, row)
         const producer = journalBatchMutationProducer(row, mutation)
         const item = journalRenderItem(itemId, revision, body, row, producer, sequenceIndex)
-        upsertItem(state, itemId, revision, item)
+        upsertItem(state, itemId, revision, item, row.fence)
       } else {
         removeItem(state, resolveItemId(state, mutation.itemId), mutation.revision)
       }
@@ -187,7 +196,8 @@ function upsertItem(
   state: JournalReducerState,
   itemId: string,
   revision: number,
-  next: AgentJournalRenderItem
+  next: AgentJournalRenderItem,
+  fence: number
 ): void {
   const tombstoned = state.tombstones.get(itemId)
   if (tombstoned !== undefined && revision <= tombstoned) {
@@ -199,6 +209,7 @@ function upsertItem(
   }
   if (!existing) {
     state.items.set(itemId, next)
+    state.itemFences.set(itemId, fence)
     state.tombstones.delete(itemId)
     return
   }
@@ -237,6 +248,7 @@ function removeItem(state: JournalReducerState, itemId: string, revision: number
   }
   state.tombstones.set(itemId, revision)
   state.items.delete(itemId)
+  state.itemFences.delete(itemId)
 }
 
 function applySubmission(
@@ -255,7 +267,7 @@ function applySubmission(
     ...(row.handoverRecorded ? { handoverRecorded: true, acceptedSequence: row.seq } : {})
   })
   const itemId = agentJournalSubmissionKey(row.clientMessageId)
-  upsertItem(state, itemId, 0, journalRenderItem(itemId, 0, row.body, row))
+  upsertItem(state, itemId, 0, journalRenderItem(itemId, 0, row.body, row), row.fence)
 }
 
 function acceptSubmissionFromProviderItem(

src/main/native-chat/agent-session-journal/journal-store.ts

@@ -234,6 +234,9 @@ export class AgentSessionJournal {
   /** Includes revisions and completion tombstones, whose timestamps disappear from render items. */
   lastActivityAt = (): number => this.state.lastActivityAt
 
+  /** Fence of the writer that created the item, while it is in the timeline. */
+  itemFence = (itemId: string): number | undefined => this.state.itemFences.get(itemId)
+
   submissions = (): AgentJournalSubmission[] => [...this.state.submissions.values()]
 
   pendingSubmissions = (): AgentJournalSubmission[] =>

src/main/native-chat/agent-session-wire/structured-agent-session-conversation-open.ts

@@ -10,6 +10,7 @@
 
 import type { AgentJournalResetReason } from '../../../shared/agent-session-journal-types'
 import { journalDirectoryFor } from '../agent-session-journal/journal-paths'
+import type { AgentSessionJournal } from '../agent-session-journal/journal-store'
 import { openAgentSessionJournalWithRecovery } from './agent-session-journal-recovery'
 import { computeAgentSessionPayloadFingerprint } from '../../../shared/agent-session-mutation-envelope'
 import type { AgentSessionRecord } from '../../../shared/agent-session-record'
@@ -103,30 +104,54 @@ export async function openStructuredAgentSessionConversationJournal(
   } catch (error) {
     deps.onEventSinkError?.({ sessionId, error })
   }
-  try {
-    // No child in this process writes to a journal nobody had open, so whatever it shows running
-    // belongs to a generation that is gone, whatever the lease still claims. Settled before any
-    // reader or child sees it.
-    if (!options.acquisition) {
-      await settleStaleStructuredAgentSessionState({
-        journal: opened.journal,
-        sessionId,
-        fence,
-        acquisitionGeneration: null,
-        deathEvidence: record.lease.deathEvidence ?? null,
-        failureTextContext: structuredAgentSessionFailureWordsContext(record)
-      })
-    }
-  } catch (error) {
-    // Best effort: the next acquire re-derives it.
-    deps.onEventSinkError?.({ sessionId, error })
+  // No child in this process writes to a journal nobody had open, so whatever it shows running
+  // belongs to a generation that is gone, whatever the lease still claims. Settled before any
+  // reader or child sees it.
+  if (!options.acquisition) {
+    await settleGoneGeneration(deps, record, opened.journal)
   }
   return {
     session: { journal: opened.journal, params, child: null },
     reset: opened.recovery?.reset ?? null
   }
 }
 
+/**
+ * The open's settle again, for a conversation already open: a proof of death written since it
+ * opened (the startup reconcile, a recovery) revises what the open could only call `unverifiable`.
+ * A record holds a proof only while released, so no child here is writing. A no-op once revised.
+ */
+export async function resettleOpenStructuredAgentSessionConversation(
+  deps: StructuredAgentSessionConversationOpenDeps,
+  sessionId: string,
+  session: StructuredAgentSessionHostSession | undefined
+): Promise<void> {
+  const record = deps.store.getRecord(sessionId)
+  if (session && record?.lease.deathEvidence) {
+    await settleGoneGeneration(deps, record, session.journal)
+  }
+}
+
+async function settleGoneGeneration(
+  deps: Pick<StructuredAgentSessionConversationOpenDeps, 'onEventSinkError'>,
+  record: AgentSessionRecord,
+  journal: AgentSessionJournal
+): Promise<void> {
+  try {
+    await settleStaleStructuredAgentSessionState({
+      journal,
+      sessionId: record.sessionId,
+      fence: record.lease.runtimeFence,
+      acquisitionGeneration: null,
+      deathEvidence: record.lease.deathEvidence ?? null,
+      failureTextContext: structuredAgentSessionFailureWordsContext(record)
+    })
+  } catch (error) {
+    // Best effort: the next open or acquire re-derives it.
+    deps.onEventSinkError?.({ sessionId: record.sessionId, error })
+  }
+}
+
 export function attachParamsForRecord(
   record: AgentSessionRecord,
   input: {