From bd3789ef24ade43dbb9551f72d8a8bc7589459c5 Mon Sep 17 00:00:00 2001 From: Brennan Benson <79079362+brennanb2025@users.noreply.github.com> Date: Mon, 5 Oct 2026 07:41:41 -0700 Subject: [PATCH] refactor(native-chat): publish only where each submission was sent A rejected send's own row now moves to its rejection (#24710), so the row that resolved a send is the item's own position and needs no second copy. Keep submittedSequence, the one journal-order record of where a send was sent. --- .../journal-dispatch-reducer.ts | 5 -- .../journal-submission-fold.ts | 1 - .../journal-submission-positions.test.ts | 75 +++++-------------- src/shared/agent-session-journal-schemas.ts | 1 - src/shared/agent-session-journal-types.ts | 7 +- ...ubmission-positions-downgrade.unit.test.ts | 25 +++---- 6 files changed, 30 insertions(+), 84 deletions(-) diff --git a/src/main/native-chat/agent-session-journal/journal-dispatch-reducer.ts b/src/main/native-chat/agent-session-journal/journal-dispatch-reducer.ts index 5e4574b60eb..c7c2a671654 100644 --- a/src/main/native-chat/agent-session-journal/journal-dispatch-reducer.ts +++ b/src/main/native-chat/agent-session-journal/journal-dispatch-reducer.ts @@ -35,11 +35,6 @@ export function applyJournalDispatchRow( delete submission.rejection } submission.resolvedAt = row.state === 'pending' ? null : row.ts - if (row.state === 'pending') { - delete submission.resolvedSequence - } else { - submission.resolvedSequence = row.seq - } if (row.state === 'pending') { submission.handedOverAt = row.ts placeHandedOverMessage(state, submission, row) diff --git a/src/main/native-chat/agent-session-journal/journal-submission-fold.ts b/src/main/native-chat/agent-session-journal/journal-submission-fold.ts index c04e354fb89..b00bdd36de2 100644 --- a/src/main/native-chat/agent-session-journal/journal-submission-fold.ts +++ b/src/main/native-chat/agent-session-journal/journal-submission-fold.ts @@ -113,7 +113,6 @@ export function acceptSubmissionFromProviderItem( submission.providerItemId = providerItemId submission.reason = null submission.resolvedAt = row.ts - submission.resolvedSequence = row.seq delete submission.recovered state.receipts.set(submission.clientMessageId, { clientMessageId: submission.clientMessageId, diff --git a/src/main/native-chat/agent-session-journal/journal-submission-positions.test.ts b/src/main/native-chat/agent-session-journal/journal-submission-positions.test.ts index c2b6eacc5bb..eac42c79113 100644 --- a/src/main/native-chat/agent-session-journal/journal-submission-positions.test.ts +++ b/src/main/native-chat/agent-session-journal/journal-submission-positions.test.ts @@ -1,6 +1,6 @@ -// Each submission carries where the journal wrote its row and the row that resolved it, so a client -// can place a send by journal order instead of guessing from clocks. Derived on every fold: a -// replay of the same rows gives the same values, and a history page carries them. +// Each submission carries where the journal wrote its row. A rejected send's own row moves to its +// rejection, so this is the only journal-order record of where it was sent. Derived on every fold: a +// replay of the same rows gives the same value, and a history page carries it. import { mkdtemp, rm } from 'node:fs/promises' import { tmpdir } from 'node:os' @@ -8,12 +8,12 @@ import { join } from 'node:path' import { afterEach, describe, expect, it } from 'vitest' import { agentSessionFailureFact } from '../../../shared/agent-session-failure' import { agentSessionFailureWords } from '../../../shared/agent-session-failure-words' +import { agentJournalSubmissionKey } from '../../../shared/agent-session-journal-item-key' import { AGENT_JOURNAL_THREAD_SCOPE, type AgentJournalMessageItem, type AgentSessionJournalIdentity } from '../../../shared/agent-session-journal-types' -import { structuredAgentSessionPayloadFingerprint } from '../../../shared/structured-agent-session-mutation' import { readAgentSessionHydrationPage } from '../agent-session-wire/agent-session-history-page' import { createTrackedJournalOpener } from './journal-host-database-test-support' @@ -69,78 +69,39 @@ async function sendHandedOverThenWithdrawn() { return { journal, submitted, handedOver, pending, withdrawn } } -describe("a submission's journal positions", () => { - it('are its own row and the row that resolved it, with none while it is pending', async () => { +describe("a submission's journal position", () => { + it('is its own row, which a take-back does not move', async () => { const { journal, submitted, handedOver, pending, withdrawn } = await sendHandedOverThenWithdrawn() expect(pending).toMatchObject({ submittedSequence: submitted.sequence }) - expect(pending).not.toHaveProperty('resolvedSequence') expect(handedOver.sequence).toBeGreaterThan(submitted.sequence) expect(journal.submission('send-1')).toMatchObject({ dispatchState: 'rejected', - submittedSequence: submitted.sequence, - resolvedSequence: withdrawn.sequence + submittedSequence: submitted.sequence }) + // Anti-vacuous: the send's own row moved to its rejection, so only this field says where it was sent. + expect( + journal.snapshot().items.find((item) => item.itemId === agentJournalSubmissionKey('send-1')) + ?.sequence + ).toBe(withdrawn.sequence) }) - it('come back the same from a replay of the stored rows', async () => { - const { journal, submitted, withdrawn } = await sendHandedOverThenWithdrawn() + it('comes back the same from a replay of the stored rows', async () => { + const { journal, submitted } = await sendHandedOverThenWithdrawn() await journals.closeAll() const replayed = await journals.open({ identity: IDENTITY, stateDirectory: root! }) - expect(replayed.submission('send-1')).toMatchObject({ - submittedSequence: submitted.sequence, - resolvedSequence: withdrawn.sequence - }) + expect(replayed.submission('send-1')).toMatchObject({ submittedSequence: submitted.sequence }) expect(journal).not.toBe(replayed) }) - it("move to the provider's echo when that is what accepts a send left in doubt", async () => { - root = await mkdtemp(join(tmpdir(), 'orca-submission-positions-')) - const journal = await journals.open({ identity: IDENTITY, stateDirectory: root }) - const submitted = await journal.appendSubmission({ - clientMessageId: 'send-1', - payloadFingerprint: structuredAgentSessionPayloadFingerprint({ - method: 'agentSession.send', - sessionId: IDENTITY.sessionId, - fields: { body: BODY } - }), - body: BODY, - fence: 1 - }) - const doubted = await journal.resolveDispatch({ - clientMessageId: 'send-1', - state: 'unknown', - reason: 'host_restarted', - fence: 1, - recovered: true - }) - - const echo = await journal.appendItem( - { provider: 'codex', threadId: 'thread-1', turnId: 'turn-1', ordinal: 0 }, - BODY, - { fence: 1, turnScope: AGENT_JOURNAL_THREAD_SCOPE } - ) - - expect(echo.cursor.sequence).toBeGreaterThan(doubted.sequence) - expect(journal.submission('send-1')).toMatchObject({ - dispatchState: 'accepted', - submittedSequence: submitted.sequence, - resolvedSequence: echo.cursor.sequence - }) - }) - - it('reach a client on the history page', async () => { - const { journal, submitted, withdrawn } = await sendHandedOverThenWithdrawn() + it('reaches a client on the history page', async () => { + const { journal, submitted } = await sendHandedOverThenWithdrawn() expect(readAgentSessionHydrationPage(journal).submissions).toEqual([ - expect.objectContaining({ - clientMessageId: 'send-1', - submittedSequence: submitted.sequence, - resolvedSequence: withdrawn.sequence - }) + expect.objectContaining({ clientMessageId: 'send-1', submittedSequence: submitted.sequence }) ]) }) }) diff --git a/src/shared/agent-session-journal-schemas.ts b/src/shared/agent-session-journal-schemas.ts index cf5e6ba6cf6..8a7a5149367 100644 --- a/src/shared/agent-session-journal-schemas.ts +++ b/src/shared/agent-session-journal-schemas.ts @@ -319,7 +319,6 @@ export const AgentJournalSubmissionSchema = z.object({ submittedAt: z.number(), resolvedAt: z.number().nullable(), submittedSequence: z.number().int().optional(), - resolvedSequence: z.number().int().optional(), recovered: z.literal(true).optional(), handoverRecorded: z.literal(true).optional(), handedOverAt: z.number().optional(), diff --git a/src/shared/agent-session-journal-types.ts b/src/shared/agent-session-journal-types.ts index 25863629790..28e470cb869 100644 --- a/src/shared/agent-session-journal-types.ts +++ b/src/shared/agent-session-journal-types.ts @@ -412,12 +412,9 @@ export type AgentJournalSubmission = { submittedAt: number resolvedAt: number | null /** Where the journal wrote this submission's row: its sequence, recomputed on every fold and - * never stored. Absent from hosts that predate it. */ + * never stored. Needed because a rejected send's own row moves to its rejection, which erases + * where it was sent. Absent from hosts that predate it. */ submittedSequence?: number - /** Where the journal wrote the row that resolved it, in the same order: its dispatch row, or the - * provider's echo (an item or lifecycle-batch row) when that accepted it. Absent while it is - * pending, and from hosts that predate it. */ - resolvedSequence?: number /** Set when crash reconciliation resolved the dispatch, not the provider. A live * `unknown` is a send still outstanding; a recovered one outlived its writer. */ recovered?: true diff --git a/tests/e2e/cross-version-wire/submission-positions-downgrade.unit.test.ts b/tests/e2e/cross-version-wire/submission-positions-downgrade.unit.test.ts index 75a4b4fcaaa..ef5d1eba2ec 100644 --- a/tests/e2e/cross-version-wire/submission-positions-downgrade.unit.test.ts +++ b/tests/e2e/cross-version-wire/submission-positions-downgrade.unit.test.ts @@ -10,8 +10,8 @@ import { createTrackedJournalOpener } from '../../../src/main/native-chat/agent- import { readAgentSessionHydrationPage } from '../../../src/main/native-chat/agent-session-wire/agent-session-history-page' import { importReleaseCheckoutModule, materializeReleaseCheckout } from './release-checkout' -// A released client that predates the published submission positions: it must fold and draw a -// page carrying them exactly as it draws the same page without them. +// A released client that predates the published submission position: it must fold and draw a page +// carrying it exactly as it draws the same page without it. const BASELINE_REF = 'v1.4.219' const IDENTITY: AgentSessionJournalIdentity = { @@ -34,18 +34,16 @@ function releaseExport(module: Record, name: string): T { type OldState = { submissions: Record[]; items: unknown[] } -/** Strips the published positions, as an older host's page would carry its submissions. */ -function withoutPositions(page: AgentSessionHistoryPage): AgentSessionHistoryPage { +/** Strips the published position, as an older host's page would carry its submissions. */ +function withoutPosition(page: AgentSessionHistoryPage): AgentSessionHistoryPage { return { ...page, - submissions: page.submissions.map( - ({ submittedSequence: _submitted, resolvedSequence: _resolved, ...rest }) => rest - ) + submissions: page.submissions.map(({ submittedSequence: _submitted, ...rest }) => rest) } } // Loads a real release checkout, cold extraction and transforms included. -test('a released client folds and draws a page whose submissions carry their journal positions', async () => { +test('a released client folds and draws a page whose submissions carry their journal position', async () => { const directory = mkdtempSync(join(tmpdir(), 'orca-submission-positions-downgrade-')) const journals = createTrackedJournalOpener() try { @@ -67,11 +65,10 @@ test('a released client folds and draws a page whose submissions carry their jou recovered: true }) const page = readAgentSessionHydrationPage(journal, 1) - // Anti-vacuous: the page this build publishes does carry both positions. + // Anti-vacuous: the page this build publishes does carry the position. expect(page.submissions.find((entry) => entry.clientMessageId === 'taken-back')).toEqual( expect.objectContaining({ - submittedSequence: expect.any(Number), - resolvedSequence: expect.any(Number) + submittedSequence: expect.any(Number) }) ) @@ -95,14 +92,12 @@ test('a released client folds and draws a page whose submissions carry their jou const draw = (from: AgentSessionHistoryPage) => { const state = reduce(empty, { type: 'history-page', page: from }) return { - submissions: state.submissions.map( - ({ submittedSequence: _s, resolvedSequence: _r, ...rest }) => rest - ), + submissions: state.submissions.map(({ submittedSequence: _s, ...rest }) => rest), messages: project(state.items, [], state.submissions) } } - expect(draw(page)).toEqual(draw(withoutPositions(page))) + expect(draw(page)).toEqual(draw(withoutPosition(page))) } finally { await journals.closeAll() rmSync(directory, { recursive: true, force: true })