fix(native-chat): a failed agent start fails only its own message

config/scripts/vitest-sqlite-database-runtime-files.mjs

@@ -31,6 +31,7 @@ export const SQLITE_DATABASE_RUNTIME_INCLUDE = [
   'src/main/runtime/claude-structured-resumed-start-failure.test.ts',
   'src/main/runtime/claude-structured-send-during-startup.test.ts',
   'src/main/runtime/claude-structured-send-restart-dies-before-dispatch.test.ts',
+  'src/main/runtime/claude-structured-send-signed-out-run.test.ts',
   'src/main/runtime/claude-structured-session-integration.test.ts',
   'src/main/runtime/claude-structured-startup-fault-is-not-an-exit.test.ts',
   'src/main/runtime/exit-provenance-audit.test.ts',

config/scripts/vitest-sqlite-runtime-files.mjs

@@ -124,6 +124,7 @@ export const SQLITE_RUNTIME_INCLUDE = [
   'src/main/native-chat/agent-session-wire/structured-agent-session-settled-attach-retry.test.ts',
   'src/main/native-chat/agent-session-wire/structured-agent-session-shutdown-cut-row.test.ts',
   'src/main/native-chat/agent-session-wire/structured-agent-session-stale-turn-verdict.test.ts',
+  'src/main/native-chat/agent-session-wire/structured-agent-session-start-failure-run-lane.test.ts',
   'src/main/native-chat/agent-session-wire/structured-agent-session-start-failure-writer.test.ts',
   'src/main/native-chat/agent-session-wire/structured-agent-session-startup-reconcile-failure.test.ts',
   'src/main/native-chat/agent-session-wire/structured-agent-session-status-feed-child-work.test.ts',

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

@@ -61,6 +61,7 @@ export function applyJournalDispatchRow(
   }
   if (row.state === 'accepted') {
     noteTurnAccepted(state, submission)
+    state.latestAcceptedSequence = Math.max(state.latestAcceptedSequence, row.seq)
   }
   if (row.state !== 'accepted' || !row.providerItemId) {
     return

src/main/native-chat/agent-session-journal/journal-lifecycle-batch-appender.ts

@@ -11,9 +11,11 @@ import type {
   JournalResolvedLifecycleBatchInput
 } from './journal-store-contracts'
 import type { JournalRow } from './journal-row-schema'
-import { journalQueuedRejectionRowBuilders } from './journal-pending-submission-recovery'
+import { journalBatchRejectionRowBuilders } from './journal-pending-submission-recovery'
+import { withoutRestatedStartFailureRows } from './journal-start-failure-run'
 
 const SETTLEMENT_ALREADY_APPLIED = new Error('journal_settlement_already_applied')
+const NOTHING_LEFT_TO_WRITE = new Error('journal_lifecycle_batch_nothing_left')
 
 export class JournalLifecycleBatchAppender {
   constructor(
@@ -28,25 +30,25 @@ export class JournalLifecycleBatchAppender {
   ) {}
 
   append(input: JournalLifecycleBatchInput): Promise<AgentJournalCursor> {
-    const { rejectsQueued } = input
-    if (rejectsQueued) {
-      // Planned on the lane: the sends queued then, and this batch unless it already landed. With
-      // none left (a Stop withdrew them first) it failed no one, so nothing is written.
+    const { rejects } = input
+    if (rejects) {
+      // Planned on the lane: the send if it is still `which` then, and this batch unless it already
+      // landed or nothing is left once its run's row is counted. With the send settled meanwhile it
+      // failed no one, so nothing is written.
       return this.deps
         .enqueueRows(() => {
-          const rejections = journalQueuedRejectionRowBuilders(
-            this.deps.state,
-            input.fence,
-            rejectsQueued
-          )
-          return rejections.length === 0 || this.wasApplied(input.settlementId)
+          const rejections = journalBatchRejectionRowBuilders(this.deps.state, input.fence, rejects)
+          const mutations = withoutRestatedStartFailureRows(this.deps.state(), input.mutations)
+          return rejections.length === 0 ||
+            mutations.length === 0 ||
+            this.wasApplied(input.settlementId)
             ? rejections
             : [
                 ...rejections,
                 journalLifecycleBatchRowBuilder(
                   this.deps.state,
                   input.settlementId,
-                  input.mutations,
+                  mutations,
                   input
                 )
               ]
@@ -56,22 +58,26 @@ export class JournalLifecycleBatchAppender {
     if (this.wasApplied(input.settlementId)) {
       return Promise.resolve(this.deps.cursor())
     }
-    const build = journalLifecycleBatchRowBuilder(
-      this.deps.state,
-      input.settlementId,
-      input.mutations,
-      input
-    )
     return this.deps
       .enqueue((seq, ts) => {
         if (this.wasApplied(input.settlementId)) {
           throw SETTLEMENT_ALREADY_APPLIED
         }
-        return build(seq, ts)
+        const mutations = withoutRestatedStartFailureRows(this.deps.state(), input.mutations)
+        // Its only row restated its run's: nothing to write. An empty batch still fails its bound.
+        if (mutations.length === 0 && input.mutations.length > 0) {
+          throw NOTHING_LEFT_TO_WRITE
+        }
+        return journalLifecycleBatchRowBuilder(
+          this.deps.state,
+          input.settlementId,
+          mutations,
+          input
+        )(seq, ts)
       })
       .then((row) => ({ epoch: row.epoch, sequence: row.seq }))
       .catch((error: unknown) => {
-        if (error === SETTLEMENT_ALREADY_APPLIED) {
+        if (error === SETTLEMENT_ALREADY_APPLIED || error === NOTHING_LEFT_TO_WRITE) {
           return this.deps.cursor()
         }
         throw error

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

@@ -6,6 +6,7 @@ import type { JournalReducerState } from './journal-reducer'
 import { journalDispatchRowBuilder } from './journal-row-builders'
 import type { JournalRow } from './journal-row-schema'
 import type { AgentSessionJournal } from './journal-store'
+import type { JournalBatchRejection } 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
@@ -92,20 +93,22 @@ export async function rejectJournalQueuedSubmissions(
   return queued.map((entry) => entry.clientMessageId)
 }
 
-/** Rows rejecting every submission still queued, read from `state` when called: for an append
- *  that must carry them with what follows, in one transaction. */
-export function journalQueuedRejectionRowBuilders(
+/** The row rejecting that one submission, read from `state` when called, or none when `which` no
+ *  longer holds for it: for an append that must carry it with what follows, in one transaction. */
+export function journalBatchRejectionRowBuilders(
   state: () => JournalReducerState,
   fence: number,
-  rejection: AgentJournalDispatchRejection
+  { clientMessageId, which, ...rejection }: JournalBatchRejection
 ): ((seq: number, ts: number) => JournalRow)[] {
-  return [...state().submissions.values()].filter(isQueuedAgentJournalSubmission).map((entry) =>
-    journalDispatchRowBuilder(state, {
-      clientMessageId: entry.clientMessageId,
-      state: 'rejected',
-      ...rejection,
-      fence,
-      recovered: true
-    })
-  )
+  const submission = state().submissions.get(clientMessageId)
+  return submission && which(submission)
+    ? [
+        journalDispatchRowBuilder(state, {
+          clientMessageId,
+          state: 'rejected',
+          ...rejection,
+          fence
+        })
+      ]
+    : []
 }

src/main/native-chat/agent-session-journal/journal-queued-rejection-batch.test.ts

@@ -1,5 +1,6 @@
-// A failed start's row and the queued messages it failed land in ONE append, the messages first:
-// no reader meets one without the other, and the messages sit above the row that says why.
+// A failed start's row and the queued message it failed land in ONE append, the message first: no
+// reader meets one without the other, the message sits above the row that says why, and a message
+// queued behind it is left for its own start.
 
 import { mkdtemp, rm } from 'node:fs/promises'
 import { tmpdir } from 'node:os'
@@ -13,7 +14,9 @@ import {
   AGENT_JOURNAL_THREAD_SCOPE,
   type AgentSessionJournalIdentity
 } from '../../../shared/agent-session-journal-types'
+import { isQueuedAgentJournalSubmission } from '../../../shared/agent-session-queued-submission'
 import type { AgentSessionJournal } from './journal-store'
+import { MAX_JOURNAL_LIFECYCLE_BATCH_MUTATIONS } from './journal-row-schema'
 import {
   closeTestJournalHostDatabases,
   createTrackedJournalOpener
@@ -30,7 +33,7 @@ const IDENTITY: AgentSessionJournalIdentity = {
 const START_FAILED = agentSessionFailureWords(agentSessionFailureFact('providerStartFailed'), {
   surface: 'rejection'
 })
-const ERROR_ROW = { provider: 'orca', clientMessageId: 'start-failure:gen-1' } as const
+const ERROR_ROW = { provider: 'orca', clientMessageId: 'start-failure:first' } as const
 
 let root: string
 let clock = 1_000
@@ -66,7 +69,7 @@ async function openWithQueued(...ids: string[]): Promise<AgentSessionJournal> {
 
 function startFailureBatch(mutations = 1) {
   return {
-    settlementId: 'start-failure:gen-1',
+    settlementId: 'start-failure:first',
     fence: 0,
     recovered: true as const,
     mutations: Array.from({ length: mutations }, () => ({
@@ -75,44 +78,55 @@ function startFailureBatch(mutations = 1) {
       body: { kind: 'status' as const, tone: 'error' as const, text: 'Claude did not start.' },
       turnScope: AGENT_JOURNAL_THREAD_SCOPE
     })),
-    rejectsQueued: START_FAILED
+    rejects: { ...START_FAILED, clientMessageId: 'first', which: isQueuedAgentJournalSubmission }
   }
 }
 
-it('writes the rejections first and the row after them, and draws the messages above it', async () => {
+it('writes the rejection first and the row right after it, leaving the message behind it queued', async () => {
   const journal = await openWithQueued('first', 'second')
   const before = journal.cursor().sequence
 
   await journal.appendLifecycleBatch(startFailureBatch())
 
-  expect(journal.cursor().sequence).toBe(before + 3)
-  expect(journal.submissions().map((entry) => entry.dispatchState)).toEqual([
-    'rejected',
-    'rejected'
-  ])
-  const order = journal.snapshot().items.map((item) => item.itemId)
-  expect(order).toEqual([
-    agentJournalSubmissionKey('first'),
-    agentJournalSubmissionKey('second'),
-    'orca:start-failure%3Agen-1'
-  ])
+  expect(journal.cursor().sequence).toBe(before + 2)
+  expect(journal.submissions().map((entry) => entry.dispatchState)).toEqual(['rejected', 'pending'])
+  const placed = new Map(journal.snapshot().items.map((item) => [item.itemId, item]))
+  const message = placed.get(agentJournalSubmissionKey('first'))
+  const row = placed.get('orca:start-failure%3Afirst')
+  // One write: the row is the next row after the rejection, at the same instant.
+  expect(row?.sequence).toBe((message?.sequence ?? 0) + 1)
+  expect(row?.observedAt).toBe(message?.observedAt)
 })
 
 it('writes neither when the row cannot be written', async () => {
   const journal = await openWithQueued('first')
   const before = journal.cursor().sequence
 
-  // An empty batch breaks the row's bound, so the transaction rolls back as a whole.
-  await expect(journal.appendLifecycleBatch(startFailureBatch(0))).rejects.toThrow(
-    'journal_lifecycle_batch_mutation_bound_exceeded'
-  )
+  // A batch over the row's bound fails to build, so the transaction rolls back as a whole.
+  await expect(
+    journal.appendLifecycleBatch(startFailureBatch(MAX_JOURNAL_LIFECYCLE_BATCH_MUTATIONS + 1))
+  ).rejects.toThrow('journal_lifecycle_batch_mutation_bound_exceeded')
 
   expect(journal.cursor().sequence).toBe(before)
   expect(journal.submissions().map((entry) => entry.dispatchState)).toEqual(['pending'])
 })
 
+// A start failing as its run's row already says rejects its message and writes no row.
+it('writes only the rejection when the batch carries no row', async () => {
+  const journal = await openWithQueued('first', 'second')
+  const before = journal.cursor().sequence
+
+  await journal.appendLifecycleBatch(startFailureBatch(0))
+
+  expect(journal.cursor().sequence).toBe(before + 1)
+  expect(journal.submissions().map((entry) => entry.dispatchState)).toEqual(['rejected', 'pending'])
+  expect(journal.snapshot().items.map((item) => item.itemId)).not.toContain(
+    'orca:start-failure%3Afirst'
+  )
+})
+
 // A Stop that reaches the lane first takes the message back; the failed start then failed no one.
-it('writes nothing when a Stop withdrew every queued message first', async () => {
+it('writes nothing when a Stop withdrew the message first', async () => {
   const journal = await openWithQueued('first')
   const withdrawal = agentSessionFailureWords(agentSessionFailureFact('cancelled'), {
     surface: 'rejection'

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

@@ -63,6 +63,9 @@ export type JournalReducerState = {
   /** The submission row of the latest turn the provider accepted, whoever sent it; 0 when none.
    *  Kept as it folds so the queue's pause reads it in O(1). */
   latestAcceptedTurnSequence: number
+  /** The row at which the provider last accepted a send, with or without an item of its own; 0
+   *  when none. Kept as it folds; it ends a run of failed starts. */
+  latestAcceptedSequence: number
   /** The latest person's Stop event and Resume, what the queue's pause is derived from. */
   queuePauseMarks: JournalQueuePauseMarks
 }
@@ -84,6 +87,7 @@ export function createJournalReducerState(sessionId: string, epoch: string): Jou
     appliedSettlementIds: new Set(),
     derivedTurnScope: new JournalDerivedTurnScope(),
     latestAcceptedTurnSequence: 0,
+    latestAcceptedSequence: 0,
     queuePauseMarks: createJournalQueuePauseMarks()
   }
 }

src/main/native-chat/agent-session-journal/journal-start-failure-run.ts

@@ -0,0 +1,70 @@
+// One row speaks for a run of starts that fail alike. Until a turn is delivered, a start that fails
+// as the chat's latest start-failure row says writes no row of its own, and its rejected message is
+// read under that row. Decided on the journal's lane, from the fold every earlier write landed in,
+// as the batch carrying the row is planned; nothing marks the run.
+
+import {
+  readAgentSessionFailureFact,
+  sameAgentSessionFailureFact,
+  type AgentSessionFailureFact
+} from '../../../shared/agent-session-failure'
+import type { AgentJournalRenderItem } from '../../../shared/agent-session-journal-types'
+import { agentJournalItemKey } from '../../../shared/agent-session-journal-item-key'
+import {
+  isStructuredAgentSessionCommandStartFailureRow,
+  isStructuredAgentSessionStartFailureRow
+} from '../../../shared/structured-agent-session-start-failure-row-key'
+import type { JournalReducerState } from './journal-reducer'
+import type { JournalLifecycleMutationInput } from './journal-row-builders'
+
+/** A batch's mutations less any start-failure row whose failure its run's row already states. */
+export function withoutRestatedStartFailureRows(
+  state: Pick<JournalReducerState, 'items' | 'latestAcceptedSequence'>,
+  mutations: readonly JournalLifecycleMutationInput[]
+): readonly JournalLifecycleMutationInput[] {
+  return mutations.filter((mutation) => {
+    if (mutation.kind !== 'item' || mutation.body.kind !== 'status') {
+      return true
+    }
+    const itemId = agentJournalItemKey(mutation.identity)
+    if (
+      !isStructuredAgentSessionStartFailureRow(itemId) ||
+      // A command's failed start always says so: a client that hides its message has only this.
+      isStructuredAgentSessionCommandStartFailureRow(itemId, (id) => state.items.get(id)?.body)
+    ) {
+      return true
+    }
+    const fact = readAgentSessionFailureFact(mutation.body.failure)
+    return fact === undefined || !journalStartFailureAlreadyStated(state, fact)
+  })
+}
+
+/** Whether the latest start-failure row states this failure, with no turn delivered since. */
+function journalStartFailureAlreadyStated(
+  state: Pick<JournalReducerState, 'items' | 'latestAcceptedSequence'>,
+  failure: AgentSessionFailureFact
+): boolean {
+  let latest: AgentJournalRenderItem | undefined
+  for (const item of state.items.values()) {
+    if (
+      isStructuredAgentSessionStartFailureRow(item.itemId) &&
+      (latest === undefined || item.sequence > latest.sequence)
+    ) {
+      latest = item
+    }
+  }
+  const stated =
+    latest?.body.kind === 'status' ? readAgentSessionFailureFact(latest.body.failure) : undefined
+  return (
+    latest !== undefined &&
+    stated !== undefined &&
+    sameAgentSessionFailureFact(stated, failure) &&
+    // A command's row names the command's next step, so it starts no run for a message.
+    !isStructuredAgentSessionCommandStartFailureRow(
+      latest.itemId,
+      (id) => state.items.get(id)?.body
+    ) &&
+    // Any send accepted after the row, a command too, ends its run: the next failure is news.
+    state.latestAcceptedSequence < latest.sequence
+  )
+}

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

@@ -8,6 +8,7 @@ import type {
   AgentJournalProducerLinkage,
   AgentJournalResetReason,
   AgentJournalRowAttribution,
+  AgentJournalSubmission,
   AgentJournalTurnScope,
   AgentSessionJournalIdentity
 } from '../../../shared/agent-session-journal-types'
@@ -74,15 +75,21 @@ export type JournalLifecycleBatchInput = {
   mutations: readonly JournalLifecycleMutationInput[]
   fence: number
   recovered?: true
-  /** Rejects the sends still queued with this first, in the same append: a failed start's row
-   *  follows the messages it failed, and no reader meets one without the other. With none still
-   *  queued, the batch is not written either. */
-  rejectsQueued?: AgentJournalDispatchRejection
+  /** Rejects that send first, in the same append: a failed start's row follows the message it
+   *  failed, and no reader meets one without the other. */
+  rejects?: JournalBatchRejection
+}
+
+/** One send a batch rejects, when `which` still holds for it on the lane. When it no longer does —
+ *  a Stop withdrew it, another writer settled it — the batch is not written either. */
+export type JournalBatchRejection = AgentJournalDispatchRejection & {
+  clientMessageId: string
+  which: (submission: AgentJournalSubmission) => boolean
 }
 
 export type JournalResolvedLifecycleBatchInput = Omit<
   JournalLifecycleBatchInput,
-  'mutations' | 'rejectsQueued'
+  'mutations' | 'rejects'
 > & {
   /** Read from the fold with every earlier write landed; may return none. */
   resolve: () => readonly JournalLifecycleMutationInput[]

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

@@ -113,6 +113,7 @@ export function acceptSubmissionFromProviderItem(
   submission.fence = row.fence
   submission.dispatchState = 'accepted'
   noteTurnAccepted(state, submission)
+  state.latestAcceptedSequence = Math.max(state.latestAcceptedSequence, row.seq)
   submission.providerItemId = providerItemId
   submission.reason = null
   submission.resolvedAt = row.ts