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 1/8] 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 89722959799d..a52ff73288c3 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 2c9c6cfcd040..04323e36bb2f 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 000000000000..298aeef48d7c --- /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 000000000000..c8e574490660 --- /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 035bb7d1d01c..cf5e6ba6cf66 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 e5f8125a5c0f..20fbf6f58d71 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 000000000000..75a4b4fcaaa7 --- /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) From 58a1ade5048669b1aadbac65799a6913152a9981 Mon Sep 17 00:00:00 2001 From: Brennan Benson <79079362+brennanb2025@users.noreply.github.com> Date: Sat, 3 Oct 2026 13:46:16 -0700 Subject: [PATCH 2/8] test(native-chat): pin a submission's resolved position on the provider-echo accept path Also say which rows can resolve a submission, and that the submission schema, not the shared type, omits acceptedSequence. --- .../journal-submission-positions.test.ts | 36 +++++++++++++++++++ src/shared/agent-session-journal-types.ts | 9 ++--- 2 files changed, 41 insertions(+), 4 deletions(-) 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 298aeef48d7c..c2b6eacc5bbd 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 @@ -13,6 +13,7 @@ import { 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' @@ -96,6 +97,41 @@ describe("a submission's journal positions", () => { 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() diff --git a/src/shared/agent-session-journal-types.ts b/src/shared/agent-session-journal-types.ts index 20fbf6f58d71..258636297900 100644 --- a/src/shared/agent-session-journal-types.ts +++ b/src/shared/agent-session-journal-types.ts @@ -414,8 +414,9 @@ export type AgentJournalSubmission = { /** 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. */ + /** 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. */ @@ -426,8 +427,8 @@ export type AgentJournalSubmission = { /** 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. 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. */ + * only on a send accepted for later handover. The snapshot still carries it, but the submission + * schema omits it; 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`. */ 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 3/8] 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 5e4574b60ebf..c7c2a6716546 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 c04e354fb890..b00bdd36de24 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 c2b6eacc5bbd..eac42c79113c 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 cf5e6ba6cf66..8a7a51493675 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 258636297900..28e470cb869e 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 75a4b4fcaaa7..ef5d1eba2ec7 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 }) From 93749070e48b624c7c2742caf1cf9945ffcb55b4 Mon Sep 17 00:00:00 2001 From: Brennan Benson <79079362+brennanb2025@users.noreply.github.com> Date: Mon, 5 Oct 2026 08:50:55 -0700 Subject: [PATCH 4/8] feat(native-chat): publish the turn a withdrawn Codex send was answered into A send Codex answered into a turn that then ended without taking it is rejected by that turn's end. The rejection row now names that turn's record, and the submission carries it as answeredInTurnItemId, so a client can tell which turn the send belonged to without guessing from journal order or clocks. Absent on every other send. --- .../codex/codex-structured-session-adapter.ts | 2 +- .../codex/codex-structured-session-state.ts | 5 +- ...dex-structured-turn-end-settlement.test.ts | 85 ++++++++++++++++++- .../codex-structured-turn-end-settlement.ts | 20 +++-- src/main/codex/codex-structured-turn-start.ts | 16 +++- .../journal-dispatch-reducer.ts | 9 ++ .../journal-row-builders.ts | 3 + .../journal-row-schema.ts | 4 + .../journal-store-contracts.ts | 5 +- .../journal-submission-positions.test.ts | 63 +++++++++++++- .../structured-agent-session-adapter.ts | 8 +- .../structured-agent-session-late-dispatch.ts | 6 +- .../structured-agent-session-turns.ts | 1 + ...-session-codex-turn-end-settlement.test.ts | 73 ++++++++++++++++ src/shared/agent-session-journal-schemas.ts | 1 + src/shared/agent-session-journal-types.ts | 4 + ...ubmission-positions-downgrade.unit.test.ts | 40 +++++++-- 17 files changed, 319 insertions(+), 26 deletions(-) diff --git a/src/main/codex/codex-structured-session-adapter.ts b/src/main/codex/codex-structured-session-adapter.ts index 13a687a72cdc..1184ada8c741 100644 --- a/src/main/codex/codex-structured-session-adapter.ts +++ b/src/main/codex/codex-structured-session-adapter.ts @@ -132,7 +132,7 @@ export class CodexStructuredSessionAdapter implements StructuredAgentSessionAdap } if (event.type === 'notification') { // Only an admitted turn end settles; a refused one settles on the retry that lands. - settleCodexSendsInEndedTurn(session, event.method, event.params, (settlement) => + settleCodexSendsInEndedTurn(session, event, (settlement) => this.deps.onDispatchSettledLate?.({ sessionId: event.sessionId, ...settlement }) ) // After the admission check, so a refused frame is observed by the child records diff --git a/src/main/codex/codex-structured-session-state.ts b/src/main/codex/codex-structured-session-state.ts index a47cb5a78fe2..277a2719673a 100644 --- a/src/main/codex/codex-structured-session-state.ts +++ b/src/main/codex/codex-structured-session-state.ts @@ -84,7 +84,10 @@ export type CodexStructuredSessionAdapterDeps = { onDispatchSettledLate?: ( input: { sessionId: string; clientMessageId: string } & ( | { providerIdentity: AgentJournalItemIdentity } - | ({ state: 'rejected' } & AgentJournalDispatchRejection) + | ({ + state: 'rejected' + answeredInTurn: AgentJournalItemIdentity + } & AgentJournalDispatchRejection) ) ) => void /** Codex reported its thread not running with no turn open: a send whose diff --git a/src/main/codex/codex-structured-turn-end-settlement.test.ts b/src/main/codex/codex-structured-turn-end-settlement.test.ts index 38cc648f79d9..ccb9fa7ba456 100644 --- a/src/main/codex/codex-structured-turn-end-settlement.test.ts +++ b/src/main/codex/codex-structured-turn-end-settlement.test.ts @@ -1,5 +1,9 @@ import { describe, expect, it, vi } from 'vitest' -import type { AgentJournalItemBody } from '../../shared/agent-session-journal-types' +import type { + AgentJournalItemBody, + AgentJournalItemIdentity +} from '../../shared/agent-session-journal-types' +import { agentJournalItemKey } from '../../shared/agent-session-journal-item-key' import { classifyDispatchRejection } from '../../shared/structured-agent-session-dispatch-rejection' import { createCodexDispatchEchoes } from './codex-structured-dispatch-echo' import { @@ -20,11 +24,17 @@ async function turnEndRig() { Object.assign(codex.routes, turns.routes) const settlements: LateSettlement[] = [] const bodies: AgentJournalItemBody[] = [] + const turnRecords: AgentJournalItemIdentity[] = [] const adapter = await acquiredCodexAdapter({ codex, settlements, sink: { - appendItem: (_identity, body) => bodies.push(body), + appendItem: (identity, body) => { + bodies.push(body) + if (body.kind === 'turn') { + turnRecords.push(identity) + } + }, appendTombstone: () => {}, publish: () => {} } @@ -46,7 +56,18 @@ async function turnEndRig() { const settledIds = () => settlements.map(({ clientMessageId }) => clientMessageId) const categoryOf = (settlement: LateSettlement | undefined) => settlement && 'state' in settlement ? classifyDispatchRejection(settlement).category : null - return { codex, turns, adapter, send, sendAndOpen, settlements, settledIds, categoryOf, bodies } + return { + codex, + turns, + adapter, + send, + sendAndOpen, + settlements, + settledIds, + categoryOf, + bodies, + turnRecords + } } describe('a Codex send its turn ended without echoing', () => { @@ -286,6 +307,64 @@ describe('a Codex send its turn ended without echoing', () => { }) }) +describe('the turn a withdrawn Codex send names', () => { + it('is the record of the turn it opened, when that turn is interrupted', async () => { + const rig = await turnEndRig() + await rig.sendAndOpen('client-1') + + rig.turns.end('interrupted') + + expect(rig.turnRecords.length).toBeGreaterThan(0) + expect(new Set(rig.turnRecords.map(agentJournalItemKey)).size).toBe(1) + expect(rig.settlements).toEqual([ + expect.objectContaining({ clientMessageId: 'client-1', answeredInTurn: rig.turnRecords[0] }) + ]) + }) + + it('is the running turn for a send steered into it', async () => { + const rig = await turnEndRig() + await rig.sendAndOpen('client-1') + await rig.send('client-2') + + rig.turns.end('interrupted') + + expect(rig.settlements.map((settlement) => [settlement.clientMessageId, settlement])).toEqual([ + ['client-1', expect.objectContaining({ answeredInTurn: rig.turnRecords[0] })], + ['client-2', expect.objectContaining({ answeredInTurn: rig.turnRecords[0] })] + ]) + }) + + it('is the ended turn whose end was read before the answer', async () => { + const rig = await turnEndRig() + const release = rig.turns.holdNextAnswer() + const sending = rig.send('client-1') + await vi.waitFor(() => expect(rig.turns.turnId).toBe('turn-1')) + rig.turns.start() + rig.turns.end('interrupted') + release() + + await expect(sending).resolves.toMatchObject({ + state: 'rejected', + answeredInTurn: rig.turnRecords[0] + }) + }) + + it('is not named when its turn completes without echoing it: the send stays pending', async () => { + const rig = await turnEndRig() + const release = rig.turns.holdNextAnswer() + const sending = rig.send('client-1') + await vi.waitFor(() => expect(rig.turns.turnId).toBe('turn-1')) + rig.turns.start() + rig.turns.end('completed') + release() + + await expect(sending).resolves.toEqual({ state: 'admitted' }) + await rig.sendAndOpen('client-2') + rig.turns.end('completed') + expect(rig.settlements).toEqual([]) + }) +}) + describe('a send bound to a turn', () => { it('dies with the settlement its turn end makes', () => { const echoes = createCodexDispatchEchoes() diff --git a/src/main/codex/codex-structured-turn-end-settlement.ts b/src/main/codex/codex-structured-turn-end-settlement.ts index 657006449331..931f4d41a702 100644 --- a/src/main/codex/codex-structured-turn-end-settlement.ts +++ b/src/main/codex/codex-structured-turn-end-settlement.ts @@ -17,7 +17,9 @@ import { agentSessionFailureWords, type AgentJournalDispatchRejection } from '../../shared/agent-session-failure-words' +import type { AgentJournalItemIdentity } from '../../shared/agent-session-journal-types' import type { CodexTurnEnd } from './codex-structured-dispatch-echo' +import { codexTurnLifecycleIdentity } from './codex-structured-journal-translation-turns' import type { CodexSession } from './codex-structured-session-state' import { readCodexThreadId, @@ -41,6 +43,8 @@ export function codexDispatchRejection( export type CodexTurnEndSettlement = { clientMessageId: string state: 'rejected' + /** The turn Codex answered the send into, whose end settled it. */ + answeredInTurn: AgentJournalItemIdentity } & AgentJournalDispatchRejection function errorDetail(params: unknown): ProviderDiagnostic | undefined { @@ -80,19 +84,23 @@ export function codexTurnEndRejection(end: CodexTurnEnd): AgentJournalDispatchRe /** Settles the sends bound to the turn this admitted notification ended. */ export function settleCodexSendsInEndedTurn( session: Pick, - method: string, - params: unknown, + frame: { sessionId: string; method: string; params: unknown }, settle: (settlement: CodexTurnEndSettlement) => void ): void { - const turnId = readCodexTurnId(params) - const end = readCodexTurnEnd(method, params) - if (!turnId || !end || (readCodexThreadId(params) ?? session.threadId) !== session.threadId) { + const turnId = readCodexTurnId(frame.params) + const end = readCodexTurnEnd(frame.method, frame.params) + if ( + !turnId || + !end || + (readCodexThreadId(frame.params) ?? session.threadId) !== session.threadId + ) { return } const rejection = codexTurnEndRejection(end) + const answeredInTurn = codexTurnLifecycleIdentity(frame.sessionId, turnId) for (const clientMessageId of session.dispatchEchoes.endTurn(session.threadId, turnId, end)) { if (rejection) { - settle({ clientMessageId, state: 'rejected', ...rejection }) + settle({ clientMessageId, state: 'rejected', answeredInTurn, ...rejection }) } } } diff --git a/src/main/codex/codex-structured-turn-start.ts b/src/main/codex/codex-structured-turn-start.ts index 7f0a2ec0c239..0aace8ee54b3 100644 --- a/src/main/codex/codex-structured-turn-start.ts +++ b/src/main/codex/codex-structured-turn-start.ts @@ -8,6 +8,7 @@ import { } from './codex-app-server-connection' import { isCodexAppServerUnsupportedError } from './codex-app-server-session' import type { CodexDispatchEchoes } from './codex-structured-dispatch-echo' +import { codexTurnLifecycleIdentity } from './codex-structured-journal-translation-turns' import { readCodexTurnId } from './codex-structured-thread-facts' import { codexRunningOrOpeningTurn, @@ -178,7 +179,12 @@ export async function startCodexTurn( */ export async function dispatchCodexTurn( session: CodexTurnHost, - input: { clientMessageId: string; body: AgentJournalMessageItem; requestedAt?: number }, + input: { + sessionId: string + clientMessageId: string + body: AgentJournalMessageItem + requestedAt?: number + }, timeoutMs: number | undefined ): Promise { let answer: { turnId: string | null } | false @@ -208,5 +214,11 @@ export async function dispatchCodexTurn( ? session.dispatchEchoes.bindTurn(input.clientMessageId, session.threadId, answer.turnId) : null const rejection = endedFirst ? codexTurnEndRejection(endedFirst) : null - return rejection ? { state: 'rejected', ...rejection } : { state: 'admitted' } + return rejection && answer.turnId + ? { + state: 'rejected', + answeredInTurn: codexTurnLifecycleIdentity(input.sessionId, answer.turnId), + ...rejection + } + : { state: 'admitted' } } 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 c7c2a6716546..c2f7d45327a5 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 @@ -34,6 +34,15 @@ export function applyJournalDispatchRow( } else { delete submission.rejection } + if ( + row.state === 'rejected' && + typeof row.answeredInTurnItemId === 'string' && + row.answeredInTurnItemId + ) { + submission.answeredInTurnItemId = row.answeredInTurnItemId + } else { + delete submission.answeredInTurnItemId + } submission.resolvedAt = row.state === 'pending' ? null : row.ts if (row.state === 'pending') { submission.handedOverAt = row.ts diff --git a/src/main/native-chat/agent-session-journal/journal-row-builders.ts b/src/main/native-chat/agent-session-journal/journal-row-builders.ts index 0865df3a41fa..6873db0fd46b 100644 --- a/src/main/native-chat/agent-session-journal/journal-row-builders.ts +++ b/src/main/native-chat/agent-session-journal/journal-row-builders.ts @@ -110,6 +110,9 @@ export function journalDispatchRowBuilder( providerItemId, reason: boundedDispatchReason(input), ...(input.state === 'rejected' ? { rejection: input.rejection } : {}), + ...(input.state === 'rejected' && input.answeredInTurn + ? { answeredInTurnItemId: agentJournalItemKey(input.answeredInTurn) } + : {}), ...journalRowBase(state().epoch, seq, input.fence, ts), ...(input.recovered ? { recovered: input.recovered } : {}), ...(input.state === 'pending' ? { turnScope: input.turnScope } : {}) diff --git a/src/main/native-chat/agent-session-journal/journal-row-schema.ts b/src/main/native-chat/agent-session-journal/journal-row-schema.ts index 9276b1a8c05f..194290542d93 100644 --- a/src/main/native-chat/agent-session-journal/journal-row-schema.ts +++ b/src/main/native-chat/agent-session-journal/journal-row-schema.ts @@ -150,6 +150,10 @@ export type JournalDispatchRow = JournalRowBase & { /** On `rejected`: why, typed. Older readers keep the key and ignore it; a malformed one is * dropped when read, never the row. */ rejection?: AgentSessionFailureFact + /** On `rejected`: the turn record a Codex send was answered into (`turn/start` or `turn/steer`) + * when that turn's end settled the send. Absent on every other row. Older readers keep the key + * and ignore it. */ + answeredInTurnItemId?: string } /** An item mutation may name its own producer, because one batch can CREATE 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 index cb86a7a8e5da..b2d2914b318b 100644 --- a/src/main/native-chat/agent-session-journal/journal-store-contracts.ts +++ b/src/main/native-chat/agent-session-journal/journal-store-contracts.ts @@ -40,7 +40,10 @@ export type ResolveDispatchInput = { | { state: 'pending'; turnScope: AgentJournalTurnScope } /** `reason` is what released clients print, `rejection` what newer ones read: both from * `agentSessionFailureWords`, never written by hand. */ - | ({ state: 'rejected' } & AgentJournalDispatchRejection) + | ({ + state: 'rejected' + answeredInTurn?: AgentJournalItemIdentity + } & AgentJournalDispatchRejection) | { state: 'unknown'; reason?: string | null } ) 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 eac42c79113c..31cfee3a25c9 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,7 @@ // 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. +// replay of the same rows gives the same value, and a history page carries it. A rejection a turn's +// end made also names that turn. import { mkdtemp, rm } from 'node:fs/promises' import { tmpdir } from 'node:os' @@ -8,7 +9,10 @@ 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 { + agentJournalItemKey, + agentJournalSubmissionKey +} from '../../../shared/agent-session-journal-item-key' import { AGENT_JOURNAL_THREAD_SCOPE, type AgentJournalMessageItem, @@ -69,6 +73,61 @@ async function sendHandedOverThenWithdrawn() { return { journal, submitted, handedOver, pending, withdrawn } } +const TURN = { + provider: 'legacy', + agent: 'codex', + sessionId: 'session-1', + recordId: 'turn-lifecycle:turn-1' +} as const + +/** A send its turn ended without taking: the rejection names that turn. */ +async function sendWithdrawnByItsTurnEnd() { + root = await mkdtemp(join(tmpdir(), 'orca-submission-positions-')) + const journal = await journals.open({ identity: IDENTITY, stateDirectory: root }) + await journal.appendSubmission({ + clientMessageId: 'send-1', + payloadFingerprint: 'send-1', + body: BODY, + fence: 1 + }) + await journal.resolveDispatch({ + clientMessageId: 'send-1', + state: 'rejected', + ...agentSessionFailureWords(agentSessionFailureFact('cancelled'), { surface: 'rejection' }), + answeredInTurn: TURN, + fence: 1 + }) + return journal +} + +describe('the turn a rejected submission was answered into', () => { + it('is the turn record its rejection named, on the snapshot, a replay, and the page', async () => { + const journal = await sendWithdrawnByItsTurnEnd() + const turnItemId = agentJournalItemKey(TURN) + + expect(journal.submission('send-1')).toMatchObject({ + dispatchState: 'rejected', + answeredInTurnItemId: turnItemId + }) + expect(readAgentSessionHydrationPage(journal).submissions).toEqual([ + expect.objectContaining({ answeredInTurnItemId: turnItemId }) + ]) + await journals.closeAll() + const replayed = await journals.open({ identity: IDENTITY, stateDirectory: root! }) + expect(replayed.submission('send-1')).toMatchObject({ answeredInTurnItemId: turnItemId }) + }) + + it('is absent on a take-back that names no turn, as on rows from older hosts', async () => { + const { journal } = await sendHandedOverThenWithdrawn() + + expect(journal.submission('send-1')).toMatchObject({ dispatchState: 'rejected' }) + expect(journal.submission('send-1')).not.toHaveProperty('answeredInTurnItemId') + expect(readAgentSessionHydrationPage(journal).submissions[0]).not.toHaveProperty( + 'answeredInTurnItemId' + ) + }) +}) + 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 } = diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-adapter.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-adapter.ts index 318bada1fe52..2aacb4d96333 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-adapter.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-adapter.ts @@ -176,8 +176,12 @@ export type AgentSessionDispatchOutcome = * anything and never promotes this to `unknown`. */ | { state: 'admitted' } - /** Words from `agentSessionFailureWords`, never written by hand. */ - | ({ state: 'rejected' } & AgentJournalDispatchRejection) + /** Words from `agentSessionFailureWords`, never written by hand. `answeredInTurn`: the turn the + * provider answered the send into, which ended before the answer was read. */ + | ({ + state: 'rejected' + answeredInTurn?: AgentJournalItemIdentity + } & AgentJournalDispatchRejection) /** The call did not settle. Never re-send on the user's behalf. */ | { state: 'unknown'; reason: string } diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-late-dispatch.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-late-dispatch.ts index ba0ab9231261..4384d5fd5610 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-late-dispatch.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-late-dispatch.ts @@ -12,7 +12,10 @@ export async function settleStructuredAgentSessionLateDispatch( clientMessageId: string } & ( | { providerIdentity: AgentJournalItemIdentity } - | ({ state: 'rejected' } & AgentJournalDispatchRejection) + | ({ + state: 'rejected' + answeredInTurn?: AgentJournalItemIdentity + } & AgentJournalDispatchRejection) | { state: 'unknown'; reason: string } ) ): Promise { @@ -43,6 +46,7 @@ export async function settleStructuredAgentSessionLateDispatch( state: 'rejected', reason: input.reason, rejection: input.rejection, + ...(input.answeredInTurn ? { answeredInTurn: input.answeredInTurn } : {}), fence } ) 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 629d7e717903..a291943fc0fd 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 @@ -231,6 +231,7 @@ export async function handOverSubmission( state: 'rejected', reason: outcome.reason, rejection: outcome.rejection, + ...(outcome.answeredInTurn ? { answeredInTurn: outcome.answeredInTurn } : {}), fence: ctx.fence } : { clientMessageId, state: 'unknown', reason: outcome.reason, fence: ctx.fence } diff --git a/src/main/runtime/structured-agent-session-codex-turn-end-settlement.test.ts b/src/main/runtime/structured-agent-session-codex-turn-end-settlement.test.ts index 57e632a6c85c..ccc922a70e56 100644 --- a/src/main/runtime/structured-agent-session-codex-turn-end-settlement.test.ts +++ b/src/main/runtime/structured-agent-session-codex-turn-end-settlement.test.ts @@ -304,6 +304,79 @@ describe('a Codex send its turn ended without taking it', () => { }) }) +describe('the turn a withdrawn Codex send was answered into', () => { + async function answeredInto(clientMessageId: string) { + await host.flushStreamedEvents(SESSION) + const snapshot = await host.journalSnapshot(SESSION) + const page = await host.history({ sessionId: SESSION, direction: 'tail' }) + const onPage = page.ok + ? page.page.submissions.find((entry) => entry.clientMessageId === clientMessageId) + : undefined + return { + turnRecords: snapshot.items.flatMap((item) => + item.body.kind === 'turn' ? [item.itemId] : [] + ), + named: snapshot.submissions.find((entry) => entry.clientMessageId === clientMessageId) + ?.answeredInTurnItemId, + onPage: onPage?.answeredInTurnItemId + } + } + + it("is that turn's record, for the send that opened it and for one steered into it", async () => { + const opening = await send('look around') + await vi.waitFor(() => expect(answers).toBe(1)) + turns.start() + const steered = await send('and check the tests') + await vi.waitFor(() => expect(steers).toBe(1)) + + await stop('turn-1') + await vi.waitFor(async () => + expect(verdictOf((await settled()).submissions, steered)).toBe('withdrawn') + ) + + const { turnRecords } = await answeredInto(opening) + expect(turnRecords).toHaveLength(1) + expect(await answeredInto(opening)).toMatchObject({ + named: turnRecords[0], + onPage: turnRecords[0] + }) + expect(await answeredInto(steered)).toMatchObject({ + named: turnRecords[0], + onPage: turnRecords[0] + }) + }) + + it('is not named on a send the turn took', async () => { + const opening = await send('look around') + await vi.waitFor(() => expect(answers).toBe(1)) + turns.start() + turns.echo(opening) + + await stop('turn-1') + await vi.waitFor(async () => + expect(verdictOf((await settled()).submissions, opening)).toBe('accepted') + ) + + expect(await answeredInto(opening)).toMatchObject({ named: undefined, onPage: undefined }) + }) + + it('is named when the answer is read after that turn ended', async () => { + const release = turns.holdNextAnswer() + const sent = await send('look around') + await vi.waitFor(() => expect(turns.turnId).toBe('turn-1')) + turns.start() + turns.end('interrupted') + release() + + await vi.waitFor(async () => + expect(verdictOf((await settled()).submissions, sent)).toBe('withdrawn') + ) + const { turnRecords, named } = await answeredInto(sent) + expect(turnRecords).toHaveLength(1) + expect(named).toBe(turnRecords[0]) + }) +}) + describe('a queued card sent now into the turn a Stop ends', () => { async function handoffs(messageId: string): Promise { return (await settled()).submissions.filter((entry) => entry.queuedMessageId === messageId) diff --git a/src/shared/agent-session-journal-schemas.ts b/src/shared/agent-session-journal-schemas.ts index 8a7a51493675..00d88b74de56 100644 --- a/src/shared/agent-session-journal-schemas.ts +++ b/src/shared/agent-session-journal-schemas.ts @@ -319,6 +319,7 @@ export const AgentJournalSubmissionSchema = z.object({ submittedAt: z.number(), resolvedAt: z.number().nullable(), submittedSequence: z.number().int().optional(), + answeredInTurnItemId: z.string().min(1).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 28e470cb869e..4e21ee5f2e86 100644 --- a/src/shared/agent-session-journal-types.ts +++ b/src/shared/agent-session-journal-types.ts @@ -415,6 +415,10 @@ export type AgentJournalSubmission = { * 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 + /** On `rejected`: the item id of the turn record a Codex send was answered into, when that turn + * ended without taking it. Absent on every other send: accepted ones (the echo places them), + * Claude, queued take-backs, restart recovery, and rows from hosts that predate it. */ + answeredInTurnItemId?: string /** 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 ef5d1eba2ec7..46ddf9715dea 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 position: it must fold and draw a page -// carrying it exactly as it draws the same page without it. +// A released client that predates the published submission position and answered turn: 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 = { @@ -34,16 +34,19 @@ function releaseExport(module: Record, name: string): T { type OldState = { submissions: Record[]; items: unknown[] } -/** Strips the published position, as an older host's page would carry its submissions. */ +/** Strips the published position and answered turn, as an older host's page would carry its + * submissions. */ function withoutPosition(page: AgentSessionHistoryPage): AgentSessionHistoryPage { return { ...page, - submissions: page.submissions.map(({ submittedSequence: _submitted, ...rest }) => rest) + submissions: page.submissions.map( + ({ submittedSequence: _submitted, answeredInTurnItemId: _turn, ...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 position', async () => { +test('a released client folds and draws a page whose submissions carry their journal position and answered turn', async () => { const directory = mkdtempSync(join(tmpdir(), 'orca-submission-positions-downgrade-')) const journals = createTrackedJournalOpener() try { @@ -64,13 +67,34 @@ test('a released client folds and draws a page whose submissions carry their jou fence: 1, recovered: true }) + await journal.appendSubmission({ + clientMessageId: 'turn-ended', + payloadFingerprint: 'turn-ended', + body: { kind: 'message', role: 'user', blocks: [{ type: 'text', text: 'turn-ended' }] }, + fence: 1 + }) + await journal.resolveDispatch({ + clientMessageId: 'turn-ended', + state: 'rejected', + ...agentSessionFailureWords(agentSessionFailureFact('cancelled'), { surface: 'rejection' }), + answeredInTurn: { + provider: 'legacy', + agent: 'codex', + sessionId: IDENTITY.sessionId, + recordId: 'turn-lifecycle:turn-1' + }, + fence: 1 + }) const page = readAgentSessionHydrationPage(journal, 1) - // Anti-vacuous: the page this build publishes does carry the position. + // Anti-vacuous: the page this build publishes does carry both. expect(page.submissions.find((entry) => entry.clientMessageId === 'taken-back')).toEqual( expect.objectContaining({ submittedSequence: expect.any(Number) }) ) + expect(page.submissions.find((entry) => entry.clientMessageId === 'turn-ended')).toEqual( + expect.objectContaining({ answeredInTurnItemId: expect.any(String) }) + ) const checkout = await materializeReleaseCheckout(BASELINE_REF) const reducer = await importReleaseCheckoutModule( @@ -92,7 +116,9 @@ 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, ...rest }) => rest), + submissions: state.submissions.map( + ({ submittedSequence: _s, answeredInTurnItemId: _t, ...rest }) => rest + ), messages: project(state.items, [], state.submissions) } } From ab2826ee1854079b0f46640b535cb5d0301af7a9 Mon Sep 17 00:00:00 2001 From: Brennan Benson <79079362+brennanb2025@users.noreply.github.com> Date: Mon, 5 Oct 2026 09:02:33 -0700 Subject: [PATCH 5/8] feat(native-chat): say how a withdrawn Codex send joined the turn it names The answered turn becomes { turnItemId, via }: 'start' when Codex answered the send's turn/start with that turn, 'steer' when Orca steered it into that running turn. A client can then tell the turn's opener from a send steered into it even when the opener is on an older page. --- .../codex-structured-dispatch-echo.test.ts | 14 ++++---- .../codex/codex-structured-dispatch-echo.ts | 36 +++++++++++++------ .../codex/codex-structured-session-state.ts | 3 +- ...dex-structured-turn-end-settlement.test.ts | 29 ++++++++++----- .../codex-structured-turn-end-settlement.ts | 16 +++++---- src/main/codex/codex-structured-turn-start.ts | 27 +++++++++----- .../journal-dispatch-reducer.ts | 25 +++++++++---- .../journal-row-builders.ts | 7 +++- .../journal-row-schema.ts | 9 ++--- .../journal-store-contracts.ts | 3 +- .../journal-submission-positions.test.ts | 14 ++++---- .../structured-agent-session-adapter.ts | 3 +- .../structured-agent-session-late-dispatch.ts | 7 ++-- ...-session-codex-turn-end-settlement.test.ts | 22 ++++++------ src/shared/agent-session-journal-schemas.ts | 3 +- src/shared/agent-session-journal-types.ts | 20 ++++++++--- ...ubmission-positions-downgrade.unit.test.ts | 19 ++++++---- 17 files changed, 171 insertions(+), 86 deletions(-) diff --git a/src/main/codex/codex-structured-dispatch-echo.test.ts b/src/main/codex/codex-structured-dispatch-echo.test.ts index c7eb43bf96f4..1eeb8186ad98 100644 --- a/src/main/codex/codex-structured-dispatch-echo.test.ts +++ b/src/main/codex/codex-structured-dispatch-echo.test.ts @@ -133,9 +133,9 @@ describe('the turn Codex answered a send into but has not opened', () => { it('is the turn the latest armed send was answered into', () => { const echoes = createCodexDispatchEchoes() echoes.arm('client-1') - echoes.bindTurn('client-1', 'thread-1', 'turn-1') + echoes.bindTurn('client-1', 'thread-1', 'turn-1', 'start') echoes.arm('client-2') - echoes.bindTurn('client-2', 'thread-1', 'turn-2') + echoes.bindTurn('client-2', 'thread-1', 'turn-2', 'start') expect(echoes.answeredUnopenedTurn('thread-1', NONE_OPEN)).toBe('turn-2') }) @@ -143,7 +143,7 @@ describe('the turn Codex answered a send into but has not opened', () => { it('is none once Codex opened that turn', () => { const echoes = createCodexDispatchEchoes() echoes.arm('client-1') - echoes.bindTurn('client-1', 'thread-1', 'turn-1') + echoes.bindTurn('client-1', 'thread-1', 'turn-1', 'start') expect(echoes.answeredUnopenedTurn('thread-1', new Set(['turn-1']))).toBeNull() }) @@ -151,7 +151,7 @@ describe('the turn Codex answered a send into but has not opened', () => { it('is none once that turn ended, even with its send still armed for an echo', () => { const echoes = createCodexDispatchEchoes() echoes.arm('client-1') - echoes.bindTurn('client-1', 'thread-1', 'turn-1') + echoes.bindTurn('client-1', 'thread-1', 'turn-1', 'start') // A completed end leaves its unechoed send armed. expect(echoes.endTurn('thread-1', 'turn-1', { status: 'completed' })).toEqual([]) @@ -161,9 +161,9 @@ describe('the turn Codex answered a send into but has not opened', () => { it('skips a turn a wait left unopened, and still names an earlier one', () => { const echoes = createCodexDispatchEchoes() echoes.arm('client-1') - echoes.bindTurn('client-1', 'thread-1', 'turn-1') + echoes.bindTurn('client-1', 'thread-1', 'turn-1', 'start') echoes.arm('client-2') - echoes.bindTurn('client-2', 'thread-1', 'turn-2') + echoes.bindTurn('client-2', 'thread-1', 'turn-2', 'start') echoes.leftUnopened('thread-1', 'turn-2') @@ -174,7 +174,7 @@ describe('the turn Codex answered a send into but has not opened', () => { const echoes = createCodexDispatchEchoes() echoes.arm('client-1') echoes.arm('client-2') - echoes.bindTurn('client-2', 'thread-2', 'turn-2') + echoes.bindTurn('client-2', 'thread-2', 'turn-2', 'start') expect(echoes.answeredUnopenedTurn('thread-1', NONE_OPEN)).toBeNull() }) diff --git a/src/main/codex/codex-structured-dispatch-echo.ts b/src/main/codex/codex-structured-dispatch-echo.ts index 775cac5bca4f..e72cb3c63a96 100644 --- a/src/main/codex/codex-structured-dispatch-echo.ts +++ b/src/main/codex/codex-structured-dispatch-echo.ts @@ -1,5 +1,8 @@ import type { ProviderDiagnostic } from '../../shared/agent-session-failure' -import type { AgentJournalItemIdentity } from '../../shared/agent-session-journal-types' +import type { + AgentJournalItemIdentity, + AgentJournalTurnJoin +} from '../../shared/agent-session-journal-types' /** Sends awaiting their echo. One bound to a turn that ended without taking it settles from that * end; any other whose echo never arrives is retired by the journal's recovery on exit. */ @@ -35,10 +38,15 @@ export type CodexDispatchEchoes = { /** Drops an armed send whose write never reached the provider. */ disarm: (clientMessageId: string) => void /** - * Binds a send to the turn Codex answered it into. Returns that turn's end when the answer is - * read after it; a send that end settles is no longer armed. + * Binds a send to the turn Codex answered it into, and how it joined that turn. Returns that + * turn's end when the answer is read after it; a send that end settles is no longer armed. */ - bindTurn: (clientMessageId: string, threadId: string, turnId: string) => CodexTurnEnd | null + bindTurn: ( + clientMessageId: string, + threadId: string, + turnId: string, + via: AgentJournalTurnJoin + ) => CodexTurnEnd | null /** The turn the latest armed send was answered into that is neither in `openTurnIds`, ended, nor * left unopened through a wait: one Codex has picked for the send but not opened. */ answeredUnopenedTurn: (threadId: string, openTurnIds: ReadonlySet) => string | null @@ -49,7 +57,11 @@ export type CodexDispatchEchoes = { * completed, which echoes its pending input first, so one it never echoed waits for recovery. An * interrupt withdraws an un-echoed send, steered or the turn's own input: neither reached history. */ - endTurn: (threadId: string, turnId: string, end: CodexTurnEnd) => string[] + endTurn: ( + threadId: string, + turnId: string, + end: CodexTurnEnd + ) => { clientMessageId: string; via: AgentJournalTurnJoin }[] /** Submission origin for this exact send, retained until its echo settles it. */ requestOrigin: (clientMessageId: string) => CodexDispatchRequestOrigin | null /** Highest causal sequence assigned to a dispatch in this session. */ @@ -61,7 +73,11 @@ export type CodexDispatchEchoes = { export function createCodexDispatchEchoes(): CodexDispatchEchoes { const armed = new Map< string, - { requestedAt: number | null; sequence: number; turn?: { threadId: string; turnId: string } } + { + requestedAt: number | null + sequence: number + turn?: { threadId: string; turnId: string; via: AgentJournalTurnJoin } + } >() const endedTurns = new Map() const unopenedTurns = new Set() @@ -85,12 +101,12 @@ export function createCodexDispatchEchoes(): CodexDispatchEchoes { }, settle: (clientMessageId) => armed.delete(clientMessageId), disarm: (clientMessageId) => void armed.delete(clientMessageId), - bindTurn: (clientMessageId, threadId, turnId) => { + bindTurn: (clientMessageId, threadId, turnId, via) => { const entry = armed.get(clientMessageId) if (!entry) { return null } - entry.turn = { threadId, turnId } + entry.turn = { threadId, turnId, via } const end = endedTurns.get(turnKey(threadId, turnId)) ?? null if (end && settles(end)) { armed.delete(clientMessageId) @@ -132,10 +148,10 @@ export function createCodexDispatchEchoes(): CodexDispatchEchoes { } const settled = [...armed].flatMap(([clientMessageId, entry]) => entry.turn && turnKey(entry.turn.threadId, entry.turn.turnId) === turn - ? [clientMessageId] + ? [{ clientMessageId, via: entry.turn.via }] : [] ) - for (const clientMessageId of settled) { + for (const { clientMessageId } of settled) { armed.delete(clientMessageId) } return settled diff --git a/src/main/codex/codex-structured-session-state.ts b/src/main/codex/codex-structured-session-state.ts index 277a2719673a..4aa6485b1bfc 100644 --- a/src/main/codex/codex-structured-session-state.ts +++ b/src/main/codex/codex-structured-session-state.ts @@ -1,4 +1,5 @@ import type { + AgentJournalAnsweredTurnIdentity, AgentJournalItemIdentity, AgentSessionJournalIdentity } from '../../shared/agent-session-journal-types' @@ -86,7 +87,7 @@ export type CodexStructuredSessionAdapterDeps = { | { providerIdentity: AgentJournalItemIdentity } | ({ state: 'rejected' - answeredInTurn: AgentJournalItemIdentity + answeredInTurn: AgentJournalAnsweredTurnIdentity } & AgentJournalDispatchRejection) ) ) => void diff --git a/src/main/codex/codex-structured-turn-end-settlement.test.ts b/src/main/codex/codex-structured-turn-end-settlement.test.ts index ccb9fa7ba456..2b6542a43336 100644 --- a/src/main/codex/codex-structured-turn-end-settlement.test.ts +++ b/src/main/codex/codex-structured-turn-end-settlement.test.ts @@ -317,7 +317,10 @@ describe('the turn a withdrawn Codex send names', () => { expect(rig.turnRecords.length).toBeGreaterThan(0) expect(new Set(rig.turnRecords.map(agentJournalItemKey)).size).toBe(1) expect(rig.settlements).toEqual([ - expect.objectContaining({ clientMessageId: 'client-1', answeredInTurn: rig.turnRecords[0] }) + expect.objectContaining({ + clientMessageId: 'client-1', + answeredInTurn: { turn: rig.turnRecords[0], via: 'start' } + }) ]) }) @@ -329,8 +332,14 @@ describe('the turn a withdrawn Codex send names', () => { rig.turns.end('interrupted') expect(rig.settlements.map((settlement) => [settlement.clientMessageId, settlement])).toEqual([ - ['client-1', expect.objectContaining({ answeredInTurn: rig.turnRecords[0] })], - ['client-2', expect.objectContaining({ answeredInTurn: rig.turnRecords[0] })] + [ + 'client-1', + expect.objectContaining({ answeredInTurn: { turn: rig.turnRecords[0], via: 'start' } }) + ], + [ + 'client-2', + expect.objectContaining({ answeredInTurn: { turn: rig.turnRecords[0], via: 'steer' } }) + ] ]) }) @@ -345,7 +354,7 @@ describe('the turn a withdrawn Codex send names', () => { await expect(sending).resolves.toMatchObject({ state: 'rejected', - answeredInTurn: rig.turnRecords[0] + answeredInTurn: { turn: rig.turnRecords[0], via: 'start' } }) }) @@ -369,9 +378,11 @@ describe('a send bound to a turn', () => { it('dies with the settlement its turn end makes', () => { const echoes = createCodexDispatchEchoes() echoes.arm('client-1') - echoes.bindTurn('client-1', 'thread-1', 'turn-1') + echoes.bindTurn('client-1', 'thread-1', 'turn-1', 'start') - expect(echoes.endTurn('thread-1', 'turn-1', { status: 'interrupted' })).toEqual(['client-1']) + expect(echoes.endTurn('thread-1', 'turn-1', { status: 'interrupted' })).toEqual([ + { clientMessageId: 'client-1', via: 'start' } + ]) expect(echoes.size).toBe(0) expect(echoes.settle('client-1')).toBe(false) }) @@ -379,20 +390,20 @@ describe('a send bound to a turn', () => { it('dies with its child, which forgets recorded turn ends too', () => { const echoes = createCodexDispatchEchoes() echoes.arm('client-1') - echoes.bindTurn('client-1', 'thread-1', 'turn-1') + echoes.bindTurn('client-1', 'thread-1', 'turn-1', 'start') echoes.endTurn('thread-2', 'turn-2', { status: 'interrupted' }) echoes.clear() expect(echoes.size).toBe(0) echoes.arm('client-2') - expect(echoes.bindTurn('client-2', 'thread-2', 'turn-2')).toBeNull() + expect(echoes.bindTurn('client-2', 'thread-2', 'turn-2', 'start')).toBeNull() }) it('is matched by thread as well as turn id', () => { const echoes = createCodexDispatchEchoes() echoes.arm('client-1') - echoes.bindTurn('client-1', 'thread-1', 'turn-1') + echoes.bindTurn('client-1', 'thread-1', 'turn-1', 'start') expect(echoes.endTurn('thread-2', 'turn-1', { status: 'interrupted' })).toEqual([]) expect(echoes.size).toBe(1) diff --git a/src/main/codex/codex-structured-turn-end-settlement.ts b/src/main/codex/codex-structured-turn-end-settlement.ts index 931f4d41a702..376d9adb6e1f 100644 --- a/src/main/codex/codex-structured-turn-end-settlement.ts +++ b/src/main/codex/codex-structured-turn-end-settlement.ts @@ -17,7 +17,7 @@ import { agentSessionFailureWords, type AgentJournalDispatchRejection } from '../../shared/agent-session-failure-words' -import type { AgentJournalItemIdentity } from '../../shared/agent-session-journal-types' +import type { AgentJournalAnsweredTurnIdentity } from '../../shared/agent-session-journal-types' import type { CodexTurnEnd } from './codex-structured-dispatch-echo' import { codexTurnLifecycleIdentity } from './codex-structured-journal-translation-turns' import type { CodexSession } from './codex-structured-session-state' @@ -43,8 +43,8 @@ export function codexDispatchRejection( export type CodexTurnEndSettlement = { clientMessageId: string state: 'rejected' - /** The turn Codex answered the send into, whose end settled it. */ - answeredInTurn: AgentJournalItemIdentity + /** The turn Codex answered the send into, whose end settled it, and how the send joined it. */ + answeredInTurn: AgentJournalAnsweredTurnIdentity } & AgentJournalDispatchRejection function errorDetail(params: unknown): ProviderDiagnostic | undefined { @@ -97,10 +97,14 @@ export function settleCodexSendsInEndedTurn( return } const rejection = codexTurnEndRejection(end) - const answeredInTurn = codexTurnLifecycleIdentity(frame.sessionId, turnId) - for (const clientMessageId of session.dispatchEchoes.endTurn(session.threadId, turnId, end)) { + const turn = codexTurnLifecycleIdentity(frame.sessionId, turnId) + for (const { clientMessageId, via } of session.dispatchEchoes.endTurn( + session.threadId, + turnId, + end + )) { if (rejection) { - settle({ clientMessageId, state: 'rejected', answeredInTurn, ...rejection }) + settle({ clientMessageId, state: 'rejected', answeredInTurn: { turn, via }, ...rejection }) } } } diff --git a/src/main/codex/codex-structured-turn-start.ts b/src/main/codex/codex-structured-turn-start.ts index 0aace8ee54b3..271def876076 100644 --- a/src/main/codex/codex-structured-turn-start.ts +++ b/src/main/codex/codex-structured-turn-start.ts @@ -1,5 +1,8 @@ import { agentSessionFailureFact, providerDiagnosticOf } from '../../shared/agent-session-failure' -import type { AgentJournalMessageItem } from '../../shared/agent-session-journal-types' +import type { + AgentJournalMessageItem, + AgentJournalTurnJoin +} from '../../shared/agent-session-journal-types' import type { NativeChatBlock } from '../../shared/native-chat-types' import type { AgentSessionDispatchOutcome } from '../native-chat/agent-session-wire/structured-agent-session-adapter' import { @@ -108,7 +111,7 @@ async function steerCodexTurn( host: CodexTurnHost, expectedTurnId: string, input: { clientMessageId: string; body: AgentJournalMessageItem; timeoutMs?: number } -): Promise<{ turnId: string } | null> { +): Promise<{ turnId: string; via: 'steer' } | null> { try { const answer = await host.connection.request( 'turn/steer', @@ -120,7 +123,7 @@ async function steerCodexTurn( }, { timeoutMs: input.timeoutMs } ) - return { turnId: readCodexTurnId(answer) ?? expectedTurnId } + return { turnId: readCodexTurnId(answer) ?? expectedTurnId, via: 'steer' } } catch (error) { if (isCodexAppServerRequestError(error) || isCodexAppServerUnsupportedError(error)) { return null @@ -142,7 +145,7 @@ export async function startCodexTurn( requestedAt?: number timeoutMs?: number } -): Promise<{ turnId: string | null } | false> { +): Promise<{ turnId: string | null; via: AgentJournalTurnJoin } | false> { // Armed before the write: the echo and `turn/started` can both land while the // response is in flight, and the start must snapshot this send in its frontier. if (!host.dispatchEchoes.arm(input.clientMessageId, input.requestedAt)) { @@ -168,7 +171,7 @@ export async function startCodexTurn( }, { timeoutMs: input.timeoutMs } ) - return { turnId: readCodexTurnId(answer) } + return { turnId: readCodexTurnId(answer), via: 'start' } } /** @@ -187,7 +190,7 @@ export async function dispatchCodexTurn( }, timeoutMs: number | undefined ): Promise { - let answer: { turnId: string | null } | false + let answer: { turnId: string | null; via: AgentJournalTurnJoin } | false try { answer = await startCodexTurn(session, { ...input, timeoutMs }) } catch (error) { @@ -211,13 +214,21 @@ export async function dispatchCodexTurn( } // An answer read after the turn it names already ended is settled by that end. const endedFirst = answer.turnId - ? session.dispatchEchoes.bindTurn(input.clientMessageId, session.threadId, answer.turnId) + ? session.dispatchEchoes.bindTurn( + input.clientMessageId, + session.threadId, + answer.turnId, + answer.via + ) : null const rejection = endedFirst ? codexTurnEndRejection(endedFirst) : null return rejection && answer.turnId ? { state: 'rejected', - answeredInTurn: codexTurnLifecycleIdentity(input.sessionId, answer.turnId), + answeredInTurn: { + turn: codexTurnLifecycleIdentity(input.sessionId, answer.turnId), + via: answer.via + }, ...rejection } : { state: 'admitted' } 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 c2f7d45327a5..b19fa2f1ce9f 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 @@ -6,6 +6,7 @@ import { type UnreadAgentSessionFailureFact } from '../../../shared/agent-session-failure' import { agentJournalSubmissionKey } from '../../../shared/agent-session-journal-item-key' +import type { AgentJournalAnsweredTurn } from '../../../shared/agent-session-journal-types' import { journalDispatchRowApplies } from './journal-dispatch-settlement' import type { JournalReducerState } from './journal-reducer' import { @@ -34,14 +35,11 @@ export function applyJournalDispatchRow( } else { delete submission.rejection } - if ( - row.state === 'rejected' && - typeof row.answeredInTurnItemId === 'string' && - row.answeredInTurnItemId - ) { - submission.answeredInTurnItemId = row.answeredInTurnItemId + const answeredInTurn = row.state === 'rejected' ? readAnsweredTurn(row.answeredInTurn) : undefined + if (answeredInTurn) { + submission.answeredInTurn = answeredInTurn } else { - delete submission.answeredInTurnItemId + delete submission.answeredInTurn } submission.resolvedAt = row.state === 'pending' ? null : row.ts if (row.state === 'pending') { @@ -70,6 +68,19 @@ export function applyJournalDispatchRow( }) } +/** A stored answered turn; one malformed, or naming a way of joining this build does not know, is + * dropped, never the row. */ +function readAnsweredTurn(value: unknown): AgentJournalAnsweredTurn | undefined { + if (typeof value !== 'object' || value === null) { + return undefined + } + const turnItemId = 'turnItemId' in value ? value.turnItemId : undefined + const via = 'via' in value ? value.via : undefined + return typeof turnItemId === 'string' && turnItemId && (via === 'start' || via === 'steer') + ? { turnItemId, via } + : undefined +} + /** A stored rejection fact, read where it can be placed; a kind it cannot place is kept as * written, so the classifier still knows a fact was there without this build claiming what it * says. Shared with the queued-draft table, whose returned card mirrors its submission. */ diff --git a/src/main/native-chat/agent-session-journal/journal-row-builders.ts b/src/main/native-chat/agent-session-journal/journal-row-builders.ts index 6873db0fd46b..4484f26a93f1 100644 --- a/src/main/native-chat/agent-session-journal/journal-row-builders.ts +++ b/src/main/native-chat/agent-session-journal/journal-row-builders.ts @@ -111,7 +111,12 @@ export function journalDispatchRowBuilder( reason: boundedDispatchReason(input), ...(input.state === 'rejected' ? { rejection: input.rejection } : {}), ...(input.state === 'rejected' && input.answeredInTurn - ? { answeredInTurnItemId: agentJournalItemKey(input.answeredInTurn) } + ? { + answeredInTurn: { + turnItemId: agentJournalItemKey(input.answeredInTurn.turn), + via: input.answeredInTurn.via + } + } : {}), ...journalRowBase(state().epoch, seq, input.fence, ts), ...(input.recovered ? { recovered: input.recovered } : {}), diff --git a/src/main/native-chat/agent-session-journal/journal-row-schema.ts b/src/main/native-chat/agent-session-journal/journal-row-schema.ts index 194290542d93..4957bd846be9 100644 --- a/src/main/native-chat/agent-session-journal/journal-row-schema.ts +++ b/src/main/native-chat/agent-session-journal/journal-row-schema.ts @@ -16,6 +16,7 @@ import type { AgentSessionFailureFact } from '../../../shared/agent-session-failure' import { AGENT_SESSION_JOURNAL_SCHEMA_VERSION, + type AgentJournalAnsweredTurn, type AgentJournalDispatchState, type AgentJournalItemBody, type AgentJournalMessageItem, @@ -150,10 +151,10 @@ export type JournalDispatchRow = JournalRowBase & { /** On `rejected`: why, typed. Older readers keep the key and ignore it; a malformed one is * dropped when read, never the row. */ rejection?: AgentSessionFailureFact - /** On `rejected`: the turn record a Codex send was answered into (`turn/start` or `turn/steer`) - * when that turn's end settled the send. Absent on every other row. Older readers keep the key - * and ignore it. */ - answeredInTurnItemId?: string + /** On `rejected`: the turn a Codex send was answered into, and how it joined it, when that + * turn's end settled the send. Absent on every other row. Older readers keep the key and ignore + * it; a malformed one is dropped when read, never the row. */ + answeredInTurn?: AgentJournalAnsweredTurn } /** An item mutation may name its own producer, because one batch can CREATE 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 index b2d2914b318b..4232e805bbe4 100644 --- a/src/main/native-chat/agent-session-journal/journal-store-contracts.ts +++ b/src/main/native-chat/agent-session-journal/journal-store-contracts.ts @@ -1,5 +1,6 @@ import type { AgentJournalDispatchRejection } from '../../../shared/agent-session-failure-words' import type { + AgentJournalAnsweredTurnIdentity, AgentJournalCursor, AgentJournalItemBody, AgentJournalItemIdentity, @@ -42,7 +43,7 @@ export type ResolveDispatchInput = { * `agentSessionFailureWords`, never written by hand. */ | ({ state: 'rejected' - answeredInTurn?: AgentJournalItemIdentity + answeredInTurn?: AgentJournalAnsweredTurnIdentity } & AgentJournalDispatchRejection) | { state: 'unknown'; reason?: string | null } ) 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 31cfee3a25c9..2ad8953ea030 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 @@ -94,7 +94,7 @@ async function sendWithdrawnByItsTurnEnd() { clientMessageId: 'send-1', state: 'rejected', ...agentSessionFailureWords(agentSessionFailureFact('cancelled'), { surface: 'rejection' }), - answeredInTurn: TURN, + answeredInTurn: { turn: TURN, via: 'start' }, fence: 1 }) return journal @@ -107,23 +107,25 @@ describe('the turn a rejected submission was answered into', () => { expect(journal.submission('send-1')).toMatchObject({ dispatchState: 'rejected', - answeredInTurnItemId: turnItemId + answeredInTurn: { turnItemId, via: 'start' } }) expect(readAgentSessionHydrationPage(journal).submissions).toEqual([ - expect.objectContaining({ answeredInTurnItemId: turnItemId }) + expect.objectContaining({ answeredInTurn: { turnItemId, via: 'start' } }) ]) await journals.closeAll() const replayed = await journals.open({ identity: IDENTITY, stateDirectory: root! }) - expect(replayed.submission('send-1')).toMatchObject({ answeredInTurnItemId: turnItemId }) + expect(replayed.submission('send-1')).toMatchObject({ + answeredInTurn: { turnItemId, via: 'start' } + }) }) it('is absent on a take-back that names no turn, as on rows from older hosts', async () => { const { journal } = await sendHandedOverThenWithdrawn() expect(journal.submission('send-1')).toMatchObject({ dispatchState: 'rejected' }) - expect(journal.submission('send-1')).not.toHaveProperty('answeredInTurnItemId') + expect(journal.submission('send-1')).not.toHaveProperty('answeredInTurn') expect(readAgentSessionHydrationPage(journal).submissions[0]).not.toHaveProperty( - 'answeredInTurnItemId' + 'answeredInTurn' ) }) }) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-adapter.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-adapter.ts index 2aacb4d96333..d08bf45b8d32 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-adapter.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-adapter.ts @@ -12,6 +12,7 @@ import type { import type { AgentSessionBackgroundTaskStops } from '../../../shared/agent-child-work-stop-targets' import type { + AgentJournalAnsweredTurnIdentity, AgentJournalItemIdentity, AgentJournalItemBody, AgentJournalMessageItem, @@ -180,7 +181,7 @@ export type AgentSessionDispatchOutcome = * provider answered the send into, which ended before the answer was read. */ | ({ state: 'rejected' - answeredInTurn?: AgentJournalItemIdentity + answeredInTurn?: AgentJournalAnsweredTurnIdentity } & AgentJournalDispatchRejection) /** The call did not settle. Never re-send on the user's behalf. */ | { state: 'unknown'; reason: string } diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-late-dispatch.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-late-dispatch.ts index 4384d5fd5610..3d38494fea93 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-late-dispatch.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-late-dispatch.ts @@ -1,4 +1,7 @@ -import type { AgentJournalItemIdentity } from '../../../shared/agent-session-journal-types' +import type { + AgentJournalAnsweredTurnIdentity, + AgentJournalItemIdentity +} from '../../../shared/agent-session-journal-types' import type { AgentJournalDispatchRejection } from '../../../shared/agent-session-failure-words' import type { StructuredAgentSessionMutationContext } from './structured-agent-session-host-mutations' import { structuredAgentSessionConversationFence } from './structured-agent-session-provider-child' @@ -14,7 +17,7 @@ export async function settleStructuredAgentSessionLateDispatch( | { providerIdentity: AgentJournalItemIdentity } | ({ state: 'rejected' - answeredInTurn?: AgentJournalItemIdentity + answeredInTurn?: AgentJournalAnsweredTurnIdentity } & AgentJournalDispatchRejection) | { state: 'unknown'; reason: string } ) diff --git a/src/main/runtime/structured-agent-session-codex-turn-end-settlement.test.ts b/src/main/runtime/structured-agent-session-codex-turn-end-settlement.test.ts index ccc922a70e56..0cd0570b83d2 100644 --- a/src/main/runtime/structured-agent-session-codex-turn-end-settlement.test.ts +++ b/src/main/runtime/structured-agent-session-codex-turn-end-settlement.test.ts @@ -317,12 +317,12 @@ describe('the turn a withdrawn Codex send was answered into', () => { item.body.kind === 'turn' ? [item.itemId] : [] ), named: snapshot.submissions.find((entry) => entry.clientMessageId === clientMessageId) - ?.answeredInTurnItemId, - onPage: onPage?.answeredInTurnItemId + ?.answeredInTurn, + onPage: onPage?.answeredInTurn } } - it("is that turn's record, for the send that opened it and for one steered into it", async () => { + it("is that turn's record, started by the send that opened it and steered by a later one", async () => { const opening = await send('look around') await vi.waitFor(() => expect(answers).toBe(1)) turns.start() @@ -336,13 +336,13 @@ describe('the turn a withdrawn Codex send was answered into', () => { const { turnRecords } = await answeredInto(opening) expect(turnRecords).toHaveLength(1) - expect(await answeredInto(opening)).toMatchObject({ - named: turnRecords[0], - onPage: turnRecords[0] - }) - expect(await answeredInto(steered)).toMatchObject({ - named: turnRecords[0], - onPage: turnRecords[0] + const started = { turnItemId: turnRecords[0], via: 'start' } + const steeredIn = { turnItemId: turnRecords[0], via: 'steer' } + expect(await answeredInto(opening)).toEqual({ turnRecords, named: started, onPage: started }) + expect(await answeredInto(steered)).toEqual({ + turnRecords, + named: steeredIn, + onPage: steeredIn }) }) @@ -373,7 +373,7 @@ describe('the turn a withdrawn Codex send was answered into', () => { ) const { turnRecords, named } = await answeredInto(sent) expect(turnRecords).toHaveLength(1) - expect(named).toBe(turnRecords[0]) + expect(named).toEqual({ turnItemId: turnRecords[0], via: 'start' }) }) }) diff --git a/src/shared/agent-session-journal-schemas.ts b/src/shared/agent-session-journal-schemas.ts index 00d88b74de56..7a21d5816a27 100644 --- a/src/shared/agent-session-journal-schemas.ts +++ b/src/shared/agent-session-journal-schemas.ts @@ -319,7 +319,8 @@ export const AgentJournalSubmissionSchema = z.object({ submittedAt: z.number(), resolvedAt: z.number().nullable(), submittedSequence: z.number().int().optional(), - answeredInTurnItemId: z.string().min(1).optional(), + // Open like `dispatchState`: a way of joining a newer host names must not drop the submission. + answeredInTurn: z.object({ turnItemId: z.string().min(1), via: z.string().min(1) }).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 4e21ee5f2e86..b6b33950de44 100644 --- a/src/shared/agent-session-journal-types.ts +++ b/src/shared/agent-session-journal-types.ts @@ -391,6 +391,18 @@ export type AgentJournalRenderItem = AgentJournalProducerLinkage & { // ─── Submissions ──────────────────────────────────────────────────────────── +/** The turn a send was answered into: its record's item id, and how the send joined it. `start`: + * the provider answered the send's start request with that turn; `steer`: Orca steered it into + * that running turn. Known limit: a start the provider silently folds into a running turn reads + * as `start`. Open: a newer host may name another way, which a reader leaves unclaimed. */ +export type AgentJournalAnsweredTurn = { turnItemId: string; via: AgentJournalTurnJoin } +export type AgentJournalTurnJoin = 'start' | 'steer' +/** The same, as a writer names it: the turn record's identity, keyed when the row is written. */ +export type AgentJournalAnsweredTurnIdentity = { + turn: AgentJournalItemIdentity + via: AgentJournalTurnJoin +} + export const AGENT_JOURNAL_DISPATCH_STATES = ['pending', 'accepted', 'rejected', 'unknown'] as const export type AgentJournalDispatchState = (typeof AGENT_JOURNAL_DISPATCH_STATES)[number] @@ -415,10 +427,10 @@ export type AgentJournalSubmission = { * 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 - /** On `rejected`: the item id of the turn record a Codex send was answered into, when that turn - * ended without taking it. Absent on every other send: accepted ones (the echo places them), - * Claude, queued take-backs, restart recovery, and rows from hosts that predate it. */ - answeredInTurnItemId?: string + /** On `rejected`: the turn a Codex send was answered into, when that turn ended without taking + * it. Absent on every other send: accepted ones (the echo places them), Claude, queued + * take-backs, restart recovery, and rows from hosts that predate it. */ + answeredInTurn?: AgentJournalAnsweredTurn /** 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 46ddf9715dea..5265cfb1af99 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 @@ -40,7 +40,7 @@ function withoutPosition(page: AgentSessionHistoryPage): AgentSessionHistoryPage return { ...page, submissions: page.submissions.map( - ({ submittedSequence: _submitted, answeredInTurnItemId: _turn, ...rest }) => rest + ({ submittedSequence: _submitted, answeredInTurn: _turn, ...rest }) => rest ) } } @@ -78,10 +78,13 @@ test('a released client folds and draws a page whose submissions carry their jou state: 'rejected', ...agentSessionFailureWords(agentSessionFailureFact('cancelled'), { surface: 'rejection' }), answeredInTurn: { - provider: 'legacy', - agent: 'codex', - sessionId: IDENTITY.sessionId, - recordId: 'turn-lifecycle:turn-1' + turn: { + provider: 'legacy', + agent: 'codex', + sessionId: IDENTITY.sessionId, + recordId: 'turn-lifecycle:turn-1' + }, + via: 'start' }, fence: 1 }) @@ -93,7 +96,9 @@ test('a released client folds and draws a page whose submissions carry their jou }) ) expect(page.submissions.find((entry) => entry.clientMessageId === 'turn-ended')).toEqual( - expect.objectContaining({ answeredInTurnItemId: expect.any(String) }) + expect.objectContaining({ + answeredInTurn: { turnItemId: expect.any(String), via: 'start' } + }) ) const checkout = await materializeReleaseCheckout(BASELINE_REF) @@ -117,7 +122,7 @@ test('a released client folds and draws a page whose submissions carry their jou const state = reduce(empty, { type: 'history-page', page: from }) return { submissions: state.submissions.map( - ({ submittedSequence: _s, answeredInTurnItemId: _t, ...rest }) => rest + ({ submittedSequence: _s, answeredInTurn: _t, ...rest }) => rest ), messages: project(state.items, [], state.submissions) } From 202138855d5cb2325eb168486ffa0e8976466c64 Mon Sep 17 00:00:00 2001 From: Brennan Benson <79079362+brennanb2025@users.noreply.github.com> Date: Mon, 5 Oct 2026 09:42:25 -0700 Subject: [PATCH 6/8] feat(native-chat): state on every rejection whether it names a turn A rejection a host writes now always carries answeredInTurn: the turn Codex answered the send into, or null for none. A reader can then tell 'answered into no turn' from a row written before the field existed, which keeps no answer and is placed by the older rules. --- .../journal-dispatch-answered-turn.test.ts | 78 ++++++ .../journal-dispatch-reducer.ts | 15 +- .../journal-row-builders.ts | 13 +- .../journal-row-schema.ts | 8 +- .../journal-submission-positions.test.ts | 14 +- src/shared/agent-session-journal-schemas.ts | 7 +- src/shared/agent-session-journal-types.ts | 10 +- ...ubmission-positions-downgrade.unit.test.ts | 247 +++++++++++------- 8 files changed, 273 insertions(+), 119 deletions(-) create mode 100644 src/main/native-chat/agent-session-journal/journal-dispatch-answered-turn.test.ts diff --git a/src/main/native-chat/agent-session-journal/journal-dispatch-answered-turn.test.ts b/src/main/native-chat/agent-session-journal/journal-dispatch-answered-turn.test.ts new file mode 100644 index 000000000000..bba802939679 --- /dev/null +++ b/src/main/native-chat/agent-session-journal/journal-dispatch-answered-turn.test.ts @@ -0,0 +1,78 @@ +// How a stored dispatch row's answered turn reads onto its submission: a named turn, none stated, +// or absent on a row written before the field existed. + +import { describe, expect, it } from 'vitest' +import { applyJournalRow, createJournalReducerState } from './journal-reducer' +import type { JournalRow } from './journal-row-schema' + +const EPOCH = 'epoch-1' + +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 } +} + +/** The submission after a send and one dispatch row, stored with the given extra keys. */ +function readAfter(dispatch: Partial>) { + const state = createJournalReducerState('session-1', EPOCH) + applyJournalRow(state, { + kind: 'submission', + clientMessageId: 'cm_1', + payloadFingerprint: 'fp_1', + providerHandle: { kind: 'codex', threadId: 'thread-1' }, + body: { kind: 'message', role: 'user', blocks: [{ type: 'text', text: 'hi' }] }, + ...base(1) + }) + applyJournalRow(state, { + kind: 'dispatch', + clientMessageId: 'cm_1', + state: 'rejected', + providerItemId: null, + reason: 'This message was withdrawn before the agent started it.', + ...base(2), + ...dispatch + }) + return state.submissions.get('cm_1') +} + +describe("a rejected dispatch row's answered turn", () => { + it('is the turn it names', () => { + expect(readAfter({ answeredInTurn: { turnItemId: 'turn-1', via: 'steer' } })).toMatchObject({ + answeredInTurn: { turnItemId: 'turn-1', via: 'steer' } + }) + }) + + it('is none when the row states none', () => { + expect(readAfter({ answeredInTurn: null })).toMatchObject({ answeredInTurn: null }) + }) + + it('is absent on a row written before the field existed', () => { + const submission = readAfter({}) + + expect(submission).toMatchObject({ dispatchState: 'rejected' }) + expect(submission).not.toHaveProperty('answeredInTurn') + }) + + it('reads as none, keeping the row, when it names a way of joining this build does not know', () => { + expect(readAfter({ answeredInTurn: { turnItemId: 'turn-1', via: 'resume' } })).toMatchObject({ + dispatchState: 'rejected', + answeredInTurn: null + }) + }) + + it('reads as none, keeping the row, when it names no turn id', () => { + expect(readAfter({ answeredInTurn: { turnItemId: '', via: 'start' } })).toMatchObject({ + dispatchState: 'rejected', + answeredInTurn: null + }) + }) + + it('is absent on any other dispatch state', () => { + const submission = readAfter({ + state: 'unknown', + answeredInTurn: { turnItemId: 'turn-1', via: 'start' } + }) + + expect(submission).toMatchObject({ dispatchState: 'unknown' }) + expect(submission).not.toHaveProperty('answeredInTurn') + }) +}) 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 b19fa2f1ce9f..3625982a24fa 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,9 +35,8 @@ export function applyJournalDispatchRow( } else { delete submission.rejection } - const answeredInTurn = row.state === 'rejected' ? readAnsweredTurn(row.answeredInTurn) : undefined - if (answeredInTurn) { - submission.answeredInTurn = answeredInTurn + if (row.state === 'rejected' && row.answeredInTurn !== undefined) { + submission.answeredInTurn = readAnsweredTurn(row.answeredInTurn) } else { delete submission.answeredInTurn } @@ -68,17 +67,17 @@ export function applyJournalDispatchRow( }) } -/** A stored answered turn; one malformed, or naming a way of joining this build does not know, is - * dropped, never the row. */ -function readAnsweredTurn(value: unknown): AgentJournalAnsweredTurn | undefined { +/** A stored answered turn. One malformed, or naming a way of joining this build does not know, is + * read as no turn: it was written knowing the field, so it is not an older row. */ +function readAnsweredTurn(value: unknown): AgentJournalAnsweredTurn | null { if (typeof value !== 'object' || value === null) { - return undefined + return null } const turnItemId = 'turnItemId' in value ? value.turnItemId : undefined const via = 'via' in value ? value.via : undefined return typeof turnItemId === 'string' && turnItemId && (via === 'start' || via === 'steer') ? { turnItemId, via } - : undefined + : null } /** A stored rejection fact, read where it can be placed; a kind it cannot place is kept as diff --git a/src/main/native-chat/agent-session-journal/journal-row-builders.ts b/src/main/native-chat/agent-session-journal/journal-row-builders.ts index 4484f26a93f1..0ffaf89e0d99 100644 --- a/src/main/native-chat/agent-session-journal/journal-row-builders.ts +++ b/src/main/native-chat/agent-session-journal/journal-row-builders.ts @@ -110,12 +110,15 @@ export function journalDispatchRowBuilder( providerItemId, reason: boundedDispatchReason(input), ...(input.state === 'rejected' ? { rejection: input.rejection } : {}), - ...(input.state === 'rejected' && input.answeredInTurn + // Every rejection states its turn, null for none, so a reader tells it from an older row. + ...(input.state === 'rejected' ? { - answeredInTurn: { - turnItemId: agentJournalItemKey(input.answeredInTurn.turn), - via: input.answeredInTurn.via - } + answeredInTurn: input.answeredInTurn + ? { + turnItemId: agentJournalItemKey(input.answeredInTurn.turn), + via: input.answeredInTurn.via + } + : null } : {}), ...journalRowBase(state().epoch, seq, input.fence, ts), diff --git a/src/main/native-chat/agent-session-journal/journal-row-schema.ts b/src/main/native-chat/agent-session-journal/journal-row-schema.ts index 4957bd846be9..6ac0fb16cedd 100644 --- a/src/main/native-chat/agent-session-journal/journal-row-schema.ts +++ b/src/main/native-chat/agent-session-journal/journal-row-schema.ts @@ -16,7 +16,6 @@ import type { AgentSessionFailureFact } from '../../../shared/agent-session-failure' import { AGENT_SESSION_JOURNAL_SCHEMA_VERSION, - type AgentJournalAnsweredTurn, type AgentJournalDispatchState, type AgentJournalItemBody, type AgentJournalMessageItem, @@ -152,9 +151,10 @@ export type JournalDispatchRow = JournalRowBase & { * dropped when read, never the row. */ rejection?: AgentSessionFailureFact /** On `rejected`: the turn a Codex send was answered into, and how it joined it, when that - * turn's end settled the send. Absent on every other row. Older readers keep the key and ignore - * it; a malformed one is dropped when read, never the row. */ - answeredInTurn?: AgentJournalAnsweredTurn + * turn's end settled the send; null on every other rejection. Absent on other rows and on rows + * written before it. `via` stays a string: a newer build may write another. Older readers keep + * the key and ignore it; one this build cannot read is read as null, never dropping the row. */ + answeredInTurn?: { turnItemId: string; via: string } | null } /** An item mutation may name its own producer, because one batch can CREATE 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 2ad8953ea030..d8aa5b4870cc 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 @@ -119,14 +119,16 @@ describe('the turn a rejected submission was answered into', () => { }) }) - it('is absent on a take-back that names no turn, as on rows from older hosts', async () => { + it('is stated as none on a take-back that names no turn, so it reads apart from an older row', async () => { const { journal } = await sendHandedOverThenWithdrawn() - expect(journal.submission('send-1')).toMatchObject({ dispatchState: 'rejected' }) - expect(journal.submission('send-1')).not.toHaveProperty('answeredInTurn') - expect(readAgentSessionHydrationPage(journal).submissions[0]).not.toHaveProperty( - 'answeredInTurn' - ) + expect(journal.submission('send-1')).toMatchObject({ + dispatchState: 'rejected', + answeredInTurn: null + }) + expect(readAgentSessionHydrationPage(journal).submissions[0]).toMatchObject({ + answeredInTurn: null + }) }) }) diff --git a/src/shared/agent-session-journal-schemas.ts b/src/shared/agent-session-journal-schemas.ts index 7a21d5816a27..39ee0b45c5f7 100644 --- a/src/shared/agent-session-journal-schemas.ts +++ b/src/shared/agent-session-journal-schemas.ts @@ -319,8 +319,11 @@ export const AgentJournalSubmissionSchema = z.object({ submittedAt: z.number(), resolvedAt: z.number().nullable(), submittedSequence: z.number().int().optional(), - // Open like `dispatchState`: a way of joining a newer host names must not drop the submission. - answeredInTurn: z.object({ turnItemId: z.string().min(1), via: z.string().min(1) }).optional(), + // `via` is a string, as `dispatchState` is: a newer host may name another way of joining. + answeredInTurn: z + .object({ turnItemId: z.string().min(1), via: z.string().min(1) }) + .nullable() + .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 b6b33950de44..f6bf89421ef0 100644 --- a/src/shared/agent-session-journal-types.ts +++ b/src/shared/agent-session-journal-types.ts @@ -394,7 +394,8 @@ export type AgentJournalRenderItem = AgentJournalProducerLinkage & { /** The turn a send was answered into: its record's item id, and how the send joined it. `start`: * the provider answered the send's start request with that turn; `steer`: Orca steered it into * that running turn. Known limit: a start the provider silently folds into a running turn reads - * as `start`. Open: a newer host may name another way, which a reader leaves unclaimed. */ + * as `start`, including into a turn no user entry opened. A newer host may name another way, + * which a reader leaves unclaimed. */ export type AgentJournalAnsweredTurn = { turnItemId: string; via: AgentJournalTurnJoin } export type AgentJournalTurnJoin = 'start' | 'steer' /** The same, as a writer names it: the turn record's identity, keyed when the row is written. */ @@ -428,9 +429,10 @@ export type AgentJournalSubmission = { * where it was sent. Absent from hosts that predate it. */ submittedSequence?: number /** On `rejected`: the turn a Codex send was answered into, when that turn ended without taking - * it. Absent on every other send: accepted ones (the echo places them), Claude, queued - * take-backs, restart recovery, and rows from hosts that predate it. */ - answeredInTurn?: AgentJournalAnsweredTurn + * it. null: the host recorded that it was answered into no turn, as every other rejection is + * (Claude, a queued message taken back before handover, restart recovery). Absent: written + * before this field existed, or not rejected. */ + answeredInTurn?: AgentJournalAnsweredTurn | null /** 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 5265cfb1af99..0945d6a77284 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 @@ -4,15 +4,23 @@ 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 { + AgentJournalSubmission, + 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 { + createTrackedJournalOpener, + type TrackedJournalOpener +} from '../../../src/main/native-chat/agent-session-journal/journal-host-database-test-support' +import { projectJournalBatch } from '../../../src/main/native-chat/agent-session-wire/agent-session-journal-batch' 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 position and answered turn: it must fold -// and draw a page carrying them exactly as it draws the same page without them. -const BASELINE_REF = 'v1.4.219' +// Released clients that predate the published submission position and answered turn: each must +// fold and draw a history page and an incremental batch carrying them exactly as it draws the same +// without them, whether the turn is named or stated as none. +const BASELINE_REFS = ['v1.4.219', 'v1.4.220'] as const const IDENTITY: AgentSessionJournalIdentity = { sessionId: 'session-positions', @@ -34,103 +42,162 @@ function releaseExport(module: Record, name: string): T { type OldState = { submissions: Record[]; items: unknown[] } -/** Strips the published position and answered turn, as an older host's page would carry its - * submissions. */ -function withoutPosition(page: AgentSessionHistoryPage): AgentSessionHistoryPage { - return { - ...page, - submissions: page.submissions.map( - ({ submittedSequence: _submitted, answeredInTurn: _turn, ...rest }) => rest - ) - } +/** Submissions as an older host would carry them. */ +function withoutNewFields( + submissions: readonly AgentJournalSubmission[] +): AgentJournalSubmission[] { + return submissions.map( + ({ submittedSequence: _submitted, answeredInTurn: _turn, ...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 position and answered turn', 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' }), +/** A chat before and after three rejections: a queued send taken back, one its turn's end + * rejected, and one refused outright. */ +async function journalWithRejections(directory: string, journals: TrackedJournalOpener) { + const journal = await journals.open({ identity: IDENTITY, stateDirectory: directory }) + const message = (text: string) => ({ + kind: 'message' as const, + role: 'user' as const, + blocks: [{ type: 'text' as const, text }] + }) + for (const id of ['taken-back', 'still-queued']) { + await journal.appendSubmission({ + clientMessageId: id, + payloadFingerprint: id, + body: message(id), fence: 1, - recovered: true + handoverRecorded: true }) + } + for (const id of ['turn-ended', 'refused']) { await journal.appendSubmission({ - clientMessageId: 'turn-ended', - payloadFingerprint: 'turn-ended', - body: { kind: 'message', role: 'user', blocks: [{ type: 'text', text: 'turn-ended' }] }, + clientMessageId: id, + payloadFingerprint: id, + body: message(id), fence: 1 }) - await journal.resolveDispatch({ - clientMessageId: 'turn-ended', - state: 'rejected', - ...agentSessionFailureWords(agentSessionFailureFact('cancelled'), { surface: 'rejection' }), - answeredInTurn: { - turn: { - provider: 'legacy', - agent: 'codex', - sessionId: IDENTITY.sessionId, - recordId: 'turn-lifecycle:turn-1' - }, - via: 'start' + } + const before = readAgentSessionHydrationPage(journal, 1) + const seen = journal.snapshot().cursor + const cancelled = agentSessionFailureWords(agentSessionFailureFact('cancelled'), { + surface: 'rejection' + }) + await journal.resolveDispatch({ + clientMessageId: 'taken-back', + state: 'rejected', + ...cancelled, + fence: 1, + recovered: true + }) + await journal.resolveDispatch({ + clientMessageId: 'turn-ended', + state: 'rejected', + ...cancelled, + answeredInTurn: { + turn: { + provider: 'legacy', + agent: 'codex', + sessionId: IDENTITY.sessionId, + recordId: 'turn-lifecycle:turn-1' }, - fence: 1 - }) - const page = readAgentSessionHydrationPage(journal, 1) - // Anti-vacuous: the page this build publishes does carry both. - expect(page.submissions.find((entry) => entry.clientMessageId === 'taken-back')).toEqual( - expect.objectContaining({ - submittedSequence: expect.any(Number) + via: 'start' + }, + fence: 1 + }) + await journal.resolveDispatch({ + clientMessageId: 'refused', + state: 'rejected', + ...agentSessionFailureWords(agentSessionFailureFact('queueFull'), { surface: 'rejection' }), + fence: 1 + }) + const since = journal.readSince(seen) + const batch = since.ok + ? projectJournalBatch({ + rows: since.rows, + snapshot: journal.snapshot(), + afterSequence: seen.sequence }) - ) - expect(page.submissions.find((entry) => entry.clientMessageId === 'turn-ended')).toEqual( - expect.objectContaining({ - answeredInTurn: { turnItemId: expect.any(String), via: 'start' } - }) - ) + : null + if (!batch?.ok) { + throw new Error('the batch did not project') + } + return { before, after: readAgentSessionHydrationPage(journal, 1), batch: batch.batch } +} + +// Loads real release checkouts, cold extraction and transforms included. +test.each(BASELINE_REFS)( + 'a released client (%s) draws a page and a batch carrying journal positions and answered turns as it draws them without', + async (ref) => { + const directory = mkdtempSync(join(tmpdir(), 'orca-submission-positions-downgrade-')) + const journals = createTrackedJournalOpener() + try { + const { before, after, batch } = await journalWithRejections(directory, journals) + // Anti-vacuous: what this build publishes carries a position, a named turn, and stated none. + expect(after.submissions).toEqual( + expect.arrayContaining([ + expect.objectContaining({ clientMessageId: 'taken-back', answeredInTurn: null }), + expect.objectContaining({ + clientMessageId: 'turn-ended', + submittedSequence: expect.any(Number), + answeredInTurn: { turnItemId: expect.any(String), via: 'start' } + }), + expect.objectContaining({ clientMessageId: 'refused', answeredInTurn: null }) + ]) + ) + expect(batch.submissions.map((entry) => entry.answeredInTurn)).toContain(null) - 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 { + const checkout = await materializeReleaseCheckout(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 = (state: OldState) => ({ submissions: state.submissions.map( ({ submittedSequence: _s, answeredInTurn: _t, ...rest }) => rest ), messages: project(state.items, [], state.submissions) - } - } + }) + const fromPage = (page: AgentSessionHistoryPage) => + draw(reduce(empty, { type: 'history-page', page })) + const fromBatch = (stripped: boolean) => + draw( + reduce(reduce(empty, { type: 'history-page', page: before }), { + type: 'event', + event: { + type: 'batch', + sessionId: IDENTITY.sessionId, + batch: stripped + ? { ...batch, submissions: withoutNewFields(batch.submissions) } + : batch, + fence: 1 + } + }) + ) - expect(draw(page)).toEqual(draw(withoutPosition(page))) - } finally { - await journals.closeAll() - rmSync(directory, { recursive: true, force: true }) - } -}, 180_000) + expect(fromPage(after)).toEqual( + fromPage({ ...after, submissions: withoutNewFields(after.submissions) }) + ) + expect(fromBatch(false)).toEqual(fromBatch(true)) + // Anti-vacuous: the batch reached the old client's submissions. + expect(fromBatch(false).submissions).toEqual( + expect.arrayContaining([expect.objectContaining({ dispatchState: 'rejected' })]) + ) + } finally { + await journals.closeAll() + rmSync(directory, { recursive: true, force: true }) + } + }, + 180_000 +) From e02f98add56b28b07161b94e38233ed7cfc7de03 Mon Sep 17 00:00:00 2001 From: Brennan Benson <79079362+brennanb2025@users.noreply.github.com> Date: Mon, 5 Oct 2026 09:44:37 -0700 Subject: [PATCH 7/8] refactor(native-chat): give the answered-turn schema its own module Keeps the journal schema file within its line limit. --- src/shared/agent-session-answered-turn-schema.ts | 7 +++++++ src/shared/agent-session-journal-schemas.ts | 7 ++----- 2 files changed, 9 insertions(+), 5 deletions(-) create mode 100644 src/shared/agent-session-answered-turn-schema.ts diff --git a/src/shared/agent-session-answered-turn-schema.ts b/src/shared/agent-session-answered-turn-schema.ts new file mode 100644 index 000000000000..44f574d9ec14 --- /dev/null +++ b/src/shared/agent-session-answered-turn-schema.ts @@ -0,0 +1,7 @@ +import { z } from 'zod' + +/** A submission's answered turn as published. `via` is a string, as `dispatchState` is: a newer + * host may name another way of joining. null: the host recorded no turn. */ +export const AgentJournalAnsweredTurnSchema = z + .object({ turnItemId: z.string().min(1), via: z.string().min(1) }) + .nullable() diff --git a/src/shared/agent-session-journal-schemas.ts b/src/shared/agent-session-journal-schemas.ts index 39ee0b45c5f7..1bbaee0330c4 100644 --- a/src/shared/agent-session-journal-schemas.ts +++ b/src/shared/agent-session-journal-schemas.ts @@ -16,6 +16,7 @@ import { z } from 'zod' import { AgentSessionContextUsageSchema } from './agent-session-context-usage-schema' import { AgentSessionFailureFactSchema } from './agent-session-failure-fact-schema' +import { AgentJournalAnsweredTurnSchema } from './agent-session-answered-turn-schema' import type { AgentJournalItemBody, AgentJournalMessageItem, @@ -319,11 +320,7 @@ export const AgentJournalSubmissionSchema = z.object({ submittedAt: z.number(), resolvedAt: z.number().nullable(), submittedSequence: z.number().int().optional(), - // `via` is a string, as `dispatchState` is: a newer host may name another way of joining. - answeredInTurn: z - .object({ turnItemId: z.string().min(1), via: z.string().min(1) }) - .nullable() - .optional(), + answeredInTurn: AgentJournalAnsweredTurnSchema.optional(), recovered: z.literal(true).optional(), handoverRecorded: z.literal(true).optional(), handedOverAt: z.number().optional(), From 4ab6cc431c01ae5c5a70982ccdbd65ea611a917f Mon Sep 17 00:00:00 2001 From: Brennan Benson <79079362+brennanb2025@users.noreply.github.com> Date: Mon, 5 Oct 2026 10:00:35 -0700 Subject: [PATCH 8/8] docs(native-chat): say an unreadable answered turn reads as none --- src/shared/agent-session-journal-types.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/shared/agent-session-journal-types.ts b/src/shared/agent-session-journal-types.ts index f6bf89421ef0..832908795a11 100644 --- a/src/shared/agent-session-journal-types.ts +++ b/src/shared/agent-session-journal-types.ts @@ -431,7 +431,7 @@ export type AgentJournalSubmission = { /** On `rejected`: the turn a Codex send was answered into, when that turn ended without taking * it. null: the host recorded that it was answered into no turn, as every other rejection is * (Claude, a queued message taken back before handover, restart recovery). Absent: written - * before this field existed, or not rejected. */ + * before this field existed, or not rejected. A stored value this build cannot read is null. */ answeredInTurn?: AgentJournalAnsweredTurn | null /** Set when crash reconciliation resolved the dispatch, not the provider. A live * `unknown` is a send still outstanding; a recovered one outlived its writer. */