From 80756ad6581585b5abf4e72cf0d55eda51dc4abd Mon Sep 17 00:00:00 2001 From: Brennan Benson <79079362+brennanb2025@users.noreply.github.com> Date: Sat, 3 Oct 2026 13:28:42 -0700 Subject: [PATCH] feat(native-chat): publish each submission's journal positions A client never learned where in the journal a send was accepted or taken back, so drawing a stopped send had to guess from its own row and from clocks, and the guesses misfired after restarts. Each projected submission now carries `submittedSequence` (its submission row) and `resolvedSequence` (the dispatch row that resolved it; absent while pending): two optional fields, computed on every fold from rows already stored, never stored themselves. Older clients ignore them. The failure-fact schema moves to its own module to keep the journal schema file within its size limit. --- .../journal-dispatch-reducer.ts | 5 + .../journal-submission-fold.ts | 2 + .../journal-submission-positions.test.ts | 110 ++++++++++++++++++ .../agent-session-failure-fact-schema.ts | 10 ++ src/shared/agent-session-journal-schemas.ts | 15 +-- src/shared/agent-session-journal-types.ts | 13 ++- ...ubmission-positions-downgrade.unit.test.ts | 110 ++++++++++++++++++ 7 files changed, 253 insertions(+), 12 deletions(-) create mode 100644 src/main/native-chat/agent-session-journal/journal-submission-positions.test.ts create mode 100644 src/shared/agent-session-failure-fact-schema.ts create mode 100644 tests/e2e/cross-version-wire/submission-positions-downgrade.unit.test.ts 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 89722959799..a52ff73288c 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 @@ -31,6 +31,11 @@ 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 2c9c6cfcd04..04323e36bb2 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 @@ -25,6 +25,7 @@ export function applyJournalSubmission( reason: null, submittedAt: row.ts, resolvedAt: null, + submittedSequence: row.seq, ...(row.handoverRecorded ? { handoverRecorded: true, acceptedSequence: row.seq } : {}), // A malformed stored link is dropped, never the row. ...(typeof row.queuedMessageId === 'string' && row.queuedMessageId.length > 0 @@ -90,6 +91,7 @@ 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 new file mode 100644 index 00000000000..298aeef48d7 --- /dev/null +++ b/src/main/native-chat/agent-session-journal/journal-submission-positions.test.ts @@ -0,0 +1,110 @@ +// 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. + +import { mkdtemp, rm } from 'node:fs/promises' +import { tmpdir } from 'node:os' +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 { + AGENT_JOURNAL_THREAD_SCOPE, + type AgentJournalMessageItem, + type AgentSessionJournalIdentity +} from '../../../shared/agent-session-journal-types' +import { readAgentSessionHydrationPage } from '../agent-session-wire/agent-session-history-page' +import { createTrackedJournalOpener } from './journal-host-database-test-support' + +const IDENTITY: AgentSessionJournalIdentity = { + sessionId: 'session-1', + workspaceId: 'workspace-1', + hostId: 'host-1', + agent: 'codex', + providerHandle: { kind: 'codex', threadId: 'thread-1' } +} +const BODY: AgentJournalMessageItem = { + kind: 'message', + role: 'user', + blocks: [{ type: 'text', text: 'look around' }] +} + +let root: string | null = null +const journals = createTrackedJournalOpener() + +afterEach(async () => { + await journals.closeAll() + if (root) { + await rm(root, { recursive: true, force: true }) + root = null + } +}) + +/** A send accepted for later handover, handed over, then taken back. */ +async function sendHandedOverThenWithdrawn() { + 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: 'send-1', + body: BODY, + fence: 1, + handoverRecorded: true + }) + const handedOver = await journal.resolveDispatch({ + clientMessageId: 'send-1', + state: 'pending', + fence: 1, + turnScope: AGENT_JOURNAL_THREAD_SCOPE + }) + const pending = { ...journal.submission('send-1') } + const withdrawn = await journal.resolveDispatch({ + clientMessageId: 'send-1', + state: 'rejected', + ...agentSessionFailureWords(agentSessionFailureFact('cancelled'), { surface: 'rejection' }), + fence: 1, + recovered: true + }) + 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 () => { + 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 + }) + }) + + it('come back the same from a replay of the stored rows', async () => { + const { journal, submitted, withdrawn } = 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(journal).not.toBe(replayed) + }) + + it('reach a client on the history page', async () => { + const { journal, submitted, withdrawn } = await sendHandedOverThenWithdrawn() + + expect(readAgentSessionHydrationPage(journal).submissions).toEqual([ + expect.objectContaining({ + clientMessageId: 'send-1', + submittedSequence: submitted.sequence, + resolvedSequence: withdrawn.sequence + }) + ]) + }) +}) diff --git a/src/shared/agent-session-failure-fact-schema.ts b/src/shared/agent-session-failure-fact-schema.ts new file mode 100644 index 00000000000..c8e57449066 --- /dev/null +++ b/src/shared/agent-session-failure-fact-schema.ts @@ -0,0 +1,10 @@ +import { z } from 'zod' + +/** A failure fact as a journal row carries it. Open like a row's `state`: a kind, audience or + * refusal detail a newer host writes must not turn the row malformed; the fact reader is where an + * unplaceable one is dropped. */ +export const AgentSessionFailureFactSchema = z.object({ + kind: z.string().min(1), + detail: z.object({ text: z.string(), audience: z.string().min(1) }).optional(), + refusal: z.object({ code: z.string().min(1), details: z.looseObject({}).optional() }).optional() +}) diff --git a/src/shared/agent-session-journal-schemas.ts b/src/shared/agent-session-journal-schemas.ts index 035bb7d1d01..cf5e6ba6cf6 100644 --- a/src/shared/agent-session-journal-schemas.ts +++ b/src/shared/agent-session-journal-schemas.ts @@ -15,6 +15,7 @@ import { z } from 'zod' import { AgentSessionContextUsageSchema } from './agent-session-context-usage-schema' +import { AgentSessionFailureFactSchema } from './agent-session-failure-fact-schema' import type { AgentJournalItemBody, AgentJournalMessageItem, @@ -201,14 +202,6 @@ const ThreadGoalState = z.union([ z.object({ state: z.string() }).refine((value) => !['set', 'cleared'].includes(value.state)) ]) -/** Open like `state`: a kind, audience or refusal detail a newer host writes must not turn the row - * malformed; the fact reader is where an unplaceable one is dropped. */ -const FailureFact = z.object({ - kind: z.string().min(1), - detail: z.object({ text: z.string(), audience: z.string().min(1) }).optional(), - refusal: z.object({ code: z.string().min(1), details: z.looseObject({}).optional() }).optional() -}) - export const AgentJournalItemBodySchema = z.discriminatedUnion('kind', [ MessageBody, z.object({ @@ -262,7 +255,7 @@ export const AgentJournalItemBodySchema = z.discriminatedUnion('kind', [ .optional(), providerFrame: ProviderFrame.optional(), threadGoal: ThreadGoalState.optional(), - failure: FailureFact.optional() + failure: AgentSessionFailureFactSchema.optional() }), z.object({ kind: z.literal('turn'), @@ -325,10 +318,12 @@ export const AgentJournalSubmissionSchema = z.object({ reason: z.string().nullable(), 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(), - rejection: FailureFact.optional(), + rejection: AgentSessionFailureFactSchema.optional(), // Listed, or the parse strips it: this schema drops unknown keys. queuedMessageId: z.string().min(1).optional() }) diff --git a/src/shared/agent-session-journal-types.ts b/src/shared/agent-session-journal-types.ts index e5f8125a5c0..20fbf6f58d7 100644 --- a/src/shared/agent-session-journal-types.ts +++ b/src/shared/agent-session-journal-types.ts @@ -411,6 +411,12 @@ export type AgentJournalSubmission = { rejection?: UnreadAgentSessionFailureFact 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. */ + submittedSequence?: number + /** Where the journal wrote the dispatch row that resolved it, in the same order; 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 @@ -419,13 +425,16 @@ export type AgentJournalSubmission = { handoverRecorded?: true /** When the host handed it to the provider (its `dispatch{pending}` row). */ handedOverAt?: number - /** Host-only: the submission row's sequence, which tells which host process accepted it. */ + /** Host-only: the submission row's sequence, which tells which host process accepted it. Set + * only on a send accepted for later handover. The snapshot still carries it, undeclared to + * clients; no released client reads it, and clients read `submittedSequence` instead. */ acceptedSequence?: number /** The queued draft this submission hands off; absent for a direct send. Read this, never * a draft id compared with `clientMessageId`. */ queuedMessageId?: string /** Host-only: who asked for this turn — a person over the client send RPC, or Orca itself. - * A person's turn is what ends a Stop's queue pause. */ + * A person's turn is what ends a Stop's queue pause. The snapshot still carries it; no released + * client reads it. */ origin?: 'client' | 'host' } 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 new file mode 100644 index 00000000000..75a4b4fcaaa --- /dev/null +++ b/tests/e2e/cross-version-wire/submission-positions-downgrade.unit.test.ts @@ -0,0 +1,110 @@ +import { mkdtempSync, rmSync } from 'node:fs' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { expect, test } from 'vitest' +import { agentSessionFailureFact } from '../../../src/shared/agent-session-failure' +import { agentSessionFailureWords } from '../../../src/shared/agent-session-failure-words' +import type { AgentSessionJournalIdentity } from '../../../src/shared/agent-session-journal-types' +import type { AgentSessionHistoryPage } from '../../../src/shared/agent-session-wire' +import { createTrackedJournalOpener } from '../../../src/main/native-chat/agent-session-journal/journal-host-database-test-support' +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. +const BASELINE_REF = 'v1.4.219' + +const IDENTITY: AgentSessionJournalIdentity = { + sessionId: 'session-positions', + workspaceId: 'ws-1', + hostId: 'host-1', + agent: 'codex', + providerHandle: { kind: 'codex', threadId: 'thread-1' } +} + +/** A function the pinned release exports, typed as the caller calls it. */ +function releaseExport(module: Record, name: string): T { + const value = module[name] + if (typeof value !== 'function' && (typeof value !== 'object' || value === null)) { + throw new Error(`the pinned release exports no ${name}`) + } + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: a value the pinned release exports; each caller names the shape it uses, and a changed one fails the test. + return value as 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 { + return { + ...page, + submissions: page.submissions.map( + ({ submittedSequence: _submitted, resolvedSequence: _resolved, ...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 () => { + const directory = mkdtempSync(join(tmpdir(), 'orca-submission-positions-downgrade-')) + const journals = createTrackedJournalOpener() + try { + const journal = await journals.open({ identity: IDENTITY, stateDirectory: directory }) + for (const id of ['taken-back', 'still-queued']) { + await journal.appendSubmission({ + clientMessageId: id, + payloadFingerprint: id, + body: { kind: 'message', role: 'user', blocks: [{ type: 'text', text: id }] }, + fence: 1, + handoverRecorded: true + }) + } + await journal.resolveDispatch({ + clientMessageId: 'taken-back', + state: 'rejected', + ...agentSessionFailureWords(agentSessionFailureFact('cancelled'), { surface: 'rejection' }), + fence: 1, + recovered: true + }) + const page = readAgentSessionHydrationPage(journal, 1) + // Anti-vacuous: the page this build publishes does carry both positions. + expect(page.submissions.find((entry) => entry.clientMessageId === 'taken-back')).toEqual( + expect.objectContaining({ + submittedSequence: expect.any(Number), + resolvedSequence: expect.any(Number) + }) + ) + + const checkout = await materializeReleaseCheckout(BASELINE_REF) + const reducer = await importReleaseCheckoutModule( + checkout, + 'src/shared/structured-agent-session-reducer.ts' + ) + const projection = await importReleaseCheckoutModule( + checkout, + 'src/shared/structured-agent-session-message-projection.ts' + ) + const reduce = releaseExport<(state: unknown, action: unknown) => OldState>( + reducer, + 'reduceStructuredAgentSession' + ) + const empty = releaseExport(reducer, 'EMPTY_STRUCTURED_AGENT_SESSION') + const project = releaseExport< + (items: unknown[], outbox: unknown[], submissions: unknown[]) => unknown[] + >(projection, 'projectStructuredAgentSessionMessages') + const draw = (from: AgentSessionHistoryPage) => { + const state = reduce(empty, { type: 'history-page', page: from }) + return { + submissions: state.submissions.map( + ({ submittedSequence: _s, resolvedSequence: _r, ...rest }) => rest + ), + messages: project(state.items, [], state.submissions) + } + } + + expect(draw(page)).toEqual(draw(withoutPositions(page))) + } finally { + await journals.closeAll() + rmSync(directory, { recursive: true, force: true }) + } +}, 180_000)