fix(native-chat): a message whose delivery is in doubt no longer holds every later send

src/main/native-chat/agent-session-wire/structured-agent-session-codex-stop-row.test.ts

@@ -8,6 +8,12 @@ import { tmpdir } from 'node:os'
 import { join } from 'node:path'
 import { afterEach, beforeEach, describe, expect, it, vi, type MockInstance } from 'vitest'
 import { computeAgentSessionPayloadFingerprint } from '../../../shared/agent-session-mutation-envelope'
+import {
+  createStructuredAgentSessionOutboxEntry,
+  structuredAgentSessionSendMutation
+} from '../../../shared/structured-agent-session-outbox'
+import { reconcileStructuredAgentSessionOutbox } from '../../../shared/structured-agent-session-outbox-reconcile'
+import { admitStructuredAgentSessionOutboxEntry } from '../../../shared/structured-agent-session-outbox-admission'
 import { CodexAppServerRequestError } from '../../codex/codex-app-server-connection'
 import { CodexAppServerTimeoutError } from '../../codex/codex-app-server-session'
 import {
@@ -534,3 +540,166 @@ describe('a message after a Codex Stop whose exit was unproven', () => {
     expect(host['sessions'].get(SESSION)?.lastEndedChild).toMatchObject({ rootGone: true })
   })
 })
