diff --git a/src/main/acp/acp-structured-session-adapter-lifecycle.test.ts b/src/main/acp/acp-structured-session-adapter-lifecycle.test.ts index 400fe7d7561d..5161d42cecf3 100644 --- a/src/main/acp/acp-structured-session-adapter-lifecycle.test.ts +++ b/src/main/acp/acp-structured-session-adapter-lifecycle.test.ts @@ -109,7 +109,12 @@ describe('ACP Stop names a turn', () => { expect(rig.sent('session/cancel')).toHaveLength(2) expect(rig.settled).toEqual( expect.arrayContaining([ - expect.objectContaining({ clientMessageId: 'follow-up', state: 'rejected' }) + // Held by Orca, never prompted: proof it never reached the agent. + expect.objectContaining({ + clientMessageId: 'follow-up', + state: 'rejected', + rejection: { kind: 'cancelled', neverReachedAgent: true } + }) ]) ) }) @@ -221,7 +226,7 @@ describe('ACP connection loss', () => { ).toEqual({ state: 'rejected', reason: 'Grok stopped before this message was sent.', - rejection: { kind: 'providerExited' } + rejection: { kind: 'providerExited', neverReachedAgent: true } }) // The next start's stop of the old child is one Orca asked for; it does not rename the crash. await expect(rig.acquire()).rejects.toMatchObject({ diff --git a/src/main/acp/acp-structured-session-adapter.ts b/src/main/acp/acp-structured-session-adapter.ts index 7388ab04ef6c..dbaec32b2e46 100644 --- a/src/main/acp/acp-structured-session-adapter.ts +++ b/src/main/acp/acp-structured-session-adapter.ts @@ -330,13 +330,14 @@ export class AcpStructuredSessionAdapter implements StructuredAgentSessionAdapte void this.stop(session.sessionId, false) } + /** Only for a send refused before its prompt was written. */ private rejected( session: AcpStructuredSession, kind: 'providerExited' ): AgentSessionDispatchOutcome { return { state: 'rejected', - ...agentSessionFailureWords(agentSessionFailureFact(kind), { + ...agentSessionFailureWords(agentSessionFailureFact(kind, { neverReachedAgent: true }), { surface: 'rejection', agentName: acpAgentName(session.spec.agent) }) diff --git a/src/main/acp/acp-structured-turns.ts b/src/main/acp/acp-structured-turns.ts index 90e5d6b0c749..beedd11f660a 100644 --- a/src/main/acp/acp-structured-turns.ts +++ b/src/main/acp/acp-structured-turns.ts @@ -156,7 +156,10 @@ export class AcpStructuredTurns { private withdrawSteers(): boolean { const withdrawn = this.steers.splice(0) for (const send of withdrawn) { - this.reject(send.clientMessageId, agentSessionFailureFact('cancelled')) + this.reject( + send.clientMessageId, + agentSessionFailureFact('cancelled', { neverReachedAgent: true }) + ) } return withdrawn.length > 0 } diff --git a/src/main/claude/claude-structured-dispatch.ts b/src/main/claude/claude-structured-dispatch.ts index 87e706ddbbc3..cc8d06d59302 100644 --- a/src/main/claude/claude-structured-dispatch.ts +++ b/src/main/claude/claude-structured-dispatch.ts @@ -134,7 +134,11 @@ export async function dispatchClaudeTurn( } const startupFailure = claudeStartupFailureFact(session) if (startupFailure) { - return { state: 'rejected', ...claudeDispatchRejection(startupFailure) } + // Refused before the write, so the agent never saw it. + return { + state: 'rejected', + ...claudeDispatchRejection({ ...startupFailure, neverReachedAgent: true }) + } } // Read the sent content, not the journal blocks: only the mapped trailing prompt decides // whether Claude runs a command, so the two cannot disagree about which frame settles this. diff --git a/src/main/claude/claude-structured-queued-stop.test.ts b/src/main/claude/claude-structured-queued-stop.test.ts index 6f2f19e5fab7..2ffb3749c166 100644 --- a/src/main/claude/claude-structured-queued-stop.test.ts +++ b/src/main/claude/claude-structured-queued-stop.test.ts @@ -199,6 +199,16 @@ describe('Stop with a follow-up Claude queued behind the running turn', () => { expect(bySessionStart).toEqual(byInit) }) + // Claude may already hold a swept send in the stopped turn, so it is never re-sent as history. + it('settles the swept follow-up without proof it never reached Claude', async () => { + const { settlements } = await stopWithQueuedFollowUp( + claudeCli({ initProof: 'init', capabilities: CAPABILITIES }), + () => {} + ) + expect(settlements[0]).toMatchObject({ state: 'rejected' }) + expect(JSON.stringify(settlements[0])).not.toContain('neverReachedAgent') + }) + it('keeps what a turn advertised when a later frame names nothing', async () => { const result = await stopWithQueuedFollowUp( claudeCli({ initProof: 'session-start' }), diff --git a/src/main/codex/codex-structured-dispatch-admission.test.ts b/src/main/codex/codex-structured-dispatch-admission.test.ts index dc0bf170af79..5ab5f6bec98e 100644 --- a/src/main/codex/codex-structured-dispatch-admission.test.ts +++ b/src/main/codex/codex-structured-dispatch-admission.test.ts @@ -158,7 +158,8 @@ describe('codex dispatch admission', () => { reason: "Codex didn't accept this message: thread not found.", rejection: { kind: 'providerRejected', - detail: { text: 'thread not found', audience: 'person' } + detail: { text: 'thread not found', audience: 'person' }, + neverReachedAgent: true } }) diff --git a/src/main/codex/codex-structured-session-adapter.test.ts b/src/main/codex/codex-structured-session-adapter.test.ts index 3a25cdbff2f2..87f87d583bef 100644 --- a/src/main/codex/codex-structured-session-adapter.test.ts +++ b/src/main/codex/codex-structured-session-adapter.test.ts @@ -593,7 +593,7 @@ describe('CodexStructuredSessionAdapter.dispatch', () => { // Built without Codex's own words, so nothing is quoted and no detail is invented. state: 'rejected', reason: "Codex didn't accept this message.", - rejection: { kind: 'providerRejected' } + rejection: { kind: 'providerRejected', neverReachedAgent: true } }) }) 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 776ce7263103..4de871e32012 100644 --- a/src/main/codex/codex-structured-turn-end-settlement.test.ts +++ b/src/main/codex/codex-structured-turn-end-settlement.test.ts @@ -78,10 +78,12 @@ describe('a Codex send its turn ended without echoing', () => { rig.turns.end('interrupted') expect(rig.settlements).toEqual([ + // Codex took it into a turn: no proof it never reached Codex. expect.objectContaining({ sessionId: 'session-1', clientMessageId: 'client-1', - state: 'rejected' + state: 'rejected', + rejection: { kind: 'cancelled' } }) ]) expect(rig.categoryOf(rig.settlements[0])).toBe('withdrawn') diff --git a/src/main/codex/codex-structured-turn-start.ts b/src/main/codex/codex-structured-turn-start.ts index 52a653a68e3c..5ce38c4bec3a 100644 --- a/src/main/codex/codex-structured-turn-start.ts +++ b/src/main/codex/codex-structured-turn-start.ts @@ -170,11 +170,15 @@ export async function dispatchCodexTurn( if (isCodexAppServerRequestError(error) || isCodexAppServerUnsupportedError(error)) { // Codex answered and declined, so no echo for this write can arrive. session.dispatchEchoes.disarm(input.clientMessageId) - // Codex's own words, when it gave any, are the one part of the error a person can use. + // Codex's own words, when it gave any, are the one part of the error a person can use. A + // declined request recorded nothing, so stopped messages it carried are owed again. return { state: 'rejected', ...codexDispatchRejection( - agentSessionFailureFact('providerRejected', { detail: providerDiagnosticOf(error) }) + agentSessionFailureFact('providerRejected', { + detail: providerDiagnosticOf(error), + neverReachedAgent: true + }) ) } } diff --git a/src/main/codex/codex-structured-turn-steer.test.ts b/src/main/codex/codex-structured-turn-steer.test.ts index e6c654c00775..596baab29864 100644 --- a/src/main/codex/codex-structured-turn-steer.test.ts +++ b/src/main/codex/codex-structured-turn-steer.test.ts @@ -171,7 +171,8 @@ describe('a Codex send while a turn runs', () => { reason: "Codex didn't accept this message: thread not found.", rejection: { kind: 'providerRejected', - detail: { text: 'thread not found', audience: 'person' } + detail: { text: 'thread not found', audience: 'person' }, + neverReachedAgent: true } }) expect(methods()).toEqual(['thread/start', 'turn/steer', 'turn/start']) 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 60cfe328ec52..ff0f763ef7f8 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 @@ -54,6 +54,7 @@ export function applyJournalDispatchRow( if (row.state === 'pending') { submission.handedOverAt = row.ts placeHandedOverMessage(state, submission, row) + foldStoppedHistoryCarried(state, row) } else if (row.state === 'rejected') { placeRejectedMessage(state, submission, row) } @@ -77,6 +78,20 @@ export function applyJournalDispatchRow( }) } +/** A send is handed over once, so its one pending row names what it carries; a malformed id is + * dropped. */ +function foldStoppedHistoryCarried( + state: JournalReducerState, + row: Extract +): void { + const carried = Array.isArray(row.carries) + ? row.carries.filter((id): id is string => typeof id === 'string' && id.length > 0) + : [] + if (carried.length > 0) { + state.stoppedHistoryCarried.set(row.clientMessageId, carried) + } +} + /** 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 { diff --git a/src/main/native-chat/agent-session-journal/journal-reducer.ts b/src/main/native-chat/agent-session-journal/journal-reducer.ts index 4acb175e82ab..0f5c1b8a8076 100644 --- a/src/main/native-chat/agent-session-journal/journal-reducer.ts +++ b/src/main/native-chat/agent-session-journal/journal-reducer.ts @@ -71,6 +71,8 @@ export type JournalReducerState = { latestAcceptedTurnSequence: number /** The latest person's Stop event and Resume, what the queue's pause is derived from. */ queuePauseMarks: JournalQueuePauseMarks + /** Host-only: each handed-over send's stopped-message history, from its latest `pending` row. */ + stoppedHistoryCarried: Map } export function createJournalReducerState(sessionId: string, epoch: string): JournalReducerState { @@ -92,7 +94,8 @@ export function createJournalReducerState(sessionId: string, epoch: string): Jou appliedSettlementIds: new Set(), derivedTurnScope: new JournalDerivedTurnScope(), latestAcceptedTurnSequence: 0, - queuePauseMarks: createJournalQueuePauseMarks() + queuePauseMarks: createJournalQueuePauseMarks(), + stoppedHistoryCarried: new Map() } } diff --git a/src/main/native-chat/agent-session-journal/journal-restart-reconciliation.test.ts b/src/main/native-chat/agent-session-journal/journal-restart-reconciliation.test.ts index 8e541f110b00..83c583a58200 100644 --- a/src/main/native-chat/agent-session-journal/journal-restart-reconciliation.test.ts +++ b/src/main/native-chat/agent-session-journal/journal-restart-reconciliation.test.ts @@ -190,6 +190,38 @@ describe('reconcileJournalSubmissionsAgainstHistory', () => { expect(journal.submissions()[0]?.dispatchState).toBe('unknown') }) + it('leaves a send that carried stopped history unknown: its provider copy leads with that history', async () => { + const journal = await open() + for (const clientMessageId of ['stopped', 'cm_1']) { + await journal.appendSubmission({ + clientMessageId, + payloadFingerprint: digestPayload(clientMessageId), + body: userMessage(clientMessageId), + fence: 1, + handoverRecorded: true + }) + } + await journal.resolveDispatch({ + clientMessageId: 'cm_1', + state: 'pending', + fence: 1, + turnScope: AGENT_JOURNAL_THREAD_SCOPE, + carries: ['stopped'] + }) + const restarted = await open() + await restarted.markPendingSubmissionsUnknown(2) + + const settled = await reconcileJournalSubmissionsAgainstHistory({ + journal: restarted, + fence: 2, + history: window([]) + }) + + expect(settled).toEqual([]) + expect(restarted.submission('cm_1')?.dispatchState).toBe('unknown') + expect(restarted.stoppedHistoryCarried().get('cm_1')).toEqual(['stopped']) + }) + it('leaves multi-block text sends unknown because Claude joins them before recording history', async () => { const journal = await reopenAfterCrash( { diff --git a/src/main/native-chat/agent-session-journal/journal-restart-reconciliation.ts b/src/main/native-chat/agent-session-journal/journal-restart-reconciliation.ts index cc59c95893cd..b44ecc5e78f5 100644 --- a/src/main/native-chat/agent-session-journal/journal-restart-reconciliation.ts +++ b/src/main/native-chat/agent-session-journal/journal-restart-reconciliation.ts @@ -55,7 +55,10 @@ function comparableSubmissions(journal: AgentSessionJournal): AgentJournalSubmis if ( (submission.dispatchState !== 'pending' && submission.dispatchState !== 'unknown') || // Never handed over, so provider history cannot hold it. - isQueuedAgentJournalSubmission(submission) + isQueuedAgentJournalSubmission(submission) || + // The provider's copy leads with the stopped history it carried, so its fingerprint cannot + // match; like an attachment's, its absence is no evidence and it stays `unknown`. + journal.stoppedHistoryCarried().has(submission.clientMessageId) ) { return false } 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 ea08945a4d1d..c55a8398cad3 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 @@ -145,7 +145,8 @@ export function journalDispatchRowBuilder( : {}), ...journalRowBase(state().epoch, seq, input.fence, ts), ...(input.recovered ? { recovered: input.recovered } : {}), - ...(input.state === 'pending' ? { turnScope: input.turnScope } : {}) + ...(input.state === 'pending' ? { turnScope: input.turnScope } : {}), + ...(input.state === 'pending' && input.carries?.length ? { carries: [...input.carries] } : {}) }) } 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 477e8549ae64..90c449c1b2ff 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 @@ -181,6 +181,9 @@ export type JournalDispatchRow = JournalRowBase & { reason: string | null /** On `pending`: the turn the message was handed into, which becomes its row's scope. */ turnScope?: AgentJournalTurnScope + /** On `pending`: the stopped messages this send carries to the agent as history, by client + * message id in the order they were sent. Host-only; older readers keep the key and ignore it. */ + carries?: string[] /** On `rejected`: why, typed. Older readers keep the key and ignore it; a malformed one is * dropped when read, never the row. */ rejection?: AgentSessionFailureFact diff --git a/src/main/native-chat/agent-session-journal/journal-stop-acceptance.ts b/src/main/native-chat/agent-session-journal/journal-stop-acceptance.ts index 9299655d92e8..adeaa7f8e6f8 100644 --- a/src/main/native-chat/agent-session-journal/journal-stop-acceptance.ts +++ b/src/main/native-chat/agent-session-journal/journal-stop-acceptance.ts @@ -2,7 +2,8 @@ // event, and the command receipt that names that event. Nothing of it is durable unless all of it // is, so a Stop whose acceptance fails withdraws nothing and interrupts nothing. -import type { AgentJournalDispatchRejection } from '../../../shared/agent-session-failure-words' +import { agentSessionFailureFact } from '../../../shared/agent-session-failure' +import { agentSessionFailureWords } from '../../../shared/agent-session-failure-words' import type { AgentJournalCursor } from '../../../shared/agent-session-journal-types' import { journalQueuedRejectionRowBuilders } from './journal-pending-submission-recovery' import type { JournalReducerState } from './journal-reducer' @@ -13,10 +14,14 @@ import { journalStopEventRowBuilder } from './journal-stop-and-resume-rows' export type JournalStopAcceptanceInput = { event: Omit fence: number - /** What every queued send is rejected with. */ - withdrawal: AgentJournalDispatchRejection } +// Only host-held sends are withdrawn, so each carries proof it never reached the agent. +const STOP_WITHDRAWAL = agentSessionFailureWords( + agentSessionFailureFact('cancelled', { neverReachedAgent: true }), + { surface: 'rejection' } +) + export type JournalStopAcceptance = { /** Where the Stop event landed. */ mark: AgentJournalCursor @@ -54,7 +59,7 @@ export class JournalStopAcceptor { turnId = input.event.turnId ?? this.deps.liveTurnId() const event = { ...input.event, ...(turnId !== null ? { turnId } : {}) } return [ - ...journalQueuedRejectionRowBuilders(state, input.fence, input.withdrawal), + ...journalQueuedRejectionRowBuilders(state, input.fence, STOP_WITHDRAWAL), journalStopEventRowBuilder(state, event, input.fence) ] }, diff --git a/src/main/native-chat/agent-session-journal/journal-store-collaborators.ts b/src/main/native-chat/agent-session-journal/journal-store-collaborators.ts index d1287b378629..4d5009b7db41 100644 --- a/src/main/native-chat/agent-session-journal/journal-store-collaborators.ts +++ b/src/main/native-chat/agent-session-journal/journal-store-collaborators.ts @@ -8,6 +8,7 @@ import type { AgentSessionJournalIdentity } from '../../../shared/agent-session-journal-types' import type { JournalHostDatabase } from './journal-host-database' +import { JournalContextController } from './journal-context-controller' import { JournalEpochController } from './journal-epoch-controller' import { JournalItemAppender } from './journal-item-appender' import { JournalLifecycleBatchAppender } from './journal-lifecycle-batch-appender' @@ -58,6 +59,7 @@ export type JournalStoreCollaborators = { stepWriter: JournalStepWriter queuedMessages: JournalQueuedMessages stopMarks: JournalStopMarks + context: JournalContextController stops: JournalStopAcceptor /** Restores the store's state from disk. Owned here because it needs the same * collaborators the constructor just built. */ @@ -125,6 +127,11 @@ export function createJournalStoreCollaborators(host: JournalStoreHost): Journal epochController, queuedMessages, stopMarks: new JournalStopMarks({ state: host.state }), + context: new JournalContextController({ + state: host.state, + writer: rowWriter, + cards: queuedMessages + }), stops: new JournalStopAcceptor({ writer: rowWriter, state: host.state, 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 3259d488b22e..64dfe7625fec 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,12 @@ export type ResolveDispatchInput = { ( | { state: 'accepted'; providerIdentity: AgentJournalItemIdentity | null } /** The turn the message is handed into — the live root turn, or `thread` when none runs. */ - | { state: 'pending'; turnScope: AgentJournalTurnScope } + | { + state: 'pending' + turnScope: AgentJournalTurnScope + /** The stopped messages this send carries as history (`JournalDispatchRow.carries`). */ + carries?: readonly string[] + } /** `reason` is what released clients print, `rejection` what newer ones read: both from * `agentSessionFailureWords`, never written by hand. */ | ({ diff --git a/src/main/native-chat/agent-session-journal/journal-store.ts b/src/main/native-chat/agent-session-journal/journal-store.ts index cb8dc8143572..2c4bc230dcf8 100644 --- a/src/main/native-chat/agent-session-journal/journal-store.ts +++ b/src/main/native-chat/agent-session-journal/journal-store.ts @@ -74,7 +74,7 @@ import type { JournalItemAppender } from './journal-item-appender' import type { JournalLifecycleBatchAppender } from './journal-lifecycle-batch-appender' import type { JournalStepWriter } from './journal-step-writer' import type { JournalStopMarks } from './journal-stop-marks' -import { JournalContextController } from './journal-context-controller' +import type { JournalContextController } from './journal-context-controller' import type { JournalStopAcceptor } from './journal-stop-acceptance' export { AgentSessionJournalError } from './journal-write-guards' @@ -146,11 +146,7 @@ export class AgentSessionJournal { this.stopMarks = collaborators.stopMarks this.stops = collaborators.stops this.restore = collaborators.restore - this.context = new JournalContextController({ - state: () => this.state, - writer: this.rowWriter, - cards: this.queuedMessages - }) + this.context = collaborators.context } get epoch(): string { @@ -216,6 +212,10 @@ export class AgentSessionJournal { itemBody = (itemId: string): AgentJournalItemBody | null => this.state.items.get(itemId)?.body ?? null + /** Host-only: the stopped messages each handed-over send carried, by its client message id. */ + stoppedHistoryCarried = (): ReadonlyMap => + this.state.stoppedHistoryCarried + /** One reduced item with its attribution, for a writer that needs the turn a row joined. */ item = (itemId: string): AgentJournalRenderItem | null => this.state.items.get(itemId) ?? null diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-chat-stop.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-chat-stop.ts index b9ed2c93be54..20043d72b557 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-chat-stop.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-chat-stop.ts @@ -6,8 +6,6 @@ // reachable only through `mutateWithChatStop`, which queues that step in the same synchronous call // as the mutation. Once saved, a failure is answered from the Stop's receipt and noted on the chat. -import { agentSessionFailureFact } from '../../../shared/agent-session-failure' -import { agentSessionFailureWords } from '../../../shared/agent-session-failure-words' import type { AgentSessionCancelResult, AgentSessionMutationEnvelope, @@ -83,10 +81,7 @@ export function mutateWithChatStop( caller: caller.callerKey, ...(target.eventTurnId !== null ? { turnId: target.eventTurnId } : {}) }, - fence: ctx.fence, - withdrawal: agentSessionFailureWords(agentSessionFailureFact('cancelled'), { - surface: 'rejection' - }) + fence: ctx.fence }) return accepted.ok ? { ok: true, value: { settled: target.queued, turnId: accepted.value.turnId } } diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-rewind.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-rewind.test.ts index 1bbb02777243..927d725a494a 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-rewind.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-rewind.test.ts @@ -250,8 +250,9 @@ describe('host rewind', () => { ) ).toMatchObject({ ok: true }) expect((await host.journalSnapshot(HOST_TEST_SESSION)).items).toHaveLength(1) + // A person's own message keeps the words Orca recorded, not the provider's copy. expect((await host.journalSnapshot(HOST_TEST_SESSION)).items[0]?.body).toEqual( - hostTestMessage('verified history') + hostTestMessage('0') ) expect(recoverRewind).toHaveBeenCalledTimes(2) expect(rewind).toHaveBeenCalledTimes(1) @@ -534,7 +535,7 @@ describe('host rewind', () => { body })) ).toEqual([ - { itemId: agentJournalItemKey(message('kept')), body: hostTestMessage('kept from provider') }, + { itemId: agentJournalItemKey(message('kept')), body: hostTestMessage('kept') }, { itemId: agentJournalItemKey(goalRow), body: goalBody }, { itemId: agentJournalItemKey(turnRow('kept')), body: keptTurn } ]) @@ -578,7 +579,7 @@ describe('host rewind', () => { body })) ).toEqual([ - { itemId: agentJournalItemKey(message('kept')), body: hostTestMessage('kept from provider') }, + { itemId: agentJournalItemKey(message('kept')), body: hostTestMessage('kept') }, { itemId: agentJournalItemKey(newerRow), body: newerBody } ]) }) @@ -732,7 +733,7 @@ describe('host rewind', () => { body })) ).toEqual([ - { itemId: agentJournalItemKey(message('kept')), body: hostTestMessage('kept from recovery') }, + { itemId: agentJournalItemKey(message('kept')), body: hostTestMessage('kept') }, { itemId: agentJournalItemKey(goalRow), body: goalBody } ]) expect(store.getRecord(HOST_TEST_SESSION)?.rewind?.phase).toBe('completed') diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-stopped-history-selection.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-stopped-history-selection.ts new file mode 100644 index 000000000000..fa9094424041 --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-stopped-history-selection.ts @@ -0,0 +1,50 @@ +// What a handover carries of the stopped messages owed to the agent, chosen in the session's +// serialized handover step so a phone and a desktop sending at once deliver each one once. + +import type { + AgentJournalMessageItem, + AgentJournalSubmission +} from '../../../shared/agent-session-journal-types' +import { agentJournalSubmissionKey } from '../../../shared/agent-session-journal-item-key' +import { + agentSessionSendCarriesStoppedHistory, + agentSessionSendDeliversStoppedHistory, + agentSessionStoppedHistoryOwed, + selectStoppedHistory, + type StoppedHistoryDelivery +} from '../../../shared/agent-session-stopped-history' +import type { AgentSessionJournal } from '../agent-session-journal/journal-store' + +type StoppedHistoryContext = { + journal: Pick< + AgentSessionJournal, + 'submissions' | 'itemBody' | 'liveTurnScope' | 'stoppedHistoryCarried' + > & { + context: Pick + } +} + +/** What a person's words-only send that opens a turn carries of the owed messages; null for a + * steer into a running turn, which may never record it, and for any other send. */ +export function carriedStoppedHistory( + ctx: StoppedHistoryContext, + submission: AgentJournalSubmission, + body: AgentJournalMessageItem +): StoppedHistoryDelivery | null { + if ( + ctx.journal.liveTurnScope().kind !== 'thread' || + !agentSessionSendDeliversStoppedHistory(submission, body) || + !agentSessionSendCarriesStoppedHistory(body) + ) { + return null + } + return selectStoppedHistory( + agentSessionStoppedHistoryOwed({ + submissions: ctx.journal.submissions(), + carried: ctx.journal.stoppedHistoryCarried(), + clearSequence: ctx.journal.context.floor()?.sequence ?? 0, + body: (id) => ctx.journal.itemBody(agentJournalSubmissionKey(id)) + }), + body + ) +} diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-stopped-history.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-stopped-history.test.ts new file mode 100644 index 000000000000..2e0b9710633b --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-stopped-history.test.ts @@ -0,0 +1,153 @@ +// A message a person sent and then stopped before the agent started on it reaches the agent with +// their next send, at most once. Against the real host, store and journal, with an agent that stays +// starting until the test proves its start. The rig's chat is Codex's; every agent carries alike. + +import { afterEach, describe, expect, it } from 'vitest' +import { agentSessionFailureFact } from '../../../shared/agent-session-failure' +import { agentSessionFailureWords } from '../../../shared/agent-session-failure-words' +import type { AgentJournalMessageItem } from '../../../shared/agent-session-journal-types' +import { STOPPED_MESSAGE_INTERRUPTED_MARKER } from '../../../shared/agent-session-stopped-history' +import { AgentJournalSubmissionSchema } from '../../../shared/agent-session-journal-submission-schema' +import { projectStructuredAgentSessionMessages } from '../../../shared/structured-agent-session-message-projection' +import type { AgentSessionJournal } from '../agent-session-journal/journal-store' +import { + createQueuedMessageTestRig, + eventually, + type QueuedMessageTestRig +} from './structured-agent-session-queued-message-rig.test-fixture' +import { HOST_TEST_SESSION as SESSION } from './structured-agent-session-host-test-data' + +let rig: QueuedMessageTestRig + +async function start(): Promise { + rig = await createQueuedMessageTestRig({ restartable: true, starting: true }) +} + +afterEach(async () => { + await rig.dispose() +}) + +function journal(): AgentSessionJournal { + const open = rig.host.collaboratorsForTests().sessions.get(SESSION)?.journal + if (!open) { + throw new Error('the conversation is not open') + } + return open +} + +function text(body: AgentJournalMessageItem): string { + return body.blocks.flatMap((block) => (block.type === 'text' ? [block.text] : [])).join('\n') +} + +/** What a send carrying `stopped` ahead of `own` hands the agent. */ +function carrying(stopped: string[], own: string): string { + return ( + stopped.map((words) => `${words}\n\n${STOPPED_MESSAGE_INTERRUPTED_MARKER}\n\n`).join('') + own + ) +} + +/** Each dispatch's words, as the agent gets them. */ +function dispatched(): string[] { + return rig.dispatch.mock.calls.map(([input]) => text(input.body)) +} + +/** Sent while the agent starts, so the host holds them; none is handed over. */ +async function heldWhileStarting(...texts: string[]): Promise { + const ids: string[] = [] + for (const body of texts) { + const { id, result } = rig.send(body) + expect(await result).toMatchObject({ ok: true }) + ids.push(id) + } + await eventually(() => + expect(rig.host.collaboratorsForTests().sessions.get(SESSION)?.child?.phase).toBe('starting') + ) + return ids +} + +/** A Stop while the agent starts: its save withdraws what the host held, with the proof. */ +async function stopWhileStarting(...texts: string[]): Promise { + const ids = await heldWhileStarting(...texts) + expect(await rig.stop()).toMatchObject({ ok: true }) + for (const id of ids) { + expect(await rig.submission(id)).toMatchObject({ + dispatchState: 'rejected', + rejection: { kind: 'cancelled', neverReachedAgent: true } + }) + } + return ids +} + +/** The person's next send, handed to a child that proved its start. */ +async function nextSend(body: string): Promise { + const before = rig.dispatch.mock.calls.length + const { id, result } = rig.send(body) + expect(await result).toMatchObject({ ok: true }) + if (rig.host.collaboratorsForTests().sessions.get(SESSION)?.child?.phase !== 'ready') { + await rig.proveStart() + } + await eventually(() => expect(rig.dispatch.mock.calls.length).toBe(before + 1)) + return id +} + +/** The texts every client draws. */ +async function transcript(): Promise { + const page = await rig.host.history({ sessionId: SESSION, direction: 'tail' }) + if (!page.ok) { + throw new Error('history refused') + } + expect(page.page.submissions.some((entry) => 'carries' in entry)).toBe(false) + // A client reads every row, the host's proof included, with the schema it already has. + page.page.submissions.forEach((entry) => AgentJournalSubmissionSchema.parse(entry)) + return projectStructuredAgentSessionMessages(page.page.items, [], page.page.submissions).map( + (message) => + `${message.blocks.map((block) => ('text' in block ? block.text : '')).join('')}${ + message.stoppedBeforeStart ? ' (stopped)' : '' + }` + ) +} + +describe('a Stop withdraws what the agent never got, and the next send delivers it once', () => { + it('carries them at the front of the next send, in order, and never again', async () => { + await start() + const [first, second] = await stopWhileStarting('first', 'second') + const carrier = await nextSend('what did I send?') + await rig.settleAccepted(carrier, 'carrier') + await nextSend('after') + + expect(dispatched()).toEqual([carrying(['first', 'second'], 'what did I send?'), 'after']) + expect(journal().stoppedHistoryCarried().get(carrier)).toEqual([first, second]) + // The chat is unchanged: each still reads Stopped manually, the send shows its own words. + expect(await transcript()).toEqual([ + 'first (stopped)', + 'second (stopped)', + 'Stopped manually', + 'what did I send?', + 'after' + ]) + }) + + it('a carrier refused before any write owes them again; one in doubt never re-sends them', async () => { + await start() + await stopWhileStarting('first') + rig.dispatch.mockResolvedValueOnce({ + state: 'rejected', + ...agentSessionFailureWords(agentSessionFailureFact('writeFailed'), { surface: 'rejection' }) + }) + await nextSend('refused') + const doubtful = await nextSend('in doubt') + await rig.host.settleLateDispatch({ + sessionId: SESSION, + clientMessageId: doubtful, + state: 'unknown', + reason: 'reply lost' + }) + await nextSend('after') + + expect(dispatched()).toEqual([ + carrying(['first'], 'refused'), + carrying(['first'], 'in doubt'), + 'after' + ]) + }) +}) 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 2726c20f073d..829ad489f110 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 @@ -45,6 +45,8 @@ import type { StructuredAgentSessionLogger } from './structured-agent-session-lo import { isAgentSessionAttachmentExpiredError } from '../agent-session-attachments/agent-session-attachment-claims' import { CommandReceiptExistsError } from '../agent-session-journal/command-receipt-transaction' import type { StructuredAgentRegistry } from './structured-agent-registry' +import { carriedStoppedHistory } from './structured-agent-session-stopped-history-selection' +import { withStoppedHistory } from '../../../shared/agent-session-stopped-history' export { performSetOption } from './structured-agent-session-turns-options' export { performPrompt } from './structured-agent-session-turns-prompt' export { performCancel } from './structured-agent-session-turns-cancel' @@ -229,19 +231,22 @@ export async function handOverSubmission( await handOverStructuredAgentSessionCommand(ctx, submission, body) return } - // The message joins the turn running at handover, a steer, or opens its own. + // The message joins the turn running at handover, a steer, or opens its own. A new turn carries + // the stopped messages owed to the agent, named on this same row before anything is written. + const carried = carriedStoppedHistory(ctx, submission, body) await ctx.journal.resolveDispatch({ clientMessageId, state: 'pending', fence: ctx.fence, - turnScope: ctx.journal.liveTurnScope() + turnScope: ctx.journal.liveTurnScope(), + ...(carried ? { carries: carried.covers } : {}) }) // The handover row's instant on the host clock; the turn this dispatch opens records it so the // live counter never re-anchors at turn-open. const outcome = await dispatchSafely( ctx, clientMessageId, - body, + carried ? withStoppedHistory(body, carried.messages) : body, structuredAgentSessionHandoverOrigin(ctx.journal, submission) ) // An admission needs no dispatch row: the submission is already pending. diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-unopened-send-withdrawal.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-unopened-send-withdrawal.test.ts index 2b5847efc573..6c9b2a72282d 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-unopened-send-withdrawal.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-unopened-send-withdrawal.test.ts @@ -100,7 +100,12 @@ describe("a Codex child's end, or a Stop Codex took, under a person's Stop", () await withdrawCodexSendsNoTurnOpenedFor(journal, 1) expect(resolveDispatch).toHaveBeenCalledWith( - expect.objectContaining({ clientMessageId: 'send-1', state: 'rejected' }) + // Codex may have recorded it during turn setup: no proof it never reached Codex. + expect.objectContaining({ + clientMessageId: 'send-1', + state: 'rejected', + rejection: { kind: 'cancelled' } + }) ) }) diff --git a/src/main/native-chat/agent-session-wire/structured-rewind-retained-host-rows.ts b/src/main/native-chat/agent-session-wire/structured-rewind-retained-host-rows.ts index f41e83522d2c..b7c98c97a2c8 100644 --- a/src/main/native-chat/agent-session-wire/structured-rewind-retained-host-rows.ts +++ b/src/main/native-chat/agent-session-wire/structured-rewind-retained-host-rows.ts @@ -96,7 +96,7 @@ export function retainedRowReplacement(row: RetainedRow): AgentJournalProducerLi } /** A provider item the old epoch held keeps that row's scope, producer and first-seen time; only - * its body is the provider's, and a user message keeps the sender the host recorded on it. */ + * its body is the provider's, except that a user message keeps what the host recorded on it. */ function withHeldAttribution(item: RetainedRow, held: RetainedRow | undefined): RetainedRow { if (!held) { return item @@ -105,21 +105,22 @@ function withHeldAttribution(item: RetainedRow, held: RetainedRow | undefined): return { ...item, ...attribution, body: withHeldSender(item.body, heldBody) } } -/** The provider's copy of a user message, with the sender the host recorded on it put back: no - * provider ever carries it. */ +/** A person's own message keeps Orca's body, as the live fold does: the provider's copy leads with + * any stopped messages it carried. Another sender's keeps the provider's copy with that sender put + * back, since no provider ever carries it. */ function withHeldSender( provider: RetainedRow['body'], held: RetainedRow['body'] ): RetainedRow['body'] { if ( !isAdmissibleAgentJournalMessageBody(held) || - !isAdmissibleAgentJournalMessageBody(provider) + !isAdmissibleAgentJournalMessageBody(provider) || + held.role !== 'user' || + provider.role !== 'user' ) { return provider } - return held.from !== undefined && held.role === 'user' && provider.role === 'user' - ? Object.assign({}, provider, { from: held.from }) - : provider + return held.from === undefined ? held : Object.assign({}, provider, { from: held.from }) } /** Provider turn id → the item id of its turn record: the turn's own, or the command turn that diff --git a/src/main/pi/rpc-prompt-delivery.test.ts b/src/main/pi/rpc-prompt-delivery.test.ts index e59ebe4c9961..cc70676b598f 100644 --- a/src/main/pi/rpc-prompt-delivery.test.ts +++ b/src/main/pi/rpc-prompt-delivery.test.ts @@ -105,7 +105,7 @@ describe('Pi pending prompt delivery bounds and ownership', () => { expect(h.accepted).toHaveBeenLastCalledWith('retrying', 1) }) - it('settles a pending auth retry as unknown on process teardown', async () => { + it('settles a prompt waiting out a sign-in refusal as never reaching Pi on process teardown', async () => { vi.useFakeTimers() const h = rig() await h.delivery.submit('retrying', 1, { type: 'prompt', message: 'first' }) @@ -116,12 +116,34 @@ describe('Pi pending prompt delivery bounds and ownership', () => { error: 'No API key found for provider' }) h.delivery.end() - expect(h.settled).toHaveBeenCalledWith( + expect(h.settled).toHaveBeenCalledExactlyOnceWith( 'retrying', - expect.objectContaining({ state: 'unknown' }) + expect.objectContaining({ + state: 'rejected', + rejection: expect.objectContaining({ kind: 'notSignedIn', neverReachedAgent: true }) + }) ) expect(vi.getTimerCount()).toBe(0) }) + + it('settles a re-sent sign-in retry still awaiting its reply as unknown on process teardown', async () => { + vi.useFakeTimers() + const h = rig() + await h.delivery.submit('retrying', 1, { type: 'prompt', message: 'first' }) + h.delivery.reply({ + type: 'response', + command: 'prompt', + success: false, + error: 'No API key found for provider' + }) + await vi.advanceTimersByTimeAsync(250) + expect(h.send).toHaveBeenCalledTimes(2) + h.delivery.end() + expect(h.settled).toHaveBeenCalledExactlyOnceWith( + 'retrying', + expect.objectContaining({ state: 'unknown' }) + ) + }) }) it('retains Pi provider detail with sign-in guidance after bounded authentication retries', async () => { @@ -143,7 +165,9 @@ it('retains Pi provider detail with sign-in guidance after bounded authenticatio state: 'rejected', rejection: { kind: 'notSignedIn', - detail: { text: 'No API key found for anthropic', audience: 'person' } + detail: { text: 'No API key found for anthropic', audience: 'person' }, + // Pi checks sign-in before it records a prompt, so the agent never had this one. + neverReachedAgent: true } }) ) diff --git a/src/main/pi/rpc-prompt-delivery.ts b/src/main/pi/rpc-prompt-delivery.ts index d58905092266..c3e3b61399af 100644 --- a/src/main/pi/rpc-prompt-delivery.ts +++ b/src/main/pi/rpc-prompt-delivery.ts @@ -1,4 +1,8 @@ -import { agentSessionFailureFact, providerDiagnostic } from '../../shared/agent-session-failure' +import { + agentSessionFailureFact, + providerDiagnostic, + type SubmissionRejectionFact +} from '../../shared/agent-session-failure' import { agentSessionFailureWords } from '../../shared/agent-session-failure-words' import type { AgentSessionDispatchOutcome } from '../native-chat/agent-session-wire/structured-agent-session-adapter' import type { JsonlRpcRecord } from '../jsonl-rpc/peer' @@ -11,6 +15,8 @@ type Submission = { bytes: number accepted: boolean retries: number + /** Pi's last sign-in refusal, while the prompt waits to be re-sent. */ + signInRefusal?: string } type DeliveryDeps = { send: (frame: JsonlRpcRecord) => Promise @@ -31,6 +37,13 @@ export function piRpcFailureFact(error: string) { ) } +function piRejection(fact: SubmissionRejectionFact): AgentSessionDispatchOutcome { + return { + state: 'rejected', + ...agentSessionFailureWords(fact, { agentName: 'Pi', surface: 'rejection' }) + } +} + export class PiRpcPromptDelivery { private readonly acknowledgements: Submission[] = [] private readonly waiting: Submission[] = [] @@ -98,8 +111,10 @@ export class PiRpcPromptDelivery { throw new Error('Pi prompt reply has no request') } if (!reply.success) { - const fact = piRpcFailureFact(reply.error ?? 'Pi rejected the prompt') + const error = reply.error ?? 'Pi rejected the prompt' + const fact = piRpcFailureFact(error) if (!submission.accepted && fact.kind === 'notSignedIn' && submission.retries++ < 8) { + submission.signInRefusal = error this.removeWaiting(submission) const timer = setTimeout(() => { this.retries.delete(timer) @@ -111,6 +126,7 @@ export class PiRpcPromptDelivery { this.deps.failed(new Error('Pi retry lost its unaccepted prompt')) return } + submission.signInRefusal = undefined this.acknowledgements.push(submission) this.waiting.push(submission) void this.deps @@ -124,12 +140,14 @@ export class PiRpcPromptDelivery { return } this.remove(submission) - this.deps.settled(submission.id, { - state: 'rejected', - ...agentSessionFailureWords(fact, { agentName: 'Pi', surface: 'rejection' }) - }) + // Pi checks sign-in before it records a prompt (0.84.0 onward), so this one never reached it. + const settledFact = + fact.kind === 'notSignedIn' && !submission.accepted + ? { ...fact, neverReachedAgent: true as const } + : fact + this.deps.settled(submission.id, piRejection(settledFact)) if (submission.accepted) { - this.deps.rejectedAfterAcceptance(reply.error ?? 'Pi rejected the prompt') + this.deps.rejectedAfterAcceptance(error) } else { this.deps.refused?.() } @@ -150,7 +168,13 @@ export class PiRpcPromptDelivery { } this.retries.clear() for (const submission of this.pending) { - if (!submission.accepted) { + // Waiting out a sign-in refusal, not yet re-sent: every reply so far proved Pi never had it. + if (submission.signInRefusal !== undefined) { + this.deps.settled( + submission.id, + piRejection({ ...piRpcFailureFact(submission.signInRefusal), neverReachedAgent: true }) + ) + } else if (!submission.accepted) { this.deps.settled(submission.id, { state: 'unknown', reason: 'Pi ended before confirming delivery' diff --git a/src/shared/agent-session-failure.ts b/src/shared/agent-session-failure.ts index 877fc622ce7c..5ae24ee45884 100644 --- a/src/shared/agent-session-failure.ts +++ b/src/shared/agent-session-failure.ts @@ -164,6 +164,10 @@ export type AgentSessionFailureFact = { retry?: AgentSessionProviderRetry /** A safe option name from Orca's saved Arguments parser, never an error message. */ argumentProblem?: AgentSessionArgumentProblem + /** Written only where Orca proves the agent never received the message: a Stop's withdrawal is + * then owed as stopped history, and a refused carrier owes its history again. Absent means + * unknown. Host-only; clients read the kind. */ + neverReachedAgent?: true /** On `backgroundTaskNotStopped`: one named task, or every task the Stop reached. */ backgroundStop?: AgentSessionBackgroundStop } @@ -201,6 +205,7 @@ export function agentSessionFailureFact( attachment?: AgentSessionAttachmentProblem retry?: AgentSessionProviderRetry argumentProblem?: AgentSessionArgumentProblem + neverReachedAgent?: true backgroundStop?: AgentSessionBackgroundStop } = {} ): AgentSessionFailureFact & { kind: TKind } { @@ -216,6 +221,7 @@ export function agentSessionFailureFact( ...(extra.attachment ? { attachment: extra.attachment } : {}), ...(extra.retry ? { retry: extra.retry } : {}), ...(extra.argumentProblem ? { argumentProblem: extra.argumentProblem } : {}), + ...(extra.neverReachedAgent ? { neverReachedAgent: true as const } : {}), ...(extra.backgroundStop ? { backgroundStop: extra.backgroundStop } : {}) } } @@ -294,6 +300,7 @@ export function readAgentSessionFailureFact(value: unknown): AgentSessionFailure ...(attachment ? { attachment } : {}), ...(retry ? { retry } : {}), ...(argumentProblem ? { argumentProblem } : {}), + ...(value.neverReachedAgent === true ? { neverReachedAgent: true as const } : {}), ...(value.backgroundStop === 'task' || value.backgroundStop === 'all' ? { backgroundStop: value.backgroundStop } : {}) diff --git a/src/shared/agent-session-stopped-history.test.ts b/src/shared/agent-session-stopped-history.test.ts new file mode 100644 index 000000000000..6f1540e6497b --- /dev/null +++ b/src/shared/agent-session-stopped-history.test.ts @@ -0,0 +1,126 @@ +// What a Stop leaves owed to the agent: only messages Orca proves it never received, at most once. + +import { describe, expect, it } from 'vitest' +import { agentSessionFailureFact, type SubmissionRejectionKind } from './agent-session-failure' +import { agentSessionFailureWords } from './agent-session-failure-words' +import type { + AgentJournalItemBody, + AgentJournalMessageItem, + AgentJournalSubmission +} from './agent-session-journal-types' +import { MAX_PROMPT_BYTES } from './rpc-contract/structured-agent-session-message-params' +import { + agentSessionStoppedHistoryOwed, + selectStoppedHistory, + STOPPED_MESSAGE_FRAMING_BYTES, + type StoppedHistoryJournal +} from './agent-session-stopped-history' + +function message( + text: string, + extra: Partial = {} +): AgentJournalMessageItem { + return { + kind: 'message', + role: 'user', + blocks: [{ type: 'text', text }], + ...extra + } +} + +function sent( + id: string, + sequence: number, + extra: Partial = {} +): AgentJournalSubmission { + return { + clientMessageId: id, + fence: 1, + payloadFingerprint: `fp-${id}`, + dispatchState: 'pending', + providerItemId: null, + reason: null, + submittedAt: sequence, + resolvedAt: null, + handoverRecorded: true, + acceptedSequence: sequence, + origin: 'client', + source: { kind: 'user' }, + ...extra + } +} + +function rejected( + kind: SubmissionRejectionKind, + extra: { neverReachedAgent?: true } = {} +): Partial { + return { + dispatchState: 'rejected', + ...agentSessionFailureWords(agentSessionFailureFact(kind, extra), { + surface: 'rejection' + }) + } +} + +/** A Stop's withdrawal with proof the agent never got it. */ +const STOPPED = rejected('cancelled', { neverReachedAgent: true }) + +function owed( + submissions: AgentJournalSubmission[], + options: { + carried?: Record + clearSequence?: number + bodies?: Record + } = {} +): string[] { + const journal: StoppedHistoryJournal = { + submissions, + carried: new Map(Object.entries(options.carried ?? {})), + clearSequence: options.clearSequence ?? 0, + body: (id) => options.bodies?.[id] ?? message(id) + } + return agentSessionStoppedHistoryOwed(journal).map((entry) => entry.clientMessageId) +} + +describe('which stopped messages the agent is owed', () => { + it("owes only a person's own sends a Stop proved unwritten, in order, after the latest /clear", () => { + const mail = message('mail', { from: { kind: 'agent', senders: [], orchestration: null } }) + expect( + owed( + [ + sent('b', 8, STOPPED), + sent('a', 7, STOPPED), + sent('beforeClear', 2, STOPPED), + // Old journals and a Claude queue-clear carry no proof. + sent('unproven', 6, rejected('cancelled')), + sent('doubt', 6, { dispatchState: 'unknown', reason: 'doubt' }), + sent('mail', 6, STOPPED), + sent('slash', 6, STOPPED), + sent('card', 6, { ...STOPPED, queuedMessageId: 'card' }) + ], + { clearSequence: 5, bodies: { mail, slash: message('/review foo') } } + ) + ).toEqual(['a', 'b']) + }) +}) + +describe('what one delivery holds', () => { + it('fits beside the send, newest first; one too large for any send never holds back the rest', () => { + const largest = 'x'.repeat(MAX_PROMPT_BYTES - STOPPED_MESSAGE_FRAMING_BYTES - 5) + const tooLarge = 'x'.repeat(MAX_PROMPT_BYTES) + expect( + owed([sent('a', 2, STOPPED), sent('largest', 3, STOPPED), sent('tooLarge', 4, STOPPED)], { + bodies: { largest: message(largest), tooLarge: message(tooLarge) } + }) + ).toEqual(['a', 'largest']) + // The largest still owed fits beside a one-letter send; the older one is left out, covered. + const stopped = [ + { clientMessageId: 'a', words: 'first' }, + { clientMessageId: 'largest', words: largest } + ] + expect(selectStoppedHistory(stopped, message('x'))).toEqual({ + messages: [stopped[1]], + covers: ['a', 'largest'] + }) + }) +}) diff --git a/src/shared/agent-session-stopped-history.ts b/src/shared/agent-session-stopped-history.ts new file mode 100644 index 000000000000..fd951c69b451 --- /dev/null +++ b/src/shared/agent-session-stopped-history.ts @@ -0,0 +1,200 @@ +// Ordinary messages Orca proves the agent never received, owed to it at most once: derived from the +// journal on each handover (host only), and carried at the front of the person's next send that +// opens a turn, for every agent. One is covered by a carrier's `carries` unless that carrier was +// refused before anything reached the agent. Words only: a stopped message's images are not replayed. + +import { readAgentSessionFailureFact, type SubmissionRejectionKind } from './agent-session-failure' +import type { + AgentJournalItemBody, + AgentJournalMessageItem, + AgentJournalSubmission +} from './agent-session-journal-types' +import { + classifyDispatchRejection, + dispatchWasWithdrawn +} from './structured-agent-session-dispatch-rejection' +import { MAX_PROMPT_BYTES } from './rpc-contract/structured-agent-session-message-params' + +/** The record Claude writes for its own Stop, carried after each stopped message. */ +export const STOPPED_MESSAGE_INTERRUPTED_MARKER = '[Request interrupted by user]' + +export type StoppedHistoryJournal = { + submissions: readonly AgentJournalSubmission[] + /** Each handed-over send's carried history, by its client message id. */ + carried: ReadonlyMap + /** The latest /clear's sequence; 0 when none. */ + clearSequence: number + body: (clientMessageId: string) => AgentJournalItemBody | null +} + +export type OwedStoppedMessage = { + clientMessageId: string + words: string +} + +/** A person's ordinary message: not a command, a goal, mail another agent sent, or a typed slash + * command, which only reads right as the start of a prompt. */ +function isOrdinaryPersonMessage( + body: AgentJournalItemBody | null | undefined +): body is AgentJournalMessageItem { + if ( + body?.kind !== 'message' || + body.role !== 'user' || + body.command !== undefined || + body.sentAs !== undefined || + body.from !== undefined + ) { + return false + } + const firstText = body.blocks.find((block) => block.type === 'text' && block.text.length > 0) + return !(firstText?.type === 'text' && firstText.text.startsWith('/')) +} + +/** A Stop took it back where Orca proves the agent never received it. */ +function withdrawnUnwritten(submission: AgentJournalSubmission): boolean { + return ( + dispatchWasWithdrawn(submission) && + !submission.answeredInTurn && + readAgentSessionFailureFact(submission.rejection)?.neverReachedAgent === true + ) +} + +// Raised by Orca before any byte of a words-only send reached the agent; any other kind may follow +// a provider write. +const REFUSED_BEFORE_WRITE: ReadonlySet = new Set([ + 'queueFull', + 'writeFailed' +]) + +/** Refused before anything reached the agent: what it carried is owed again. */ +function refusedBeforeWrite(carrier: AgentJournalSubmission): boolean { + if (carrier.dispatchState !== 'rejected' || carrier.answeredInTurn) { + return false + } + // Proven unwritten where it was refused: a Stop's withdrawal, a child gone before the write, or a + // sign-in Pi checks before recording the prompt. + if (readAgentSessionFailureFact(carrier.rejection)?.neverReachedAgent === true) { + return true + } + const { kind } = classifyDispatchRejection(carrier) + return kind !== undefined && REFUSED_BEFORE_WRITE.has(kind) +} + +const encoder = new TextEncoder() + +/** Words as a send's size limit counts them. */ +function wordsBytes(words: string): number { + return encoder.encode(JSON.stringify(words)).length +} + +/** Per message, the framing `withStoppedHistory` writes beyond its words. */ +export const STOPPED_MESSAGE_FRAMING_BYTES = wordsBytes( + `\n\n${STOPPED_MESSAGE_INTERRUPTED_MARKER}\n\n` +) + +/** What a stopped message takes of the send it rides with. */ +function deliveryBytes(words: string): number { + return wordsBytes(words) + STOPPED_MESSAGE_FRAMING_BYTES +} + +/** The stopped messages owed to the agent, in the order they were sent. One too large to deliver + * beside even a one-letter send is not owed, so it never holds the others back. */ +export function agentSessionStoppedHistoryOwed( + journal: StoppedHistoryJournal +): OwedStoppedMessage[] { + const byId = new Map(journal.submissions.map((entry) => [entry.clientMessageId, entry])) + const covered = new Set() + for (const [carrierId, carried] of journal.carried) { + const carrier = byId.get(carrierId) + if (carrier && !refusedBeforeWrite(carrier)) { + carried.forEach((id) => covered.add(id)) + } + } + return journal.submissions + .flatMap((submission) => { + if ( + submission.acceptedSequence === undefined || + submission.acceptedSequence <= journal.clearSequence || + submission.origin !== 'client' || + submission.source?.kind !== 'user' || + submission.queuedMessageId !== undefined || + covered.has(submission.clientMessageId) || + !withdrawnUnwritten(submission) + ) { + return [] + } + const { clientMessageId, acceptedSequence } = submission + const body = journal.body(clientMessageId) + const words = isOrdinaryPersonMessage(body) ? messageWords(body) : '' + return words.length > 0 && deliveryBytes(words) + wordsBytes('x') <= MAX_PROMPT_BYTES + ? [{ clientMessageId, words, acceptedSequence }] + : [] + }) + .sort((left, right) => left.acceptedSequence - right.acceptedSequence) + .map(({ clientMessageId, words }) => ({ clientMessageId, words })) +} + +/** A message's words as an agent joins its text blocks. */ +function messageWords(body: AgentJournalMessageItem): string { + return body.blocks + .flatMap((block) => (block.type === 'text' && block.text.length > 0 ? [block.text] : [])) + .join('\n') +} + +/** One delivery: the newest owed message, and older ones while they fit, in the order sent. With + * the send's own words they never exceed what one send can hold (`MAX_PROMPT_BYTES`), so the agent + * is handed nothing larger than a send the person could make; if the newest doesn't fit, all stay + * owed. It covers every owed message, those left out included, so none arrives after a newer one. */ +export type StoppedHistoryDelivery = { + messages: OwedStoppedMessage[] + covers: string[] +} + +export function selectStoppedHistory( + owed: readonly OwedStoppedMessage[], + send: AgentJournalMessageItem +): StoppedHistoryDelivery | null { + const messages: OwedStoppedMessage[] = [] + let room = MAX_PROMPT_BYTES - wordsBytes(messageWords(send)) + for (const entry of owed.toReversed()) { + const cost = deliveryBytes(entry.words) + if (cost <= room) { + room -= cost + messages.unshift(entry) + } else if (messages.length === 0) { + return null + } + } + return messages.length > 0 + ? { messages, covers: owed.map((entry) => entry.clientMessageId) } + : null +} + +/** Whether this send's handover delivers the stopped messages owed: a person's ordinary message, a + * queued card's hand-off included. */ +export function agentSessionSendDeliversStoppedHistory( + submission: Pick, + body: AgentJournalItemBody | null | undefined +): body is AgentJournalMessageItem { + // A card's hand-off records no sender; its body's missing `from` says it is the person's. + const persons = submission.source?.kind === 'user' || submission.queuedMessageId !== undefined + return persons && isOrdinaryPersonMessage(body) +} + +/** Whether this send can carry stopped messages: words only, so the carried words can never take + * it past an agent's limits, which its attachments' unread size could. */ +export function agentSessionSendCarriesStoppedHistory(body: AgentJournalMessageItem): boolean { + return body.blocks.every((block) => block.type === 'text') && messageWords(body).length > 0 +} + +/** A words-only send's body with the delivered stopped messages ahead of its own words, each + * followed by the record Claude writes for its own Stop. */ +export function withStoppedHistory( + body: AgentJournalMessageItem, + messages: readonly OwedStoppedMessage[] +): AgentJournalMessageItem { + const carried = messages + .map((entry) => `${entry.words}\n\n${STOPPED_MESSAGE_INTERRUPTED_MARKER}\n\n`) + .join('') + return { ...body, blocks: [{ type: 'text', text: `${carried}${messageWords(body)}` }] } +}