fix(native-chat): retire settled dispatch ownership

src/main/native-chat/agent-session-journal/journal-row-schema.ts

@@ -26,7 +26,7 @@ type JournalRowBase = {
   fence: number
   /** Observed (provider or host) timestamp. Ordering is by `seq`, not by this. */
   ts: number
-  /** Set when crash reconciliation appended the row after the fact. */
+  /** Set when host lifecycle reconciliation appends a row instead of a live provider event. */
   recovered?: true
 }
 

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

@@ -317,12 +317,18 @@ export class AgentSessionJournal {
       capturePrecedingPendingSubmissions: () => string[]
     ) => Promise<JournalOrderedAppendResult<T>>
   ): Promise<T> {
-    const settlesTurn = bodies.some((body) => {
-      const turn = readAgentJournalTurn(body)
-      return turn !== null && turn.state !== 'running'
-    })
+    const terminalTurnIds = new Set(
+      bodies.flatMap((body) => {
+        const turn = readAgentJournalTurn(body)
+        return turn !== null && turn.state !== 'running' ? [turn.turnId] : []
+      })
+    )
     const capturePrecedingPendingSubmissions = (): string[] => {
-      if (!settlesTurn) {
+      if (terminalTurnIds.size === 0) {
+        return []
+      }
+      const activeTurnId = this.activeTurnId()
+      if (activeTurnId === null || !terminalTurnIds.has(activeTurnId)) {
         return []
       }
       return this.submissions()
@@ -332,7 +338,7 @@ export class AgentSessionJournal {
         .map((submission) => submission.clientMessageId)
     }
     const result = await append(capturePrecedingPendingSubmissions)
-    if (!result.appended || !settlesTurn) {
+    if (!result.appended || result.precedingPendingSubmissionIds.length === 0) {
       return result.value
     }
     await Promise.all(
@@ -342,7 +348,8 @@ export class AgentSessionJournal {
             clientMessageId,
             state: 'unknown',
             reason: DISPATCH_DOUBT_TURN_SETTLED,
-            fence
+            fence,
+            recovered: true
           })
         } catch (error) {
           console.warn(

src/main/native-chat/agent-session-journal/journal-turn-settlement.test.ts

@@ -121,12 +121,13 @@ describe('turn settlement dispatch resolution', () => {
       expect.objectContaining({
         clientMessageId: 'message-1',
         dispatchState: 'unknown',
-        reason: 'turn_settled_before_acknowledgement'
+        reason: 'turn_settled_before_acknowledgement',
+        recovered: true
       })
     ])
     const afterSettlement = journal.readSince(settlement)
     expect(afterSettlement.ok && afterSettlement.rows).toEqual([
-      expect.objectContaining({ kind: 'dispatch', state: 'unknown' })
+      expect.objectContaining({ kind: 'dispatch', state: 'unknown', recovered: true })
     ])
   })
 
@@ -153,7 +154,8 @@ describe('turn settlement dispatch resolution', () => {
 
     expect(journal.submissions()[0]).toMatchObject({
       dispatchState: 'unknown',
-      reason: 'turn_settled_before_acknowledgement'
+      reason: 'turn_settled_before_acknowledgement',
+      recovered: true
     })
   })
 
@@ -233,6 +235,77 @@ describe('turn settlement dispatch resolution', () => {
     })
   })
 
+  it('does not let a terminal record for another turn settle the active turn submission', async () => {
+    const journal = await open()
+    await openTurn(journal, 'old-turn')
+    await journal.appendLifecycleBatch(terminalTurn('old-turn', 'completed'))
+    await openTurn(journal, 'active-turn')
+    await appendPending(journal, 'active-message')
+
+    await journal.appendItem(
+      turnIdentity('old-turn'),
+      { kind: 'turn', turnId: 'old-turn', state: 'completed' },
+      { fence: 1 }
+    )
+
+    expect(journal.activeTurnId()).toBe('active-turn')
+    expect(journal.submissions()[0]).toMatchObject({ dispatchState: 'pending' })
+  })
+
+  it('does not let a repeated direct terminal record settle a later submission', async () => {
+    const journal = await open()
+    await openTurn(journal, 'turn-repeated')
+    await appendPending(journal, 'original')
+    const identity = turnIdentity('turn-repeated')
+    const completed = {
+      kind: 'turn' as const,
+      turnId: 'turn-repeated',
+      state: 'completed' as const
+    }
+    await journal.appendItem(identity, completed, { fence: 1 })
+    await appendPending(journal, 'later')
+
+    await journal.appendItem(identity, completed, { fence: 1 })
+
+    expect(journal.submissions().find((entry) => entry.clientMessageId === 'later')).toMatchObject({
+      dispatchState: 'pending'
+    })
+  })
+
+  it.each(['accepted', 'rejected'] as const)(
+    'lets late %s evidence narrow a turn-settled unknown',
+    async (state) => {
+      const journal = await open()
+      await openTurn(journal, `turn-late-${state}`)
+      await appendPending(journal, `late-${state}`)
+      await journal.appendLifecycleBatch(terminalTurn(`turn-late-${state}`, 'completed'))
+
+      await journal.resolveDispatch(
+        state === 'accepted'
+          ? {
+              clientMessageId: `late-${state}`,
+              state,
+              providerIdentity: {
+                provider: 'codex',
+                threadId: 'thread-1',
+                turnId: `turn-late-${state}`,
+                ordinal: 0
+              },
+              fence: 1
+            }
+          : {
+              clientMessageId: `late-${state}`,
+              state,
+              reason: 'provider later rejected the send',
+              fence: 1
+            }
+      )
+
+      expect(journal.submissions()[0]).toMatchObject({ dispatchState: state })
+      expect(journal.submissions()[0]).not.toHaveProperty('recovered')
+    }
+  )
+
   it('does not settle a pending submission from another fence', async () => {
     const journal = await open()
     await openTurn(journal, 'turn-fenced')
@@ -287,4 +360,26 @@ describe('turn settlement dispatch resolution', () => {
     expect(journal.activeTurnId()).toBeNull()
     await expect(appendPending(journal, 'next-send')).resolves.toBeUndefined()
   })
+
+  it('retires the crash window through the existing next-attach reconciliation', async () => {
+    const journal = await open()
+    await openTurn(journal, 'turn-crash-window')
+    await appendPending(journal, 'unwritten-settlement')
+    const resolution = vi
+      .spyOn(journal, 'resolveDispatch')
+      .mockRejectedValueOnce(new Error('process exited before dispatch row'))
+    vi.spyOn(console, 'warn').mockImplementation(() => undefined)
+    await journal.appendLifecycleBatch(terminalTurn('turn-crash-window', 'completed'))
+    resolution.mockRestore()
+    await journal.close()
+
+    const restarted = await open()
+    expect(restarted.submissions()[0]).toMatchObject({ dispatchState: 'pending' })
+    await restarted.markPendingSubmissionsUnknown(2)
+
+    expect(restarted.submissions()[0]).toMatchObject({
+      dispatchState: 'unknown',
+      recovered: true
+    })
+  })
 })

src/main/native-chat/agent-session-wire/structured-conversation-command-admission.test.ts

@@ -1,24 +1,55 @@
 import { describe, expect, it } from 'vitest'
 import type { AgentSessionRecord } from '../../../shared/agent-session-record'
+import type {
+  AgentJournalSnapshot,
+  AgentJournalSubmission
+} from '../../../shared/agent-session-journal-types'
 import type { AgentSessionBackgroundTaskState } from '../../../shared/agent-session-wire'
 import { conversationCommandBlocked } from './structured-conversation-command-admission'
-import type { AgentSessionTurnContext } from './structured-agent-session-turns'
 
-function contextWith(
-  backgroundTasks: AgentSessionBackgroundTaskState | null
-): AgentSessionTurnContext {
+function snapshot(items: AgentJournalSnapshot['items'] = []): AgentJournalSnapshot {
   return {
     sessionId: 'session-1',
+    cursor: { epoch: 'epoch-1', sequence: 0 },
+    items,
+    submissions: []
+  }
+}
+
+function contextWith(
+  backgroundTasks: AgentSessionBackgroundTaskState | null,
+  submissions: AgentJournalSubmission[] = []
+): Parameters<typeof conversationCommandBlocked>[0] {
+  return {
+    sessionId: 'session-1',
+    fence: 1,
     journal: {
-      snapshot: () => ({ items: [] }),
-      submissions: () => []
+      snapshot: () => snapshot(),
+      submissions: () => submissions
     },
     adapter: { backgroundTaskState: () => backgroundTasks }
-  } as unknown as AgentSessionTurnContext
+  }
 }
 
 const RECORD = { lease: {} } as unknown as AgentSessionRecord
 
+function submission(
+  dispatchState: AgentJournalSubmission['dispatchState'],
+  overrides: Partial<AgentJournalSubmission> = {}
+): AgentJournalSubmission {
+  return {
+    clientMessageId: 'message-1',
+    fence: 1,
+    payloadFingerprint: 'fingerprint',
+    dispatchState,
+    providerItemId: null,
+    reason: null,
+    submittedAt: 1,
+    resolvedAt: dispatchState === 'pending' ? null : 2,
+    ...overrides
+  }
+}
+
 describe('conversationCommandBlocked background tasks', () => {
   it('admits the command when nothing is being monitored', () => {
     expect(conversationCommandBlocked(contextWith(null), RECORD)).toBeNull()
@@ -52,19 +83,44 @@ describe('conversationCommandBlocked background tasks', () => {
     // so a live fan-out never re-labels the reason or blocks anything new.
     const ctx = contextWith({ state: 'monitoring', supportsTaskStop: true })
     ctx.journal.snapshot = () =>
-      ({
-        items: [
-          {
-            id: 'turn-1',
-            body: {
-              kind: 'status',
-              turnLifecycle: { turnId: 'turn-1', state: 'running' }
-            }
+      snapshot([
+        {
+          itemId: 'turn-1',
+          revision: 1,
+          sequence: 1,
+          observedAt: 1,
+          body: {
+            kind: 'status',
+            text: 'Working',
+            turnLifecycle: { turnId: 'turn-1', state: 'running' }
           }
-        ]
-      }) as unknown as ReturnType<typeof ctx.journal.snapshot>
+        }
+      ])
     expect(conversationCommandBlocked(ctx, RECORD)).toBe(
       'Wait for the current turn to finish before using this command.'
     )
   })
 })
+
+describe('conversationCommandBlocked dispatch ownership', () => {
+  it.each(['pending', 'unknown'] as const)('blocks a live %s dispatch', (dispatchState) => {
+    expect(conversationCommandBlocked(contextWith(null, [submission(dispatchState)]), RECORD)).toBe(
+      'Resolve pending or unconfirmed messages before using this command.'
+    )
+  })
+
+  it('admits a command after turn settlement retires an unconfirmed dispatch', () => {
+    const retired = submission('unknown', {
+      reason: 'turn_settled_before_acknowledgement',
+      recovered: true
+    })
+
+    expect(conversationCommandBlocked(contextWith(null, [retired]), RECORD)).toBeNull()
+  })
+
+  it('does not let an unanswered dispatch from an older owner block the current fence', () => {
+    expect(
+      conversationCommandBlocked(contextWith(null, [submission('pending', { fence: 0 })]), RECORD)
+    ).toBeNull()
+  })
+})

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

@@ -1,9 +1,19 @@
 import type { AgentSessionRecord } from '../../../shared/agent-session-record'
-import { activeStructuredAgentSessionTurnId } from '../../../shared/structured-agent-session-projection'
+import {
+  activeStructuredAgentSessionTurnId,
+  hasUnansweredStructuredAgentSessionDispatch
+} from '../../../shared/structured-agent-session-projection'
 import type { AgentSessionTurnContext } from './structured-agent-session-turns'
 
+type ConversationCommandAdmissionContext = {
+  sessionId: AgentSessionTurnContext['sessionId']
+  fence: AgentSessionTurnContext['fence']
+  journal: Pick<AgentSessionTurnContext['journal'], 'snapshot' | 'submissions'>
+  adapter: Pick<AgentSessionTurnContext['adapter'], 'backgroundTaskState'>
+}
+
 export function conversationCommandBlocked(
-  ctx: AgentSessionTurnContext,
+  ctx: ConversationCommandAdmissionContext,
   record: AgentSessionRecord
 ): string | null {
   const items = ctx.journal.snapshot().items
@@ -47,11 +57,7 @@ export function conversationCommandBlocked(
       ? 'Stop background tasks before using this command.'
       : 'Wait for background tasks to finish before using this command.'
   }
-  if (
-    ctx.journal
-      .submissions()
-      .some((entry) => entry.dispatchState === 'pending' || entry.dispatchState === 'unknown')
-  ) {
+  if (hasUnansweredStructuredAgentSessionDispatch(ctx.journal.submissions(), ctx.fence)) {
     return 'Resolve pending or unconfirmed messages before using this command.'
   }
   return null

src/shared/agent-session-journal-types.ts

@@ -247,8 +247,8 @@ export type AgentJournalSubmission = {
   reason: string | null
   submittedAt: number
   resolvedAt: number | null
-  /** Set when crash reconciliation resolved the dispatch, not the provider. A live
-   *  `unknown` is a send still outstanding; a recovered one outlived its writer. */
+  /** Set when host lifecycle evidence proves the dispatch has no live owner. A live
+   *  `unknown` is still outstanding; a recovered one remains delivery-ambiguous but retired. */
   recovered?: true
 }
 

src/shared/structured-agent-session-projection.test.ts

@@ -225,6 +225,25 @@ describe('structured agent session status projection', () => {
     })
   })
 
+  it('reads a turn-settled unknown as retired work for every shared status projection', () => {
+    const asked = item('asked', 1, {
+      kind: 'message',
+      role: 'user',
+      blocks: [{ type: 'text', text: 'go' }]
+    })
+    const retired = {
+      ...submission('m1', 'unknown'),
+      reason: 'turn_settled_before_acknowledgement',
+      recovered: true as const
+    }
+
+    expect(hasUnansweredStructuredAgentSessionDispatch([retired], 1)).toBe(false)
+    expect(projectStructuredAgentSessionStatus([asked], [retired], 1)).toBe('idle')
+    expect(projectStructuredAgentSessionStatusSummary([asked], [retired], 1)).toMatchObject({
+      status: 'idle'
+    })
+  })
+
   it('carries the running tool and the newest assistant prose the sidebar row shows', () => {
     const ask = item('ask', 1, {
       kind: 'message',

src/shared/structured-agent-session-projection.ts

@@ -187,8 +187,8 @@ export function hasPersistedStructuredAgentSessionTurn(
  * every session list, so the send itself is the evidence.
  *
  * A live `unknown` still counts because an ambiguous adapter reply does not prove the provider
- * stopped. A recovered `unknown` does not — it outlived the host generation that sent it, so
- * there is nothing still running to report.
+ * stopped. A recovered `unknown` does not — host lifecycle evidence retired its execution owner,
+ * so there is nothing still running to report.
  */
 export function hasUnansweredStructuredAgentSessionDispatch(
   submissions: readonly AgentJournalSubmission[],