+
+// A follow-up steered into an open Codex turn and not yet echoed, when a Stop's interrupt fails.
+// The desktop's outbox is fed the host's own rows, as its reconcile and drain read them: nothing
+// the host keeps in doubt may hold the next message, which goes to a fresh Codex when the Stop
+// ended the child, or to the same one when it did not.
+describe('a follow-up Codex never echoed, when a Stop whose interrupt failed', () => {
+  /** A running turn, then a follow-up Codex took as a steer and has not echoed. */
+  async function steeredFollowUp(): Promise<string> {
+    await runningTurn()
+    codex.routes['turn/steer'] = turns.routes['turn/steer']
+    const followUp = await send('and also this')
+    if (!followUp.ok) {
+      throw new Error(JSON.stringify(followUp.refusal))
+    }
+    await vi.waitFor(() =>
+      expect(codex.connections.at(-1)!.calls.some((call) => call.method === 'turn/steer')).toBe(
+        true
+      )
+    )
+    return followUp.value.clientMessageId
+  }
+
+  function queued(clientMessageId: string, text: string) {
+    return createStructuredAgentSessionOutboxEntry({
+      clientMessageId,
+      sessionId: SESSION,
+      text,
+      attachments: [],
+      queuedAt: NOW
+    })
+  }
+
+  async function followUpRow(followUpId: string) {
+    const { submissions } = await host.journalSnapshot(SESSION)
+    return submissions.find((row) => row.clientMessageId === followUpId)
+  }
+
+  /** The desktop's queue as it held them, the follow-up on its way, read through the host's rows:
+   *  the next message is admitted and reaches Codex as a turn of its own. Resolves with the queue
+   *  as it stands once that message is on its way. */
+  async function nextMessageGoesOut(followUpId: string, child: number) {
+    const { submissions } = await host.journalSnapshot(SESSION)
+    const next = queued(hostTestOperationId(), 'carry on')
+    const outbox = reconcileStructuredAgentSessionOutbox(
+      [{ ...queued(followUpId, 'and also this'), state: 'dispatching', lastAttemptAt: NOW }, next],
+      submissions,
+      []
+    )
+    expect(admitStructuredAgentSessionOutboxEntry(outbox)).toEqual({
+      state: 'dispatch',
+      entry: next
+    })
+    launch.resumeThreadId = THREAD
+    expect(await host.send(CALLER, structuredAgentSessionSendMutation(next, 1))).toMatchObject({
+      ok: true
+    })
+    await vi.waitFor(() => expect(codex.connections).toHaveLength(child + 1))
+    await vi.waitFor(() =>
+      expect(
+        codex.connections[child]!.calls.some(
+          (call) => call.method === 'turn/start' && JSON.stringify(call.params).includes('carry on')
+        )
+      ).toBe(true)
+    )
+    return outbox.map((entry) =>
+      entry === next ? { ...entry, state: 'dispatching' as const, lastAttemptAt: NOW } : entry
+    )
+  }
+
+  /** Read through the host's rows as they stand now, the queue admits what is sent after. */
+  async function queueStaysOpen(outbox: ReturnType<typeof reconcileStructuredAgentSessionOutbox>) {
+    const { submissions } = await host.journalSnapshot(SESSION)
+    const later = queued(hostTestOperationId(), 'and then')
+    expect(
+      admitStructuredAgentSessionOutboxEntry(
+        reconcileStructuredAgentSessionOutbox([...outbox, later], submissions, [])
+      )
+    ).toEqual({ state: 'dispatch', entry: later })
+  }
+
+  /** As the real connection: once a close begins it refuses every request, proven or not. The
+   *  first close does not prove the exit. */
+  function firstCloseUnproven(): void {
+    const old = codex.connections.at(-1)!
+    const close = old.close
+    const request = old.request
+    let unproven = 1
+    old.close = async () => {
+      old.closed = true
+      if (unproven === 0) {
+        return close()
+      }
+      unproven -= 1
+      return false
+    }
+    old.request = (method, params) =>
+      old.closed
+        ? Promise.reject(new Error('codex app-server is closing'))
+        : request(method, params)
+  }
+
+  it.each(['internal error', 'unanswered'] as const)(
+    'stays in doubt when the Stop ends the child (%s), and the next message goes to a fresh Codex',
+    async (failure) => {
+      const followUpId = await steeredFollowUp()
+      codex.routes['turn/interrupt'] = () => {
+        throw interruptFailure(failure)
+      }
+
+      await stop()
+      await host.flushStreamedEvents(SESSION)
+      expect(childEndedByStop()).toBe(true)
+      expect(await followUpRow(followUpId)).toMatchObject({
+        dispatchState: 'unknown',
+        recovered: true
+      })
+      await queueStaysOpen(await nextMessageGoesOut(followUpId, 1))
+    }
+  )
+
+  // Codex marks the turn ended before it writes the refusal, so the turn's end follows it.
+  it('keeps the child when Codex had no turn to interrupt, and the next message goes to it', async () => {
+    const followUpId = await steeredFollowUp()
+    codex.routes['turn/interrupt'] = () => {
+      setTimeout(() => turns.end('completed'), 2)
+      throw refused('no active turn to interrupt')
+    }
+
+    await stop()
+    await vi.waitFor(async () => expect((await journalRows()).turns).toEqual(['completed']))
+    await host.flushStreamedEvents(SESSION)
+    expect(childEndedByStop()).toBe(false)
+    // Nothing settled it, so the desktop's entry stays on its way, which holds nothing up.
+    expect(await followUpRow(followUpId)).toMatchObject({ dispatchState: 'pending' })
+    await queueStaysOpen(await nextMessageGoesOut(followUpId, 0))
+  })
+
+  // The stop's unproven close is joined, and proven, by the next send, which settles the follow-up
+  // in doubt only then.
+  it.each(['internal error', 'unanswered'] as const)(
+    'stays in doubt once a Stop that could not prove the exit (%s) is proven, and the queue never stops',
+    async (failure) => {
+      const followUpId = await steeredFollowUp()
+      codex.routes['turn/interrupt'] = () => {
+        throw interruptFailure(failure)
+      }
+      firstCloseUnproven()
+
+      await stop()
+      await host.flushStreamedEvents(SESSION)
+      expect(host['sessions'].get(SESSION)?.child?.close).toBeDefined()
+      expect(await followUpRow(followUpId)).toMatchObject({ dispatchState: 'pending' })
+
+      const outbox = await nextMessageGoesOut(followUpId, 1)
+      expect(host['sessions'].get(SESSION)?.child?.close).toBeUndefined()
+      expect(await followUpRow(followUpId)).toMatchObject({
+        dispatchState: 'unknown',
+        recovered: true
+      })
+      await queueStaysOpen(outbox)
+    }
+  )
+})

