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.
This commit is contained in:
Brennan Benson
2026-10-03 13:28:42 -07:00
parent 752871bd67
commit 80756ad658
7 changed files with 253 additions and 12 deletions
@@ -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)
@@ -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,
@@ -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
})
])
})
})
@@ -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()
})
+5 -10
View File
@@ -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()
})
+11 -2
View File
@@ -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'
}
@@ -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<T>(module: Record<string, unknown>, 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<string, unknown>[]; 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<unknown>(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)