fix(native-chat): reconcile queued Codex send echoes

mobile/src/session/mobile-structured-session-timeline.test.ts

@@ -40,6 +40,28 @@ const OUTBOX: MobileStructuredOutboxEntry = {
 }
 
 describe('mobile structured session timeline', () => {
+  it.each([5, 10])('renders %i authoritative mobile sends exactly once', (sendCount) => {
+    const items = Array.from(
+      { length: sendCount },
+      (_, index): AgentJournalRenderItem => ({
+        itemId: `orca:client-${index}`,
+        revision: 1,
+        sequence: index + 1,
+        observedAt: index + 1,
+        body: {
+          kind: 'message',
+          role: 'user',
+          blocks: [{ type: 'text', text: `RAPID_${index + 1}` }]
+        }
+      })
+    )
+
+    const rows = buildMobileStructuredTimeline(items, [])
+    expect(
+      rows.filter((row) => row.kind === 'message' && row.message.role === 'user')
+    ).toHaveLength(sendCount)
+  })
+
   it('keeps pending prompts as cards and unknown sends as their original bubble', () => {
     const rows = buildMobileStructuredTimeline([APPROVAL], [OUTBOX])
 

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

@@ -1,6 +1,7 @@
 import { describe, expect, it } from 'vitest'
 import { agentJournalSubmissionKey } from '../../../shared/agent-session-journal-item-key'
 import type { AgentJournalMessageItem } from '../../../shared/agent-session-journal-types'
+import { structuredAgentSessionPayloadFingerprint } from '../../../shared/structured-agent-session-mutation'
 import {
   applyJournalRow,
   createJournalReducerState,
@@ -16,6 +17,18 @@ function text(value: string): AgentJournalMessageItem {
   return { kind: 'message', role: 'assistant', blocks: [{ type: 'text', text: value }] }
 }
 
+function userText(value: string): AgentJournalMessageItem {
+  return { kind: 'message', role: 'user', blocks: [{ type: 'text', text: value }] }
+}
+
+function sendFingerprint(body: AgentJournalMessageItem): string {
+  return structuredAgentSessionPayloadFingerprint({
+    method: 'agentSession.send',
+    sessionId: 'session-1',
+    fields: { body }
+  })
+}
+
 function base(seq: number): { v: number; epoch: string; seq: number; fence: number; ts: number } {
   return { v: 1, epoch: EPOCH, seq, fence: 1, ts: 1_000 + seq }
 }
@@ -178,6 +191,91 @@ describe('submission and dispatch state machine', () => {
     expect(items[0]?.revision).toBe(1)
   })
 
+  it('adopts a provider echo that arrives before dispatch settles', () => {
+    const body = userText('early echo')
+    const state = fold([
+      {
+        kind: 'submission',
+        clientMessageId: 'early-client',
+        payloadFingerprint: sendFingerprint(body),
+        providerHandle: { kind: 'codex', threadId: 'thread-1' },
+        body,
+        ...base(1)
+      },
+      {
+        kind: 'item',
+        itemId: 'codex:thread-1:root-turn:2',
+        revision: 1,
+        body,
+        ...base(2)
+      },
+      {
+        kind: 'dispatch',
+        clientMessageId: 'early-client',
+        state: 'accepted',
+        providerItemId: 'codex:thread-1:predicted-turn:0',
+        reason: null,
+        ...base(3)
+      }
+    ])
+
+    expect(renderJournalState(state).items).toMatchObject([
+      { itemId: agentJournalSubmissionKey('early-client'), revision: 1, sequence: 1 }
+    ])
+  })
+
+  it.each([5, 10])(
+    'reconciles %i rapid sends across an interleaved cancel when Codex reuses the root turn',
+    (count) => {
+      const rows: JournalRow[] = []
+      for (let index = 0; index < count; index += 1) {
+        const body = userText(`RAPID_${index + 1}`)
+        rows.push(
+          {
+            kind: 'submission',
+            clientMessageId: `client-${index}`,
+            payloadFingerprint: sendFingerprint(body),
+            providerHandle: { kind: 'codex', threadId: 'thread-1' },
+            body,
+            ...base(rows.length + 1)
+          },
+          {
+            kind: 'dispatch',
+            clientMessageId: `client-${index}`,
+            state: 'accepted',
+            providerItemId: `codex:thread-1:predicted-turn-${index}:0`,
+            reason: null,
+            ...base(rows.length + 2)
+          }
+        )
+      }
+      rows.push({
+        kind: 'item',
+        itemId: 'orca:cancel-between-sends',
+        revision: 1,
+        body: { kind: 'status', text: 'Cancelled an earlier turn.' },
+        ...base(rows.length + 1)
+      })
+      for (let index = 0; index < count; index += 1) {
+        rows.push({
+          kind: 'item',
+          itemId: `codex:thread-1:root-turn:${index}`,
+          revision: 1,
+          body: userText(`RAPID_${index + 1}`),
+          ...base(rows.length + 1)
+        })
+      }
+
+      const messages = renderJournalState(fold(rows)).items.filter(
+        (item) => item.body.kind === 'message' && item.body.role === 'user'
+      )
+      expect(messages).toHaveLength(count)
+      expect(messages.map((item) => item.itemId)).toEqual(
+        Array.from({ length: count }, (_, index) => agentJournalSubmissionKey(`client-${index}`))
+      )
+    }
+  )
+
   it('treats rejected as terminal', () => {
     const state = fold([
       submission,

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

@@ -13,7 +13,11 @@ import type {
   AgentJournalSnapshot,
   AgentJournalSubmission
 } from '../../../shared/agent-session-journal-types'
-import { agentJournalSubmissionKey } from '../../../shared/agent-session-journal-item-key'
+import {
+  agentJournalSubmissionKey,
+  parseAgentJournalItemKey
+} from '../../../shared/agent-session-journal-item-key'
+import { structuredAgentSessionPayloadFingerprint } from '../../../shared/structured-agent-session-mutation'
 import type { JournalRow } from './journal-row-schema'
 
 export type JournalReducerState = {
@@ -55,7 +59,7 @@ export function applyJournalRow(state: JournalReducerState, row: JournalRow): vo
     return
   }
   if (row.kind === 'item') {
-    const itemId = resolveItemId(state, row.itemId)
+    const itemId = resolveJournalItemId(state, row.itemId, row.body)
     upsertItem(state, itemId, row.revision, {
       itemId,
       revision: row.revision,
@@ -77,6 +81,47 @@ export function applyJournalRow(state: JournalReducerState, row: JournalRow): vo
   applyDispatch(state, row)
 }
 
+export function resolveJournalItemId(
+  state: JournalReducerState,
+  itemId: string,
+  body?: AgentJournalRenderItem['body']
+): string {
+  const aliased = state.aliases.get(itemId)
+  if (aliased) {
+    return aliased
+  }
+  const identity = parseAgentJournalItemKey(itemId)
+  if (
+    !body ||
+    body.kind !== 'message' ||
+    body.role !== 'user' ||
+    !identity ||
+    identity.provider === 'orca'
+  ) {
+    return itemId
+  }
+  const fingerprint = structuredAgentSessionPayloadFingerprint({
+    method: 'agentSession.send',
+    sessionId: state.sessionId,
+    fields: { body }
+  })
+  // Exact payload plus queue order preserves repeated identical sends one-for-one.
+  const submission = [...state.submissions.values()]
+    .sort((left, right) => left.submittedAt - right.submittedAt)
+    .find((candidate) => {
+      if (candidate.dispatchState === 'rejected' || candidate.payloadFingerprint !== fingerprint) {
+        return false
+      }
+      return state.items.get(agentJournalSubmissionKey(candidate.clientMessageId))?.revision === 0
+    })
+  if (!submission) {
+    return itemId
+  }
+  const submissionId = agentJournalSubmissionKey(submission.clientMessageId)
+  state.aliases.set(itemId, submissionId)
+  return submissionId
+}
+
 function resolveItemId(state: JournalReducerState, itemId: string): string {
   return state.aliases.get(itemId) ?? itemId
 }

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

@@ -0,0 +1,37 @@
+import type {
+  AgentJournalCursor,
+  AgentJournalItemIdentity,
+  AgentJournalResetReason,
+  AgentSessionJournalIdentity
+} from '../../../shared/agent-session-journal-types'
+import type { JournalCompactionPolicy } from './journal-compaction'
+import type { JournalPayloadLimits } from './journal-payload-bounds'
+import type { JournalRow } from './journal-row-schema'
+
+export type AgentSessionJournalOptions = {
+  identity: AgentSessionJournalIdentity
+  journalDir: string
+  limits?: JournalPayloadLimits
+  compaction?: JournalCompactionPolicy
+  now?: () => number
+  mintEpoch?: () => string
+}
+
+export type JournalReadSince =
+  | { ok: true; rows: JournalRow[]; cursor: AgentJournalCursor }
+  | { ok: false; reset: AgentJournalResetReason }
+
+export type ResolveDispatchInput = {
+  clientMessageId: string
+  fence: number
+  recovered?: true
+} & (
+  | { state: 'accepted'; providerIdentity: AgentJournalItemIdentity }
+  | { state: 'rejected' | 'unknown'; reason?: string | null }
+)
+
+export type JournalAppendResult = {
+  cursor: AgentJournalCursor
+  itemId: string
+  revision: number
+}

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

@@ -12,7 +12,6 @@ import type {
   AgentJournalItemBody,
   AgentJournalItemIdentity,
   AgentJournalMessageItem,
-  AgentJournalResetReason,
   AgentJournalSnapshot,
   AgentJournalSubmission,
   AgentSessionJournalIdentity
@@ -24,13 +23,20 @@ import { resolveJournalResume, type JournalResume } from './journal-cursor'
 import { publishNewEpoch } from './journal-epoch-rollover'
 import { appendJournalRows, ensureJournalDir } from './journal-log-file'
 import { loadJournal } from './journal-open'
-import { DEFAULT_JOURNAL_PAYLOAD_LIMITS, type JournalPayloadLimits } from './journal-payload-bounds'
+import { DEFAULT_JOURNAL_PAYLOAD_LIMITS } from './journal-payload-bounds'
 import {
   applyJournalRow,
   createJournalReducerState,
   renderJournalState,
+  resolveJournalItemId,
   type JournalReducerState
 } from './journal-reducer'
+import type {
+  AgentSessionJournalOptions,
+  JournalAppendResult,
+  JournalReadSince,
+  ResolveDispatchInput
+} from './journal-store-contracts'
 import {
   journalRowByteLength,
   type AgentJournalEpochReason,
@@ -44,35 +50,6 @@ import {
 
 export { AgentSessionJournalError } from './journal-write-guards'
 
-export type AgentSessionJournalOptions = {
-  identity: AgentSessionJournalIdentity
-  journalDir: string
-  limits?: JournalPayloadLimits
-  compaction?: JournalCompactionPolicy
-  now?: () => number
-  mintEpoch?: () => string
-}
-
-export type JournalReadSince =
-  | { ok: true; rows: JournalRow[]; cursor: AgentJournalCursor }
-  | { ok: false; reset: AgentJournalResetReason }
-
-export type ResolveDispatchInput = {
-  clientMessageId: string
-  fence: number
-  /** Set when crash reconciliation, not the live dispatch, settled this. */
-  recovered?: true
-} & (
-  | { state: 'accepted'; providerIdentity: AgentJournalItemIdentity }
-  | { state: 'rejected' | 'unknown'; reason?: string | null }
-)
-
-export type JournalAppendResult = {
-  cursor: AgentJournalCursor
-  itemId: string
-  revision: number
-}
-
 export async function openAgentSessionJournal(
   options: AgentSessionJournalOptions
 ): Promise<AgentSessionJournal> {
@@ -169,6 +146,9 @@ export class AgentSessionJournal {
     return this.state.receipts.get(clientMessageId) ?? null
   }
 
+  canonicalItemId(itemId: string): string {
+    return resolveJournalItemId(this.state, itemId)
+  }
   readSince(cursor: AgentJournalCursor): JournalReadSince {
     const resume = this.resume(cursor)
     if (!resume.ok) {

src/main/native-chat/agent-session-wire/agent-session-history-page.test.ts

@@ -6,9 +6,11 @@ import { agentJournalSubmissionKey } from '../../../shared/agent-session-journal
 import type {
   AgentJournalItemBody,
   AgentJournalItemIdentity,
+  AgentJournalMessageItem,
   AgentSessionJournalIdentity
 } from '../../../shared/agent-session-journal-types'
 import { AGENT_SESSION_HISTORY_MAX_LIMIT } from '../../../shared/agent-session-wire'
+import { structuredAgentSessionPayloadFingerprint } from '../../../shared/structured-agent-session-mutation'
 import {
   openAgentSessionJournal,
   type AgentSessionJournal
@@ -244,4 +246,52 @@ describe('projectJournalBatch', () => {
     expect(projected.batch.removedItemIds).toHaveLength(1)
     expect(projected.batch.items).toHaveLength(0)
   })
+
+  it('publishes a mismatched provider echo under its submission slot', async () => {
+    const message: AgentJournalMessageItem = {
+      kind: 'message',
+      role: 'user',
+      blocks: [{ type: 'text', text: 'queued follow-up' }]
+    }
+    await journal.appendSubmission({
+      clientMessageId: 'client-follow-up',
+      payloadFingerprint: structuredAgentSessionPayloadFingerprint({
+        method: 'agentSession.send',
+        sessionId: IDENTITY.sessionId,
+        fields: { body: message }
+      }),
+      body: message,
+      fence: 1
+    })
+    const cursor = journal.cursor()
+    await journal.resolveDispatch({
+      clientMessageId: 'client-follow-up',
+      state: 'accepted',
+      providerIdentity: {
+        provider: 'codex',
+        threadId: 'thread-1',
+        turnId: 'predicted',
+        ordinal: 0
+      },
+      fence: 1
+    })
+    await journal.appendItem(
+      { provider: 'codex', threadId: 'thread-1', turnId: 'root-turn', ordinal: 2 },
+      message,
+      { fence: 1 }
+    )
+
+    const page = readAgentSessionHistory(journal, {
+      sessionId: IDENTITY.sessionId,
+      direction: 'after',
+      cursor
+    })
+    if (!page.ok) {
+      throw new Error(`expected a page, got reset ${page.reset}`)
+    }
+    expect(page.page.items).toMatchObject([
+      { itemId: agentJournalSubmissionKey('client-follow-up'), revision: 1 }
+    ])
+    expect(page.page.removedItemIds).toEqual([])
+  })
 })

src/main/native-chat/agent-session-wire/agent-session-history-page.ts

@@ -91,7 +91,12 @@ function readForward(
     return { ok: false, reset: since.reset, snapshot }
   }
   const rows = since.rows.slice(0, limit)
-  const projected = projectJournalBatch({ rows, snapshot, afterSequence: cursor.sequence })
+  const projected = projectJournalBatch({
+    rows,
+    snapshot,
+    afterSequence: cursor.sequence,
+    canonicalItemId: (itemId) => journal.canonicalItemId(itemId)
+  })
   if (!projected.ok) {
     return { ok: false, reset: projected.reset, snapshot }
   }

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

@@ -22,6 +22,7 @@ export function projectJournalBatch(input: {
   snapshot: AgentJournalSnapshot
   /** Sequence the subscriber has already applied. */
   afterSequence: number
+  canonicalItemId?: (itemId: string) => string
 }): JournalBatchProjection {
   const gap = findSequenceGap(
     input.rows.map((row) => row.seq),
@@ -35,7 +36,9 @@ export function projectJournalBatch(input: {
   const touchedClientMessageIds = new Set<string>()
   for (const row of input.rows) {
     if (row.kind === 'item' || row.kind === 'tombstone') {
-      touchedItemIds.add(aliases.get(row.itemId) ?? row.itemId)
+      touchedItemIds.add(
+        input.canonicalItemId?.(row.itemId) ?? aliases.get(row.itemId) ?? row.itemId
+      )
       continue
     }
     if (row.kind === 'submission' || row.kind === 'dispatch') {

src/main/native-chat/agent-session-wire/structured-agent-session-send-idempotency.test.ts

@@ -0,0 +1,81 @@
+import { mkdtemp, rm } from 'node:fs/promises'
+import { tmpdir } from 'node:os'
+import { join } from 'node:path'
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
+import type { AgentJournalMessageItem } from '../../../shared/agent-session-journal-types'
+import { structuredAgentSessionPayloadFingerprint } from '../../../shared/structured-agent-session-mutation'
+import {
+  openAgentSessionJournal,
+  type AgentSessionJournal
+} from '../agent-session-journal/journal-store'
+import type { StructuredAgentSessionAdapter } from './structured-agent-session-adapter'
+import { performSend, type AgentSessionTurnContext } from './structured-agent-session-turns'
+
+let root: string
+let journal: AgentSessionJournal
+
+beforeEach(async () => {
+  root = await mkdtemp(join(tmpdir(), 'orca-send-idempotency-'))
+  journal = await openAgentSessionJournal({
+    identity: {
+      sessionId: 'session-1',
+      workspaceId: 'workspace-1',
+      hostId: 'local',
+      agent: 'codex',
+      providerHandle: { kind: 'codex', threadId: 'thread-1' }
+    },
+    journalDir: root
+  })
+})
+
+afterEach(async () => {
+  await rm(root, { recursive: true, force: true })
+})
+
+describe('structured send idempotency', () => {
+  it('does not redispatch one send id reused across caller ledgers', async () => {
+    const body: AgentJournalMessageItem = {
+      kind: 'message',
+      role: 'user',
+      blocks: [{ type: 'text', text: 'one durable send' }]
+    }
+    const dispatch = vi.fn(async () => ({
+      state: 'accepted' as const,
+      providerIdentity: {
+        provider: 'codex' as const,
+        threadId: 'thread-1',
+        turnId: 'turn-1',
+        ordinal: 0
+      }
+    }))
+    const context: AgentSessionTurnContext = {
+      sessionId: 'session-1',
+      journal,
+      fence: 1,
+      adapter: { dispatch } as unknown as StructuredAgentSessionAdapter,
+      persistOptions: async () => undefined,
+      resolvedBy: 'caller',
+      publish: vi.fn(),
+      now: () => 1
+    }
+    const input = {
+      clientMessageId: 'shared-send-id',
+      payloadFingerprint: structuredAgentSessionPayloadFingerprint({
+        method: 'agentSession.send',
+        sessionId: 'session-1',
+        fields: { body }
+      }),
+      body
+    }
+
+    await performSend(context, input)
+    const replay = await performSend(context, input)
+
+    expect(replay).toMatchObject({
+      ok: true,
+      value: { clientMessageId: 'shared-send-id', submission: { dispatchState: 'accepted' } }
+    })
+    expect(dispatch).toHaveBeenCalledOnce()
+    expect(journal.submissions()).toHaveLength(1)
+  })
+})

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

@@ -179,7 +179,8 @@ export class AgentSessionSubscribers {
     const projected = projectJournalBatch({
       rows: since.rows,
       snapshot,
-      afterSequence: subscriber.cursor.sequence
+      afterSequence: subscriber.cursor.sequence,
+      canonicalItemId: (itemId) => journal.canonicalItemId(itemId)
     })
     if (!projected.ok) {
       this.emit(subscriber, {

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

@@ -92,6 +92,15 @@ export async function performSend(
   const existing = ctx.journal
     .submissions()
     .find((entry) => entry.clientMessageId === input.clientMessageId)
+  if (existing && existing.payloadFingerprint !== input.payloadFingerprint) {
+    return invalid(`Message id ${input.clientMessageId} was already used for another send.`)
+  }
+  if (existing && !(input.retryUnknown && existing.dispatchState === 'unknown')) {
+    return {
+      ok: true,
+      value: { clientMessageId: input.clientMessageId, submission: existing }
+    }
+  }
   if (!(input.retryUnknown && existing?.dispatchState === 'unknown')) {
     await ctx.journal.appendSubmission({ ...input, fence: ctx.fence })
     ctx.publish()

src/renderer/src/components/native-chat/structured-agent-session-message-projection.test.ts

@@ -6,8 +6,6 @@ import type {
 import { createStructuredAgentSessionOutboxEntry } from '../../../../shared/structured-agent-session-outbox'
 import { projectStructuredAgentSessionMessages } from './structured-agent-session-message-projection'
 
-const SEND_COUNT = 8
-
 function submission(index: number): AgentJournalSubmission {
   return {
     clientMessageId: `client-${index}`,
@@ -32,8 +30,8 @@ function item(index: number): AgentJournalRenderItem {
 }
 
 describe('structured agent session message projection', () => {
-  it('renders N rapid accepted sends once when journal and optimistic updates interleave', () => {
-    const outbox = Array.from({ length: SEND_COUNT }, (_, index) =>
+  it.each([5, 10])('renders %i rapid accepted desktop sends exactly once', (sendCount) => {
+    const outbox = Array.from({ length: sendCount }, (_, index) =>
       createStructuredAgentSessionOutboxEntry({
         clientMessageId: `client-${index}`,
         sessionId: 'session-1',
@@ -43,14 +41,14 @@ describe('structured agent session message projection', () => {
       })
     )
     const messages = projectStructuredAgentSessionMessages(
-      Array.from({ length: SEND_COUNT }, (_, index) => item(index)),
+      Array.from({ length: sendCount }, (_, index) => item(index)),
       outbox,
-      Array.from({ length: SEND_COUNT }, (_, index) => submission(SEND_COUNT - index - 1))
+      Array.from({ length: sendCount }, (_, index) => submission(sendCount - index - 1))
     )
 
-    expect(messages).toHaveLength(SEND_COUNT)
+    expect(messages.filter((message) => message.role === 'user')).toHaveLength(sendCount)
     expect(messages.map((message) => message.id)).toEqual(
-      Array.from({ length: SEND_COUNT }, (_, index) => `journal-${index}`)
+      Array.from({ length: sendCount }, (_, index) => `journal-${index}`)
     )
   })