src/renderer/src/components/native-chat/NativeChatStructuredSessionDelivery.probe.test.tsx

@@ -119,7 +119,7 @@ describe('NativeChatStructuredSession delivery probe', () => {
     )
   }, 20000)
 
-  it('parks a host-confirmed unknown instead of probing it', async () => {
+  it('never probes a host-confirmed unknown, and sends what follows it', async () => {
     useProbeClock()
     mocks.mode = 'outbox'
     mocks.call.mockRejectedValueOnce(new Error('socket closed')).mockResolvedValue({
@@ -147,7 +147,7 @@ describe('NativeChatStructuredSession delivery probe', () => {
     expect(mocks.call).toHaveBeenCalledOnce()
 
     const sent = mocks.call.mock.calls[0]?.[2] as { envelope: { clientOperationId: string } }
-    // The host now reports an unresolved unknown: another replay is the user's call.
+    // The host now reports an unresolved unknown: it never sends it again, so nothing waits on it.
     mocks.submissions = [
       {
         clientMessageId: sent.envelope.clientOperationId,
@@ -160,15 +160,16 @@ describe('NativeChatStructuredSession delivery probe', () => {
         resolvedAt: null
       }
     ]
-    // Queue a second message purely to re-render so the effect observes the
-    // new submissions; it must stay wedged behind the parked head.
+    // The second message also re-renders, so the effects observe the new submissions.
     await act(async () => {
       send?.('second', [])
     })
     await advanceProbeClock(3000)
-    // From the row on, only the user's Retry moves it, so it says so.
-    expect(screen.getByText('Message delivery is unconfirmed.')).toBeTruthy()
-    expect(mocks.call).toHaveBeenCalledOnce()
+    const texts = mocks.call.mock.calls.map(
+      (call) => (call[2] as { body?: { blocks?: { text?: string }[] } })?.body?.blocks?.[0]?.text
+    )
+    expect(texts).toEqual(['first', 'second'])
+    expect(screen.queryByText('Message delivery is unconfirmed.')).toBeNull()
   }, 20000)
 
   it('still probes while streaming batches rebuild the submissions array', async () => {

src/renderer/src/components/native-chat/NativeChatStructuredSessionDelivery.test.tsx

@@ -313,7 +313,8 @@ describe('NativeChatStructuredSession delivery', () => {
     ['a recovered unknown', { recovered: true }],
     ["an older host's recovered unknown", { reason: 'host_restarted_before_acknowledgement' }]
   ])(
-    'says a send reopened mid-send is unconfirmed, with its Retry, once the journal holds %s',
+    // Recorded in doubt, the host never sends it again: its row draws it, and nothing waits on it.
+    'says nothing of a send reopened mid-send once the journal holds %s, and never resends it',
     async (label, patch) => {
       mocks.mode = 'outbox'
       const sessionId = `session-reopened-${label.replace(/\W+/g, '-')}`
@@ -334,12 +335,12 @@ describe('NativeChatStructuredSession delivery', () => {
 
       renderSession(sessionId)
 
-      await waitFor(() => expect(screen.getByText('Message delivery is unconfirmed.')).toBeTruthy())
-      expect(screen.getByRole('button', { name: /Retry/ })).toBeTruthy()
-      expect(screen.queryByText('Sending…')).toBeNull()
       await act(async () => {
         await new Promise((resolve) => setTimeout(resolve, 1500))
       })
+      expect(screen.queryByText('Message delivery is unconfirmed.')).toBeNull()
+      expect(screen.queryByRole('button', { name: /Retry/ })).toBeNull()
+      expect(screen.queryByText('Sending…')).toBeNull()
       expect(mocks.call).not.toHaveBeenCalled()
     },
     10000

src/renderer/src/components/native-chat/structured-agent-session-delivery-notices.rejected-retry.test.ts

@@ -0,0 +1,87 @@
+// Which message the host recorded and then rejected has a Retry: only one no agent ever took, on a
+// host that can queue that same message again. Every other one is drawn where it was rejected, in
+// the host's words, and sending it again is a new message.
+
+import { describe, expect, it, vi } from 'vitest'
+import { agentSessionFailureWords } from '../../../../shared/agent-session-failure-words'
+import type { SubmissionRejectionFact } from '../../../../shared/agent-session-failure'
+import { agentJournalSubmissionKey } from '../../../../shared/agent-session-journal-item-key'
+import type {
+  AgentJournalRenderItem,
+  AgentJournalSubmission
+} from '../../../../shared/agent-session-journal-types'
+import { structuredAgentSessionDeliveryNotices } from './structured-agent-session-delivery-notices'
+import { projectStructuredAgentSessionMessages } from './structured-agent-session-message-projection'
+
+const ID = 'op-message'
+const KEY = agentJournalSubmissionKey(ID)
+const ITEM: AgentJournalRenderItem = {
+  itemId: KEY,
+  revision: 0,
+  sequence: 5,
+  observedAt: 5,
+  body: { kind: 'message', role: 'user', blocks: [{ type: 'text', text: 'hello' }] }
+}
+
+function rejected(
+  fact: SubmissionRejectionFact,
+  patch: Partial<AgentJournalSubmission> = {}
+): AgentJournalSubmission {
+  return {
+    clientMessageId: ID,
+    fence: 1,
+    payloadFingerprint: 'fingerprint',
+    dispatchState: 'rejected',
+    providerItemId: null,
+    submittedAt: 4,
+    resolvedAt: 7,
+    handoverRecorded: true,
+    ...agentSessionFailureWords(fact, { surface: 'rejection', agentName: 'Codex' }),
+    ...patch
+  }
+}
+
+function noticeOn(submission: AgentJournalSubmission, retriesInPlace: boolean) {
+  const retry = vi.fn()
+  const notice = structuredAgentSessionDeliveryNotices(
+    [],
+    'Codex',
+    retry,
+    [submission],
+    [],
+    new Set(),
+    [],
+    new Set(),
+    [],
+    retriesInPlace
+  ).get(KEY)
+  return { notice, retry }
+}
+
+describe('a message the host recorded and then rejected', () => {
+  it("is drawn in place with the host's words and no Retry, even on a host that retries in place", () => {
+    // Handed to the agent, which never got it: the agent may have seen it, so no Retry.
+    const undelivered = rejected({ kind: 'notDelivered' }, { handedOverAt: 5 })
+
+    expect(projectStructuredAgentSessionMessages([ITEM], [], [undelivered], [])).toEqual([
+      expect.objectContaining({ id: KEY, unsent: true })
+    ])
+    const { notice } = noticeOn(undelivered, true)
+    expect(notice).toEqual({ text: 'This message was not delivered. Send it again to continue.' })
+  })
+})
+
+describe('a failed start no agent took', () => {
+  const failedStart = rejected({ kind: 'providerStartFailed' })
+
+  it('gets Retry in place when the host advertises retry-message', () => {
+    const { notice, retry } = noticeOn(failedStart, true)
+    expect(notice?.text).toBe('Codex stopped before it finished starting.')
+    notice?.onRetry?.()
+    expect(retry).toHaveBeenCalledExactlyOnceWith(ID)
+  })
+
+  it('gets no Retry on a host that does not advertise retry-message', () => {
+    expect(noticeOn(failedStart, false).notice).not.toHaveProperty('onRetry')
+  })
+})