From 6bb062827e633fea1dc365f122ed72ce9ddea896 Mon Sep 17 00:00:00 2001 From: Brennan Benson <79079362+brennanb2025@users.noreply.github.com> Date: Fri, 14 Aug 2026 18:18:22 -0700 Subject: [PATCH] fix(native-chat): reconcile queued Codex send echoes --- ...mobile-structured-session-timeline.test.ts | 22 +++++ .../journal-reducer.test.ts | 98 +++++++++++++++++++ .../agent-session-journal/journal-reducer.ts | 49 +++++++++- .../journal-store-contracts.ts | 37 +++++++ .../agent-session-journal/journal-store.ts | 42 +++----- .../agent-session-history-page.test.ts | 50 ++++++++++ .../agent-session-history-page.ts | 7 +- .../agent-session-journal-batch.ts | 5 +- ...red-agent-session-send-idempotency.test.ts | 81 +++++++++++++++ .../structured-agent-session-subscribers.ts | 3 +- .../structured-agent-session-turns.ts | 9 ++ ...d-agent-session-message-projection.test.ts | 14 ++- 12 files changed, 373 insertions(+), 44 deletions(-) create mode 100644 src/main/native-chat/agent-session-journal/journal-store-contracts.ts create mode 100644 src/main/native-chat/agent-session-wire/structured-agent-session-send-idempotency.test.ts diff --git a/mobile/src/session/mobile-structured-session-timeline.test.ts b/mobile/src/session/mobile-structured-session-timeline.test.ts index bb0b041df08..8e75f385e5c 100644 --- a/mobile/src/session/mobile-structured-session-timeline.test.ts +++ b/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]) diff --git a/src/main/native-chat/agent-session-journal/journal-reducer.test.ts b/src/main/native-chat/agent-session-journal/journal-reducer.test.ts index 3ebb1dffc85..1be6f256406 100644 --- a/src/main/native-chat/agent-session-journal/journal-reducer.test.ts +++ b/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, diff --git a/src/main/native-chat/agent-session-journal/journal-reducer.ts b/src/main/native-chat/agent-session-journal/journal-reducer.ts index 31898d4c5e7..c311d003a2b 100644 --- a/src/main/native-chat/agent-session-journal/journal-reducer.ts +++ b/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 } diff --git a/src/main/native-chat/agent-session-journal/journal-store-contracts.ts b/src/main/native-chat/agent-session-journal/journal-store-contracts.ts new file mode 100644 index 00000000000..1959c224ba4 --- /dev/null +++ b/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 +} diff --git a/src/main/native-chat/agent-session-journal/journal-store.ts b/src/main/native-chat/agent-session-journal/journal-store.ts index d4cb6358454..6591bfb08da 100644 --- a/src/main/native-chat/agent-session-journal/journal-store.ts +++ b/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 { @@ -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) { diff --git a/src/main/native-chat/agent-session-wire/agent-session-history-page.test.ts b/src/main/native-chat/agent-session-wire/agent-session-history-page.test.ts index daeeffff16b..22344a5cf61 100644 --- a/src/main/native-chat/agent-session-wire/agent-session-history-page.test.ts +++ b/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([]) + }) }) diff --git a/src/main/native-chat/agent-session-wire/agent-session-history-page.ts b/src/main/native-chat/agent-session-wire/agent-session-history-page.ts index bbda0446c81..85663b50bee 100644 --- a/src/main/native-chat/agent-session-wire/agent-session-history-page.ts +++ b/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 } } diff --git a/src/main/native-chat/agent-session-wire/agent-session-journal-batch.ts b/src/main/native-chat/agent-session-wire/agent-session-journal-batch.ts index 96a3c262c8b..61c811f6762 100644 --- a/src/main/native-chat/agent-session-wire/agent-session-journal-batch.ts +++ b/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() 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') { diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-send-idempotency.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-send-idempotency.test.ts new file mode 100644 index 00000000000..8b6a73fd57b --- /dev/null +++ b/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) + }) +}) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-subscribers.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-subscribers.ts index 59003786ede..4906a268ab2 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-subscribers.ts +++ b/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, { diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-turns.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-turns.ts index 4356551d77e..d454c5338cf 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-turns.ts +++ b/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() diff --git a/src/renderer/src/components/native-chat/structured-agent-session-message-projection.test.ts b/src/renderer/src/components/native-chat/structured-agent-session-message-projection.test.ts index 84e820d6300..e7d30cb5e0b 100644 --- a/src/renderer/src/components/native-chat/structured-agent-session-message-projection.test.ts +++ b/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}`) ) })