fix(native-chat): tie each rejected message to the failed start that rejected it

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

@@ -30,6 +30,15 @@ export function applyJournalDispatchRow(
   } else {
     delete submission.rejection
   }
+  if (
+    row.state === 'rejected' &&
+    typeof row.rejectedByStartKey === 'string' &&
+    row.rejectedByStartKey
+  ) {
+    submission.rejectedByStartKey = row.rejectedByStartKey
+  } else {
+    delete submission.rejectedByStartKey
+  }
   submission.resolvedAt = row.state === 'pending' ? null : row.ts
   if (row.state === 'pending') {
     submission.handedOverAt = row.ts

src/main/native-chat/agent-session-journal/journal-pending-submission-recovery.ts

@@ -1,8 +1,8 @@
-import type { AgentJournalDispatchRejection } from '../../../shared/agent-session-failure-words'
 import type { AgentJournalSubmission } from '../../../shared/agent-session-journal-types'
 import { isQueuedAgentJournalSubmission } from '../../../shared/agent-session-queued-submission'
 import { DISPATCH_DOUBT_HOST_RESTARTED } from './journal-dispatch-doubt-reasons'
 import type { AgentSessionJournal } from './journal-store'
+import type { JournalDispatchRejectionInput } from './journal-store-contracts'
 
 /** Settles every submission a process fact left unanswerable. Doubt is never
  *  proof of non-delivery, so nothing here ever becomes re-deliverable. A queued
@@ -41,7 +41,7 @@ export async function markJournalPendingSubmissionsUnknown(
 export async function rejectJournalPendingSubmissions(
   journal: AgentSessionJournal,
   fence: number,
-  rejection: AgentJournalDispatchRejection
+  rejection: JournalDispatchRejectionInput
 ): Promise<string[]> {
   const unwritten = journal
     .submissions()
@@ -67,7 +67,7 @@ export async function rejectJournalPendingSubmissions(
 export async function rejectJournalQueuedSubmissions(
   journal: AgentSessionJournal,
   fence: number,
-  rejection: AgentJournalDispatchRejection,
+  rejection: JournalDispatchRejectionInput,
   which: (submission: AgentJournalSubmission) => boolean = () => true
 ): Promise<string[]> {
   const queued = journal

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

@@ -10,7 +10,7 @@ import {
   createJournalReducerState,
   type JournalReducerState
 } from './journal-reducer'
-import type { JournalRow } from './journal-row-schema'
+import { parseJournalRow, serializeJournalRow, type JournalRow } from './journal-row-schema'
 
 const EPOCH = 'epoch-1'
 
@@ -97,4 +97,53 @@ describe('a rejected dispatch', () => {
     expect(malformed).not.toHaveProperty('rejection')
     expect(malformed && classifyDispatchRejection(malformed)).toMatchObject({ kind: 'queueFull' })
   })
+
+  // Set only by a failed start's writer; a host from before it, or any other writer, writes none.
+  describe('the start that rejected it', () => {
+    const rejectedBy = (rejectedByStartKey: unknown, state = 'rejected'): JournalRow => {
+      const row: JournalRow = {
+        kind: 'dispatch',
+        clientMessageId: 'cm_1',
+        state: state === 'rejected' ? 'rejected' : 'unknown',
+        providerItemId: null,
+        reason: "Codex couldn't start.",
+        rejection: { kind: 'startFailed' },
+        ...base(2)
+      }
+      // A row read from disk carries whatever the host that wrote it did.
+      return Object.assign(row, { rejectedByStartKey })
+    }
+
+    it('copies it onto a rejected submission, beside the fact it leaves alone', () => {
+      const settled = fold([submission, rejectedBy('generation-2')]).submissions.get('cm_1')
+      expect(settled).toMatchObject({
+        dispatchState: 'rejected',
+        rejection: { kind: 'startFailed' },
+        rejectedByStartKey: 'generation-2'
+      })
+      const unnamed = fold([submission, rejectedBy(undefined)]).submissions.get('cm_1')
+      expect(unnamed).not.toHaveProperty('rejectedByStartKey')
+      // Identity, not the situation: the verdict is the same whichever start it names.
+      expect(settled && classifyDispatchRejection(settled)).toEqual(
+        unnamed && classifyDispatchRejection(unnamed)
+      )
+      const doubt = fold([submission, rejectedBy('generation-2', 'unknown')]).submissions.get(
+        'cm_1'
+      )
+      expect(doubt).not.toHaveProperty('rejectedByStartKey')
+    })
+
+    it('drops a malformed one when read, never the row', () => {
+      for (const malformed of ['', 7, { generation: 'generation-2' }]) {
+        const parsed = parseJournalRow(serializeJournalRow(rejectedBy(malformed)))
+        expect(parsed.ok).toBe(true)
+        const settled = parsed.ok ? fold([submission, parsed.row]).submissions.get('cm_1') : null
+        expect(settled).toMatchObject({
+          dispatchState: 'rejected',
+          rejection: { kind: 'startFailed' }
+        })
+        expect(settled).not.toHaveProperty('rejectedByStartKey')
+      }
+    })
+  })
 })

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

@@ -109,6 +109,9 @@ export function journalDispatchRowBuilder(
     providerItemId,
     reason: boundedDispatchReason(input),
     ...(input.state === 'rejected' ? { rejection: input.rejection } : {}),
+    ...(input.state === 'rejected' && input.rejectedByStartKey
+      ? { rejectedByStartKey: input.rejectedByStartKey }
+      : {}),
     ...journalRowBase(state().epoch, seq, input.fence, ts),
     ...(input.recovered ? { recovered: input.recovered } : {}),
     ...(input.state === 'pending' ? { turnScope: input.turnScope } : {})

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

@@ -110,6 +110,9 @@ export type JournalDispatchRow = JournalRowBase & {
   /** On `rejected`: why, typed. Older readers keep the key and ignore it; a malformed one is
    *  dropped when read, never the row. */
   rejection?: AgentSessionFailureFact
+  /** On `rejected` by a failed start: that start's key. Older readers keep the key and ignore it;
+   *  anything but a non-empty string is dropped when read, never the row. */
+  rejectedByStartKey?: string
 }
 
 /** An item mutation may name its own producer, because one batch can CREATE

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

@@ -40,10 +40,16 @@ export type ResolveDispatchInput = {
     | { state: 'pending'; turnScope: AgentJournalTurnScope }
     /** `reason` is what released clients print, `rejection` what newer ones read: both from
      *  `agentSessionFailureWords`, never written by hand. */
