diff --git a/config/scripts/vitest-sqlite-runtime-files.mjs b/config/scripts/vitest-sqlite-runtime-files.mjs index db4db00ed174..3c9d738d8ba9 100644 --- a/config/scripts/vitest-sqlite-runtime-files.mjs +++ b/config/scripts/vitest-sqlite-runtime-files.mjs @@ -120,11 +120,13 @@ export const SQLITE_RUNTIME_INCLUDE = [ 'src/main/native-chat/agent-session-wire/structured-agent-session-owed-work-release.test.ts', 'src/main/native-chat/agent-session-wire/structured-agent-session-pre-spawn-first-answer.test.ts', 'src/main/native-chat/agent-session-wire/structured-agent-session-processless-reservation.test.ts', + 'src/main/native-chat/agent-session-wire/structured-agent-session-prompt-answer-receipts.test.ts', 'src/main/native-chat/agent-session-wire/structured-agent-session-prompt-cancel.test.ts', 'src/main/native-chat/agent-session-wire/structured-agent-session-provider-child-record.test.ts', 'src/main/native-chat/agent-session-wire/structured-agent-session-provider-restore.test.ts', 'src/main/native-chat/agent-session-wire/structured-agent-session-provider-started-catalog.test.ts', 'src/main/native-chat/agent-session-wire/structured-agent-session-provider-started.test.ts', + 'src/main/native-chat/agent-session-wire/structured-agent-session-queued-action-receipts.test.ts', 'src/main/native-chat/agent-session-wire/structured-agent-session-queued-gate.test.ts', 'src/main/native-chat/agent-session-wire/structured-agent-session-queued-newer-orca.test.ts', 'src/main/native-chat/agent-session-wire/structured-agent-session-queued-pause-lift.test.ts', diff --git a/src/main/native-chat/agent-session-journal/command-receipt-schema.ts b/src/main/native-chat/agent-session-journal/command-receipt-schema.ts index 2502300785a0..e61de46588e0 100644 --- a/src/main/native-chat/agent-session-journal/command-receipt-schema.ts +++ b/src/main/native-chat/agent-session-journal/command-receipt-schema.ts @@ -18,7 +18,7 @@ export function commandReceiptScope( return operationIdScope === 'global' ? { kind: 'global' } : { kind: 'caller', callerKey } } -/** The pointer locates the result; if its row is gone, the command stays spent: read what remains, else unknown, never rerun. */ +/** What a replay answers from: a pointer (if its row is gone the command stays spent: read what remains, else unknown, never rerun), or the outcome itself. */ export const commandReceiptResultSchema = z.discriminatedUnion('kind', [ z.strictObject({ kind: z.literal('journal-row'), @@ -29,6 +29,28 @@ export const commandReceiptResultSchema = z.discriminatedUnion('kind', [ kind: z.literal('queued-draft'), messageId: z.string().min(1) }), + /** An accepted prompt answer, kept whole: no rewind or later revision changes its replay. Its + * values are the resolution the prompt's row holds, bounded as that row is. */ + z.strictObject({ + kind: z.literal('prompt-answer'), + itemId: z.string().min(1), + revision: z.int().positive(), + resolution: z.strictObject({ + state: z.literal('resolved'), + selectedOptionId: z.string().nullable(), + answers: z + .array( + z.strictObject({ + questionId: z.string(), + optionIds: z.array(z.string()), + other: z.string().optional() + }) + ) + .optional(), + resolvedBy: z.string().nullable(), + resolvedAt: z.number().nullable() + }) + }), z.strictObject({ kind: z.literal('no-op'), outcome: z.discriminatedUnion('kind', [ @@ -37,7 +59,12 @@ export const commandReceiptResultSchema = z.discriminatedUnion('kind', [ cancelled: z.literal(false), turnId: z.string().optional() }), - z.strictObject({ kind: z.literal('queue-resume'), resumed: z.literal(false) }) + z.strictObject({ kind: z.literal('queue-resume'), resumed: z.literal(false) }), + z.strictObject({ + kind: z.literal('queue-delete'), + messageId: z.string().min(1), + disposition: z.enum(['dispatched', 'withdrawn', 'missing']) + }) ]) }) ]) diff --git a/src/main/native-chat/agent-session-journal/journal-item-appender.ts b/src/main/native-chat/agent-session-journal/journal-item-appender.ts index 1faec275ffd7..c3b7d65c0a38 100644 --- a/src/main/native-chat/agent-session-journal/journal-item-appender.ts +++ b/src/main/native-chat/agent-session-journal/journal-item-appender.ts @@ -7,6 +7,7 @@ import { journalItemRowBuilder } from './journal-row-builders' import type { JournalReducerState } from './journal-reducer' import type { JournalAppendResult, JournalItemAppendOptions } from './journal-store-contracts' import type { JournalRow } from './journal-row-schema' +import type { JournalOperationReceipt } from './journal-row-writer' // A module-level Error's lazy stack retains the bundle's startup code and read buffer. class NothingResolvedError extends Error {} @@ -17,18 +18,22 @@ export class JournalItemAppender { constructor( private readonly deps: { state: () => JournalReducerState - enqueue: (build: (seq: number, ts: number) => JournalRow) => Promise + enqueue: ( + build: (seq: number, ts: number) => JournalRow, + receipt?: JournalOperationReceipt + ) => Promise } ) {} append( identity: AgentJournalItemIdentity, body: AgentJournalItemBody, - options: JournalItemAppendOptions + options: JournalItemAppendOptions, + receipt?: JournalOperationReceipt ): Promise { const itemId = agentJournalItemKey(identity) return this.deps - .enqueue(journalItemRowBuilder(this.deps.state, identity, body, options)) + .enqueue(journalItemRowBuilder(this.deps.state, identity, body, options), receipt) .then((row) => appendResult(row, itemId)) } diff --git a/src/main/native-chat/agent-session-journal/journal-open-failure.ts b/src/main/native-chat/agent-session-journal/journal-open-failure.ts index 73349ab09391..69ab0973e75d 100644 --- a/src/main/native-chat/agent-session-journal/journal-open-failure.ts +++ b/src/main/native-chat/agent-session-journal/journal-open-failure.ts @@ -115,6 +115,20 @@ export function journalOpenRefusal(error: unknown): AgentSessionWireRefusal { return refuse('agent_session_journal_unreadable', { reason }, message) } +/** A newer Orca's journal refuses a write with the words a send gets there. */ +export async function refusingNewerOrcaJournal( + run: Promise +): Promise { + try { + return await run + } catch (error) { + if (isJournalWrittenByNewerOrca(error)) { + return { ok: false, refusal: journalOpenRefusal(error) } + } + throw error + } +} + /** The same refusal, thrown: for a host that could not open the journal at all. It keeps what * failed as its cause, for the log. */ export function journalOpenRefusalError(error: unknown): AgentSessionRefusalError { diff --git a/src/main/native-chat/agent-session-journal/journal-queued-messages.ts b/src/main/native-chat/agent-session-journal/journal-queued-messages.ts index 1a66fc156bf7..e87fb450154c 100644 --- a/src/main/native-chat/agent-session-journal/journal-queued-messages.ts +++ b/src/main/native-chat/agent-session-journal/journal-queued-messages.ts @@ -11,7 +11,7 @@ import { AGENT_SESSION_OPERATION_FUTURE_SKEW_MS } from '../../../shared/agent-session-host-authority' import type { JournalRow } from './journal-row-schema' -import type { JournalOperationReceipt } from './journal-row-writer' +import { writeReceiptIfChanged, type JournalOperationReceipt } from './journal-row-writer' import type { JournalQueuedMessagesDeps, JournalSubmissionConsume } from './journal-store-contracts' import { holdQueuedMessages } from './queued-message-holds' import { @@ -92,6 +92,7 @@ export class JournalQueuedMessages { return this.transact( (db) => updateQueuedMessageText(db, this.deps.sessionId, input), (result) => result.status === 'updated', + undefined, () => this.editLeases.retire(input.messageId) ) } @@ -133,11 +134,10 @@ export class JournalQueuedMessages { queuedAt: { epoch, sequence: lastSequence }, now: this.deps.now() }) - receipt?.write(db) return row }, () => inserted, - receipt?.committed + receipt ) } @@ -199,11 +199,11 @@ export class JournalQueuedMessages { } /** Compare-and-transition waiting ∪ returned rows to op-stamped tombstones, - * kept only so a replay of the settling operation answers "spent". */ - withdraw(input: { - messageIds: readonly string[] - settledByOp: string - }): Promise { + * kept only so a replay of the settling operation answers "spent". `receipt`: with a withdrawal. */ + withdraw( + input: { messageIds: readonly string[]; settledByOp: string }, + receipt?: JournalOperationReceipt + ): Promise { if (input.messageIds.length === 0) { // Delete races and empty carries land here; neither may cost a write transaction. return Promise.resolve([]) @@ -215,7 +215,8 @@ export class JournalQueuedMessages { sessionId: this.deps.sessionId, now: this.deps.now() }), - (withdrawn) => withdrawn.length > 0 + (withdrawn) => withdrawn.length > 0, + receipt ) } @@ -234,17 +235,19 @@ export class JournalQueuedMessages { }).length } - /** One standalone draft-table transaction on the journal's queue; one that - * changed rows bumps the revision and notifies after COMMIT, `adopted` first. */ + /** One standalone draft-table transaction on the journal's queue; one that changed rows commits + * `receipt` with them, bumps the revision and notifies after COMMIT, receipt and `adopted` first. */ private transact( run: (db: Database.Database) => JournalWriteResult, changed: (result: T) => boolean, + receipt?: JournalOperationReceipt, adopted?: () => void ): Promise { return this.deps.serialize(() => { assertJournalWritable(this.deps.readOnly(), this.deps.sessionId) - const result = this.deps.database().transaction(run) + const result = this.deps.database().transaction(writeReceiptIfChanged(run, changed, receipt)) if (changed(result)) { + receipt?.committed() adopted?.() this.changeRevision++ this.deps.committed() diff --git a/src/main/native-chat/agent-session-journal/journal-row-writer.ts b/src/main/native-chat/agent-session-journal/journal-row-writer.ts index 6ff9985e21d6..4975155676b7 100644 --- a/src/main/native-chat/agent-session-journal/journal-row-writer.ts +++ b/src/main/native-chat/agent-session-journal/journal-row-writer.ts @@ -28,6 +28,21 @@ export function composeJournalOperationReceipts( } } +/** `run`, with `receipt` written in its transaction only when `changed` says it wrote something. */ +export function writeReceiptIfChanged( + run: (db: Database.Database) => T, + changed: (result: T) => boolean, + receipt?: JournalOperationReceipt +): (db: Database.Database) => T { + return (db) => { + const result = run(db) + if (receipt && changed(result)) { + receipt.write(db) + } + return result + } +} + export type JournalRowWriterDeps = { sessionId: string now: () => number 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 f5bc7b062a51..7ef45140b5af 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 @@ -16,7 +16,7 @@ import { JournalQueuedMessages } from './journal-queued-messages' import { JournalStopMarks } from './journal-stop-marks' import { journalQueuePauseRestatement } from './queued-message-pause' import type { JournalReducerState } from './journal-reducer' -import { JournalRowWriter } from './journal-row-writer' +import { JournalRowWriter, type JournalOperationReceipt } from './journal-row-writer' import { JournalStepWriter } from './journal-step-writer' import { restoreJournalStore } from './journal-store-restore' import { JournalSubmissionWriter } from './journal-submission-writer' @@ -42,7 +42,10 @@ export type JournalStoreHost = { adopt: (loaded: JournalLoad) => void commit: (rows: readonly JournalRow[], savedAt: number) => void journal: () => AgentSessionJournal - enqueue: (build: (seq: number, ts: number) => JournalRow) => Promise + enqueue: ( + build: (seq: number, ts: number) => JournalRow, + receipt?: JournalOperationReceipt + ) => Promise } export type JournalStoreCollaborators = { 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 5d485217b06d..8fe8c970b0a7 100644 --- a/src/main/native-chat/agent-session-journal/journal-store.ts +++ b/src/main/native-chat/agent-session-journal/journal-store.ts @@ -131,7 +131,7 @@ export class AgentSessionJournal { }, notifyCommitted: () => this.onCommitted?.(), journal: () => this, - enqueue: (build) => this.rowWriter.enqueue(build) + enqueue: (build, receipt) => this.rowWriter.enqueue(build, undefined, receipt) }) this.rowWriter = collaborators.rowWriter this.epochController = collaborators.epochController @@ -285,9 +285,10 @@ export class AgentSessionJournal { appendItem( identity: AgentJournalItemIdentity, body: AgentJournalItemBody, - options: JournalItemAppendOptions + options: JournalItemAppendOptions, + receipt?: JournalOperationReceipt ): Promise { - return this.itemAppender.append(identity, body, options) + return this.itemAppender.append(identity, body, options, receipt) } /** An upsert whose row is chosen from the fold at its own turn in the queue; null writes nothing. */ @@ -296,12 +297,10 @@ export class AgentSessionJournal { appendTombstone( identity: AgentJournalItemIdentity, - options: JournalTombstoneInput + { fence }: JournalTombstoneInput ): Promise { const itemId = agentJournalItemKey(identity) - return this.rowWriter.append( - journalTombstoneRowBuilder(() => this.state, itemId, options.fence) - ) + return this.rowWriter.append(journalTombstoneRowBuilder(() => this.state, itemId, fence)) } /** A Stop that took effect, timed by its row (`JournalStopEvent`). */ @@ -310,8 +309,9 @@ export class AgentSessionJournal { } /** A person's Resume of the queue. */ - appendQueueResume(fence: number): Promise { - return this.rowWriter.append(journalQueueResumeRowBuilder(() => this.state, fence)) + appendQueueResume(fence: number, receipt?: JournalOperationReceipt): Promise { + const build = journalQueueResumeRowBuilder(() => this.state, fence) + return this.rowWriter.append(build, undefined, receipt) } /** This open found waiting cards an earlier handle wrote (`queued-message-pause.ts`). */ diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-command-admission.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-command-admission.ts index 2de6745fe0d2..41e433f1ca24 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-command-admission.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-command-admission.ts @@ -1,5 +1,8 @@ import { parseAgentSessionOperationTimestamp } from '../../../shared/agent-session-host-authority' -import { agentSessionLedgerRefusal } from '../../../shared/agent-session-mutation-envelope' +import { + agentSessionLedgerRefusal, + refuseUnlessWriterAdmitted +} from '../../../shared/agent-session-mutation-envelope' import { agentSessionRefusalFromReference, isAgentSessionRefusalError @@ -15,6 +18,7 @@ import { refuseAgentSessionMutation } from './structured-agent-session-mutation-refusals' import type { AgentSessionMutationRequest } from './structured-agent-session-mutation-admission' +import type { MutationCommandReceipt } from './structured-agent-session-mutation-plans' import { mutationTurnContext } from './structured-agent-session-mutation-turn-context' import { runCommandReceiptMutation } from './structured-agent-session-command-receipt' import { @@ -22,10 +26,12 @@ import { resolveAgentSessionReplayOutcome } from './structured-agent-session-replay-outcome' -/** Submission-accepting plans use their effect's receipt, never a ledger reservation. */ +/** A plan with a command receipt accepts through it, committed with its effect, never a ledger + * reservation. */ export async function admitCommandReceiptMutation( request: AgentSessionMutationRequest, - hostFingerprint: string + hostFingerprint: string, + commandReceipt: MutationCommandReceipt ): Promise> { const { store, envelope, plan, callerKey } = request if (parseAgentSessionOperationTimestamp(envelope.clientOperationId) === null) { @@ -68,7 +74,11 @@ export async function admitCommandReceiptMutation( if (!journal || !current) { return refuseAgentSessionMutation(AGENT_SESSION_NOT_ATTACHED) } - // These plans write only to the conversation, whose journal enforces the execution-host fence. + // A conversation write is fenced by its journal; anything the provider must carry needs the lease. + const leaseRefusal = plan.conversationWrite ? null : refuseUnlessWriterAdmitted(current.lease) + if (leaseRefusal) { + return refuseAgentSessionMutation(leaseRefusal) + } const context = mutationTurnContext(request, journal, current) try { const outcome = await runCommandReceiptMutation({ @@ -78,6 +88,7 @@ export async function admitCommandReceiptMutation( wakeDelivery: request.wakeDelivery, envelope, plan, + commandReceipt, context }) if ('committedReceipt' in outcome) { @@ -151,7 +162,7 @@ async function answerCommandReceipt( const replay = resolveAgentSessionReplayOutcome({ operationId: envelope.clientOperationId, outcome, - reconstruct: () => plan.replay(context, outcome), + reconstruct: () => plan.replay(context, outcome, receipt.result), rerunWhenReplayMissing: plan.rerunWhenReplayMissing?.(context), recoverUnknownFromDurableState: plan.recoverUnknownFromDurableState }) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-command-receipt.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-command-receipt.ts index 763506a2c319..e011265fafb1 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-command-receipt.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-command-receipt.ts @@ -1,60 +1,52 @@ import type { AgentSessionMutationEnvelope } from '../../../shared/agent-session-wire' +import { isAgentSessionRefusalError } from '../../../shared/agent-session-wire-refusals' import type { AgentSessionRecordStore } from '../../runtime/agent-session-record-store' import { commandReceiptScope, - type CommandReceipt + type CommandReceipt, + type CommandReceiptResult } from '../agent-session-journal/command-receipt-schema' -import { buildCommandReceiptTransaction } from '../agent-session-journal/command-receipt-transaction' -import { composeJournalOperationReceipts } from '../agent-session-journal/journal-row-writer' -import type { MutationPlan } from './structured-agent-session-mutation-plans' +import { + buildCommandReceiptTransaction, + CommandReceiptExistsError +} from '../agent-session-journal/command-receipt-transaction' +import { + composeJournalOperationReceipts, + type JournalOperationReceipt +} from '../agent-session-journal/journal-row-writer' +import type { JournalRow } from '../agent-session-journal/journal-row-schema' +import type { + MutationCommandReceipt, + MutationPlan +} from './structured-agent-session-mutation-plans' +import { agentSessionOperationOutcomeUnknown } from './structured-agent-session-replay-outcome' import type { AgentSessionTurnContext, TurnOutcome } from './structured-agent-session-turns' -/** Acceptance identity and compatibility bookkeeping share the submission or draft's commit. */ -export async function runCommandReceiptMutation(input: { +type CommandReceiptIdentity = { store: AgentSessionRecordStore operationCallerKey: string fingerprint: string envelope: AgentSessionMutationEnvelope - plan: MutationPlan + plan: { method: string; operationIdScope?: 'global' } context: AgentSessionTurnContext - wakeDelivery?: (sessionId: string) => void -}): Promise | { committedReceipt: CommandReceipt }> { - const { store, operationCallerKey, fingerprint, envelope, plan, context, wakeDelivery } = input +} + +/** Acceptance identity and compatibility bookkeeping share the commit of the command's effect. */ +export async function runCommandReceiptMutation( + input: CommandReceiptIdentity & { + plan: MutationPlan + commandReceipt: MutationCommandReceipt + wakeDelivery?: (sessionId: string) => void + } +): Promise | { committedReceipt: CommandReceipt }> { + const { envelope, plan, context, commandReceipt, wakeDelivery } = input let acceptedReceipt: CommandReceipt | undefined let submissionWritten = false - const receipt = composeJournalOperationReceipts( - buildCommandReceiptTransaction( - commandReceiptScope(operationCallerKey, plan.operationIdScope), - (row) => { - if (row && row.kind !== 'submission') { - throw new Error('Send acceptance requires a submission') - } - submissionWritten = row !== undefined - acceptedReceipt = { - operationId: envelope.clientOperationId, - sessionId: envelope.sessionId, - callerKey: operationCallerKey, - method: plan.method, - fingerprint, - status: 'accepted', - acceptedAt: context.now(), - result: row - ? { kind: 'journal-row', epoch: row.epoch, sequence: row.seq } - : { kind: 'queued-draft', messageId: envelope.clientOperationId } - } - return acceptedReceipt - } - ), - // TEMPORARY: ledger co-write preserves cross-family identity until every mutation uses receipts. - store.operationOutcomeReceipt({ - callerKey: operationCallerKey, - operationId: envelope.clientOperationId, - fingerprint, - now: context.now(), - ...(plan.operationIdScope ? { operationIdScope: plan.operationIdScope } : {}), - outcome: { status: 'succeeded', sessionId: envelope.sessionId } - }) - ) + const receipt = acceptanceReceipt(input, (row) => { + submissionWritten = row?.kind === 'submission' + acceptedReceipt = acceptedCommandReceipt(input, commandReceipt.result(row)) + return acceptedReceipt + }) let committed = false try { const ran = await plan.run({ @@ -67,7 +59,14 @@ export async function runCommandReceiptMutation(input: { } } }) - if (!committed || ran.ok) { + if (!committed) { + const unwritten = ran.ok ? commandReceipt.unwritten?.(ran.value, context) : null + return unwritten && + !(await commitNoOpReceipt(input, acceptedCommandReceipt(input, unwritten))) + ? unknownOutcome(input) + : ran + } + if (ran.ok) { return ran } context.logger.warn('publishing an accepted command failed; replaying its receipt', { @@ -104,3 +103,75 @@ export async function runCommandReceiptMutation(input: { } return { committedReceipt: acceptedReceipt } } + +function acceptedCommandReceipt( + { envelope, operationCallerKey, plan, fingerprint, context }: CommandReceiptIdentity, + result: CommandReceiptResult +): CommandReceipt { + return { + operationId: envelope.clientOperationId, + sessionId: envelope.sessionId, + callerKey: operationCallerKey, + method: plan.method, + fingerprint, + status: 'accepted', + acceptedAt: context.now(), + result + } +} + +function acceptanceReceipt( + input: CommandReceiptIdentity, + receipt: (row?: JournalRow) => CommandReceipt +): JournalOperationReceipt { + const { store, operationCallerKey, envelope, plan, fingerprint, context } = input + return composeJournalOperationReceipts( + buildCommandReceiptTransaction( + commandReceiptScope(operationCallerKey, plan.operationIdScope), + receipt + ), + // TEMPORARY: ledger co-write preserves cross-family identity until every mutation uses receipts. + store.operationOutcomeReceipt({ + callerKey: operationCallerKey, + operationId: envelope.clientOperationId, + fingerprint, + now: context.now(), + ...(plan.operationIdScope ? { operationIdScope: plan.operationIdScope } : {}), + outcome: { status: 'succeeded', sessionId: envelope.sessionId } + }) + ) +} + +function unknownOutcome(input: CommandReceiptIdentity): TurnOutcome { + return { + ok: false, + refusal: agentSessionOperationOutcomeUnknown(input.envelope.clientOperationId) + } +} + +/** A no-op is answered only once its receipt commits, alone. An identity verdict or a classified + * refusal throws for admission to answer; an unclassified storage failure is logged and answered + * unknown, with nothing recorded, so a retry decides afresh. */ +async function commitNoOpReceipt( + input: CommandReceiptIdentity, + receipt: CommandReceipt +): Promise { + try { + await input.store.commitOperationReceipt(acceptanceReceipt(input, () => receipt)) + return true + } catch (error) { + if (error instanceof CommandReceiptExistsError || isAgentSessionRefusalError(error)) { + throw error + } + input.context.logger.warn( + 'recording a command that changed nothing failed; answering unknown', + { + scope: 'command-receipt-unwritten', + sessionId: input.envelope.sessionId, + operationId: input.envelope.clientOperationId, + error + } + ) + return false + } +} diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-host.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-host.test.ts index e92b33aac1c1..155e6cbb1d00 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-host.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-host.test.ts @@ -460,7 +460,7 @@ describe('respondToPrompt', () => { expect(answerPrompt).not.toHaveBeenCalled() }) - it("does not turn a recorded refusal into another client's successful answer", async () => { + it("does not turn a refused answer into another client's successful answer", async () => { await attach() const prompt = await seedApproval() const rejectedFields = { @@ -485,10 +485,12 @@ describe('respondToPrompt', () => { } ) + // A refusal records nothing, so the retry is judged afresh: the prompt has moved on. expect(await host.respondToPrompt(CALLER, rejected)).toMatchObject({ ok: false, - refusal: { code: 'agent_session_operation_invalid' } + refusal: { code: 'agent_session_item_revision_stale' } }) + expect(answerPrompt).toHaveBeenCalledOnce() }) it('keeps the answer and reports it undelivered when the provider callback throws', async () => { diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-mutation-admission.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-mutation-admission.ts index aec938b85fe6..54a1d769a001 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-mutation-admission.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-mutation-admission.ts @@ -1,5 +1,5 @@ // The one route every mutating agent-session call takes: recompute the fingerprint, -// dispatch submission-accepting plans to their command receipt, otherwise admit +// accept a plan that declares a command receipt through it, otherwise admit // through the ledger and check the lease. Methods cannot grow their own admission rules. // // Ledger-backed admission is two-phase when a call brings `prepareSession`. The ledger's @@ -98,8 +98,8 @@ export async function admitAndRunAgentSessionMutation( if (conflict) { return refuseAgentSessionMutation(conflict) } - if (plan.acceptsWithCommandReceipt) { - return admitCommandReceiptMutation(request, hostFingerprint) + if (plan.commandReceipt) { + return admitCommandReceiptMutation(request, hostFingerprint, plan.commandReceipt) } if (request.prepareSession) { const ledger = request.store.evaluateMutationOperation({ diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-mutation-plans.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-mutation-plans.ts index 6d4520f9c9d8..04688a8c491a 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-mutation-plans.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-mutation-plans.ts @@ -1,8 +1,9 @@ // One plan per mutating method: what it fingerprints, what it does, and how its // answer is rebuilt on a replay. // -// Send and /compact prove acceptance by their command receipt, the rest by their ledger row; the -// journal projects the current answer. +// A plan with `commandReceipt` proves acceptance by that receipt, the rest by their ledger row. A +// prompt answer's replay answers from the resolution its receipt keeps; the journal projects the +// others' current answer. import type { AgentJournalMessageItem } from '../../../shared/agent-session-journal-types' import { @@ -38,9 +39,17 @@ import { type AgentSessionTurnContext, type TurnOutcome } from './structured-agent-session-turns' -import type { AgentSessionPromptRequest } from './structured-agent-session-turns-prompt' +import { + acceptedPromptAnswer, + promptAnswerReceipt, + promptRowAnswer, + type AgentSessionPromptRequest +} from './structured-agent-session-turns-prompt' import { queuedSendAnswer } from './structured-agent-session-queued-send-answer' import type { JournalOperationReceipt } from '../agent-session-journal/journal-row-writer' +import type { CommandReceiptResult } from '../agent-session-journal/command-receipt-schema' +import type { JournalRow } from '../agent-session-journal/journal-row-schema' +import { refusingNewerOrcaJournal } from '../agent-session-journal/journal-open-failure' export type MutationPlan = { method: string @@ -51,13 +60,17 @@ export type MutationPlan = { /** Still runs, decided from the committed ledger, when its ledger row cannot be written. */ runsWithoutLedgerRow?: true run: (ctx: AgentSessionTurnContext) => Promise> - replay: (ctx: AgentSessionTurnContext, outcome: AgentSessionOperationOutcome) => TValue | null + /** `receipt`: the accepted command receipt's result, for a plan with `commandReceipt`. */ + replay: ( + ctx: AgentSessionTurnContext, + outcome: AgentSessionOperationOutcome, + receipt?: CommandReceiptResult + ) => TValue | null rerunWhenReplayMissing?: (ctx: AgentSessionTurnContext) => boolean recoverUnknownFromDurableState?: boolean } & ( | { - /** Accepts with a submission or draft; its command receipt commits in that write's transaction. */ - acceptsWithCommandReceipt: true + commandReceipt: MutationCommandReceipt settlesWithWrite?: never successReceipt?: never settledOutcome?: never @@ -65,17 +78,47 @@ export type MutationPlan = { | { /** Commits success with its row; paths without a committed receipt use fallback settlement. */ settlesWithWrite: true - acceptsWithCommandReceipt?: never + commandReceipt?: never successReceipt?: () => JournalOperationReceipt settledOutcome?: (value: TValue) => AgentSessionOperationOutcome } | { settlesWithWrite?: never - acceptsWithCommandReceipt?: never + commandReceipt?: never settledOutcome?: (value: TValue) => AgentSessionOperationOutcome } ) +/** Accepted through its own command receipt, inserted if absent in the transaction of its effect. */ +export type MutationCommandReceipt = { + /** Inside the effect's transaction: the journal row it wrote, or none for a draft-table write. */ + result: (row: JournalRow | undefined) => CommandReceiptResult + /** A run that wrote nothing: the receipt committed alone, before the answer goes out (a no-op's + * answer, or the earlier write it acknowledged), or null to record nothing. */ + unwritten?: (value: TValue, ctx: AgentSessionTurnContext) => CommandReceiptResult | null +} + +/** The journal row an accepted command wrote, which must be of the kind it accepts with. */ +export function journalRowReceiptResult( + row: JournalRow | undefined, + kind: JournalRow['kind'] +): CommandReceiptResult { + if (row?.kind !== kind) { + throw new Error(`an accepted command requires its ${kind} row`) + } + return { kind: 'journal-row', epoch: row.epoch, sequence: row.seq } +} + +/** A send accepts with its submission, or a draft held while the agent works, keyed by its id. */ +function submissionOrDraftReceipt(clientMessageId: string): MutationCommandReceipt { + return { + result: (row) => + row + ? journalRowReceiptResult(row, 'submission') + : { kind: 'queued-draft', messageId: clientMessageId } + } +} + /** Who a send is from, read off what it carries, the one place it is decided: the person's own * send, another agent's message (its body names the sender), or neither, such as a dispatch * preamble or a restart continuation. */ @@ -107,7 +150,7 @@ export function sendPlan(params: { method: 'agentSession.send', operationIdScope: 'global', conversationWrite: true, - acceptsWithCommandReceipt: true, + commandReceipt: submissionOrDraftReceipt(clientMessageId), // `delivery` joins the OPERATION fingerprint only; the submission row keeps // the body-only fingerprint the reducer's echo-aliasing recomputes. fields: { @@ -192,7 +235,7 @@ export function conversationCommandPlan(params: { method: 'agentSession.conversationCommand', operationIdScope: 'global', conversationWrite: true, - acceptsWithCommandReceipt: true, + commandReceipt: submissionOrDraftReceipt(clientMessageId), fields: { command: STRUCTURED_AGENT_SESSION_COMPACT_COMMAND, ...(params.delivery ? { delivery: params.delivery } : {}) @@ -281,17 +324,13 @@ export function promptPlan( optionId: params.optionId, answers: params.answers }, - run: (ctx) => performPrompt(ctx, params), - replay: (ctx) => { - const item = ctx.journal.snapshot().items.find((entry) => entry.itemId === params.itemId) - const body = item?.body - if (!item || !body || (body.kind !== 'approval' && body.kind !== 'question')) { - return null - } - return body.resolution.state === 'pending' - ? null - : { itemId: item.itemId, revision: item.revision, resolution: body.resolution } - } + // The resolved revision's answer, or the answer the prompt already held, kept in the receipt. + commandReceipt: { + result: (row) => promptAnswerReceipt(promptRowAnswer(row)), + unwritten: (value) => promptAnswerReceipt(value) + }, + run: (ctx) => refusingNewerOrcaJournal(performPrompt(ctx, params)), + replay: (_ctx, _outcome, receipt) => acceptedPromptAnswer(receipt) } } diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-operation-settlement.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-operation-settlement.test.ts index ce8e68798fbf..a3a2880e8430 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-operation-settlement.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-operation-settlement.test.ts @@ -16,12 +16,27 @@ import { HOST_TEST_THREAD as THREAD } from './structured-agent-session-host-test-data' import type { AgentSessionTurnContext } from './structured-agent-session-turns' -import { sendPlan } from './structured-agent-session-mutation-plans' +import { + sendPlan, + type MutationCommandReceipt, + type MutationPlan +} from './structured-agent-session-mutation-plans' import { runCommandReceiptMutation } from './structured-agent-session-command-receipt' import { createStructuredAgentSessionLogger } from './structured-agent-session-logger' import { codexProviderHandle } from '../../../shared/agent-session-provider-handle-encoding' import { NO_STRUCTURED_AGENTS } from './structured-agent-session-adapter-router-test-support' +const DRAFT_RECEIPT: MutationCommandReceipt = { + result: () => ({ kind: 'queued-draft', messageId: 'draft' }) +} + +function withReceipt(plan: MutationPlan) { + if (!plan.commandReceipt) { + throw new Error('expected a plan that accepts through its command receipt') + } + return { plan, commandReceipt: plan.commandReceipt } +} + async function context(): Promise { await attach() return { @@ -68,12 +83,13 @@ it('rethrows a throw from a plan answered by its write, writing nothing and leav plan: { method: 'agentSession.send', fields: {}, - settlesWithWrite: true, + commandReceipt: DRAFT_RECEIPT, run: async () => { throw refusal }, replay: () => null - } + }, + commandReceipt: DRAFT_RECEIPT }).catch((error: unknown) => error) expect(result).toBe(refusal) expect(writes).not.toHaveBeenCalled() @@ -82,11 +98,11 @@ it('rethrows a throw from a plan answered by its write, writing nothing and leav }) it.each([ - { plan: 'answered by its write', settlesWithWrite: true as const, failures: 0 }, - { plan: 'settled after its run', settlesWithWrite: undefined, failures: 2 } + { plan: 'answered by its write', acceptsWithReceipt: true, failures: 0 }, + { plan: 'settled after its run', acceptsWithReceipt: false, failures: 2 } ])( 'preserves a proven refusal of a plan $plan through $failures failed bookkeeping writes', - async ({ settlesWithWrite, failures }) => { + async ({ acceptsWithReceipt, failures }) => { const ctx = await context() const { store, dispatch } = hostTestState() const warning = vi.spyOn(console, 'warn').mockImplementation(() => {}) @@ -105,11 +121,12 @@ it.each([ context: ctx, plan: { method: 'agentSession.send', fields: {}, run, replay: () => null } } - const result = settlesWithWrite + const result = acceptsWithReceipt ? await runCommandReceiptMutation({ ...input, fingerprint: 'test-fingerprint', - plan: { ...input.plan, settlesWithWrite } + plan: { ...input.plan, commandReceipt: DRAFT_RECEIPT }, + commandReceipt: DRAFT_RECEIPT }) : await runSettledAgentSessionMutation(input) expect(result).toEqual(refusal) @@ -135,7 +152,7 @@ it('accepts without touching the provider', async () => { fingerprint: 'test-fingerprint', envelope: operation, context: ctx, - plan: sendPlan({ envelope: operation, body, beforeRun }) + ...withReceipt(sendPlan({ envelope: operation, body, beforeRun })) }) expect(result).toMatchObject({ ok: true }) expect(beforeRun).toHaveBeenCalledOnce() @@ -162,7 +179,7 @@ it('refuses a superseded send at acceptance, recording and dispatching nothing', fingerprint: 'test-fingerprint', envelope: operation, context: ctx, - plan: sendPlan({ envelope: operation, body, beforeRun }) + ...withReceipt(sendPlan({ envelope: operation, body, beforeRun })) }).catch((error: unknown) => error) expect(result).toBeInstanceOf(AgentSessionPreDispatchError) expect(writes).not.toHaveBeenCalled() @@ -189,10 +206,10 @@ it('returns the pre-acceptance refusal without recording a receipt', async () => fingerprint: 'test-fingerprint', envelope: operation, context: ctx, - plan: { + ...withReceipt({ ...sendPlan({ envelope: operation, body }), run: async () => ({ ok: false, refusal }) - } + }) }) ).toMatchObject({ ok: false, refusal }) expect(writes).not.toHaveBeenCalled() diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-operation-settlement.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-operation-settlement.ts index 56c49b42d6c0..d5de0fbf324b 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-operation-settlement.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-operation-settlement.ts @@ -16,7 +16,7 @@ export async function runSettledAgentSessionMutation(input: { store: AgentSessionRecordStore operationCallerKey: string envelope: AgentSessionMutationEnvelope - plan: Exclude, { acceptsWithCommandReceipt: true }> + plan: Exclude, { commandReceipt: object }> context: AgentSessionTurnContext }): Promise> { const operation = { diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-prompt-answer-receipts.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-prompt-answer-receipts.test.ts new file mode 100644 index 000000000000..87ad1fc95a11 --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-prompt-answer-receipts.test.ts @@ -0,0 +1,397 @@ +// A prompt answer accepts through its own command receipt, committed with the resolved revision +// before the provider hears the answer: a retry answers from that receipt and the provider hears +// each answer once. + +import { beforeEach, describe, expect, it, type Mock } from 'vitest' +import { AGENT_SESSION_MAX_NEW_OPERATION_AGE_MS } from '../../../shared/agent-session-host-authority' +import { + AGENT_JOURNAL_THREAD_SCOPE, + type AgentJournalItemBody +} from '../../../shared/agent-session-journal-types' +import { agentJournalItemKey } from '../../../shared/agent-session-journal-item-key' +import type { AgentSessionRecordStore } from '../../runtime/agent-session-record-store' +import { + openTestJournalHostDatabase, + SAVED_BY_NEWER_ORCA +} from '../agent-session-journal/journal-host-database-test-support' +import type { StructuredAgentSessionAdapter } from './structured-agent-session-adapter' +import type { StructuredAgentSessionHost } from './structured-agent-session-host' +import { + attach, + CALLER, + envelope, + hostTestState, + seedApproval +} from './structured-agent-session-host-test-harness' +import { + HOST_TEST_NOW as NOW, + HOST_TEST_SESSION as SESSION, + HOST_TEST_THREAD as THREAD +} from './structured-agent-session-host-test-data' + +const OLD = NOW - AGENT_SESSION_MAX_NEW_OPERATION_AGE_MS - 60_000 +const FUTURE = NOW + 6 * 60_000 +const CONFLICT = { + ok: false, + refusal: { code: 'agent_session_operation_conflict', details: { reason: 'operationIdReused' } } +} +const UNKNOWN = { ok: false, refusal: { code: 'agent_session_operation_unknown' } } +/** The provider identity `seedApproval` raises its prompt under. */ +const PROMPT = { provider: 'codex' as const, threadId: THREAD, turnId: 'turn-1', ordinal: 99 } + +let root: string +let store: AgentSessionRecordStore +let host: StructuredAgentSessionHost +let answerPrompt: Mock +let acquire: Mock +let serial = 0 + +beforeEach(() => { + ;({ root, store, host, answerPrompt, acquire } = hostTestState()) +}) + +function opId(at = NOW): string { + serial += 1 + return `${at}-${serial.toString(16).padStart(32, 'd')}` +} + +async function seeded() { + await attach() + return seedApproval() +} + +function answer( + prompt: { itemId: string; revision: number }, + clientOperationId: string, + optionId = 'allow' +) { + const fields = { itemId: prompt.itemId, expectedRevision: prompt.revision, optionId } + return { + envelope: envelope('agentSession.respondTo:approval', fields, { clientOperationId }), + kind: 'approval' as const, + ...fields + } +} + +function receipt(id: string, callerKey = CALLER.callerKey) { + return store.readCommandReceipt({ kind: 'caller', callerKey }, id) +} + +function journal() { + const open = host.collaboratorsForTests().sessions.get(SESSION)?.journal + if (!open) { + throw new Error('expected an open conversation') + } + return open +} + +function promptBody(itemId: string): AgentJournalItemBody { + const body = journal().itemBody(itemId) + if (!body) { + throw new Error(`no item ${itemId}`) + } + return body +} + +/** The provider writes the answered prompt again, as a later echo of it would. */ +async function reviseFromProvider(itemId: string): Promise { + const body = promptBody(itemId) + if (body.kind !== 'approval') { + throw new Error('expected an approval') + } + const events = acquire.mock.calls.at(-1)?.[0].events + events?.appendItem( + PROMPT, + { + ...body, + resolution: { ...body.resolution, resolvedBy: 'provider-echo', resolvedAt: NOW + 1 } + }, + { turnScope: AGENT_JOURNAL_THREAD_SCOPE } + ) + await host.flushStreamedEvents(SESSION) +} + +/** What a rewind does to the conversation: a new epoch, the prompt rebuilt into it as it stands. */ +async function rewindKeepingPrompt(itemId: string): Promise { + const fence = store.getRecord(SESSION)!.lease.runtimeFence + await journal().replaceEpochItems('handle_forked', fence, [ + { identity: PROMPT, body: promptBody(itemId), turnScope: AGENT_JOURNAL_THREAD_SCOPE } + ]) +} + +function failReceiptWrites(): () => void { + const db = openTestJournalHostDatabase(root).db + db.exec(`CREATE TRIGGER fail_prompt_receipt BEFORE INSERT ON agent_session_command_receipts + BEGIN SELECT RAISE(ABORT, 'receipt unavailable'); END`) + return () => db.exec('DROP TRIGGER fail_prompt_receipt') +} + +async function resolution(itemId: string) { + const page = await host.history({ sessionId: SESSION, direction: 'tail' }) + const item = page.ok ? page.page.items.find((entry) => entry.itemId === itemId) : undefined + return item?.body.kind === 'approval' ? item.body.resolution.state : undefined +} + +describe('a prompt answer', () => { + it.each([ + ['older than a day', OLD], + ['six minutes ahead', FUTURE] + ])('accepts an id %s with its resolved revision', async (_case, at) => { + const prompt = await seeded() + const id = opId(at) + expect(await host.respondToPrompt(CALLER, answer(prompt, id))).toMatchObject({ + ok: true, + replayed: false, + value: { itemId: prompt.itemId, resolution: { state: 'resolved' } } + }) + expect(receipt(id)).toMatchObject({ + verdict: 'readable', + receipt: { method: 'agentSession.respondTo:approval', result: { kind: 'prompt-answer' } } + }) + expect(answerPrompt).toHaveBeenCalledOnce() + }) + + it('delivers once for the same id twice, in turn or at once', async () => { + const prompt = await seeded() + const params = answer(prompt, opId()) + const [first, second] = await Promise.all([ + host.respondToPrompt(CALLER, params), + host.respondToPrompt(CALLER, params) + ]) + const third = await host.respondToPrompt(CALLER, params) + expect(first).toMatchObject({ ok: true, replayed: false }) + expect(second).toMatchObject({ + ok: true, + replayed: true, + value: { revision: prompt.revision + 1 } + }) + expect(third).toMatchObject({ + ok: true, + replayed: true, + value: { revision: prompt.revision + 1 } + }) + expect(answerPrompt).toHaveBeenCalledOnce() + }) + + it('refuses the same id with another choice', async () => { + const prompt = await seeded() + const id = opId() + await host.respondToPrompt(CALLER, answer(prompt, id)) + expect(await host.respondToPrompt(CALLER, answer(prompt, id, 'deny'))).toMatchObject(CONFLICT) + expect(answerPrompt).toHaveBeenCalledOnce() + }) + + it('points another id at the revision that already holds its choice and delivers nothing', async () => { + const prompt = await seeded() + await host.respondToPrompt(CALLER, answer(prompt, opId())) + const alias = opId() + const other = { callerKey: 'client-2' } + expect(await host.respondToPrompt(other, answer(prompt, alias))).toMatchObject({ + ok: true, + replayed: false, + value: { revision: prompt.revision + 1 } + }) + expect(receipt(alias, other.callerKey)).toMatchObject({ + receipt: { + result: { + kind: 'prompt-answer', + itemId: prompt.itemId, + revision: prompt.revision + 1, + resolution: { state: 'resolved', selectedOptionId: 'allow' } + } + } + }) + expect(await host.respondToPrompt(other, answer(prompt, alias))).toMatchObject({ + ok: true, + replayed: true + }) + expect(answerPrompt).toHaveBeenCalledOnce() + }) + + it('rolls the revision back when its receipt cannot be saved', async () => { + const prompt = await seeded() + const id = opId() + const restore = failReceiptWrites() + await expect(host.respondToPrompt(CALLER, answer(prompt, id))).rejects.toThrow( + 'receipt unavailable' + ) + restore() + expect(await resolution(prompt.itemId)).toBe('pending') + expect(receipt(id)).toEqual({ verdict: 'absent' }) + expect(store.listOperationRows().find((row) => row.operationId === id)).toBeUndefined() + }) + + it('answers an unreadable receipt as unknown without asking the provider', async () => { + const prompt = await seeded() + const id = opId() + openTestJournalHostDatabase(root) + .db.prepare( + `INSERT INTO agent_session_command_receipts (operation_id, session_id, caller_key, method, + fingerprint, status, result_json, rejection_json, accepted_at) + VALUES (?, ?, ?, 'agentSession.respondTo:approval', 'x', 'accepted', '{', NULL, 0)` + ) + .run(id, SESSION, CALLER.callerKey) + expect(await host.respondToPrompt(CALLER, answer(prompt, id))).toMatchObject({ + ok: false, + refusal: { code: 'agent_session_operation_unknown' } + }) + expect(answerPrompt).not.toHaveBeenCalled() + expect(await resolution(prompt.itemId)).toBe('pending') + }) + + it("refuses with the update words when a newer Orca's journal holds the chat", async () => { + const prompt = await seeded() + const id = opId() + Object.defineProperty(openTestJournalHostDatabase(root), 'readOnly', { value: true }) + expect(await host.respondToPrompt(CALLER, answer(prompt, id))).toMatchObject({ + ok: false, + ...SAVED_BY_NEWER_ORCA + }) + expect(receipt(id)).toEqual({ verdict: 'absent' }) + }) + + it('records nothing for a refusal before acceptance', async () => { + const prompt = await seeded() + const id = opId() + expect(await host.respondToPrompt(CALLER, answer(prompt, id, 'deny'))).toMatchObject({ + ok: false, + refusal: { code: 'agent_session_operation_invalid' } + }) + expect(receipt(id)).toEqual({ verdict: 'absent' }) + expect(store.listOperationRows().find((row) => row.operationId === id)).toBeUndefined() + }) + + it('replays the resolution it was accepted with after the provider revises the prompt', async () => { + const prompt = await seeded() + const params = answer(prompt, opId()) + const first = await host.respondToPrompt(CALLER, params) + expect(first).toMatchObject({ ok: true, replayed: false }) + await reviseFromProvider(prompt.itemId) + expect(journal().item(prompt.itemId)?.revision).toBe(prompt.revision + 2) + const replay = await host.respondToPrompt(CALLER, params) + expect(replay).toEqual({ ...first, replayed: true, cursor: expect.anything() }) + expect(answerPrompt).toHaveBeenCalledOnce() + }) + + it('replays the revision another id acknowledged after the provider revises the prompt', async () => { + const prompt = await seeded() + await host.respondToPrompt(CALLER, answer(prompt, opId())) + const other = { callerKey: 'client-2' } + const params = answer(prompt, opId()) + const first = await host.respondToPrompt(other, params) + await reviseFromProvider(prompt.itemId) + const replay = await host.respondToPrompt(other, params) + expect(replay).toEqual({ ...first, replayed: true, cursor: expect.anything() }) + expect(replay).toMatchObject({ + value: { revision: prompt.revision + 1, resolution: { resolvedBy: CALLER.callerKey } } + }) + }) + + it.each([ + ['its own answer', CALLER], + ['the answer another id acknowledged', { callerKey: 'client-2' }] + ])( + 'replays the accepted answer after a rewind, never answering again: %s', + async (_case, caller) => { + const prompt = await seeded() + const params = answer(prompt, opId()) + if (caller !== CALLER) { + await host.respondToPrompt(CALLER, answer(prompt, opId())) + } + const first = await host.respondToPrompt(caller, params) + expect(first).toMatchObject({ ok: true, replayed: false }) + await rewindKeepingPrompt(prompt.itemId) + expect(journal().item(prompt.itemId)?.revision).not.toBe(prompt.revision + 1) + expect(await host.respondToPrompt(caller, params)).toEqual({ + ...first, + replayed: true, + cursor: expect.anything() + }) + expect(answerPrompt).toHaveBeenCalledOnce() + } + ) + + it('replays a grouped answer whole after a rewind, free text included', async () => { + await attach() + const question = { ...PROMPT, ordinal: 100 } + acquire.mock.calls.at(-1)?.[0].events?.appendItem( + question, + { + kind: 'question', + question: '2 grouped questions', + options: [], + questions: [ + { + id: 'q1', + question: 'Targets', + multiSelect: true, + options: [ + { id: 'web', label: 'Web' }, + { id: 'mobile', label: 'Mobile' } + ] + }, + { id: 'q2', question: 'Host', multiSelect: false, options: [], freeTextQuestionId: 'q2' } + ], + resolution: { state: 'pending', selectedOptionId: null, resolvedBy: null, resolvedAt: null } + }, + { turnScope: AGENT_JOURNAL_THREAD_SCOPE } + ) + await host.flushStreamedEvents(SESSION) + const itemId = agentJournalItemKey(question) + const answers = [ + { questionId: 'q1', optionIds: ['web', 'mobile'] }, + { questionId: 'q2', optionIds: [], other: 'SSH host' } + ] + const fields = { itemId, expectedRevision: journal().item(itemId)!.revision, answers } + const params = { + envelope: envelope('agentSession.respondTo:question', fields, { clientOperationId: opId() }), + kind: 'question' as const, + ...fields + } + const first = await host.respondToPrompt(CALLER, params) + expect(first).toMatchObject({ ok: true, value: { resolution: { answers } } }) + await journal().replaceEpochItems( + 'handle_forked', + store.getRecord(SESSION)!.lease.runtimeFence, + [] + ) + expect(await host.respondToPrompt(CALLER, params)).toEqual({ + ...first, + replayed: true, + cursor: expect.anything() + }) + expect(answerPrompt).toHaveBeenCalledOnce() + }) + + it('answers unknown, never success, when acknowledging an answer cannot be recorded', async () => { + const prompt = await seeded() + await host.respondToPrompt(CALLER, answer(prompt, opId())) + const other = { callerKey: 'client-2' } + const id = opId() + const restore = failReceiptWrites() + expect(await host.respondToPrompt(other, answer(prompt, id))).toMatchObject(UNKNOWN) + restore() + expect(receipt(id, other.callerKey)).toEqual({ verdict: 'absent' }) + expect(answerPrompt).toHaveBeenCalledOnce() + }) + + it('refuses a Stop reusing its id, and an answer reusing a Stop id', async () => { + const prompt = await seeded() + const answered = opId() + await host.respondToPrompt(CALLER, answer(prompt, answered)) + expect( + await host.cancel(CALLER, { + envelope: envelope('agentSession.cancel', {}, { clientOperationId: answered }) + }) + ).toMatchObject(CONFLICT) + const second = await seedApproval('approve') + const stopped = opId() + await host.cancel(CALLER, { + envelope: envelope('agentSession.cancel', {}, { clientOperationId: stopped }) + }) + expect(await host.respondToPrompt(CALLER, answer(second, stopped, 'approve'))).toMatchObject( + CONFLICT + ) + expect(receipt(stopped)).toEqual({ verdict: 'absent' }) + }) +}) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-queued-action-receipts.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-queued-action-receipts.test.ts new file mode 100644 index 000000000000..06d3bd164bc6 --- /dev/null +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-queued-action-receipts.test.ts @@ -0,0 +1,551 @@ +// Send now, Delete and Resume accept through their own command receipt, committed with the draft or +// row they change: no ledger policy decides them, a retry answers from that receipt, and an id +// another command already used is refused. + +import { afterEach, beforeEach, describe, expect, it } from 'vitest' +import { AGENT_SESSION_MAX_NEW_OPERATION_AGE_MS } from '../../../shared/agent-session-host-authority' +import { openTestJournalHostDatabase } from '../agent-session-journal/journal-host-database-test-support' +import { + createQueuedMessageTestRig, + QUEUED_RIG_CALLER as CALLER, + type QueuedMessageTestRig +} from './structured-agent-session-queued-message-rig.test-fixture' +import { + HOST_TEST_NOW as NOW, + HOST_TEST_SESSION as SESSION +} from './structured-agent-session-host-test-data' + +const CALLER_SCOPE = { kind: 'caller', callerKey: CALLER.callerKey } as const +const OLD = NOW - AGENT_SESSION_MAX_NEW_OPERATION_AGE_MS - 60_000 +const FUTURE = NOW + 6 * 60_000 + +let rig: QueuedMessageTestRig +let serial = 0 + +beforeEach(async () => { + rig = await createQueuedMessageTestRig() +}) + +afterEach(() => rig.dispose()) + +function opId(at = NOW): string { + serial += 1 + return `${at}-${serial.toString(16).padStart(32, 'e')}` +} + +async function queuedDraft(text: string): Promise { + const queued = await rig.send(text, 'queue-if-active').result + if (!queued.ok || !('queued' in queued.value)) { + throw new Error('expected a queued receipt') + } + return queued.value.queued.messageId +} + +function receipt(id: string) { + return rig.store.readCommandReceipt(CALLER_SCOPE, id) +} + +function ledgerRow(id: string) { + return rig.store.listOperationRows().find((row) => row.operationId === id) +} + +function failReceiptWrites(): () => void { + const db = openTestJournalHostDatabase(rig.root).db + db.exec(`CREATE TRIGGER fail_queued_receipt BEFORE INSERT ON agent_session_command_receipts + BEGIN SELECT RAISE(ABORT, 'receipt unavailable'); END`) + return () => db.exec('DROP TRIGGER fail_queued_receipt') +} + +function writeUnreadableReceipt(id: string, method: string): void { + openTestJournalHostDatabase(rig.root) + .db.prepare( + `INSERT INTO agent_session_command_receipts (operation_id, session_id, caller_key, method, + fingerprint, status, result_json, rejection_json, accepted_at) + VALUES (?, ?, ?, ?, 'x', 'accepted', '{', NULL, 0)` + ) + .run(id, SESSION, CALLER.callerKey, method) +} + +/** What a rewind does to the conversation: a new epoch rebuilt with none of the old rows. */ +async function rewindConversation(): Promise { + const journal = rig.host.collaboratorsForTests().sessions.get(SESSION)?.journal + if (!journal) { + throw new Error('expected an open conversation') + } + await journal.replaceEpochItems( + 'handle_forked', + rig.store.getRecord(SESSION)!.lease.runtimeFence, + [] + ) +} + +/** A stop that leaves a waiting card paused, which only a Resume lifts. */ +async function pausedQueue(): Promise { + await rig.workingSend() + await queuedDraft('held by stop') + await rig.stop() + expect(await rig.queuePause()).toEqual({ reason: 'stopped' }) +} + +const UNKNOWN = { ok: false, refusal: { code: 'agent_session_operation_unknown' } } +const CONFLICT = { + ok: false, + refusal: { code: 'agent_session_operation_conflict', details: { reason: 'operationIdReused' } } +} + +describe('Send now', () => { + it.each([ + ['older than a day', OLD], + ['six minutes ahead', FUTURE] + ])('accepts an id %s and hands the card off once', async (_case, at) => { + await rig.workingSend() + const draftId = await queuedDraft('send me now') + const id = opId(at) + expect(await rig.sendNow(draftId, id)).toMatchObject({ + ok: true, + replayed: false, + value: { clientMessageId: id, submission: { queuedMessageId: draftId } } + }) + expect(receipt(id)).toMatchObject({ + verdict: 'readable', + receipt: { method: 'agentSession.queuedMessageSend', result: { kind: 'journal-row' } } + }) + expect(ledgerRow(id)?.outcome).toMatchObject({ status: 'succeeded' }) + }) + + it('answers the same id twice, in turn or at once, from one hand-off', async () => { + await rig.workingSend() + const draftId = await queuedDraft('send me once') + const id = opId() + const [first, second] = await Promise.all([rig.sendNow(draftId, id), rig.sendNow(draftId, id)]) + const third = await rig.sendNow(draftId, id) + expect(first).toMatchObject({ ok: true, replayed: false }) + expect(second).toMatchObject({ ok: true, replayed: true, value: { clientMessageId: id } }) + expect(third).toMatchObject({ ok: true, replayed: true, value: { clientMessageId: id } }) + const handoffs = (await rig.host.journalSnapshot(SESSION)).submissions.filter( + (entry) => entry.queuedMessageId === draftId + ) + expect(handoffs).toHaveLength(1) + }) + + it('points another id at the hand-off that already took the card', async () => { + await rig.workingSend() + const draftId = await queuedDraft('already sent') + const first = opId() + await rig.sendNow(draftId, first) + const alias = opId() + expect(await rig.sendNow(draftId, alias)).toMatchObject({ + ok: true, + replayed: false, + value: { clientMessageId: first } + }) + expect(receipt(alias)).toMatchObject({ receipt: { result: { kind: 'journal-row' } } }) + expect(await rig.sendNow(draftId, alias)).toMatchObject({ + ok: true, + replayed: true, + value: { clientMessageId: first } + }) + }) + + it('replays the hand-off it acknowledged, not a later one that took the card after it came back', async () => { + await rig.workingSend() + const draftId = await queuedDraft('card') + const first = opId() + expect(await rig.sendNow(draftId, first)).toMatchObject({ value: { clientMessageId: first } }) + const ack = opId() + expect(await rig.sendNow(draftId, ack)).toMatchObject({ + ok: true, + replayed: false, + value: { clientMessageId: first } + }) + // The provider refuses the first hand-off: the card comes back, and another Send takes it. + await rig.settleRejected(first, 'nope') + const later = opId() + expect(await rig.sendNow(draftId, later)).toMatchObject({ + ok: true, + value: { clientMessageId: later } + }) + expect(await rig.sendNow(draftId, ack)).toMatchObject({ + ok: true, + replayed: true, + value: { clientMessageId: first } + }) + }) + + it('replays its hand-off once the dispatched card is pruned', async () => { + await rig.workingSend() + const draftId = await queuedDraft('pruned card') + const first = opId() + await rig.sendNow(draftId, first) + const ack = opId() + await rig.sendNow(draftId, ack) + openTestJournalHostDatabase(rig.root) + .db.prepare('DELETE FROM queued_messages WHERE session_id = ? AND message_id = ?') + .run(SESSION, draftId) + for (const id of [first, ack]) { + expect(await rig.sendNow(draftId, id)).toMatchObject({ + ok: true, + replayed: true, + value: { clientMessageId: first } + }) + } + }) + + it('refuses the same id naming another card', async () => { + await rig.workingSend() + const first = await queuedDraft('first card') + const second = await queuedDraft('second card') + const id = opId() + await rig.sendNow(first, id) + expect(await rig.sendNow(second, id)).toMatchObject({ + ok: false, + refusal: { + code: 'agent_session_operation_conflict', + details: { reason: 'operationIdReused' } + } + }) + expect(await rig.drafts()).toContainEqual(expect.objectContaining({ messageId: second })) + }) + + it('rolls the hand-off back when its receipt cannot be saved, then sends on a retry', async () => { + await rig.workingSend() + const draftId = await queuedDraft('roll me back') + const id = opId() + const restore = failReceiptWrites() + await expect(rig.sendNow(draftId, id)).rejects.toThrow('receipt unavailable') + expect(await rig.handoff(draftId)).toBeUndefined() + expect(await rig.drafts()).toContainEqual(expect.objectContaining({ messageId: draftId })) + expect(receipt(id)).toEqual({ verdict: 'absent' }) + expect(ledgerRow(id)).toBeUndefined() + restore() + expect(await rig.sendNow(draftId, id)).toMatchObject({ ok: true, replayed: false }) + }) + + it('records nothing for a refusal before acceptance', async () => { + const id = opId() + expect(await rig.sendNow('no-such-card', id)).toMatchObject({ + ok: false, + refusal: { code: 'agent_session_operation_invalid' } + }) + expect(receipt(id)).toEqual({ verdict: 'absent' }) + expect(ledgerRow(id)).toBeUndefined() + }) + + it('answers unknown, never success, when acknowledging a hand-off cannot be recorded', async () => { + await rig.workingSend() + const draftId = await queuedDraft('sent by another') + const first = opId() + await rig.sendNow(draftId, first) + const alias = opId() + const restore = failReceiptWrites() + expect(await rig.sendNow(draftId, alias)).toMatchObject(UNKNOWN) + restore() + expect(receipt(alias)).toEqual({ verdict: 'absent' }) + expect(ledgerRow(alias)).toBeUndefined() + const handoffs = (await rig.host.journalSnapshot(SESSION)).submissions.filter( + (entry) => entry.queuedMessageId === draftId + ) + expect(handoffs).toHaveLength(1) + }) + + it('answers an unreadable receipt as unknown and hands nothing off', async () => { + await rig.workingSend() + const draftId = await queuedDraft('unknown send') + const id = opId() + writeUnreadableReceipt(id, 'agentSession.queuedMessageSend') + expect(await rig.sendNow(draftId, id)).toMatchObject(UNKNOWN) + expect(await rig.handoff(draftId)).toBeUndefined() + }) + + it('answers a retry after a rewind dropped its hand-off as spent, and sends nothing again', async () => { + await rig.workingSend() + const draftId = await queuedDraft('rewound away') + const id = opId() + expect(await rig.sendNow(draftId, id)).toMatchObject({ ok: true, replayed: false }) + await rewindConversation() + expect(await rig.sendNow(draftId, id)).toMatchObject({ + ok: false, + refusal: { code: 'agent_session_operation_unknown', details: { reason: 'resultLost' } } + }) + expect(await rig.handoff(draftId)).toBeUndefined() + }) +}) + +describe('Delete', () => { + it('withdraws once under an old or future id and replays the withdrawal', async () => { + await rig.workingSend() + const draftId = await queuedDraft('delete me') + const id = opId(OLD) + expect(await rig.deleteQueued(draftId, id)).toMatchObject({ + ok: true, + replayed: false, + value: { deleted: true, messageId: draftId } + }) + expect(receipt(id)).toMatchObject({ + receipt: { result: { kind: 'queued-draft', messageId: draftId } } + }) + expect(await rig.deleteQueued(draftId, id)).toMatchObject({ + ok: true, + replayed: true, + value: { deleted: true, messageId: draftId } + }) + const future = opId(FUTURE) + expect(await rig.deleteQueued(draftId, future)).toMatchObject({ + ok: true, + replayed: false, + value: { deleted: false, disposition: 'withdrawn' } + }) + }) + + it.each(['missing', 'withdrawn'] as const)( + "replays a no-op's first answer when the card was %s", + async (disposition) => { + await rig.workingSend() + const draftId = disposition === 'missing' ? 'no-such-card' : await queuedDraft('gone') + if (disposition === 'withdrawn') { + await rig.deleteQueued(draftId) + } + const id = opId() + const answer = { deleted: false, messageId: draftId, disposition } + expect(await rig.deleteQueued(draftId, id)).toMatchObject({ + ok: true, + replayed: false, + value: answer + }) + expect(receipt(id)).toMatchObject({ + receipt: { + result: { + kind: 'no-op', + outcome: { kind: 'queue-delete', messageId: draftId, disposition } + } + } + }) + expect(await rig.deleteQueued(draftId, id)).toEqual( + expect.objectContaining({ ok: true, replayed: true, value: answer }) + ) + } + ) + + it('answers concurrent copies of one id with one withdrawal', async () => { + await rig.workingSend() + const draftId = await queuedDraft('delete once') + const id = opId() + const [first, second] = await Promise.all([ + rig.deleteQueued(draftId, id), + rig.deleteQueued(draftId, id) + ]) + expect(first).toMatchObject({ ok: true, replayed: false, value: { deleted: true } }) + expect(second).toMatchObject({ ok: true, replayed: true, value: { deleted: true } }) + }) + + it('keeps the card when its receipt cannot be saved', async () => { + await rig.workingSend() + const draftId = await queuedDraft('keep me') + const id = opId() + const restore = failReceiptWrites() + await expect(rig.deleteQueued(draftId, id)).rejects.toThrow('receipt unavailable') + restore() + expect(await rig.drafts()).toContainEqual( + expect.objectContaining({ messageId: draftId, state: 'waiting' }) + ) + expect(ledgerRow(id)).toBeUndefined() + }) + + it('answers an unreadable receipt as unknown and withdraws nothing', async () => { + await rig.workingSend() + const draftId = await queuedDraft('unknown delete') + const id = opId() + writeUnreadableReceipt(id, 'agentSession.queuedMessageDelete') + expect(await rig.deleteQueued(draftId, id)).toMatchObject(UNKNOWN) + expect(await rig.drafts()).toContainEqual( + expect.objectContaining({ messageId: draftId, state: 'waiting' }) + ) + }) + + it('refuses the same id naming another card and keeps that card', async () => { + await rig.workingSend() + const first = await queuedDraft('first card') + const second = await queuedDraft('second card') + const id = opId() + await rig.deleteQueued(first, id) + expect(await rig.deleteQueued(second, id)).toMatchObject(CONFLICT) + expect(await rig.drafts()).toContainEqual( + expect.objectContaining({ messageId: second, state: 'waiting' }) + ) + }) + + it('answers unknown, never success, when a no-op cannot be recorded, and records nothing', async () => { + await rig.workingSend() + const id = opId() + const restore = failReceiptWrites() + expect(await rig.deleteQueued('no-such-card', id)).toMatchObject(UNKNOWN) + restore() + expect(receipt(id)).toEqual({ verdict: 'absent' }) + expect(ledgerRow(id)).toBeUndefined() + }) + + it('answers a retry after a rewind from its receipt, withdrawing nothing again', async () => { + await rig.workingSend() + const draftId = await queuedDraft('deleted then rewound') + const id = opId() + await rig.deleteQueued(draftId, id) + await rewindConversation() + expect(await rig.deleteQueued(draftId, id)).toMatchObject({ + ok: true, + replayed: true, + value: { deleted: true, messageId: draftId } + }) + }) +}) + +describe('Resume', () => { + it('lifts the pause once under an old id and never lifts a later pause under it', async () => { + await rig.workingSend() + await queuedDraft('held by stop') + await queuedDraft('held by the second stop') + await rig.stop() + const id = opId(OLD) + expect(await rig.resume(id)).toMatchObject({ + ok: true, + replayed: false, + value: { resumed: true } + }) + expect(receipt(id)).toMatchObject({ receipt: { result: { kind: 'journal-row' } } }) + await rig.stop() + expect(await rig.queuePause()).toEqual({ reason: 'stopped' }) + expect(await rig.resume(id)).toMatchObject({ + ok: true, + replayed: true, + value: { resumed: false } + }) + expect(await rig.queuePause()).toEqual({ reason: 'stopped' }) + }) + + it('records a no-op Resume and replays it', async () => { + const id = opId(FUTURE) + expect(await rig.resume(id)).toMatchObject({ + ok: true, + replayed: false, + value: { resumed: false } + }) + expect(receipt(id)).toMatchObject({ + receipt: { result: { kind: 'no-op', outcome: { kind: 'queue-resume', resumed: false } } } + }) + expect(await rig.resume(id)).toMatchObject({ + ok: true, + replayed: true, + value: { resumed: false } + }) + }) + + it('keeps the pause when its receipt cannot be saved', async () => { + await rig.workingSend() + await queuedDraft('still held') + await rig.stop() + const restore = failReceiptWrites() + await expect(rig.resume(opId())).rejects.toThrow('receipt unavailable') + restore() + expect(await rig.queuePause()).toEqual({ reason: 'stopped' }) + }) + + it('answers concurrent copies of one id with one Resume', async () => { + await pausedQueue() + const id = opId() + const [first, second] = await Promise.all([rig.resume(id), rig.resume(id)]) + expect(first).toMatchObject({ ok: true, replayed: false, value: { resumed: true } }) + expect(second).toMatchObject({ ok: true, replayed: true, value: { resumed: false } }) + expect(await rig.queuePause()).toBeNull() + }) + + it('answers unknown, never success, when a no-op cannot be recorded, and records nothing', async () => { + const id = opId() + const restore = failReceiptWrites() + expect(await rig.resume(id)).toMatchObject(UNKNOWN) + restore() + expect(receipt(id)).toEqual({ verdict: 'absent' }) + expect(ledgerRow(id)).toBeUndefined() + }) + + it("answers a no-op on a newer Orca's records with that refusal, not unknown", async () => { + await rig.workingSend() + const id = opId() + Object.defineProperty(openTestJournalHostDatabase(rig.root), 'readOnly', { value: true }) + expect(await rig.resume(id)).toMatchObject({ + ok: false, + refusal: { code: 'agent_session_journal_unreadable' } + }) + }) + + it('answers an unreadable receipt as unknown and lifts nothing', async () => { + await pausedQueue() + const id = opId() + writeUnreadableReceipt(id, 'agentSession.queuedMessagesResume') + expect(await rig.resume(id)).toMatchObject(UNKNOWN) + expect(await rig.queuePause()).toEqual({ reason: 'stopped' }) + }) + + it('answers a retry after a rewind from its receipt and never lifts the pause a later Stop set', async () => { + await pausedQueue() + const id = opId() + expect(await rig.resume(id)).toMatchObject({ ok: true, value: { resumed: true } }) + await rewindConversation() + await rig.stop() + expect(await rig.queuePause()).toEqual({ reason: 'stopped' }) + expect(await rig.resume(id)).toMatchObject({ + ok: true, + replayed: true, + value: { resumed: false } + }) + expect(await rig.queuePause()).toEqual({ reason: 'stopped' }) + }) +}) + +describe.each([ + ['Delete', (id: string) => rig.deleteQueued('some-card', id)], + ['Resume', (id: string) => rig.resume(id)] +])('%s refused before acceptance', (_action, act) => { + it('records nothing', async () => { + const id = 'not-an-operation-id' + expect(await act(id)).toMatchObject({ + ok: false, + refusal: { code: 'agent_session_operation_invalid' } + }) + expect(receipt(id)).toEqual({ verdict: 'absent' }) + expect(ledgerRow(id)).toBeUndefined() + }) +}) + +describe('an id reused across commands', () => { + const conflict = CONFLICT + + it('refuses a Stop reusing a Delete id, through the compatibility ledger row', async () => { + await rig.workingSend() + const draftId = await queuedDraft('delete then stop') + const id = opId() + await rig.deleteQueued(draftId, id) + expect(await rig.stop(id)).toMatchObject(conflict) + }) + + it('refuses a Delete or Resume reusing a Stop id, committing nothing', async () => { + await rig.workingSend() + const draftId = await queuedDraft('stop then delete') + const id = opId() + await rig.stop(id) + expect(await rig.deleteQueued(draftId, id)).toMatchObject(conflict) + expect(await rig.resume(id)).toMatchObject(conflict) + expect(await rig.queuePause()).toEqual({ reason: 'stopped' }) + expect(receipt(id)).toEqual({ verdict: 'absent' }) + }) + + it('refuses a Send now reusing a Delete id, and the reverse', async () => { + await rig.workingSend() + const first = await queuedDraft('first') + const second = await queuedDraft('second') + const deleteId = opId() + await rig.deleteQueued(first, deleteId) + expect(await rig.sendNow(second, deleteId)).toMatchObject(conflict) + const sendId = opId() + await rig.sendNow(second, sendId) + expect(await rig.deleteQueued(second, sendId)).toMatchObject(conflict) + }) +}) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-queued-mutations.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-queued-mutations.ts index ab8a9384f091..a3fee6e2a339 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-queued-mutations.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-queued-mutations.ts @@ -1,8 +1,7 @@ -// `agentSession.queuedMessageSend` / `agentSession.queuedMessageDelete`, and -// mutations of the conversation's queued drafts. Settling -// operations stamp op-scoped tombstone receipts, so a lost acknowledgement -// replays from the rows themselves — never from the operation ledger, which -// records only that an operation happened. No mutation returns draft text: +// `agentSession.queuedMessageSend` / `agentSession.queuedMessageDelete` / Resume: mutations of +// the conversation's queued drafts. Each action's command receipt commits with the row it +// changes, so a lost acknowledgement replays from that receipt and the rows it points at, never +// from the operation ledger. No mutation returns draft text: // the published list is the one authority a client renders. import type { AgentJournalSubmission } from '../../../shared/agent-session-journal-types' @@ -16,12 +15,14 @@ import type { } from '../../../shared/agent-session-wire' import type { AgentSessionJournal } from '../agent-session-journal/journal-store' import { QueuedMessageNotConsumableError } from '../agent-session-journal/queued-message-consume-error' -import { - isJournalWrittenByNewerOrca, - journalOpenRefusal -} from '../agent-session-journal/journal-open-failure' +import { refusingNewerOrcaJournal } from '../agent-session-journal/journal-open-failure' import type { QueuedMessageRow } from '../agent-session-journal/queued-message-table' -import type { MutationPlan } from './structured-agent-session-mutation-plans' +import { + journalRowReceiptResult, + type MutationPlan +} from './structured-agent-session-mutation-plans' +import type { JournalOperationReceipt } from '../agent-session-journal/journal-row-writer' +import type { CommandReceiptResult } from '../agent-session-journal/command-receipt-schema' import { structuredQueueHold } from './structured-agent-session-queued-messages' import { resumeStructuredQueue, @@ -52,6 +53,17 @@ function submissionFor( return ctx.journal.submissions().find((entry) => entry.clientMessageId === clientMessageId) } +/** Where the journal wrote a hand-off's submission, so a replay answers with that one. */ +function submissionRowReceipt( + ctx: AgentSessionTurnContext, + submission: AgentJournalSubmission | undefined +): CommandReceiptResult { + if (submission?.submittedSequence === undefined) { + throw new Error('an acknowledged hand-off requires its submission row') + } + return { kind: 'journal-row', epoch: ctx.journal.epoch, sequence: submission.submittedSequence } +} + /** One transaction, stamped with the operation's caller-scoped key so a replay * answers "spent" from the receipts. Withdrawal retires any hold in the same * UPDATE, and the store's commit notification publishes the change. */ @@ -62,12 +74,16 @@ export async function withdrawQueuedMessagesForOperation( messageIds: readonly string[] callerKey: string operationId: string - } + }, + receipt?: JournalOperationReceipt ): Promise { - return journal.queuedMessages.withdraw({ - messageIds: input.messageIds, - settledByOp: agentSessionOperationKey(input.callerKey, input.operationId) - }) + return journal.queuedMessages.withdraw( + { + messageIds: input.messageIds, + settledByOp: agentSessionOperationKey(input.callerKey, input.operationId) + }, + receipt + ) } /** Draft actions run like any mutation: admitted on the session's lane, the @@ -87,20 +103,6 @@ export function mutateQueued( ) } -/** A newer Orca's journal refuses a draft write with the words a send gets there. */ -async function refusingNewerOrcaJournal( - run: Promise> -): Promise> { - try { - return await run - } catch (error) { - if (isJournalWrittenByNewerOrca(error)) { - return { ok: false, refusal: journalOpenRefusal(error) } - } - throw error - } -} - /** * Send-now. It overrides ONLY queue policy — FIFO order, pause, the busy-turn * wait — through the same send block and pending-prompt gates as any send; @@ -119,13 +121,13 @@ export function sendQueuedStructuredAgentMessage( method: 'agentSession.queuedMessageSend', fields: { messageId }, conversationWrite: true, + // Its own submission's row, or the row of the hand-off that already took the card. + commandReceipt: { + result: (row) => journalRowReceiptResult(row, 'submission'), + unwritten: (value, ctx) => + submissionRowReceipt(ctx, 'submission' in value ? value.submission : undefined) + }, run: async (ctx): Promise> => { - // A rerun of this operation after it consumed the card (its answer never - // settled): answer with the submission it made, never append it again. - const consumedHere = submissionFor(ctx, operationId) - if (consumedHere?.queuedMessageId === messageId) { - return { ok: true, value: { clientMessageId: operationId, submission: consumedHere } } - } // The one queue gate; Send-now's override set is exactly `working` (plus // FIFO order and the stored hold, which the consume below clears). const record = context.deps.store.getRecord(ctx.sessionId) @@ -172,7 +174,8 @@ export function sendQueuedStructuredAgentMessage( expect: row.state, settledByOp: agentSessionOperationKey(ctx.resolvedBy, operationId), hostInstance: structuredAgentSessionHostInstance() - } + }, + ctx.operationReceipt ) } catch (error) { if (error instanceof QueuedMessageNotConsumableError) { @@ -184,19 +187,17 @@ export function sendQueuedStructuredAgentMessage( if (!submission) { throw new Error('agent_session_submission_lost') } - context.wakeDelivery(ctx.sessionId) return { ok: true, value: { clientMessageId: submissionId, submission } } }, - replay: (ctx) => { - const opKey = agentSessionOperationKey(ctx.resolvedBy, operationId) - const row = ctx.journal.queuedMessages - .receipts(opKey) - .find((receipt) => receipt.messageId === messageId) - if (!row || row.state !== 'dispatched') { - return null - } - const submission = row.consumedAs === null ? undefined : submissionFor(ctx, row.consumedAs) - return submission ? { clientMessageId: submission.clientMessageId, submission } : null + // The hand-off the receipt points at, never whichever one the card holds now. + replay: (ctx, _outcome, receipt) => { + const submission = + receipt?.kind === 'journal-row' && receipt.epoch === ctx.journal.epoch + ? ctx.journal.submissions().find((entry) => entry.submittedSequence === receipt.sequence) + : undefined + return submission?.queuedMessageId === messageId + ? { clientMessageId: submission.clientMessageId, submission } + : null } } return mutateQueued(context, caller, params.envelope, plan) @@ -216,6 +217,17 @@ export function deleteQueuedStructuredAgentMessage( method: 'agentSession.queuedMessageDelete', fields: { messageId }, conversationWrite: true, + // The withdrawn card, or a no-op's answer when there was nothing left to withdraw. + commandReceipt: { + result: () => ({ kind: 'queued-draft', messageId }), + unwritten: (value) => + value.deleted + ? null + : { + kind: 'no-op', + outcome: { kind: 'queue-delete', messageId, disposition: value.disposition } + } + }, run: async (ctx): Promise> => { const row = ctx.journal.queuedMessages.get(messageId) if (!row) { @@ -230,21 +242,26 @@ export function deleteQueuedStructuredAgentMessage( // The withdrawal notifies through the journal's commit listener, which // also re-derives the drain — deleting a returned card can unblock the // drafts behind it. - const withdrawn = await withdrawQueuedMessagesForOperation(ctx.journal, { - sessionId: ctx.sessionId, - messageIds: [messageId], - callerKey: ctx.resolvedBy, - operationId - }) + const withdrawn = await withdrawQueuedMessagesForOperation( + ctx.journal, + { + sessionId: ctx.sessionId, + messageIds: [messageId], + callerKey: ctx.resolvedBy, + operationId + }, + ctx.operationReceipt + ) return withdrawn.length > 0 ? { ok: true, value: { deleted: true, messageId } } : { ok: true, value: { deleted: false, messageId, disposition: 'withdrawn' } } }, - replay: (ctx) => { - const replayed = ctx.journal.queuedMessages - .receipts(agentSessionOperationKey(ctx.resolvedBy, operationId)) - .some((row) => row.messageId === messageId && row.state === 'withdrawn') - return replayed ? { deleted: true, messageId } : null + // The receipt alone answers, so a pruned tombstone cannot lose it. + replay: (_ctx, _outcome, receipt) => { + if (receipt?.kind === 'no-op' && receipt.outcome.kind === 'queue-delete') { + return { deleted: false, messageId, disposition: receipt.outcome.disposition } + } + return receipt?.kind === 'queued-draft' ? { deleted: true, messageId } : null } } return mutateQueued(context, caller, params.envelope, plan) @@ -262,11 +279,17 @@ export function resumeStructuredAgentQueue( method: 'agentSession.queuedMessagesResume', fields: {}, conversationWrite: true, + commandReceipt: { + result: (row) => journalRowReceiptResult(row, 'tombstone'), + unwritten: () => ({ kind: 'no-op', outcome: { kind: 'queue-resume', resumed: false } }) + }, // The Resume row notifies through the journal's commit listener, which publishes the // lifted pause and wakes the drain. run: async (ctx) => ({ ok: true, - value: { resumed: await resumeStructuredQueue(ctx.journal, ctx.fence) } + value: { + resumed: await resumeStructuredQueue(ctx.journal, ctx.fence, ctx.operationReceipt) + } }), // Like Stop's replay: the Resume already ran, so this one lifts nothing. replay: () => ({ resumed: false }) diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-queued-pause.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-queued-pause.ts index e59a56a7c795..9dd963bb43bf 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-queued-pause.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-queued-pause.ts @@ -5,6 +5,7 @@ import { randomUUID } from 'node:crypto' import type { AgentSessionJournal } from '../agent-session-journal/journal-store' import type { DerivedQueuePause } from '../agent-session-journal/queued-message-pause' +import type { JournalOperationReceipt } from '../agent-session-journal/journal-row-writer' import type { StructuredAgentSessionLogger } from './structured-agent-session-logger' /** A per-process id, minted once per host process like the runtime's own @@ -56,11 +57,12 @@ export async function markStructuredQueueReopen( /** Resume: a journal row that ends every pause. Returns whether the queue was paused. */ export async function resumeStructuredQueue( journal: Pick, - fence: number + fence: number, + receipt?: JournalOperationReceipt ): Promise { if (structuredQueuePauses(journal).length === 0) { return false } - await journal.appendQueueResume(fence) + await journal.appendQueueResume(fence, receipt) return true } diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-turns-prompt.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-turns-prompt.ts index 3a2b9a205f28..a7bbf523c3ce 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-turns-prompt.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-turns-prompt.ts @@ -23,6 +23,8 @@ import { AgentSessionPromptAnswerRejectedError, AgentSessionPromptUnavailableError } from './structured-agent-session-adapter' +import type { CommandReceiptResult } from '../agent-session-journal/command-receipt-schema' +import type { JournalRow } from '../agent-session-journal/journal-row-schema' import { settledPrompt, validatePendingPrompt } from './structured-agent-session-prompt-state' import type { AgentSessionTurnContext, TurnOutcome } from './structured-agent-session-turns' @@ -132,7 +134,9 @@ export async function performPrompt( identity, { ...prompt, resolution }, // A revision: the prompt keeps the turn it was raised in. - { fence: ctx.fence, turnScope: ctx.journal.liveTurnScope() } + { fence: ctx.fence, turnScope: ctx.journal.liveTurnScope() }, + // The answer's identity commits here, before the adapter tells the provider. + ctx.operationReceipt ) } }) @@ -167,3 +171,49 @@ export async function performPrompt( value: { itemId: appended.itemId, revision: appended.revision, resolution } } } + +/** The answer a resolved prompt row carries, the one its writer was accepted with. */ +export function promptRowAnswer(row: JournalRow | undefined): AgentSessionPromptResult { + const body = row?.kind === 'item' ? row.body : undefined + if (row?.kind !== 'item' || (body?.kind !== 'approval' && body?.kind !== 'question')) { + throw new Error('an accepted prompt answer requires its prompt row') + } + return { itemId: row.itemId, revision: row.revision, resolution: body.resolution } +} + +/** What a receipt keeps of an accepted answer: everything its replay says, so the journal's + * later revisions or a rewind never change it. */ +export function promptAnswerReceipt(answer: AgentSessionPromptResult): CommandReceiptResult { + const { state, selectedOptionId, answers, resolvedBy, resolvedAt } = answer.resolution + if (state !== 'resolved') { + throw new Error('an accepted prompt answer is resolved') + } + return { + kind: 'prompt-answer', + itemId: answer.itemId, + revision: answer.revision, + resolution: { + state, + selectedOptionId, + ...(answers + ? { + answers: answers.map(({ questionId, optionIds, other }) => ({ + questionId, + optionIds, + ...(other === undefined ? {} : { other }) + })) + } + : {}), + resolvedBy, + resolvedAt + } + } +} + +export function acceptedPromptAnswer( + receipt: CommandReceiptResult | undefined +): AgentSessionPromptResult | null { + return receipt?.kind === 'prompt-answer' + ? { itemId: receipt.itemId, revision: receipt.revision, resolution: receipt.resolution } + : null +} 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 b9ca710a1fd2..cb3d87e72e95 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 @@ -68,7 +68,7 @@ export type AgentSessionTurnContext = { publish: () => void /** Who a Stop's refusal row names. */ failureTextContext?: AgentSessionFailureWordsContext - /** The operation's success, committed with the row that accepts it (`settlesWithWrite` or `acceptsWithCommandReceipt`). */ + /** The operation's success, committed with the row that accepts it (`settlesWithWrite` or `commandReceipt`). */ operationReceipt?: JournalOperationReceipt now: () => number } diff --git a/src/main/runtime/agent-session-record-store.ts b/src/main/runtime/agent-session-record-store.ts index 64045ebbb9ab..5601d2a2e00c 100644 --- a/src/main/runtime/agent-session-record-store.ts +++ b/src/main/runtime/agent-session-record-store.ts @@ -69,6 +69,7 @@ import { showAgentSessionTabs } from './agent-session-tab-table' import type { JournalHostDatabase } from '../native-chat/agent-session-journal/journal-host-database' +import type { JournalOperationReceipt } from '../native-chat/agent-session-journal/journal-row-writer' import { agentSessionOperationOutcomeReceipt } from './agent-session-operation-receipt' import { loadAgentSessionStoreRows } from './agent-session-record-rows' import { AgentSessionStoreTransactions } from './agent-session-store-transactions' @@ -311,6 +312,9 @@ export class AgentSessionRecordStore { operationOutcomeReceipt = (args: Parameters[1]) => agentSessionOperationOutcomeReceipt(this.transactions, args) + commitOperationReceipt = (receipt: JournalOperationReceipt) => + this.transactions.commitReceipt(receipt) + readCommandReceipt = (...args: Parameters) => this.transactions.readCommandReceipt(...args) diff --git a/src/main/runtime/agent-session-store-transactions.ts b/src/main/runtime/agent-session-store-transactions.ts index 60fb9d9a2241..f68b9a1b6a82 100644 --- a/src/main/runtime/agent-session-store-transactions.ts +++ b/src/main/runtime/agent-session-store-transactions.ts @@ -175,6 +175,16 @@ export class AgentSessionStoreTransactions { } } + /** A receipt no journal write carries (a command that changed nothing), in its own transaction. */ + commitReceipt(receipt: JournalOperationReceipt): Promise { + const run = this.queue.then(() => { + this.journalDatabase.transaction((db) => receipt.write(db)) + receipt.committed() + }) + this.queue = run.catch(() => {}) + return run + } + private commit(apply: (draft: AgentSessionStoreState) => T, inMemoryWhenReadOnly: boolean): T { const readOnly = this.journalDatabase.readOnly if (readOnly && !inMemoryWhenReadOnly) { diff --git a/src/shared/agent-session-mutation-envelope.ts b/src/shared/agent-session-mutation-envelope.ts index a3acf31af4f1..dc783baa0a1a 100644 --- a/src/shared/agent-session-mutation-envelope.ts +++ b/src/shared/agent-session-mutation-envelope.ts @@ -132,7 +132,9 @@ export function agentSessionLedgerRefusal( /** Why the single admission oracle said no, mapped to what the client can do * about it. The predicate itself is never re-implemented here. */ -function refuseUnlessWriterAdmitted(lease: AgentSessionLease): AgentSessionWireRefusal | null { +export function refuseUnlessWriterAdmitted( + lease: AgentSessionLease +): AgentSessionWireRefusal | null { if (agentSessionLeaseAdmitsWriter(lease)) { return null }