diff --git a/.changeset/harness-atomic-input-rejection.md b/.changeset/harness-atomic-input-rejection.md new file mode 100644 index 0000000000..3edcfb122c --- /dev/null +++ b/.changeset/harness-atomic-input-rejection.md @@ -0,0 +1,5 @@ +--- +'@tanstack/ai-harness': patch +--- + +Store a rejected input and its `harness.input.rejected` record in one log append. The rejected receipt resolves after that append is stored, and a crash no longer runs the input as pending work. diff --git a/docs/config.json b/docs/config.json index 5ebdf1256b..c8bb974f35 100644 --- a/docs/config.json +++ b/docs/config.json @@ -1122,7 +1122,7 @@ "label": "Send inputs safely", "to": "harness/inputs", "addedAt": "2026-10-13", - "updatedAt": "2026-10-22" + "updatedAt": "2026-10-30" }, { "label": "Run a tool without a model", @@ -1133,7 +1133,7 @@ "label": "Durable sessions", "to": "harness/durable-sessions", "addedAt": "2026-10-13", - "updatedAt": "2026-10-22" + "updatedAt": "2026-10-30" }, { "label": "Run background work on workers", diff --git a/docs/harness/durable-sessions.md b/docs/harness/durable-sessions.md index a9d5143e4d..36e9fdd43c 100644 --- a/docs/harness/durable-sessions.md +++ b/docs/harness/durable-sessions.md @@ -78,7 +78,7 @@ A host that stops also stops renewing its ownership of the thread. When the owne 4. The time limit passed: the input ends `failed`, with the error code `timeout`. 5. Else the turn runs again as the next attempt, and clients get a `harness.operation.resumed` event. -An input that was stored but never ran, runs now. `session.settled({ inputId })` resolves to how an input ended. See [Send inputs safely](./inputs). +An input that was stored but never ran, runs now. A rejected input is not pending work. When the receipt resolves, the log has `harness.input.rejected`. Recovery does not run that input. `session.settled({ inputId })` resolves to how an input ended. See [Send inputs safely](./inputs). A background agent that you started with `runAgainAfterCrash: true` runs again on the next host. A background child that the model started with `background: true` runs again too, unless you set `durability: { resumeBackgroundAgents: false }`. See [Keep a background agent going after a crash](./subagents#keep-a-background-agent-going-after-a-crash). The children and groups that it started before the stop do not start again: they give their stored results. See [Children of a run that runs again](./subagents#children-of-a-run-that-runs-again). For any other background agent whose host stopped, the next host: diff --git a/docs/harness/inputs.md b/docs/harness/inputs.md index 5ce0dd0a07..20c4a69a64 100644 --- a/docs/harness/inputs.md +++ b/docs/harness/inputs.md @@ -38,6 +38,7 @@ const receipt = await turn.receipt console.log(receipt.status) // 'accepted' or 'queued' ``` +- When the session rejects a stored input, the receipt resolves after the log has `harness.input.rejected`. - The same id and the same message: you get the first input's operation, and the model is not called again. - The same id and another message: the receipt has `status: 'rejected'` and `reason: 'conflict'`, and `await turn` rejects with `InputRejectedError`. - An input to a thread that was deleted: the receipt has `status: 'rejected'` and `reason: 'deleted'`, and `await turn` rejects with `InputRejectedError`. Nothing runs. diff --git a/packages/ai-harness/src/session/inputs.ts b/packages/ai-harness/src/session/inputs.ts index 783d09bb7d..3f1e874d8c 100644 --- a/packages/ai-harness/src/session/inputs.ts +++ b/packages/ai-harness/src/session/inputs.ts @@ -44,7 +44,8 @@ import type { QueuedTurn, WaitingSteer } from './types' * How `accept` took an input. `offered`: another host owns the log, and the * input waits for that host in the input lane of the log. `refused`: the * `acceptInput` hook of the host refused it, with this reason. Nothing is - * stored. + * stored. `rejected`: the input and `harness.input.rejected` landed in one + * append, with this reason. */ export type Admission = | 'new' @@ -54,6 +55,7 @@ export type Admission = | 'offered' | 'deleted' | { refused: string } + | { rejected: string } /** What `accept` stores with an input besides the input itself. */ interface AcceptOptions { @@ -74,6 +76,12 @@ interface AcceptOptions { offer?: () => Promise /** More records for the append of a new input. */ records?: Array + /** + * Called once the input is new, before its append. A reason stores + * `harness.input.rejected` in that same append and returns + * `{ rejected: reason }`. Turn records are left out. + */ + rejectIf?: (input: HarnessInput) => string | undefined } /** The offer that the next `accept` applies. See `applyOffer`. */ @@ -625,7 +633,8 @@ export abstract class SessionInputs< * tick is a duplicate too. `user` is who sent it. The access check of * a `checked` call runs first: a refused input is not stored, and the * error rejects this call. The `turn` record goes in the append of the - * input, so a crash keeps both or neither. + * input, so a crash keeps both or neither. A rejection from `rejectIf` + * is in that append too: the receipt waits until the log has it. * * When another host owns the log, the input goes to that host through the * input lane (`offered`). An input from the lane (see `applyOffer`) @@ -655,7 +664,12 @@ export abstract class SessionInputs< offer, }) // A retry with the input as its caller sent it is the same input. - if (admission === 'new' && accepted !== input) { + // A rejection stores `accepted`, so the caller's payload is screened + // the same way. + const wasStored = + admission === 'new' || + (typeof admission === 'object' && 'rejected' in admission) + if (wasStored && accepted !== input) { this.inputRecords.screened.set( inputId, inputKey(input, storedUser(user).user), @@ -712,11 +726,17 @@ export abstract class SessionInputs< // for the next owner. if (this.closing && WORK_OPS.has(input.op)) return 'closed' this.admitted.set(inputId, inputKey(input, sender.user)) + // Before the append, so the decision and the record are one batch. + // A crash then leaves the input absent, or rejected, never pending. + const reason = options.rejectIf?.(input) try { - const records = [ - ...(options.turn ? turnRecords(inputId, options.turn) : []), - ...(options.records ?? []), - ] + const records = + reason !== undefined + ? [{ type: 'harness.input.rejected' as const, inputId, reason }] + : [ + ...(options.turn ? turnRecords(inputId, options.turn) : []), + ...(options.records ?? []), + ] await this.writer.append( [ { @@ -742,15 +762,11 @@ export abstract class SessionInputs< } // An input with no work to run (a cancel, a setting) ends once stored. if (!WORK_OPS.has(input.op)) this.inputRecords.end(inputId) - this.feed.publish( - 'session', - // The feed scrubs the session's secrets from the message. - customEvent(HARNESS_EVENTS.inputAccepted, { - inputId, - op: input.op, - ...('message' in input ? { message: input.message } : {}), - }), - ) + this.publishAccepted(inputId, input) + if (reason !== undefined) { + this.recordRejection(inputId, reason) + return { rejected: reason } + } return 'new' } @@ -1174,11 +1190,28 @@ export abstract class SessionInputs< ) } - /** Refuse an input. Returns its rejected receipt. */ + /** Refuse an input that is already in the log. Returns its rejected receipt. */ protected reject(inputId: string, reason: string) { void this.writer .append([{ type: 'harness.input.rejected', inputId, reason }]) .catch(() => {}) + return this.recordRejection(inputId, reason) + } + + /** Tell listeners the input is in the log. The feed scrubs secrets. */ + private publishAccepted(inputId: string, input: HarnessInput) { + this.feed.publish( + 'session', + customEvent(HARNESS_EVENTS.inputAccepted, { + inputId, + op: input.op, + ...('message' in input ? { message: input.message } : {}), + }), + ) + } + + /** Keep the rejected receipt and publish it. The log write is the caller's. */ + private recordRejection(inputId: string, reason: string): Receipt { const receipt: Receipt = { inputId, status: 'rejected', reason } this.inputRecords.reject(receipt) this.feed.publish( @@ -1348,7 +1381,9 @@ export abstract class SessionInputs< admission: Exclude, ): Receipt { if (typeof admission === 'object') { - return { inputId, status: 'rejected', reason: admission.refused } + const reason = + 'rejected' in admission ? admission.rejected : admission.refused + return { inputId, status: 'rejected', reason } } if (admission !== 'duplicate' && admission !== 'offered') { return { inputId, status: 'rejected', reason: admission } diff --git a/packages/ai-harness/src/session/requests.ts b/packages/ai-harness/src/session/requests.ts index 03f209c460..22b3c819c0 100644 --- a/packages/ai-harness/src/session/requests.ts +++ b/packages/ai-harness/src/session/requests.ts @@ -183,6 +183,31 @@ export abstract class SessionRequests< return this.startTurn(input, options) } + /** + * Why a prompt, a steer, or a follow-up is refused before it is stored. + * `undefined`: it can run. The order matches the checks that used to run + * after the input was stored. + */ + private refusalOfTurn( + accepted: HarnessInput & { op: 'prompt' | 'steer' | 'followUp' }, + runId: string | undefined, + ): string | undefined { + if (runId !== undefined && this.isRunTaken(runId)) return 'conflict' + if (accepted.op === 'prompt' && accepted.as === 'operator') { + return this.humanHandoff() ? undefined : 'no_human_wait' + } + const busy = + accepted.op === 'prompt' + ? accepted.busy + : accepted.op === 'steer' + ? 'steer' + : 'queue' + if (this.activeTurn && busy === 'reject') return 'busy' + const running = busy === 'steer' ? this.activeTurn : undefined + if (!running && this.waitsForInterrupts()) return 'pending_interrupts' + return undefined + } + /** * Start the turn of `input` (a prompt, a steer, or a follow-up), with its * operation. `wake`: the harness wakes the session with the message (a @@ -232,6 +257,12 @@ export abstract class SessionRequests< ...(traceparent !== undefined && parseTraceparent(traceparent) ? { traceparent } : {}), + rejectIf: (accepted) => + accepted.op === 'prompt' || + accepted.op === 'steer' || + accepted.op === 'followUp' + ? this.refusalOfTurn(accepted, runId) + : undefined, }).then( async ({ admission, input: accepted }) => { if (admission !== 'new') { @@ -246,25 +277,11 @@ export abstract class SessionRequests< : accepted.op === 'steer' ? 'steer' : 'queue' - // A client run id must be new in the log of this thread. - if (runId !== undefined && this.isRunTaken(runId)) { - this.reject(inputId, 'conflict') - this.refuse(operation, { - inputId, - status: 'rejected', - reason: 'conflict', - }) - return - } const isWaiting = this.activeTurn !== undefined || this.queue.length > 0 if (accepted.op === 'prompt' && accepted.as === 'operator') { // An operator reply runs no model. It needs an open handoff to - // a person. - if (!this.humanHandoff()) { - this.refuse(operation, this.reject(inputId, 'no_human_wait')) - return - } + // a person. `refusalOfTurn` already refused a missing handoff. this.answerTurn(operation, { inputId, status: isWaiting ? 'queued' : 'accepted', @@ -279,26 +296,8 @@ export abstract class SessionRequests< }) return } - if (this.activeTurn && busy === 'reject') { - this.reject(inputId, 'busy') - this.refuse(operation, { - inputId, - status: 'rejected', - reason: 'busy', - }) - return - } // A steer joins the running turn, so the receipt names that turn. const running = busy === 'steer' ? this.activeTurn : undefined - if (!running && this.waitsForInterrupts()) { - this.reject(inputId, 'pending_interrupts') - this.refuse(operation, { - inputId, - status: 'rejected', - reason: 'pending_interrupts', - }) - return - } // Without canJoin, a steer of another sender never joins (see // claimJoins): it waits, and runs as its own operation. const joins = @@ -395,7 +394,12 @@ export abstract class SessionRequests< const operation = this.createTurnOperation() this.bindTurn(inputId, operation) void this.admitting(() => - this.accept(inputId, input, user).then( + this.accept(inputId, input, user, { + rejectIf: (accepted) => + accepted.op === 'tool' && this.waitsForInterrupts() + ? 'pending_interrupts' + : undefined, + }).then( ({ admission, input: accepted }) => { if (admission !== 'new') { this.answerDuplicate(operation, inputId, admission) @@ -403,16 +407,6 @@ export abstract class SessionRequests< } // `acceptInput` can change the input. const { name, args, meta } = accepted - // As a prompt: the open interrupts are answered first. - if (this.waitsForInterrupts()) { - this.reject(inputId, 'pending_interrupts') - this.refuse(operation, { - inputId, - status: 'rejected', - reason: 'pending_interrupts', - }) - return - } const isWaiting = this.activeTurn !== undefined || this.queue.length > 0 this.answerTurn(operation, { @@ -555,13 +549,17 @@ export abstract class SessionRequests< inputId, { op: 'reset', ...(sent !== undefined ? { note: sent } : {}) }, user, + { + rejectIf: (accepted) => + accepted.op === 'reset' && this.waitsForInterrupts() + ? 'pending_interrupts' + : undefined, + }, ) if (admission !== 'new') return this.duplicateReceipt(inputId, admission) // `acceptInput` can change the input. const text = input.note?.trim() ? input.note : undefined const isRunning = this.activeTurn !== undefined - if (this.waitsForInterrupts()) - return this.reject(inputId, 'pending_interrupts') const operation = this.queueReset(inputId, text, user, 'front') return this.keep({ inputId, @@ -604,6 +602,71 @@ export abstract class SessionRequests< return undefined } + /** + * Why a resolve is refused before it is stored. `undefined`: it can run. + * On the background-wait path that can run, the wait ids are claimed in + * `claimed` before the append, so a second resolve sees them. + */ + private refusalOfResume( + resume: Extract['resume'], + runId: string | undefined, + claimed: { ids: Array; runId: string | undefined }, + ): string | undefined { + const isTaken = + runId !== undefined && + (this.operations.has(runId) || + this.retiredIds.has(runId) || + this.isRunTaken(runId)) + if (isTaken) return 'conflict' + const logged = resume.map((item) => this.loggedWait(item.interruptId)) + const open = logged.flatMap((state) => + state !== undefined && + state.outcome === undefined && + state.live === undefined + ? [state] + : [], + ) + const interrupts = this.logWriter?.state.thread.interrupts + // A `step.ask` question and the `human` wait of a handoff have no + // interrupt: `answer` checks their schema, keeps a secret out of the + // log, and ends the handoff. + if ( + open.some( + ({ wait }) => interrupts?.get(wait.waitId)?.status !== 'pending', + ) + ) { + return 'use_answer' + } + if (open.length > 0) { + // The access check saw the run of the first item: every item must + // be a wait of that run. + const backgroundRunId = open[0]?.wait.runId + const isOneRun = + open.length === resume.length && + open.every(({ wait }) => wait.runId === backgroundRunId) + if (!isOneRun) return 'mixed_runs' + // Another answer ends one of these waits now. + if (open.some(({ wait }) => this.answering.has(wait.waitId))) { + return 'unknown_question' + } + const ids = open.map(({ wait }) => wait.waitId) + for (const id of ids) this.answering.add(id) + claimed.ids = ids + claimed.runId = backgroundRunId + return undefined + } + if (!this.interrupted) return 'no_pending_interrupts' + const answers = this.interrupted + const isEnding = this.activeTurn?.id === answers.runId + if ( + (this.activeTurn && !isEnding) || + this.queue.some((turn) => turn.answers?.runId === answers.runId) + ) { + return 'busy' + } + return undefined + } + /** * @internal `resolve`, with the AG-UI `runId` of a `POST run` request: the * turn runs as it. The HTTP handler and an offered resolve call it. @@ -628,141 +691,114 @@ export abstract class SessionRequests< const inputId = options.inputId ?? createInputId() const user = this.actor(options.user) const runId = options.runId - const { admission, input } = await this.accept( - inputId, - { op: 'resolve', resume: options.resume }, - user, - runId !== undefined ? { runId } : {}, - ) - // `acceptInput` can change the input. - const { resume } = input - if (admission !== 'new') { - const receipt = this.duplicateReceipt(inputId, admission) - // The owner runs an offered resume as `runId`. - return admission === 'offered' && - runId !== undefined && - receipt.status !== 'rejected' - ? { ...receipt, operationId: runId } - : receipt + const claimed: { ids: Array; runId: string | undefined } = { + ids: [], + runId: undefined, } - // A run id must not take the id of another operation or stored run. - const isTaken = - runId !== undefined && - (this.operations.has(runId) || - this.retiredIds.has(runId) || - this.isRunTaken(runId)) - if (isTaken) { - this.reject(inputId, 'conflict') - return { inputId, status: 'rejected', reason: 'conflict' } + const release = () => { + for (const id of claimed.ids) this.answering.delete(id) + claimed.ids.length = 0 } - // The waits of a background agent run: the log ends them, and the - // run goes on (see `resumeWaitingRuns`). - const logged = resume.map((item) => this.loggedWait(item.interruptId)) - const open = logged.flatMap((state) => - state !== undefined && - state.outcome === undefined && - state.live === undefined - ? [state] - : [], - ) - const interrupts = this.logWriter?.state.thread.interrupts - // A `step.ask` question and the `human` wait of a handoff have no - // interrupt: `answer` checks their schema, keeps a secret out of the - // log, and ends the handoff. - if ( - open.some( - ({ wait }) => interrupts?.get(wait.waitId)?.status !== 'pending', + try { + const { admission, input } = await this.accept( + inputId, + { op: 'resolve', resume: options.resume }, + user, + { + ...(runId !== undefined ? { runId } : {}), + rejectIf: (accepted) => + accepted.op === 'resolve' + ? this.refusalOfResume(accepted.resume, runId, claimed) + : undefined, + }, ) - ) { - this.reject(inputId, 'use_answer') - return { inputId, status: 'rejected', reason: 'use_answer' } - } - if (open.length > 0) { - // The access check saw the run of the first item: every item must - // be a wait of that run. - const runId = open[0]?.wait.runId - const isOneRun = - open.length === resume.length && - open.every(({ wait }) => wait.runId === runId) - if (!isOneRun) { - this.reject(inputId, 'mixed_runs') - return { inputId, status: 'rejected', reason: 'mixed_runs' } + // `acceptInput` can change the input. + const { resume } = input + if (admission !== 'new') { + release() + const receipt = this.duplicateReceipt(inputId, admission) + // The owner runs an offered resume as `runId`. + return admission === 'offered' && + runId !== undefined && + receipt.status !== 'rejected' + ? { ...receipt, operationId: runId } + : receipt } - // Another answer ends one of these waits now. - if (open.some(({ wait }) => this.answering.has(wait.waitId))) { - this.reject(inputId, 'unknown_question') - return { inputId, status: 'rejected', reason: 'unknown_question' } + if (claimed.ids.length > 0) { + // The receipt names the run of the wait, not the client's run id. + const backgroundRunId = claimed.runId + try { + // The input ends here, in the append that ends the waits. + await this.settleWith( + [{ inputId, outcome: 'completed' }], + resume.flatMap((item) => + item.status === 'resolved' + ? this.waitEndRecords( + item.interruptId, + 'answered', + item.payload, + ) + : this.waitEndRecords(item.interruptId, 'cancelled'), + ), + ) + } finally { + release() + } + for (const item of resume) await this.afterWaitEnd(item.interruptId) + return this.keep({ + inputId, + status: 'accepted', + ...(backgroundRunId !== undefined + ? { operationId: backgroundRunId } + : {}), + }) + } + const answers = this.interrupted + if (!answers) { + return this.reject(inputId, 'no_pending_interrupts') + } + // A client answers on RUN_FINISHED, while the interrupted turn still + // ends. The resolve runs right after it. + const isEnding = this.activeTurn?.id === answers.runId + if ( + (this.activeTurn && !isEnding) || + this.queue.some((turn) => turn.answers?.runId === answers.runId) + ) { + return this.reject(inputId, 'busy') } - const ids = open.map(({ wait }) => wait.waitId) - for (const id of ids) this.answering.add(id) - try { - // The input ends here, in the append that ends the waits. - await this.settleWith( - [{ inputId, outcome: 'completed' }], - resume.flatMap((item) => - item.status === 'resolved' - ? this.waitEndRecords( - item.interruptId, - 'answered', - item.payload, - ) - : this.waitEndRecords(item.interruptId, 'cancelled'), - ), + // A cancelled sign-in fails its tool when the turn runs it again. + for (const entry of resume) { + const connector = signInConnector( + answers.interrupts.find((item) => item.id === entry.interruptId), ) - } finally { - for (const id of ids) this.answering.delete(id) + if (connector && entry.status === 'cancelled') + this.declinedSignIns.add(connector) } - for (const item of resume) await this.afterWaitEnd(item.interruptId) + const operation = this.createTurnOperation(runId) + this.bindTurn(inputId, operation) + // The turn goes on as its sender, with the context its input sent: + // an owner who answers the turn of a member does not lend the member + // the owner's keys. A copy from an older version has no sender. + const turn: QueuedTurn = { + operation, + resume, + inputId, + answers, + user: answers.user ?? user, + ...(user ? { sender: user } : {}), + context: answers.context, + } + if (isEnding) this.queue.unshift(turn) + else this.enqueueTurn(turn) return this.keep({ inputId, status: 'accepted', - ...(runId !== undefined ? { operationId: runId } : {}), + operationId: operation.id, }) + } catch (error) { + release() + throw error } - const answers = this.interrupted - if (!answers) { - this.reject(inputId, 'no_pending_interrupts') - return { inputId, status: 'rejected', reason: 'no_pending_interrupts' } - } - // A client answers on RUN_FINISHED, while the interrupted turn still - // ends. The resolve runs right after it. - const isEnding = this.activeTurn?.id === answers.runId - if ( - (this.activeTurn && !isEnding) || - this.queue.some((turn) => turn.answers?.runId === answers.runId) - ) { - this.reject(inputId, 'busy') - return { inputId, status: 'rejected', reason: 'busy' } - } - // A cancelled sign-in fails its tool when the turn runs it again. - for (const entry of resume) { - const connector = signInConnector( - answers.interrupts.find((item) => item.id === entry.interruptId), - ) - if (connector && entry.status === 'cancelled') - this.declinedSignIns.add(connector) - } - const operation = this.createTurnOperation(runId) - this.bindTurn(inputId, operation) - // The turn goes on as its sender, with the context its input sent: - // an owner who answers the turn of a member does not lend the member - // the owner's keys. A copy from an older version has no sender. - const turn: QueuedTurn = { - operation, - resume, - inputId, - answers, - user: answers.user ?? user, - ...(user ? { sender: user } : {}), - context: answers.context, - } - if (isEnding) this.queue.unshift(turn) - else this.enqueueTurn(turn) - return this.keep({ - inputId, - status: 'accepted', - operationId: operation.id, - }) }) } diff --git a/packages/ai-harness/tests/rejection-durability.test.ts b/packages/ai-harness/tests/rejection-durability.test.ts new file mode 100644 index 0000000000..2ff06ae093 --- /dev/null +++ b/packages/ai-harness/tests/rejection-durability.test.ts @@ -0,0 +1,525 @@ +import { describe, expect, it, vi } from 'vitest' +import { z } from 'zod' +import { toolDefinition } from '@tanstack/ai' +import type { AnyTool } from '@tanstack/ai' +import { memoryLogState, memoryLogStore } from '@tanstack/ai-persistence' +import { + InputRejectedError, + createHarnessHost, + defineHarness, + durableTool, + logIdOf, +} from '../src' +import { + CRASHING_OWNERSHIP, + gate, + mockAdapter, + ownershipExpired, + text, + toolCall, + wrapLogStore, +} from './helpers' +import type { LogRecord, LogStore } from '@tanstack/ai-persistence' +import type { SessionImpl } from '../src/session/harness-session' +import type { HarnessSession, Operation, Receipt } from '../src' +import type { Reply } from './helpers' + +const waitFor = ( + check: () => T | Promise, + options?: { timeout?: number; interval?: number }, +) => vi.waitFor(check, { timeout: 5_000, ...options }) + +const THREAD = 't1' +const LOG = logIdOf({ logId: THREAD }) + +const openScreen = toolDefinition({ + name: 'openScreen', + description: 'Open a screen in the browser. Runs on the client.', + inputSchema: z.object({ route: z.string() }), + outputSchema: z.object({ opened: z.boolean() }), +}) + +const ping = toolDefinition({ + name: 'ping', + description: 'Ping', + inputSchema: z.object({}), +}).server(async () => 'pong') + +const askCode = durableTool( + toolDefinition({ + name: 'code', + description: 'Ask for a code', + inputSchema: z.object({}), + }), + async (_args, { step }) => + step.ask({ + name: 'code', + message: 'Code?', + schema: z.string(), + }), +) + +/** The next rejection append waits, or the next few rejection appends throw. */ +function watchRejections(inner: LogStore) { + let waiting: + | { records?: ReadonlyArray; release: () => void } + | undefined + let failuresLeft = 0 + const log = wrapLogStore(inner, { + append: async (batch) => { + const rejects = batch.records.some( + (record) => record.type === 'harness.input.rejected', + ) + if (rejects && failuresLeft > 0) { + failuresLeft -= 1 + throw new Error('disk full') + } + if (rejects && waiting && waiting.records === undefined) { + waiting.records = batch.records + await new Promise((resolve) => { + waiting!.release = resolve + }) + } + return inner.append(batch) + }, + }) + return { + log, + holdNext() { + const slot: { + records?: ReadonlyArray + release: () => void + } = { release: () => {} } + waiting = slot + return slot + }, + failNext(times: number) { + failuresLeft = times + }, + } +} + +const typesOf = async (log: LogStore, inputId: string) => + (await log.read({ logId: LOG })) + .map((entry) => entry.record) + .filter((record) => record.inputId === inputId) + .map((record) => record.type) + +async function openOn( + log: LogStore, + replies: Array, + tools: Array = [openScreen, ping], + ownership?: typeof CRASHING_OWNERSHIP, +) { + const { adapter, calls } = mockAdapter(replies) + const host = createHarnessHost({ + persistence: { stores: { log } }, + ...(ownership ? { ownership } : {}), + }) + const session = await host.open( + defineHarness({ + name: 'test/rejection-durability', + adapter, + tools, + }), + { threadId: THREAD }, + ) + return { host, session, calls, log } +} + +/** Stop on a client tool, so the thread waits for an interrupt. */ +async function interrupt(session: HarnessSession) { + const turn = await session.send('Open settings.', { inputId: 'ask' }) + expect(turn.interrupts?.length).toBeGreaterThan(0) +} + +/** The refusal lands in one append, and the receipt matches `reason`. */ +async function expectAtomic( + watch: ReturnType, + run: () => Operation | Promise, + reason: string, +) { + const slot = watch.holdNext() + const result = run() + if ('receipt' in result) { + result.then( + () => {}, + () => {}, + ) + } + const receipt = 'receipt' in result ? result.receipt : result + await waitFor(() => expect(slot.records).toBeDefined()) + expect(slot.records?.map((record) => record.type)).toEqual([ + 'harness.input', + 'harness.input.rejected', + ]) + slot.release() + await expect(receipt).resolves.toMatchObject({ status: 'rejected', reason }) +} + +describe('a rejected input is durable before the receipt', () => { + it('does not resolve the receipt before the log has the rejection', async () => { + const inner = memoryLogStore() + const watch = watchRejections(inner) + const { host, session } = await openOn(watch.log, [ + () => toolCall('openScreen', { route: '/settings' }), + ]) + await interrupt(session) + + const slot = watch.holdNext() + let sawReceipt = false + const followUp = session.send('And then?', { + mode: 'followUp', + inputId: 'reply-1', + }) + followUp.then( + () => {}, + () => {}, + ) + const receipt = followUp.receipt.then((value) => { + sawReceipt = true + return value + }) + await waitFor(() => expect(slot.records).toBeDefined()) + expect(sawReceipt).toBe(false) + expect(slot.records?.map((record) => record.type)).toEqual([ + 'harness.input', + 'harness.input.rejected', + ]) + expect(await typesOf(inner, 'reply-1')).toEqual([]) + + slot.release() + expect(await receipt).toMatchObject({ + inputId: 'reply-1', + status: 'rejected', + reason: 'pending_interrupts', + }) + expect(await typesOf(inner, 'reply-1')).toEqual([ + 'harness.input', + 'harness.input.rejected', + ]) + await host.close() + }) + + it('does not recover an input whose rejection never landed', async () => { + const state = memoryLogState() + const inner = memoryLogStore({ state }) + const watch = watchRejections(inner) + const { host, session } = await openOn( + watch.log, + [() => toolCall('openScreen', { route: '/settings' })], + [openScreen], + CRASHING_OWNERSHIP, + ) + await interrupt(session) + + const slot = watch.holdNext() + const followUp = session.send('And then?', { + mode: 'followUp', + inputId: 'reply-1', + }) + followUp.then( + () => {}, + () => {}, + ) + await waitFor(() => expect(slot.records).toBeDefined()) + await ownershipExpired() + + const { adapter, calls } = mockAdapter([() => text('Recovered.')]) + const next = createHarnessHost({ + persistence: { stores: { log: memoryLogStore({ state }) } }, + }) + const recovered = await next.open( + defineHarness({ + name: 'test/rejection-durability', + adapter, + tools: [openScreen], + }), + { threadId: THREAD }, + ) + expect(await typesOf(inner, 'reply-1')).toEqual([]) + expect(calls).toHaveLength(0) + expect(recovered.snapshot().status).not.toBe('running') + + slot.release() + await next.close() + await host.close().catch(() => {}) + }) + + it('fails the operation when the rejection append fails', async () => { + const inner = memoryLogStore() + const watch = watchRejections(inner) + const { host, session } = await openOn(watch.log, [ + () => toolCall('openScreen', { route: '/settings' }), + ]) + await interrupt(session) + watch.failNext(2) + + const followUp = session.send('And then?', { + mode: 'followUp', + inputId: 'reply-1', + }) + followUp.then( + () => {}, + () => {}, + ) + const receipt = await followUp.receipt + expect(receipt.status).toBe('rejected') + if (receipt.status === 'rejected') { + expect(receipt.reason).not.toBe('pending_interrupts') + } + await expect(followUp).rejects.toThrow() + expect(await typesOf(inner, 'reply-1')).toEqual([]) + await host.close().catch(() => {}) + }) + + it('replays a rejection, and conflicts when the payload differs', async () => { + const { host, session } = await openOn(memoryLogStore(), [ + () => toolCall('openScreen', { route: '/settings' }), + ]) + await interrupt(session) + const first = session.send('And then?', { + mode: 'followUp', + inputId: 'reply-1', + }) + first.then( + () => {}, + () => {}, + ) + expect(await first.receipt).toMatchObject({ + status: 'rejected', + reason: 'pending_interrupts', + }) + + const same = session.send('And then?', { + mode: 'followUp', + inputId: 'reply-1', + }) + same.then( + () => {}, + () => {}, + ) + expect(await same.receipt).toMatchObject({ + status: 'rejected', + reason: 'pending_interrupts', + }) + await expect(same).rejects.toBeInstanceOf(InputRejectedError) + + const other = session.send('A different message.', { + mode: 'followUp', + inputId: 'reply-1', + }) + other.then( + () => {}, + () => {}, + ) + expect(await other.receipt).toMatchObject({ + status: 'rejected', + reason: 'conflict', + }) + await expect(other).rejects.toBeInstanceOf(InputRejectedError) + await host.close() + }) + + it('replays a rejection through the offer lane', async () => { + const state = memoryLogState() + const owner = await openOn(memoryLogStore({ state }), [ + () => toolCall('openScreen', { route: '/settings' }), + ]) + const otherHost = createHarnessHost({ + persistence: { stores: { log: memoryLogStore({ state }) } }, + }) + const other = await otherHost.open( + defineHarness({ + name: 'test/rejection-durability', + adapter: mockAdapter([]).adapter, + }), + { threadId: THREAD }, + ) + await interrupt(owner.session) + + const offered = other.send('And then?', { + mode: 'followUp', + inputId: 'reply-1', + }) + offered.then( + () => {}, + () => {}, + ) + await expect( + other.settled({ + inputId: 'reply-1', + signal: AbortSignal.timeout(3_000), + }), + ).rejects.toBeInstanceOf(InputRejectedError) + + const replay = async (message: string, reason: string) => { + const sent = other.send(message, { + mode: 'followUp', + inputId: 'reply-1', + }) + sent.then( + () => {}, + () => {}, + ) + const receipt = await sent.receipt + if (receipt.status === 'rejected') { + expect(receipt.reason).toBe(reason) + return + } + await expect( + other.settled({ + inputId: 'reply-1', + signal: AbortSignal.timeout(3_000), + }), + ).rejects.toMatchObject({ receipt: { reason } }) + } + await replay('And then?', 'pending_interrupts') + await replay('A different message.', 'conflict') + + await owner.host.close() + await otherHost.close() + }) + + it('writes each post-admission refusal in the same append as the input', async () => { + const inner = memoryLogStore() + const watch = watchRejections(inner) + const waiting = await openOn(watch.log, [ + () => toolCall('openScreen', { route: '/settings' }), + ]) + await interrupt(waiting.session) + await expectAtomic( + watch, + () => + waiting.session.send('Next.', { mode: 'followUp', inputId: 'follow' }), + 'pending_interrupts', + ) + await expectAtomic( + watch, + () => waiting.session.send('Next prompt.', { inputId: 'prompt' }), + 'pending_interrupts', + ) + await expectAtomic( + watch, + () => waiting.session.runTool({ name: 'ping', inputId: 'tool' }), + 'pending_interrupts', + ) + await expectAtomic( + watch, + () => waiting.session.reset({ inputId: 'reset' }), + 'pending_interrupts', + ) + await waiting.host.close() + + const busyLog = memoryLogStore() + const busyWatch = watchRejections(busyLog) + const release = gate() + const busy = await openOn(busyWatch.log, [ + () => + (async function* () { + await release.opened + yield* text('Later.') + })(), + ]) + const running = busy.session.send('Go.', { inputId: 'go' }) + await running.receipt + await expectAtomic( + busyWatch, + () => busy.session.send('No.', { inputId: 'no', busy: 'reject' }), + 'busy', + ) + release.open() + await running + await busy.host.close() + + const idleLog = memoryLogStore() + const idleWatch = watchRejections(idleLog) + const idle = await openOn(idleWatch.log, [() => text('Hi.')]) + await expectAtomic( + idleWatch, + () => + idle.session.send('I refunded it.', { + as: 'operator', + inputId: 'operator', + }), + 'no_human_wait', + ) + await expectAtomic( + idleWatch, + () => + idle.session.resolve({ + resume: [{ interruptId: 'missing', status: 'cancelled' }], + inputId: 'none', + }), + 'no_pending_interrupts', + ) + await idle.host.close() + + const askLog = memoryLogStore() + const askWatch = watchRejections(askLog) + const asking = await openOn( + askWatch.log, + [() => toolCall('code', {}, 'call-1'), () => text('ok')], + [askCode], + ) + const turn = asking.session.send('Ask me.', { inputId: 'ask-code' }) + turn.then( + () => {}, + () => {}, + ) + const question = await waitFor(() => { + const [wait] = asking.session.snapshot().pendingWaits + expect(wait).toBeDefined() + return wait! + }) + await expectAtomic( + askWatch, + () => + asking.session.resolve({ + resume: [ + { + interruptId: question.waitId, + status: 'resolved', + payload: 'secret', + }, + ], + inputId: 'use-answer', + }), + 'use_answer', + ) + await asking.session.answer({ questionId: question.waitId, value: 'key' }) + await turn + await asking.host.close() + + // A run id that the log already has is a conflict on the next host. + const takenLog = memoryLogStore() + const takenWatch = watchRejections(takenLog) + const firstHost = await openOn(takenWatch.log, [() => text('Hi.')]) + await (firstHost.session as unknown as SessionImpl).sendForRun('Hi.', { + runId: 'run-1', + }) + await firstHost.host.close() + const nextHost = await openOn(takenWatch.log, [() => text('no')]) + const again = (message: string, inputId: string) => + (nextHost.session as unknown as SessionImpl).sendForRun(message, { + runId: 'run-1', + inputId, + }) + await expectAtomic( + takenWatch, + () => again('Again.', 'run-again'), + 'conflict', + ) + await expectAtomic( + takenWatch, + () => + (nextHost.session as unknown as SessionImpl).resolveForRun({ + resume: [{ interruptId: 'missing', status: 'cancelled' }], + runId: 'run-1', + inputId: 'resolve-run', + }), + 'conflict', + ) + await nextHost.host.close() + }) +})