-    | ({ state: 'rejected' } & AgentJournalDispatchRejection)
+    | ({ state: 'rejected' } & JournalDispatchRejectionInput)
     | { state: 'unknown'; reason?: string | null }
   )
 
+/** A rejection, and the failed start that wrote it when it is one: only that start's writer
+ *  names it, so its messages pair with its row by the row's own key. */
+export type JournalDispatchRejectionInput = AgentJournalDispatchRejection & {
+  rejectedByStartKey?: string
+}
+
 export type JournalAppendResult = {
   cursor: AgentJournalCursor
   itemId: string

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

@@ -1,7 +1,6 @@
 // Append-only journal store for one agent session. It owns the chat's fold and write queue, and no
 // connection: every statement goes through the host's one journal database.
 
-import type { AgentJournalDispatchRejection } from '../../../shared/agent-session-failure-words'
 import { randomUUID } from 'node:crypto'
 import type {
   AgentJournalAcceptanceReceipt,
@@ -48,6 +47,7 @@ import {
 import type {
   AgentSessionJournalOptions,
   JournalAppendResult,
+  JournalDispatchRejectionInput,
   JournalItemAppendOptions,
   JournalItemLinkageVisitor,
   JournalLifecycleBatchInput,
@@ -347,15 +347,15 @@ export class AgentSessionJournal {
   /** Reject unanswered sends after an owner that never proved its start ended: none was written. */
   async rejectPendingSubmissions(
     fence: number,
-    rejection: AgentJournalDispatchRejection
+    rejection: JournalDispatchRejectionInput
   ): Promise<string[]> {
     return rejectJournalPendingSubmissions(this, fence, rejection)
   }
 
   /** Reject sends accepted but never handed over, optionally only those `which` names. */
   async rejectQueuedSubmissions(
     fence: number,
-    rejection: AgentJournalDispatchRejection,
+    rejection: JournalDispatchRejectionInput,
     which?: (submission: AgentJournalSubmission) => boolean
   ): Promise<string[]> {
     return rejectJournalQueuedSubmissions(this, fence, rejection, which)

src/main/native-chat/agent-session-wire/structured-agent-session-accept-then-deliver.test.ts

@@ -336,11 +336,13 @@ describe('a start the chat needed and did not get', () => {
       refusal: { code: 'agent_session_operation_invalid', details: { ownerVerdict: 'exited' } }
     }
     expect(await errorFailures()).toEqual([failure])
+    // No child was published, so the start is keyed by the oldest message it was for.
     for (const id of [first, second]) {
       expect(await submission(id)).toMatchObject({
         dispatchState: 'rejected',
         reason: rows[0],
-        rejection: failure
+        rejection: failure,
+        rejectedByStartKey: first
       })
     }
 
@@ -444,6 +446,7 @@ describe('a start the chat needed and did not get', () => {
     await eventually(async () => expect((await submission(id))?.dispatchState).toBe('rejected'))
     expect(await errorRows()).toEqual([row.text])
     expect(await errorFailures()).toEqual([row.failure])
+    expect(await submission(id)).toMatchObject({ rejectedByStartKey: id })
     const framedRows = events.flatMap((event) =>
       event.type === 'batch' || event.type === 'snapshot'
         ? (event.type === 'batch' ? event.batch.items : event.page.items).filter(
@@ -535,6 +538,8 @@ describe('what an earlier host process left behind', () => {
         surface: 'rejection'
       })
     })
+    // No start rejected it, so it names none.
+    expect(await submission('queued')).not.toHaveProperty('rejectedByStartKey')
     expect(dispatch).not.toHaveBeenCalled()
   })
 

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

@@ -25,6 +25,7 @@ import {
   AGENT_JOURNAL_THREAD_SCOPE,
   type AgentJournalItemIdentity,
   type AgentJournalMessageItem,
+  type AgentJournalStatusItem,
   type AgentJournalSubmission
 } from '../../../shared/agent-session-journal-types'
 import type { AgentSessionConversationCommand } from '../../../shared/agent-session-conversation-command'
@@ -41,7 +42,6 @@ import type {
   StructuredAgentSessionAdapter,
   StructuredAgentSessionProviderChildPhase
 } from './structured-agent-session-adapter'
-import { structuredAgentSessionStartFailure } from './structured-agent-session-failure-text'
 import { conversationCommandBlocked } from './structured-conversation-command-admission'
 
 export const STRUCTURED_AGENT_SESSION_COMPACT_COMMAND = 'compact'
@@ -157,12 +157,14 @@ export type StructuredAgentSessionCommandHandoverContext = {
   now: () => number
 }
 
-/** Refuses the command, or opens its turn and sends it. */
+/** Refuses the command, or opens its turn and sends it. Returns the cause when the child's start
+ *  failed at the handover, leaving the command pending for the caller to record that start's
+ *  failure for it and everything queued behind it. */
 export async function handOverStructuredAgentSessionCommand(
   ctx: StructuredAgentSessionCommandHandoverContext,
   submission: AgentJournalSubmission,
   body: AgentJournalMessageItem
-): Promise<void> {
+): Promise<{ error: unknown } | null> {
   const { clientMessageId } = submission
   // Provider frames already received decide whether a turn is running.
   await ctx.flushStreamedEvents()
@@ -174,7 +176,7 @@ export async function handOverStructuredAgentSessionCommand(
       ...agentSessionFailureWords(blocked, { ...ctx.failureTextContext, surface: 'rejection' }),
       fence: ctx.fence
     })
-    return
+    return null
   }
   const turn = structuredAgentSessionCommandTurn(clientMessageId)
   await ctx.journal.resolveDispatch({
@@ -204,20 +206,16 @@ export async function handOverStructuredAgentSessionCommand(
       command: { clientMessageId, ...turn, running }
     })
   } catch (error) {
-    // A child that had not proven its start took nothing, so the command provably did not run. Any
-    // other throw is a lost reply: the command may have run.
-    const unsent =
-      ctx.providerChildPhase?.() === 'starting'
-        ? {
-            state: 'rejected' as const,
-            ...structuredAgentSessionStartFailure({ error }, ctx.failureTextContext)
-          }
-        : {
-            state: 'unknown' as const,
-            reason: error instanceof Error ? error.message : String(error)
-          }
-    await settleUnsentCommand(ctx, clientMessageId, unsent)
-    return
+    // A child that had not proven its start took nothing: its start failed. Any other throw is a
+    // lost reply: the command may have run.
+    if (ctx.providerChildPhase?.() === 'starting') {
+      return { error }
+    }
+    await settleUnsentCommand(ctx, clientMessageId, {
+      state: 'unknown',
+      reason: error instanceof Error ? error.message : String(error)
+    })
+    return null
   }
   if (admission.state === 'rejected') {
     // The provider refused the compaction itself: its row reads as the compaction failing.
@@ -231,6 +229,7 @@ export async function handOverStructuredAgentSessionCommand(
     // An unknown write leaves the turn to the provider's end or the child's: it may have run.
     await ctx.journal.resolveDispatch({ clientMessageId, ...admission, fence: ctx.fence })
   }
+  return null
 }
 
 /** Where the turn a handed-over submission runs in starts counting: its handover, so time spent
@@ -256,28 +255,52 @@ async function settleUnsentCommand(
     | { state: 'unknown'; reason: string },
   /** What the result row reports, when it is not the rejection's own fact. */
   rowFailure?: AgentSessionFailureFact
+): Promise<void> {
+  await ctx.journal.resolveDispatch({ clientMessageId, ...unsent, fence: ctx.fence })
+  await endUnsentCommandTurn(
+    ctx,
+    clientMessageId,
+    unsent.state === 'rejected'
+      ? {
+          kind: 'status',
+          ...agentSessionFailureWords(rowFailure ?? unsent.rejection, {
+            ...ctx.failureTextContext,
+            surface: 'row'
+          }),
+          tone: 'error'
+        }
+      : null
+  )
+}
+
+/** A command whose start failed at its handover, once that start's writer has rejected it: its
+ *  turn ends failed, and the start's row is its one row. */
+export async function endStructuredAgentSessionCommandStartFailure(
+  ctx: Pick<StructuredAgentSessionCommandHandoverContext, 'journal' | 'fence' | 'now'>,
+  clientMessageId: string
+): Promise<void> {
+  await endUnsentCommandTurn(ctx, clientMessageId, 'failed')
+}
+
+/** Ends the command's turn if it still runs: failed when the command was refused (`result` is its
+ *  row, if it has one of its own), else unverifiable. */
+async function endUnsentCommandTurn(
+  ctx: Pick<StructuredAgentSessionCommandHandoverContext, 'journal' | 'fence' | 'now'>,
+  clientMessageId: string,
+  result: AgentJournalStatusItem | 'failed' | null
 ): Promise<void> {
   const turn = structuredAgentSessionCommandTurn(clientMessageId)
   const running = readAgentJournalTurn(ctx.journal.itemBody(turn.itemId) ?? undefined)
-  await ctx.journal.resolveDispatch({ clientMessageId, ...unsent, fence: ctx.fence })
   if (running?.state !== 'running') {
     return
   }
-  const refused = unsent.state === 'rejected'
   const mutations: JournalLifecycleMutationInput[] = [
-    ...(refused
+    ...(result && result !== 'failed'
       ? [
           {
             kind: 'item' as const,
             identity: turn.resultIdentity,
-            body: {
-              kind: 'status' as const,
-              ...agentSessionFailureWords(rowFailure ?? unsent.rejection, {
-                ...ctx.failureTextContext,
-                surface: 'row'
-              }),
-              tone: 'error' as const
-            },
+            body: result,
             turnScope: { kind: 'turn' as const, turnItemId: turn.itemId }
           }
         ]
@@ -287,7 +310,7 @@ async function settleUnsentCommand(
       identity: turn.identity,
       body: agentJournalTurnBody({
         ...running,
-        ...(refused
+        ...(result
           ? { state: 'completed', outcome: 'failure', completedAt: ctx.now() }
           : { state: 'unverifiable' })
       }),