mirror of
https://github.com/stablyai/orca.git
synced 2026-10-12 08:02:37 +00:00
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.
This commit is contained in:
@@ -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)
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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 })
|
||||
])
|
||||
})
|
||||
})
|
||||
|
||||
@@ -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(),
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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<T>(module: Record<string, unknown>, name: string): 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 {
|
||||
/** 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 })
|
||||
|
||||
Reference in New Issue
Block a user