fix(native-chat): settle a structured send the provider proves it received after the ack window (#19140)

* fix(native-chat): settle a structured send the provider proves it received after the ack window A send waits a bounded window for the provider to echo the message it was given. On timeout the dispatch resolves `unknown`. The echo that arrives later IS matched — `recoverLateIdentity` uses it to repair the session's turn identity — but nothing tells the journal, and `unknown` is terminal there. The submission stays unknown for the life of the session. Two consequences, both reachable on any ordinary session: - The composer renders "Message delivery is unconfirmed." with a Retry, forever, for a message that was delivered and answered. - Retry redispatches, because the host only replays a recorded outcome unless `retryUnknown` is set, which that button is the only thing that sets. So the banner is a duplicate delivery armed and waiting for a click — and a user who believes the banner and resends is doing exactly that by hand. Every send made while a turn is already running takes this path: the provider does not echo a queued message until the running turn ends, which is far past the 10s ack window. Sends made while idle are unaffected, which is why this reads as intermittent. Carry the `clientMessageId` on the dispatch waiter and settle the journal submission `accepted` when the late echo proves delivery. Deliberately unfenced against the dispatch sequence: that fence decides which turn owns the identity, while delivery is settled either way. Already-terminal rows are untouched. * fix(native-chat): persist late dispatch receipts before session close --------- Co-authored-by: Merge Sim <sim@local>
2026-09-23 16:02:24 +00:00 · 2026-09-06 16:41:09 -07:00
parent ade9718557
commit ad4dc353f3
9 changed files with 366 additions and 11 deletions
@@ -87,6 +87,60 @@ describe('Claude structured dispatch image limits', () => {
    expect(session.activeTurnSequence).toBe(session.dispatchSequence)
  })

+  it('settles the send a timed-out replay proves was delivered', async () => {
+    const session = sessionFor()
+    const settled = vi.fn()
+    const dispatched = dispatchClaudeTurn(
+      session,
+      { clientMessageId: 'client-1', body: userMessage([{ type: 'text', text: 'one' }]) },
+      500
+    )
+    await vi.waitFor(() => expect(session.dispatchWaiters).toHaveLength(1))
+    const sentUuid = (session.dispatchWaiters[0] as { sentUuid?: string }).sentUuid
+    await expect(dispatched).resolves.toMatchObject({ state: 'unknown' })
+
+    resolveClaudeReplayWaiter(session, userReplayFrame(sentUuid!, 'one'), settled)
+    expect(settled).toHaveBeenCalledWith({
+      clientMessageId: 'client-1',
+      providerIdentity: { provider: 'claude', sessionId: 'provider-session', uuid: sentUuid }
+    })
+  })
+
+  it('settles a superseded dispatch even though it no longer owns the turn identity', async () => {
+    const session = sessionFor()
+    const settled = vi.fn()
+    const first = dispatchClaudeTurn(
+      session,
+      { clientMessageId: 'client-1', body: userMessage([{ type: 'text', text: 'one' }]) },
+      500
+    )
+    await vi.waitFor(() => expect(session.dispatchWaiters).toHaveLength(1))
+    const firstUuid = (session.dispatchWaiters[0] as { sentUuid?: string }).sentUuid
+    await expect(first).resolves.toMatchObject({ state: 'unknown' })
+
+    const second = dispatchClaudeTurn(
+      session,
+      { clientMessageId: 'client-2', body: userMessage([{ type: 'text', text: 'two' }]) },
+      100
+    )
+    await vi.waitFor(() => expect(session.dispatchWaiters).toHaveLength(1))
+    const secondUuid = (session.dispatchWaiters[0] as { sentUuid?: string }).sentUuid
+
+    // The stale replay must not claim the active turn, but the message it names
+    // did land, so the send it came from is delivered and must stop reading as
+    // unconfirmed — that banner is what makes a user resend a duplicate.
+    expect(resolveClaudeReplayWaiter(session, userReplayFrame(firstUuid!, 'one'), settled)).toBe(
+      false
+    )
+    expect(settled).toHaveBeenCalledWith({
+      clientMessageId: 'client-1',
+      providerIdentity: { provider: 'claude', sessionId: 'provider-session', uuid: firstUuid }
+    })
+    resolveClaudeReplayWaiter(session, userReplayFrame(secondUuid!, 'two'), settled)
+    await expect(second).resolves.toMatchObject({ state: 'accepted' })
+    expect(settled).toHaveBeenCalledTimes(1)
+  })
+
  it('never lets a late replay for dispatch A resolve dispatch B', async () => {
    const session = sessionFor()
    const first = dispatchClaudeTurn(
@@ -1,5 +1,8 @@
 import { randomUUID } from 'node:crypto'
-import type { AgentJournalMessageItem } from '../../shared/agent-session-journal-types'
+import type {
+  AgentJournalItemIdentity,
+  AgentJournalMessageItem
+} from '../../shared/agent-session-journal-types'
 import type { AgentSessionDispatchOutcome } from '../native-chat/agent-session-wire/structured-agent-session-adapter'
 import {
  claudeHasReplayContent,
@@ -14,9 +17,16 @@ import {

 const MAX_RETIRED_DISPATCH_WAITERS = 64

+/** A dispatch whose ack window expired, proven delivered by this replay. */
+export type ClaudeLateDispatchSettlement = (input: {
+  clientMessageId: string
+  providerIdentity: AgentJournalItemIdentity
+}) => void
+
 export function resolveClaudeReplayWaiter(
  session: ClaudeSession,
-  message: Record<string, unknown>
+  message: Record<string, unknown>,
+  onSettledLate?: ClaudeLateDispatchSettlement
 ): boolean {
  const envelope = readClaudeMessageEnvelope(message)
  const isUserReplay =
@@ -52,7 +62,7 @@ export function resolveClaudeReplayWaiter(
    )
    if (retired) {
      forgetRetiredWaiter(session, retired)
-      return recoverLateIdentity(session, retired, uuid, isUserReplay)
+      return recoverLateIdentity(session, retired, uuid, isUserReplay, onSettledLate)
    }
    return false
  }
@@ -65,7 +75,7 @@ export function resolveClaudeReplayWaiter(
  const retired = session.retiredDispatchWaiters.find((candidate) => candidate.sentUuid === uuid)
  if (retired) {
    forgetRetiredWaiter(session, retired)
-    return recoverLateIdentity(session, retired, uuid, isUserReplay)
+    return recoverLateIdentity(session, retired, uuid, isUserReplay, onSettledLate)
  }

  if (isUserReplay) {
@@ -89,7 +99,7 @@ export function resolveClaudeReplayWaiter(
      if (lateCompatible.length === 1) {
        const [candidate] = lateCompatible
        forgetRetiredWaiter(session, candidate!)
-        return recoverLateIdentity(session, candidate!, uuid, true)
+        return recoverLateIdentity(session, candidate!, uuid, true, onSettledLate)
      }
    }
    return false
@@ -138,11 +148,19 @@ function recoverLateIdentity(
  session: ClaudeSession,
  waiter: ClaudeDispatchWaiter,
  uuid: string,
-  isUserReplay: boolean
+  isUserReplay: boolean,
+  onSettledLate?: ClaudeLateDispatchSettlement
 ): boolean {
  if (!isUserReplay && !waiter.acceptsResult) {
    return false
  }
+  // The provider acted on this dispatch, so the send it came from is delivered.
+  // Unfenced on purpose: the dispatch-sequence check below only decides which
+  // turn owns the identity, while delivery is settled for good either way.
+  onSettledLate?.({
+    clientMessageId: waiter.clientMessageId,
+    providerIdentity: { provider: 'claude', sessionId: session.providerSessionId, uuid }
+  })
  if (waiter.dispatchSequence === session.dispatchSequence) {
    session.activeTurnId = uuid
    session.activeTurnSequence = waiter.dispatchSequence
@@ -155,12 +173,14 @@ function waitForReplay(
  timeoutMs: number,
  acceptsResult: boolean,
  sentUuid: string,
-  replayContentKey: string
+  replayContentKey: string,
+  clientMessageId: string
 ): { waiter: ClaudeDispatchWaiter; promise: Promise<string | null> } {
  let waiter!: ClaudeDispatchWaiter
  const promise = new Promise<string | null>((resolve) => {
    waiter = {
      acceptsResult,
+      clientMessageId,
      sentUuid,
      dispatchSequence: session.dispatchSequence,
      replayContentKey,
@@ -220,7 +240,8 @@ export async function dispatchClaudeTurn(
    timeoutMs,
    acceptsResult,
    sentUuid,
-    claudeDispatchContentKey(content)
+    claudeDispatchContentKey(content),
+    input.clientMessageId
  )
  const replayed = replay.promise
  try {
@@ -109,7 +109,11 @@ export async function acquireClaudeSession({
    if (liveSession) {
      liveSession.leafUuid = observedLeafUuid
    }
-    const startsTurn = liveSession ? resolveClaudeReplayWaiter(liveSession, message) : false
+    const startsTurn = liveSession
+      ? resolveClaudeReplayWaiter(liveSession, message, (settlement) =>
+          deps.onDispatchSettledLate?.({ sessionId, ...settlement })
+        )
+      : false
    callbacks.deliver(attempt, sessionId, () =>
      callbacks.emit(liveSession, input.events, {
        type: 'message',
@@ -1,4 +1,7 @@
-import type { AgentSessionJournalIdentity } from '../../shared/agent-session-journal-types'
+import type {
+  AgentJournalItemIdentity,
+  AgentSessionJournalIdentity
+} from '../../shared/agent-session-journal-types'
 import type { StructuredAgentSessionEventSink } from '../native-chat/agent-session-wire/structured-agent-session-event-sink'
 import type {
  ClaudeStreamJsonConnection,
@@ -56,6 +59,12 @@ export type ClaudeStructuredSessionAdapterDeps = {
    identity: AgentSessionJournalIdentity
  }) => Promise<ClaudeStructuredLaunch>
  onEvent?: (event: ClaudeStructuredSessionEvent) => void
+  /** A dispatch whose ack timed out, proven delivered by a later provider replay. */
+  onDispatchSettledLate?: (input: {
+    sessionId: string
+    clientMessageId: string
+    providerIdentity: AgentJournalItemIdentity
+  }) => void
  onBackgroundTasksChanged?: (
    sessionId: string,
    state: AgentSessionBackgroundTaskState | null
@@ -87,6 +96,9 @@ export type ClaudeDispatchWaiter = {
  resolve: (uuid: string | null) => void
  timer: ReturnType<typeof setTimeout>
  acceptsResult: boolean
+  /** Carried so a replay that lands after the ack window can settle the journal
+   *  submission this dispatch came from, not just the in-memory turn identity. */
+  clientMessageId: string
  /** Client uuid echoed by Claude so a replay is tied to its own dispatch. */
  sentUuid: string
  /** Sequence used to fence a late identity from a newer dispatch. */
@@ -5,7 +5,10 @@
 // they share one path here rather than five copies in the host. The host keeps attach, holds and
 // teardown; this is the surface that assumes those already happened.

-import type { AgentJournalMessageItem } from '../../../shared/agent-session-journal-types'
+import type {
+  AgentJournalItemIdentity,
+  AgentJournalMessageItem
+} from '../../../shared/agent-session-journal-types'
 import type {
  AgentSessionCancelResult,
  AgentSessionMutationEnvelope,
@@ -118,3 +121,26 @@ export function readStructuredAgentSessionOptions(
    return context.deps.adapter.readOptions({ sessionId, fence: session.fence })
  })
 }
+
+/** Settle provider-proven delivery independently of an in-flight client mutation. */
+export async function settleStructuredAgentSessionLateDispatch(
+  context: StructuredAgentSessionMutationContext,
+  input: {
+    sessionId: string
+    clientMessageId: string
+    providerIdentity: AgentJournalItemIdentity
+  }
+): Promise<void> {
+  const session = context.sessions.get(input.sessionId)
+  if (!session) {
+    return
+  }
+  // The journal queue drains before close; the host queue would defer this past teardown.
+  await session.journal.resolveDispatch({
+    clientMessageId: input.clientMessageId,
+    state: 'accepted',
+    providerIdentity: input.providerIdentity,
+    fence: session.fence
+  })
+  context.publish(input.sessionId, session.journal)
+}
@@ -38,6 +38,7 @@ import {
  respondToStructuredAgentSessionPrompt,
  sendStructuredAgentSessionTurn,
  setStructuredAgentSessionOption,
+  settleStructuredAgentSessionLateDispatch,
  type StructuredAgentSessionMutationContext
 } from './structured-agent-session-host-mutations'
 import { tearDownStructuredAgentSessionHost } from './structured-agent-session-host-teardown'
@@ -331,6 +332,9 @@ export class StructuredAgentSessionHost {
  subscribe = (input: AgentSessionSubscribeInput): (() => void) =>
    this.backgroundTasks.subscribe(input)

+  settleLateDispatch = (input: Parameters<typeof settleStructuredAgentSessionLateDispatch>[1]) =>
+    settleStructuredAgentSessionLateDispatch(this.mutationContext(), input)
+
  publishBackgroundTaskState: StructuredAgentSessionBackgroundTaskChannel['publish'] = (
    sessionId,
    state
@@ -0,0 +1,224 @@
+import { mkdtemp, rm } from 'node:fs/promises'
+import { tmpdir } from 'node:os'
+import { join } from 'node:path'
+import { afterEach, beforeEach, describe, expect, it, vi, type Mock } from 'vitest'
+import { computeAgentSessionPayloadFingerprint } from '../../../shared/agent-session-mutation-envelope'
+import type {
+  AgentSessionMutationEnvelope,
+  AgentSessionSubscribeEvent
+} from '../../../shared/agent-session-wire'
+import { AgentSessionRecordStore } from '../../runtime/agent-session-record-store'
+import type {
+  AgentSessionDispatchOutcome,
+  StructuredAgentSessionAdapter
+} from './structured-agent-session-adapter'
+import { StructuredAgentSessionHost } from './structured-agent-session-host'
+import {
+  HOST_TEST_NOW as NOW,
+  HOST_TEST_SESSION as SESSION,
+  HOST_TEST_THREAD as THREAD,
+  hostTestAttachParams,
+  hostTestMessage,
+  hostTestOperationId,
+  resetHostTestOperationIds
+} from './structured-agent-session-host-test-data'
+
+const CALLER = { callerKey: 'client-1' }
+
+let root: string
+let store: AgentSessionRecordStore
+let host: StructuredAgentSessionHost
+let dispatch: Mock<StructuredAgentSessionAdapter['dispatch']>
+let closeSession: Mock<NonNullable<StructuredAgentSessionAdapter['closeSession']>>
+
+function accepted(): AgentSessionDispatchOutcome {
+  return {
+    state: 'accepted',
+    providerIdentity: { provider: 'codex', threadId: THREAD, turnId: 'turn-1', ordinal: 1 }
+  }
+}
+
+function sendParams(text: string): {
+  envelope: AgentSessionMutationEnvelope
+  body: ReturnType<typeof hostTestMessage>
+} {
+  const body = hostTestMessage(text)
+  return {
+    envelope: {
+      sessionId: SESSION,
+      clientOperationId: hostTestOperationId(),
+      expectedRuntimeFence: store.getRecord(SESSION)?.lease.runtimeFence ?? 1,
+      payloadFingerprint: computeAgentSessionPayloadFingerprint({
+        method: 'agentSession.send',
+        sessionId: SESSION,
+        fields: { body }
+      })
+    },
+    body
+  }
+}
+
+function submissions(): unknown {
+  const state = host.history({ sessionId: SESSION, direction: 'tail' })
+  return state.ok ? state.page.submissions : null
+}
+
+beforeEach(async () => {
+  root = await mkdtemp(join(tmpdir(), 'orca-wire-late-settle-'))
+  resetHostTestOperationIds()
+  dispatch = vi.fn(async () => accepted())
+  closeSession = vi.fn(async () => true)
+  store = await AgentSessionRecordStore.open({ directory: join(root, 'store'), hostId: 'local' })
+  host = new StructuredAgentSessionHost({
+    store,
+    adapter: {
+      acquire: vi.fn(async ({ fence }) => ({
+        process: {
+          hostId: 'local',
+          pid: 4242,
+          processStartTimeMs: 1_700_000_000_000,
+          spawnToken: store.getRecord(SESSION)?.lease.reservedSpawnToken ?? 'spawn-a'
+        },
+        link: {
+          linkId: `link-${fence}`,
+          handle: { provider: 'codex' as const, threadId: THREAD },
+          origin: 'created' as const,
+          mintedAtFence: fence,
+          observedAt: NOW
+        }
+      })),
+      releaseAcquisition: vi.fn(async () => true),
+      dispatch,
+      closeSession,
+      cancelTurn: vi.fn(async () => ({ cancelled: true })),
+      answerPrompt: vi.fn(async () => undefined),
+      setOption: vi.fn(async () => undefined)
+    },
+    journalRoot: root,
+    claimKeyId: 'key-1',
+    mintSpawnToken: () => 'spawn-a',
+    now: () => NOW
+  })
+  expect((await host.attach(CALLER, hostTestAttachParams(null))).ok).toBe(true)
+})
+
+afterEach(async () => {
+  await host.flushAllStreamedEvents()
+  await host.close(SESSION)
+  await rm(root, { recursive: true, force: true })
+})
+
+describe('settling a send the provider proves it received after the ack window', () => {
+  it('publishes acceptance during a pending send and never reopens it for retry', async () => {
+    let finishDispatch!: (outcome: AgentSessionDispatchOutcome) => void
+    dispatch.mockImplementationOnce(
+      () =>
+        new Promise((resolve) => {
+          finishDispatch = resolve
+        })
+    )
+    const events: AgentSessionSubscribeEvent[] = []
+    const unsubscribe = host.subscribe({
+      id: 'late-receipt',
+      sessionId: SESSION,
+      emit: (event) => events.push(event)
+    })
+    const params = sendParams('echo before send completes')
+    const pending = host.send(CALLER, params)
+    await vi.waitFor(() => expect(dispatch).toHaveBeenCalledTimes(1))
+    try {
+      await host.settleLateDispatch({
+        sessionId: SESSION,
+        clientMessageId: params.envelope.clientOperationId,
+        providerIdentity: { provider: 'claude', sessionId: THREAD, uuid: 'early-echo' }
+      })
+      expect(events.at(-1)).toMatchObject({
+        type: 'batch',
+        batch: {
+          submissions: [
+            { clientMessageId: params.envelope.clientOperationId, dispatchState: 'accepted' }
+          ]
+        }
+      })
+    } finally {
+      finishDispatch({ state: 'unknown', reason: 'ack timeout' })
+      unsubscribe()
+    }
+    await expect(pending).resolves.toMatchObject({
+      ok: true,
+      value: { submission: { dispatchState: 'accepted' } }
+    })
+    await expect(host.send(CALLER, { ...params, retryUnknown: true })).resolves.toMatchObject({
+      ok: true,
+      value: { submission: { dispatchState: 'accepted' } }
+    })
+    expect(dispatch).toHaveBeenCalledTimes(1)
+  })
+
+  it('persists an echo received while the provider is closing', async () => {
+    dispatch.mockResolvedValueOnce({ state: 'unknown', reason: 'ack timeout' })
+    const params = sendParams('received just before shutdown')
+    await host.send(CALLER, params)
+    let settlement: Promise<void> | undefined
+    closeSession.mockImplementationOnce(async () => {
+      settlement = host.settleLateDispatch({
+        sessionId: SESSION,
+        clientMessageId: params.envelope.clientOperationId,
+        providerIdentity: { provider: 'claude', sessionId: THREAD, uuid: 'closing-echo' }
+      })
+      void settlement.catch(() => undefined)
+      return true
+    })
+
+    await host.close(SESSION)
+    await expect(settlement).resolves.toBeUndefined()
+    await host.revealSession(SESSION)
+    expect(submissions()).toMatchObject([{ dispatchState: 'accepted' }])
+    expect(dispatch).toHaveBeenCalledTimes(1)
+  })
+
+  it('moves a durable unknown to accepted so nothing offers to send it again', async () => {
+    dispatch.mockRejectedValueOnce(new Error('socket closed'))
+    const params = sendParams('sent while a turn was running')
+    const first = await host.send(CALLER, params)
+    expect(first).toMatchObject({ ok: true, value: { submission: { dispatchState: 'unknown' } } })
+
+    await host.settleLateDispatch({
+      sessionId: SESSION,
+      clientMessageId: params.envelope.clientOperationId,
+      providerIdentity: { provider: 'claude', sessionId: THREAD, uuid: 'late-uuid' }
+    })
+
+    expect(submissions()).toMatchObject([
+      { clientMessageId: params.envelope.clientOperationId, dispatchState: 'accepted' }
+    ])
+    // The point of the fix: the client stops rendering Retry, and Retry is what
+    // was delivering the message to the agent a second time.
+    expect(dispatch).toHaveBeenCalledTimes(1)
+  })
+
+  it('leaves an already accepted send alone', async () => {
+    const params = sendParams('ordinary send')
+    await host.send(CALLER, params)
+
+    await host.settleLateDispatch({
+      sessionId: SESSION,
+      clientMessageId: params.envelope.clientOperationId,
+      providerIdentity: { provider: 'claude', sessionId: THREAD, uuid: 'a-different-uuid' }
+    })
+
+    expect(submissions()).toMatchObject([
+      { clientMessageId: params.envelope.clientOperationId, dispatchState: 'accepted' }
+    ])
+  })
+
+  it('ignores a session this host is not holding', async () => {
+    await expect(
+      host.settleLateDispatch({
+        sessionId: 'session-that-is-not-attached',
+        clientMessageId: 'whatever',
+        providerIdentity: { provider: 'claude', sessionId: THREAD, uuid: 'x' }
+      })
+    ).resolves.toBeUndefined()
+  })
+})
@@ -260,6 +260,14 @@ async function install(deps: StructuredAgentSessionRuntimeDeps): Promise<Install
      },
      onBackgroundTasksChanged: (sessionId, state) =>
        host?.publishBackgroundTaskState(sessionId, state),
+      onDispatchSettledLate: (settlement) => {
+        void host?.settleLateDispatch(settlement).catch((error) =>
+          deps.onError?.({
+            scope: `structured-agent-session-late-settlement:${settlement.sessionId}`,
+            error
+          })
+        )
+      },
      ...(deps.openClaudeConnection ? { openClaudeConnection: deps.openClaudeConnection } : {}),
      ...(deps.readProcessStartTime ? { readProcessStartTime: deps.readProcessStartTime } : {})
    })
@@ -34,6 +34,7 @@ export type StructuredClaudeRuntimeAdapterDeps = {
    sessionId: string,
    state: AgentSessionBackgroundTaskState | null
  ) => void
+  onDispatchSettledLate?: ClaudeStructuredSessionAdapterDeps['onDispatchSettledLate']
 }

 export function createStructuredClaudeRuntimeAdapter(
@@ -100,6 +101,7 @@ export function createStructuredClaudeRuntimeAdapter(
    ...(deps.onBackgroundTasksChanged
      ? { onBackgroundTasksChanged: deps.onBackgroundTasksChanged }
      : {}),
+    ...(deps.onDispatchSettledLate ? { onDispatchSettledLate: deps.onDispatchSettledLate } : {}),
    ...(deps.openClaudeConnection ? { openConnection: deps.openClaudeConnection } : {}),
    ...(deps.readProcessStartTime ? { readProcessStartTime: deps.readProcessStartTime } : {})
  })