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 1eee3bc9a0a7..4ad6d58faedc 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 @@ -52,7 +60,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. */ @@ -64,7 +76,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() @@ -101,12 +117,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) @@ -141,10 +157,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-adapter.ts b/src/main/codex/codex-structured-session-adapter.ts index 288e08b9e34f..1c5a034fd63c 100644 --- a/src/main/codex/codex-structured-session-adapter.ts +++ b/src/main/codex/codex-structured-session-adapter.ts @@ -130,7 +130,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 f70517668b54..5773b659f5e1 100644 --- a/src/main/codex/codex-structured-session-state.ts +++ b/src/main/codex/codex-structured-session-state.ts @@ -1,5 +1,6 @@ import type { StructuredAgentSessionLogger } from '../native-chat/agent-session-wire/structured-agent-session-logger' import type { + AgentJournalAnsweredTurnIdentity, AgentJournalItemIdentity, AgentSessionJournalIdentity } from '../../shared/agent-session-journal-types' @@ -87,7 +88,10 @@ export type CodexStructuredSessionAdapterDeps = { onDispatchSettledLate?: ( input: { sessionId: string; clientMessageId: string } & ( | { providerIdentity: AgentJournalItemIdentity } - | ({ state: 'rejected' } & AgentJournalDispatchRejection) + | ({ + state: 'rejected' + answeredInTurn: AgentJournalAnsweredTurnIdentity + } & 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..2b6542a43336 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,13 +307,82 @@ 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: { turn: rig.turnRecords[0], via: 'start' } + }) + ]) + }) + + 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: { turn: rig.turnRecords[0], via: 'start' } }) + ], + [ + 'client-2', + expect.objectContaining({ answeredInTurn: { turn: rig.turnRecords[0], via: 'steer' } }) + ] + ]) + }) + + 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: { turn: rig.turnRecords[0], via: 'start' } + }) + }) + + 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() 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) }) @@ -300,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 657006449331..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,9 @@ import { agentSessionFailureWords, type AgentJournalDispatchRejection } from '../../shared/agent-session-failure-words' +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' 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, and how the send joined it. */ + answeredInTurn: AgentJournalAnsweredTurnIdentity } & AgentJournalDispatchRejection function errorDetail(params: unknown): ProviderDiagnostic | undefined { @@ -80,19 +84,27 @@ 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) - 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', ...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 7f0a2ec0c239..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 { @@ -8,6 +11,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, @@ -107,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', @@ -119,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 @@ -141,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)) { @@ -167,7 +171,7 @@ export async function startCodexTurn( }, { timeoutMs: input.timeoutMs } ) - return { turnId: readCodexTurnId(answer) } + return { turnId: readCodexTurnId(answer), via: 'start' } } /** @@ -178,10 +182,15 @@ 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 + let answer: { turnId: string | null; via: AgentJournalTurnJoin } | false try { answer = await startCodexTurn(session, { ...input, timeoutMs }) } catch (error) { @@ -205,8 +214,22 @@ 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 ? { state: 'rejected', ...rejection } : { state: 'admitted' } + return rejection && answer.turnId + ? { + state: 'rejected', + 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-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 c7c2a6716546..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 @@ -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,6 +35,11 @@ export function applyJournalDispatchRow( } else { delete submission.rejection } + if (row.state === 'rejected' && row.answeredInTurn !== undefined) { + submission.answeredInTurn = readAnsweredTurn(row.answeredInTurn) + } else { + delete submission.answeredInTurn + } submission.resolvedAt = row.state === 'pending' ? null : row.ts if (row.state === 'pending') { submission.handedOverAt = row.ts @@ -61,6 +67,19 @@ export function applyJournalDispatchRow( }) } +/** 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 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 } + : null +} + /** 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 0865df3a41fa..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,6 +110,17 @@ export function journalDispatchRowBuilder( providerItemId, reason: boundedDispatchReason(input), ...(input.state === 'rejected' ? { rejection: input.rejection } : {}), + // Every rejection states its turn, null for none, so a reader tells it from an older row. + ...(input.state === 'rejected' + ? { + answeredInTurn: input.answeredInTurn + ? { + turnItemId: agentJournalItemKey(input.answeredInTurn.turn), + via: input.answeredInTurn.via + } + : null + } + : {}), ...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 f0ebac8622bf..d3cc15fbd98c 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 @@ -160,6 +160,11 @@ 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 a Codex send was answered into, and how it joined it, when that + * 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-store-contracts.ts b/src/main/native-chat/agent-session-journal/journal-store-contracts.ts index cb86a7a8e5da..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, @@ -40,7 +41,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?: AgentJournalAnsweredTurnIdentity + } & AgentJournalDispatchRejection) | { state: 'unknown'; reason?: string | null } ) 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 80e4e2c93e95..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 @@ -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 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..d8aa5b4870cc --- /dev/null +++ b/src/main/native-chat/agent-session-journal/journal-submission-positions.test.ts @@ -0,0 +1,170 @@ +// 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. A rejection a turn's +// end made also names that turn. + +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 { + agentJournalItemKey, + agentJournalSubmissionKey +} from '../../../shared/agent-session-journal-item-key' +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 } +} + +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: TURN, via: 'start' }, + 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', + answeredInTurn: { turnItemId, via: 'start' } + }) + expect(readAgentSessionHydrationPage(journal).submissions).toEqual([ + expect.objectContaining({ answeredInTurn: { turnItemId, via: 'start' } }) + ]) + await journals.closeAll() + const replayed = await journals.open({ identity: IDENTITY, stateDirectory: root! }) + expect(replayed.submission('send-1')).toMatchObject({ + answeredInTurn: { turnItemId, via: 'start' } + }) + }) + + 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', + answeredInTurn: null + }) + expect(readAgentSessionHydrationPage(journal).submissions[0]).toMatchObject({ + answeredInTurn: null + }) + }) +}) + +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(handedOver.sequence).toBeGreaterThan(submitted.sequence) + expect(journal.submission('send-1')).toMatchObject({ + dispatchState: 'rejected', + 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('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 }) + expect(journal).not.toBe(replayed) + }) + + 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 }) + ]) + }) +}) 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 9c34c1841048..9975701e7047 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, @@ -176,8 +177,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?: 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 ba0ab9231261..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' @@ -12,7 +15,10 @@ export async function settleStructuredAgentSessionLateDispatch( clientMessageId: string } & ( | { providerIdentity: AgentJournalItemIdentity } - | ({ state: 'rejected' } & AgentJournalDispatchRejection) + | ({ + state: 'rejected' + answeredInTurn?: AgentJournalAnsweredTurnIdentity + } & AgentJournalDispatchRejection) | { state: 'unknown'; reason: string } ) ): Promise { @@ -43,6 +49,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 6ce700a84350..712e05552217 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 @@ -345,6 +345,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) + ?.answeredInTurn, + onPage: onPage?.answeredInTurn + } + } + + 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() + 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) + 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 + }) + }) + + 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).toEqual({ turnItemId: turnRecords[0], via: 'start' }) + }) +}) + 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-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-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 cb5e0e573d7d..2de4731a7a61 100644 --- a/src/shared/agent-session-journal-schemas.ts +++ b/src/shared/agent-session-journal-schemas.ts @@ -27,6 +27,8 @@ 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 { knownTags, openDiscriminatedUnion } from './agent-session-journal-open-union' import type { AgentJournalItemBody, @@ -201,14 +203,6 @@ const ThreadGoalState = openDiscriminatedUnion( ]) ) -/** 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() -}) - const KnownItemBody = z.discriminatedUnion('kind', [ MessageBody, z.object({ @@ -262,7 +256,7 @@ const KnownItemBody = z.discriminatedUnion('kind', [ .optional(), providerFrame: ProviderFrame.optional(), threadGoal: ThreadGoalState.optional(), - failure: FailureFact.optional() + failure: AgentSessionFailureFactSchema.optional() }), z.object({ kind: z.literal('turn'), @@ -329,10 +323,12 @@ export const AgentJournalSubmissionSchema = z.object({ reason: z.string().nullable(), submittedAt: z.number(), resolvedAt: z.number().nullable(), + submittedSequence: z.number().int().optional(), + answeredInTurn: AgentJournalAnsweredTurnSchema.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 9613afa8a849..7ec0d82be7c0 100644 --- a/src/shared/agent-session-journal-types.ts +++ b/src/shared/agent-session-journal-types.ts @@ -404,6 +404,19 @@ 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`, 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. */ +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] @@ -424,6 +437,15 @@ 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. 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 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. 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. */ recovered?: true @@ -432,13 +454,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, 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`. */ 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..0945d6a77284 --- /dev/null +++ b/tests/e2e/cross-version-wire/submission-positions-downgrade.unit.test.ts @@ -0,0 +1,203 @@ +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 { + AgentJournalSubmission, + AgentSessionJournalIdentity +} from '../../../src/shared/agent-session-journal-types' +import type { AgentSessionHistoryPage } from '../../../src/shared/agent-session-wire' +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' + +// 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', + 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[] } + +/** Submissions as an older host would carry them. */ +function withoutNewFields( + submissions: readonly AgentJournalSubmission[] +): AgentJournalSubmission[] { + return submissions.map( + ({ submittedSequence: _submitted, answeredInTurn: _turn, ...rest }) => rest + ) +} + +/** 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, + handoverRecorded: true + }) + } + for (const id of ['turn-ended', 'refused']) { + await journal.appendSubmission({ + clientMessageId: id, + payloadFingerprint: id, + body: message(id), + fence: 1 + }) + } + 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' + }, + 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 + }) + : 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(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(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 +)