From fb524dd2ea4dbb8f09ef9f8f730606643fe0b8fe Mon Sep 17 00:00:00 2001 From: Mario Zechner Date: Wed, 28 Oct 2026 11:45:29 +0000 Subject: [PATCH] feat(durable): continue submissions --- packages/durable/CHANGELOG.md | 3 + packages/durable/README.md | 16 +- packages/durable/docs/spec.md | 89 +++- packages/durable/src/harness/generation.ts | 21 +- packages/durable/src/harness/live.ts | 7 +- packages/durable/src/harness/submissions.ts | 19 +- packages/durable/src/harness/telemetry.ts | 4 +- packages/durable/src/harness/types.ts | 42 +- packages/durable/src/session/transaction.ts | 6 +- .../src/testing/storage-conformance.ts | 48 ++ packages/durable/src/types.ts | 33 +- .../durable/test/harness-continue.test.ts | 422 ++++++++++++++++++ .../durable/test/harness-submissions.test.ts | 2 +- 13 files changed, 675 insertions(+), 37 deletions(-) create mode 100644 packages/durable/test/harness-continue.test.ts diff --git a/packages/durable/CHANGELOG.md b/packages/durable/CHANGELOG.md index 3ccc3662e..32fbab885 100644 --- a/packages/durable/CHANGELOG.md +++ b/packages/durable/CHANGELOG.md @@ -94,6 +94,8 @@ - `ReportFrom`, the second argument of a task's `runtime.report(error, from)`, is tagged by `kind`: `{ kind: "extension", extension? }` or `{ kind: "host", callback }` (was `type`). Before: `runtime.report(error, { type: "host", callback: "env" })`. After: `runtime.report(error, { kind: "host", callback: "env" })`. - `afterTool` runs on every result a tool task writes, not only on an executed call's: also on the error results the Harness writes itself, for a tool that does not resolve (`tool_unavailable`), invalid arguments (`invalid_arguments`, with `call.arguments` as the model sent them), a block (`blocked`), an interruption (`interrupted`), the restart limit (`restart_limit`), and an abort (`aborted`, `abandoned`); the diagnostic code tells them apart. Such a result is bounded with the tool's output limits after the hooks, never keeps a `structuredOutput`, and applies `control` only when the call completes; a nested call a hook turns into a success gets its bounded output as value unless the tool declares a schema. Once a tool task was cut off more than `settings.restartLimit` times its hooks no longer run, nor in the last attempt the scheduler gives its abort handler, and an abort's hooks delay `Conversation.abort()`. A non-replay-safe tool interrupted that often now ends with `restart_limit` too. Hooks that assumed the tool ran (cast `call.arguments`, read `structuredOutput`) check `result.diagnostics` first ([#11043](https://github.com/earendil-works/pi/issues/11043)) - A call to a tool its request did not offer gets a `pi.tool` task like any call, flagged `unoffered: true` in its input and its `pi.live.tools` slot, instead of a result the generation wrote without one: the task resolves nothing (a deferred tool the agent has still never runs), settles `tool_unavailable`, and runs `afterTool`, whose `control` then counts toward the round (one such call no longer prevents `terminate`). Its slot starts `pending` with a `taskId` instead of `done`, and its result entry lands in a later commit than the tool-calling answer. Controls are applied in call order. `pi.tool` is version 3 and `pi.generation` version 2 (a sequential round keeps such calls pending until their turn): an older Harness leaves both blocked (`task_too_old`) instead of resolving the flagged name ([#11043](https://github.com/earendil-works/pi/issues/11043)) +- `SubmissionRecord` gains a `type: "continue"` variant (`placed` without `entry`, `done` with `answer`, `unanswered`), so narrowing a record by `status` alone no longer guarantees an `entry`: check `type === "input"` too. A generation whose context does not end in a user message or tool results, system messages aside, now ends its run `unanswered` with the new `RunFailureReason` `nothing_to_answer` without a request; an empty conversation no longer faults its generation. `ConversationHandle.submit()` takes any `SubmissionDraft` (was `InputSubmissionDraft`), as `Conversation.submit()` does ([#11043](https://github.com/earendil-works/pi/issues/11043)) +- `RunFailureReason` (`pi.live.runFailure.reason`) gains `nothing_to_answer`, also an `unanswered` submission's `reason`: exhaustive switches over it need a case ([#11043](https://github.com/earendil-works/pi/issues/11043)) ### Added @@ -104,6 +106,7 @@ - `RunFailureReason` gains `interrupted` (the host stopped while the model was answering, with no retry left) and `restart_limit` (the run's task was cut off `settings.restartLimit` times in a row) ([#11139](https://github.com/earendil-works/pi/issues/11139)). - `onReport` gets `task_interrupted` (`reason` `restart_limit` or `interrupted`, `restarts`, and the task) when crashes end a built-in task: a tool call, a compaction, or a generation at the restart limit, or a run that ends `interrupted`. The telemetry report attributes gain the type `task_interrupted` and the reason `interrupted` ([#11139](https://github.com/earendil-works/pi/issues/11139)). - `HarnessOptions.readOnly` opens a Harness as a viewer of a session another process drives: open leaves running tasks as committed instead of recovering them, every commit with writes rejects with `ReadOnlySession` before it reaches Storage, and `resume()` and every call that would run tasks reject with it. +- Continue submissions: `submit({ type: "continue" })` starts a run on the context as it is, without a user message, for example after a run that ended without an answer (`pi.live.runFailure`). Only an idle conversation admits one (a busy one throws `ConversationBusy`; it never queues); items a failed run left queued are placed first and join the run. A context that ends in an answer, also a text answer cut off at its output limit, has nothing to continue (`nothing_to_answer`); a cut-off answer with tool calls is continued, as its calls end in error results. Tasks and tools submit one through `ConversationHandle`. The `pi.durable.submit` span starts a root trace for it, as for an input ([#11043](https://github.com/earendil-works/pi/issues/11043)) ### Changed diff --git a/packages/durable/README.md b/packages/durable/README.md index 996f484cf..ae3aae44a 100644 --- a/packages/durable/README.md +++ b/packages/durable/README.md @@ -540,9 +540,21 @@ await root.submit({ type: "write", entry: { kind: "app.note", data: "user opened - `await submission.abort(context)` withdraws a queued submission. - The [settings](#settings) `steeringMode: "all"` and `followUpMode: "all"` place every queued item at once instead of one per turn. -If a run fails, queued items stay in the inbox until the next input or queued write places them, oldest first. +If a run fails, queued items stay in the inbox until the next input, queued write, or continue places them, oldest first. -When a run ends without an answer, `docs["pi.live"].runFailure` says why until the next run starts: `model_error` and `no_model` (with the error text or the missing model in `detail`), `aborted`, `reset`, `interrupted` (the host stopped while the model answered, with no retry left), or `faulted`, `orphaned`, and `restart_limit` (the run's task could not go on). The run's inputs are in it too, settled `unanswered`; the settlement's reason is the same, except that an orphaned run's inputs carry why it was orphaned (`missing_task`, `task_too_old`, `migration_failed`) and `no_model` has no `detail` there (spec §8.2). +When a run ends without an answer, `docs["pi.live"].runFailure` says why until the next run starts: `model_error` and `no_model` (with the error text or the missing model in `detail`), `aborted`, `reset`, `interrupted` (the host stopped while the model answered, with no retry left), `faulted`, `orphaned`, and `restart_limit` (the run's task could not go on), or `nothing_to_answer` (a continue, below, found nothing to answer). The run's inputs are in it too, settled `unanswered`; the settlement's reason is the same, except that an orphaned run's inputs carry why it was orphaned (`missing_task`, `task_too_old`, `migration_failed`) and `no_model` has no `detail` there (spec §8.2). + +To resume such a run without typing a message, submit a continue. It starts a run on the context as it is, so the model answers the user message or tool results the stopped run left: + +```typescript +const live = await harness.snapshot(LiveDoc, root.id, context); +if (live?.run === undefined && live?.runFailure !== undefined) { + const resumed = await root.submit({ type: "continue" }, context); // throws ConversationBusy while busy + await resumed.wait(context); // done with the answer, or unanswered like any run +} +``` + +A continue is never queued: it throws `ConversationBusy` while a run is working, since whether there is anything to continue depends on how that run ends. Items a failed run left queued are placed first and join the run. A context with nothing to answer (empty, or ending in an answer, also a text answer cut off at its output limit) ends the run without a request, `unanswered` with `nothing_to_answer`. A cut-off answer with tool calls can be continued: its calls never ran, so the context ends in their error results. Tools can continue the conversations they own through `api.conversation(id)`, as they submit to them. An input is placed as a `pi.user` entry. To show it as something else, give it an `entry` with your kind and data; the model still sees `content` as the user message: diff --git a/packages/durable/docs/spec.md b/packages/durable/docs/spec.md index 35330bce3..2d07abfb1 100644 --- a/packages/durable/docs/spec.md +++ b/packages/durable/docs/spec.md @@ -241,6 +241,32 @@ type SubmissionRecord = readonly reason: string; readonly detail?: JsonValue; } + )) + /** A run without a user message (section 6): no entry, never queued. */ + | (SubmissionRecordBase & { + readonly type: "continue"; + } & ( + | { + readonly status: "placed"; + readonly entry?: never; + readonly answer?: never; + readonly reason?: never; + readonly detail?: never; + } + | { + readonly status: "done"; + readonly entry?: never; + readonly answer: EntryId; + readonly reason?: never; + readonly detail?: never; + } + | { + readonly status: "unanswered"; + readonly entry?: never; + readonly answer?: never; + readonly reason: string; + readonly detail?: JsonValue; + } )); type SubmissionSettlement = @@ -349,6 +375,13 @@ type SubmissionDraft = { readonly content?: never; readonly whenBusy?: never; } + /** Start a run on the context as it is, without a user message; idle only (section 6). */ + | { + readonly type: "continue"; + readonly content?: never; + readonly entry?: never; + readonly whenBusy?: never; + } ); type InputSubmissionDraft = Extract; @@ -1410,7 +1443,7 @@ interface Tx { ): Promise>; /** Raw submission record; no admission rules (busy check, inbox, placement). Hosts use `Conversation.submit()`. */ createSubmission(create: SubmissionCreate): Promise; - /** Settle a queued or placed submission; only a placed input can be answered. A settled one stays unchanged. */ + /** Settle a queued or placed submission; only a placed input or continue can be answered. A settled one stays unchanged. */ settleSubmission(id: SubmissionId, settlement: SubmissionSettlement): void; /** Newest visible entry of the conversation that carries a `head`. */ latestHeadMarker(conversationId: ConversationId): Promise<(EntryRecord & { readonly head: EntryId }) | undefined>; @@ -2795,12 +2828,14 @@ Admission and terminal transitions: | action | submission state | other writes | |---|---|---| | input, idle with empty inbox | `placed`, with user entry | create run and generation | +| continue, idle | `placed`, no entry | place queued items (final boundary); create run and generation with them and itself | +| continue, busy | none | reject with `ConversationBusy` | | display-only write, no queued reset | `done`, with entry | append entry; no run | | write, idle with empty inbox | `done`, with entry | append entry; no run | | input or other write, busy or non-empty inbox | `queued` | append inbox item | | boundary places user item | `placed`, with entry | add ID to current or successor run | | boundary places write | `done`, with entry | append entry | -| run answers | input `done`, with required answer entry | remove `run`, or hand it to a successor | +| run answers | input or continue `done`, with required answer entry | remove `run`, or hand it to a successor | | run fails or its task aborts | input `unanswered`, with reason | remove `run`; inbox per `afterFailedRun` below | | withdraw queued item | `unanswered`, reason `aborted` | remove inbox item | | stale head write | `unanswered`, reason `stale` | remove inbox item | @@ -2810,8 +2845,26 @@ for the other submission type rejects. A busy input with `whenBusy: "reject"` writes no record and reports `ConversationBusy`. An idle input queues with mode `steer` when `whenBusy` is `steer` and `followUp` otherwise. +A continue starts a run on the context as it is, without a user message, for +example after a run that ended without an answer. It is admitted only while the +conversation is idle and never queues: whether a run can continue depends on how +the current one ends, so a busy conversation rejects it with `ConversationBusy` +before any write. Its record is `placed` without an entry; an idle conversation's +queued items are placed first by a final boundary in the same commit, and its +selected user items join the run, ahead of the continue. Whether the context has +anything to answer is checked when the run's generation prepares (section 8.3), +not at admission: a context that does not end in a user message or tool results, +system messages aside, ends the run without a request, `unanswered` with +`nothing_to_answer`. An answer cut off at its output limit (`length`) is an +answer, so a text answer is not continued; one with tool calls is, because its +calls never ran and the context ends in their synthesized `missing_result` +error results (section 2.1). A host offers continue while the conversation is +idle and `pi.live.runFailure` is set (section 8.2); a stale failure (one a later +run started in its commit kept) settles `nothing_to_answer`. Each continue starts +a run with its own turn budget. + `settings.afterFailedRun` decides what a run that fails with `model_error`, -`no_model`, `interrupted`, or `restart_limit` does with the inputs queued behind it, in the commit that settles its +`no_model`, `interrupted`, `restart_limit`, or `nothing_to_answer` does with the inputs queued behind it, in the commit that settles its own inputs: `keep` (the default) leaves them queued; `continue` applies the final boundary, so the next run starts with them, as after an answer; `withdraw` settles them `unanswered` with reason `run_failed`, keeping queued @@ -3316,7 +3369,8 @@ type ToolExecutionResult = { interface ConversationHandle { readonly id: ConversationId; - submit(submission: InputSubmissionDraft, context: Context): Promise; + /** Any submission, as `Conversation.submit()` takes: a task may also continue a conversation it owns. */ + submit(submission: SubmissionDraft, context: Context): Promise; abort(context: Context): Promise; waitForIdle(context: Context): Promise; } @@ -4347,7 +4401,16 @@ type RunFailure = { taskId: TaskId; /** The run's inputs, settled `unanswered`. */ inputs: SubmissionId[]; - reason: "model_error" | "no_model" | "aborted" | "reset" | "faulted" | "orphaned" | "interrupted" | "restart_limit"; + reason: + | "model_error" + | "no_model" + | "aborted" + | "reset" + | "faulted" + | "orphaned" + | "interrupted" + | "restart_limit" + | "nothing_to_answer"; /** * `model_error`: the error text; `no_model`: which model is missing; `faulted`: the error message; `orphaned`: why * the task could not run (`missing_task`, `task_too_old`, `migration_failed`). @@ -4440,6 +4503,7 @@ that settles its inputs `unanswered`: | `orphaned` | its task was aborted while no definition could take it; `detail` is the reason | the reason | | `interrupted` | the host stopped while the model was answering, with no retry left (section 8.3); `detail` names the attempt | `interrupted` with the text | | `restart_limit` | its generation was cut off `restartLimit` times in a row (section 5.4), also while an interrupted request was charged; `detail` says in which phase | `restart_limit` with the text | +| `nothing_to_answer` | its generation found a context that does not end in a user message or tool results, system messages aside, such as a continue after an answer (section 8.3); no `detail` | `nothing_to_answer` | The next commit that starts a run removes it. A run started in the same commit, the queue's next run under `afterFailedRun: "continue"` (section 6) or after a @@ -4519,7 +4583,13 @@ The run's inputs live in `pi.live.run`, not in the task input. the settings once. When the agent has no model or `models.getModel()` does not know it, the task fails with `no_model`. When it resumes with `overflow` and its compaction did not complete with an `entryId`, the run fails with - `model_error` and the overflow text as detail. Otherwise one commit appends + `model_error` and the overflow text as detail. When the committed model + context does not end in a user message or tool results, system messages + aside (it is empty, or ends in an answer, as for a continue after one), the + run fails with `nothing_to_answer` before any section is rendered or request + sent. Only roles count: a `length` answer with only text ends the context, + while one with tool calls is followed by the error results synthesized for + its calls (section 2.1), so a continue answers those. Otherwise one commit appends the planned `pi.system` entries and moves to `request` with the new tail as `cutoff`, the agent's model and thinking level, and the settings' stream options. These stay fixed for this request attempt; a retry prepares again. `compacted` carries over to @@ -4653,9 +4723,10 @@ A tool round starts in the commit that appends the tool-calling answer: 4. `pi.live.tools` receives the round's slots (section 8.2), and `generation` is removed. -Input submissions settle `unanswered` with one of these reasons: `no_model`, -`model_error` (detail: provider error text), `aborted`, `faulted` (detail: error -message), or an orphaning blocked reason (section 5.4). Fault and orphan +Input and continue submissions settle `unanswered` with one of these reasons: +`no_model`, `model_error` (detail: provider error text), `nothing_to_answer`, +`aborted`, `faulted` (detail: error message), or an orphaning blocked reason +(section 5.4). Fault and orphan settlement convert a committed partial into an aborted `pi.assistant` entry, like the abort handler. diff --git a/packages/durable/src/harness/generation.ts b/packages/durable/src/harness/generation.ts index 56d130082..3302b467d 100644 --- a/packages/durable/src/harness/generation.ts +++ b/packages/durable/src/harness/generation.ts @@ -183,6 +183,11 @@ export const GenerationTask = defineTask message.role !== "system"); + if (last?.role !== "user" && last?.role !== "toolResult") return failNothingToAnswer(runtime, context); const shown = replaySections(view.messages); let env: ExecutionEnv | undefined; try { @@ -218,9 +223,9 @@ export const GenerationTask = defineTask { - let cutoff = (await tx.scanEntries({ conversationId }, 1)).items[0]?.id; + // The context ends in a message the model answers, so the conversation has an entry. + let cutoff = (await tx.scanEntries({ conversationId }, 1)).items[0]!.id; for (const entry of entries) cutoff = (await tx.appendEntry(SystemEntry, conversationId, entry)).id; - if (cutoff === undefined) throw new Error(`Conversation ${conversationId} has no entries to send`); // Checked in this commit, so a compaction admitted during preparation counts. if (threshold === "background" && (await tx.doc(LiveDoc, conversationId)).compactions === undefined) { await createCompaction(tx, conversationId, { reason: "threshold" }); @@ -552,6 +557,18 @@ async function failNoModel(runtime: Runtime, ref: ModelRef | undefined, context: }, context); } +/** End the run without a request: its context has nothing the model would answer. */ +async function failNothingToAnswer(runtime: Runtime, context: Context): Promise { + const message = "The conversation has nothing to answer"; + await runtime.commit(async (tx) => { + const failure = await prepareFailure(tx, runtime); + const live = await tx.doc(LiveDoc, runtime.conversationId); + await endFailedRun(tx, runtime, live, failure, { status: "unanswered", reason: "nothing_to_answer" }); + const error = { message, detail: { reason: "nothing_to_answer" } }; + return { status: "terminal", outcome: { status: "failed", error } }; + }, context); +} + /** What a failed run's end needs from the table, read before the commit's first table write. */ type Failure = { readonly policy: AfterFailedRun; readonly boundary?: Boundary }; diff --git a/packages/durable/src/harness/live.ts b/packages/durable/src/harness/live.ts index a6669e3cb..5fb3cfb46 100644 --- a/packages/durable/src/harness/live.ts +++ b/packages/durable/src/harness/live.ts @@ -100,7 +100,9 @@ export type CompactionFailure = { * - `orphaned`: the run's task was aborted while no registered definition could take it; * - `interrupted`: the host stopped while the model was answering, more often than `settings.retry` retries; * - `restart_limit`: the run's generation was cut off `settings.restartLimit` times in a row; a request or poll it was - * cut off during counts as a failed attempt first, so this ends one whose charge itself keeps failing. + * cut off during counts as a failed attempt first, so this ends one whose charge itself keeps failing; + * - `nothing_to_answer`: the context did not end in a user message or tool results (empty, or ending in an answer), so + * no request was sent, as for a continue after an answer. */ export type RunFailureReason = | "model_error" @@ -110,7 +112,8 @@ export type RunFailureReason = | "faulted" | "orphaned" | "interrupted" - | "restart_limit"; + | "restart_limit" + | "nothing_to_answer"; /** The newest run that ended without an answer (spec §8.2). */ export type RunFailure = { diff --git a/packages/durable/src/harness/submissions.ts b/packages/durable/src/harness/submissions.ts index 419c6786d..7de0054b8 100644 --- a/packages/durable/src/harness/submissions.ts +++ b/packages/durable/src/harness/submissions.ts @@ -67,7 +67,9 @@ export class Submissions { /** Admit a submission in one commit; see `admitSubmission()`. */ async submit(conversationId: ConversationId, draft: SubmissionDraft, context: Context): Promise { this.#resume(); - const parent = draft.type === "input" ? this.#telemetry : context.telemetry; + // Inputs and continues start runs: root traces linked to their callers. + const startsRun = draft.type !== "write"; + const parent = startsRun ? this.#telemetry : context.telemetry; return createTypedSpanStarter(parent, [DURABLE_TELEMETRY_SCHEMA])( "pi.durable.submit", { @@ -75,7 +77,7 @@ export class Submissions { "conversation.id": conversationId, }, async (span) => { - if (draft.type === "input") span.addLink({ context: context.telemetry }); + if (startsRun) span.addLink({ context: context.telemetry }); try { const { id, status } = await this.#session.commitWith( (tx) => admitSubmission(tx, conversationId, draft, this.#now(), this.#queueModes(), span), @@ -187,7 +189,8 @@ function checkInputEntry(entry: InputEntry): void { * known request ID returns its existing submission without writing. A display-only write (only `kind` and `data`) is * appended and settled `done` at once, busy or not, as it changes nothing the model sees; only behind a queued reset it * queues, so it lands after the reset instead of before it in the context the reset ends. A busy conversation queues anything else in - * `pi.inbox`, or rejects `whenBusy: "reject"` input with `ConversationBusy`. An idle conversation with queued items + * `pi.inbox`, or rejects `whenBusy: "reject"` input and any continue with `ConversationBusy`. An idle continue places + * queued items and starts a run with them and itself, placing no entry of its own. An idle conversation with queued items * queues it behind them and runs a final boundary. Otherwise idle input places its entry (`pi.user`, or the input's * own) and starts a run, and an idle write appends its entry and settles `done`, or `stale` when its head reaches * before the active range. @@ -232,6 +235,16 @@ export async function admitSubmission( if (run?.trace !== undefined) span.addLink({ context: span.propagation.extract(run.trace) }); } if (busy && draft.type === "input" && draft.whenBusy === "reject") throw new ConversationBusy(conversationId); + if (draft.type === "continue") { + // Never queued: whether a run can continue depends on how the current one ends. Rejected before any write. + if (busy) throw new ConversationBusy(conversationId); + const boundary = await prepareBoundary(tx, conversationId, queueModes); + const { id } = await tx.createSubmission({ ...fields, type: "continue", status: "placed" }); + // Items an earlier failed run left queued (`afterFailedRun: "keep"`) are placed first and join the run. + const { users } = await applyBoundary(tx, boundary, "final", now); + await startRun(tx, conversationId, live, [...users, id]); + return { id, status: "placed" }; + } // A boundary reads the table, so it is prepared before the first table write; a busy one needs none. const boundary = busy ? undefined : await prepareBoundary(tx, conversationId, queueModes); if (boundary === undefined || boundary.inbox.items.length > 0) { diff --git a/packages/durable/src/harness/telemetry.ts b/packages/durable/src/harness/telemetry.ts index 1cf66ef73..0c5f60ec6 100644 --- a/packages/durable/src/harness/telemetry.ts +++ b/packages/durable/src/harness/telemetry.ts @@ -52,13 +52,13 @@ export const DURABLE_TELEMETRY_SCHEMA = defineTelemetrySchema({ spans: { "pi.durable.submit": { description: - "Public submission admission and its commit. Inputs start root traces linked to their callers; writes are children. Internal admissions emit phase events only.", + "Public submission admission and its commit. Inputs and continues start root traces linked to their callers; writes are children. Internal admissions emit phase events only.", parents: { kind: "any" }, startAttributes: { "submission.type": { type: "string", required: true, - values: ["input", "write"], + values: ["input", "write", "continue"], description: "Submission type.", }, "conversation.id": { type: "number", required: true, description: "Conversation ID.", cardinality: "high" }, diff --git a/packages/durable/src/harness/types.ts b/packages/durable/src/harness/types.ts index 272af94f4..353231dac 100644 --- a/packages/durable/src/harness/types.ts +++ b/packages/durable/src/harness/types.ts @@ -63,7 +63,11 @@ export type UserInput = UserMessage["content"]; */ export type InputEntry = { readonly kind: string; readonly data?: JsonValue }; -/** Host submission: user input that may start a run, or a passive entry write. */ +/** + * Host submission: user input that may start a run, a passive entry write, or a continue: a run without a new user + * message, for an idle conversation whose context the model still owes an answer (it ends in a user message or tool + * results, for example after a run that ended without an answer). + */ export type SubmissionDraft = { readonly requestId?: string; } & ( @@ -80,6 +84,19 @@ export type SubmissionDraft = { readonly content?: never; readonly whenBusy?: never; } + | { + /** + * Start a run on the context as it is, without a user message. Only when idle: a busy conversation rejects it with + * `ConversationBusy`, and it never queues. Items kept in `pi.inbox` are placed first and join the run. A context + * with nothing to answer (empty, or ending in an answer, also a text answer cut off at its output limit) ends the + * run without a request, `unanswered` with `nothing_to_answer`. A cut-off answer with tool calls is continued: + * its calls never ran, and the context ends in their error results (`missing_result`). + */ + readonly type: "continue"; + readonly content?: never; + readonly entry?: never; + readonly whenBusy?: never; + } ); export type InputSubmissionDraft = Extract; @@ -127,12 +144,13 @@ export type ConversationAbortOptions = { export type HooksOf = K extends Task ? H : never; /** - * Invocation-bound conversation operations for tasks and tools. Rejects after the invocation ends; passive entries are - * written with ordinary transaction writes instead. + * Invocation-bound conversation operations for tasks and tools. Rejects after the invocation ends. `submit` takes any + * submission, as `Conversation.submit()` does: a subagent's conversation is an ordinary one, so a tool may also continue + * it. */ export interface ConversationHandle { readonly id: ConversationId; - submit(submission: InputSubmissionDraft, context: Context): Promise; + submit(submission: SubmissionDraft, context: Context): Promise; /** `Conversation.abort()`: withdraw queued inputs, abort the ordinary ownership scope, and wait until it is idle. */ abort(context: Context, options?: ConversationAbortOptions): Promise; /** Resolve when the conversation's ordinary ownership scope has no live non-background task. */ @@ -578,9 +596,10 @@ export type HarnessSettings = { */ readonly documentRetentionMs?: number; /** - * What a run that fails (`model_error`, `no_model`, `interrupted`, `restart_limit`) does with the inputs queued behind it: `keep` leaves them - * queued until the next input or queued write, `continue` starts the next run with them, as an answer would, and - * `withdraw` settles them `unanswered` with reason `run_failed`. Default `keep`. + * What a run that fails (`model_error`, `no_model`, `interrupted`, `restart_limit`, `nothing_to_answer`) does with the + * inputs queued behind it: `keep` leaves them queued until the next input or queued write, `continue` starts the + * next run with them, as an answer would, and `withdraw` settles them `unanswered` with reason `run_failed`. Default + * `keep`. */ readonly afterFailedRun?: AfterFailedRun; /** @@ -752,10 +771,11 @@ export interface Conversation { configure(change: AgentChange, context: Context): Promise; /** - * Durably admit user input or a passive entry write. A busy conversation, or one with queued items, queues input and - * writes that affect model context in `pi.inbox`; `whenBusy: "reject"` rejects input with `ConversationBusy` instead - * and writes nothing. A write of only `kind` and `data` (no `model`, `head`, or `edits`) is appended at once, - * even mid-run, as it changes nothing the model sees. + * Durably admit user input, a passive entry write, or a continue. A busy conversation, or one with queued items, + * queues input and writes that affect model context in `pi.inbox`; `whenBusy: "reject"` rejects input with + * `ConversationBusy` instead and writes nothing. A write of only `kind` and `data` (no `model`, `head`, or `edits`) + * is appended at once, even mid-run, as it changes nothing the model sees. A continue starts a run on the context as + * it is, after placing queued items; a busy conversation rejects it with `ConversationBusy`, as it never queues. */ submit(submission: SubmissionDraft, context: Context): Promise; /** diff --git a/packages/durable/src/session/transaction.ts b/packages/durable/src/session/transaction.ts index cd59bba27..79f91ada8 100644 --- a/packages/durable/src/session/transaction.ts +++ b/packages/durable/src/session/transaction.ts @@ -64,8 +64,8 @@ type AnyTaskRecord = TaskRecord; type SubmissionChange = SubmissionSettlement | { readonly status: "placed"; readonly entry: EntryId }; /** - * Complete record after applying one change. Placement turns a queued input `placed` and a queued write `done`; only a - * placed input can be answered. A settled record stays. + * Complete record after applying one change. Placement turns a queued input `placed` and a queued write `done`; a + * continue is never queued, so never placed. Only a placed input or continue can be answered. A settled record stays. */ function applySubmissionChange(current: SubmissionRecord, change: SubmissionChange): SubmissionRecord { if (current.status === "done" || current.status === "unanswered") return current; @@ -75,7 +75,7 @@ function applySubmissionChange(current: SubmissionRecord, change: SubmissionChan return { ...current, status, entry: change.entry } as SubmissionRecord; } if (change.status === "done" && current.status !== "placed") { - throw new Error(`Submission ${current.id} is not a placed input`); + throw new Error(`Submission ${current.id} is not a placed input or continue`); } // Queued and placed records carry no answer, reason, or detail; an unanswered input keeps its entry. return { ...current, ...change } as SubmissionRecord; diff --git a/packages/durable/src/testing/storage-conformance.ts b/packages/durable/src/testing/storage-conformance.ts index 58dcd599b..2797c0462 100644 --- a/packages/durable/src/testing/storage-conformance.ts +++ b/packages/durable/src/testing/storage-conformance.ts @@ -908,6 +908,54 @@ export function createStorageConformance(options: StorageConformanceOptions): re expect(await storage.submissionByRequest(rootId, "passive-failed", context)).toEqual(unanswered); }), + createCase(options, "stores continue submissions, which have no entry and are never queued", async (storage) => { + const rootId = await createRoot(storage); + const doneId = await storage.mintId(); + const failedId = await storage.mintId(); + const placedDone: SubmissionRecord = { + id: doneId, + conversationId: rootId, + requestId: "continue-done", + type: "continue", + status: "placed", + }; + const placedFailed: SubmissionRecord = { + id: failedId, + conversationId: rootId, + requestId: "continue-failed", + type: "continue", + status: "placed", + }; + await storage.commit( + [ + { type: "submission", value: placedDone }, + { type: "submission", value: placedFailed }, + ], + adopt, + context, + ); + expect(await storage.submission(doneId, context)).toEqual(placedDone); + expect((await storage.scanSubmissions({ status: "placed" }, 10, undefined, context)).items).toEqual([ + placedDone, + placedFailed, + ]); + + const done: SubmissionRecord = { ...placedDone, status: "done", answer: await storage.mintId() }; + const unanswered: SubmissionRecord = { ...placedFailed, status: "unanswered", reason: "nothing_to_answer" }; + await storage.commit( + [ + { type: "submission", value: done }, + { type: "submission", value: unanswered }, + ], + adopt, + context, + ); + expect(await storage.submission(doneId, context)).toEqual(done); + expect(await storage.submissionByRequest(rootId, "continue-done", context)).toEqual(done); + expect(await storage.submission(failedId, context)).toEqual(unanswered); + expect(await storage.submissionByRequest(rootId, "continue-failed", context)).toEqual(unanswered); + }), + createCase(options, "reconstructs rewindable documents and preserves half-open incarnations", async (storage) => { const rootId = await createRoot(storage); const firstId = await storage.mintId(); diff --git a/packages/durable/src/types.ts b/packages/durable/src/types.ts index 3e20ff6ac..3ea1f1ca4 100644 --- a/packages/durable/src/types.ts +++ b/packages/durable/src/types.ts @@ -491,7 +491,7 @@ type SubmissionRecordBase = { readonly requestId?: string; }; -/** Durable lifecycle of one admitted user input or passive entry write. */ +/** Durable lifecycle of one admitted user input, passive entry write, or continue. */ export type SubmissionRecord = | (SubmissionRecordBase & { readonly type: "input"; @@ -556,6 +556,35 @@ export type SubmissionRecord = readonly reason: string; readonly detail?: JsonValue; } + )) + | (SubmissionRecordBase & { + /** A run started without a user message; it places no entry and is never queued. */ + readonly type: "continue"; + } & ( + | { + /** Owned by an active run. */ + readonly status: "placed"; + readonly entry?: never; + readonly answer?: never; + readonly reason?: never; + readonly detail?: never; + } + | { + /** The run answered. */ + readonly status: "done"; + readonly entry?: never; + readonly answer: EntryId; + readonly reason?: never; + readonly detail?: never; + } + | { + /** The run ended without an answer. */ + readonly status: "unanswered"; + readonly entry?: never; + readonly answer?: never; + readonly reason: string; + readonly detail?: JsonValue; + } )); /** Terminal status staged for a submission; identity, type, and entry come from its current record. */ @@ -992,7 +1021,7 @@ export interface Tx { */ createSubmission(create: SubmissionCreate): Promise; /** - * Settle a queued or placed submission; only a placed input can be answered, and a settled submission stays + * Settle a queued or placed submission; only a placed input or continue can be answered, and a settled submission stays * unchanged. Resolved against this transaction's latest record of the submission, so it works after table writes. * Run tasks settle the inputs they answer. */ diff --git a/packages/durable/test/harness-continue.test.ts b/packages/durable/test/harness-continue.test.ts new file mode 100644 index 000000000..8b609b8eb --- /dev/null +++ b/packages/durable/test/harness-continue.test.ts @@ -0,0 +1,422 @@ +// `submit({ type: "continue" })`: a run without a new user message (#11043). +import { mkdtemp, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { withTelemetry } from "@earendil-works/chord/context"; +import { + type AssistantMessage, + type FauxResponseStep, + fauxAssistantMessage, + fauxText, + fauxToolCall, + type Message, + Type, +} from "@earendil-works/pi-ai"; +import { + AssistantEntry, + type Conversation, + ConversationBusy, + configure, + defineTool, + Harness, + InboxDoc, + LiveDoc, + MemoryStorage, + type SettledSubmissionRecord, + type Storage, + UserEntry, +} from "@earendil-works/pi-durable"; +import { InMemoryTelemetryContext } from "@earendil-works/pi-telemetry"; +import { afterEach, describe, expect, it } from "vitest"; +import { openNodeSqliteStorage } from "../src/storage/sqlite/node.ts"; +import { allEntries, type ChatSetup, chatSetup, openChat, textOf, unanswered, waitFor } from "./chat-support.ts"; +import { addTool } from "./harness-support.ts"; +import { context } from "./session-support.ts"; +import { aborted, deferred } from "./task-support.ts"; + +const directories = new Set(); +/** Storages of killed Harnesses, never closed by them: closed after each test. */ +const killed = new Set(); + +afterEach(async () => { + for (const storage of killed) await storage.close(context).catch(() => {}); + killed.clear(); + for (const directory of directories) await rm(directory, { recursive: true, force: true }); + directories.clear(); +}); + +async function sqlitePath(): Promise { + const directory = await mkdtemp(join(tmpdir(), "pi-durable-continue-")); + directories.add(directory); + return join(directory, "session.sqlite"); +} + +/** Open a chat over the SQLite file at `path` and start scheduling; `kill()` simulates a crash. */ +async function open(path: string, setup: ChatSetup) { + const inner = await openNodeSqliteStorage(path); + let dead = false; + const storage = new Proxy(inner, { + get(target, property) { + const value = Reflect.get(target, property, target); + if (typeof value !== "function") return value; + return (...args: unknown[]) => (dead ? new Promise(() => {}) : value.apply(target, args)); + }, + }); + const opened = await openChat(storage, setup); + opened.harness.resume(); + const kill = (): void => { + dead = true; + killed.add(inner); + }; + return { ...opened, kill }; +} + +const failure = (): AssistantMessage => + fauxAssistantMessage([], { stopReason: "error", errorMessage: "Invalid request" }); + +/** A response step that records the roles of the context it was asked with, then answers `text`. */ +function answering(text: string, seen: Message["role"][][]): FauxResponseStep { + return (request) => { + seen.push(request.messages.map((message) => message.role)); + return fauxAssistantMessage([fauxText(text)]); + }; +} + +async function answerText(root: Conversation, settled: SettledSubmissionRecord): Promise { + if (settled.status !== "done" || settled.type === "write") throw new Error(`Unexpected ${settled.status}`); + const answer = await root.commit((tx) => tx.entry(AssistantEntry, settled.answer), context); + return textOf(answer?.model?.[0]); +} + +async function userEntries(root: Conversation): Promise { + return (await allEntries(root)).filter((entry) => UserEntry.is(entry)).length; +} + +describe("continue", () => { + // #11043 + it("answers a run that failed, without a new user message", async () => { + const setup = chatSetup(); + const seen: Message["role"][][] = []; + setup.faux.setResponses([failure, answering("recovered", seen)]); + const { harness, root } = await openChat(new MemoryStorage(), setup); + const first = await root.submit({ type: "input", content: "Analyze this" }, context); + expect(await first.wait(context)).toMatchObject({ status: "unanswered", reason: "model_error" }); + expect((await harness.snapshot(LiveDoc, root.id, context))?.runFailure?.reason).toBe("model_error"); + + const resumed = await root.submit({ type: "continue" }, context); + expect((await resumed.status(context)).status).toBe("placed"); + const settled = await resumed.wait(context); + expect(settled.type).toBe("continue"); + expect(await answerText(root, settled)).toBe("recovered"); + expect(seen.at(-1)?.at(-1)).toBe("user"); + // The failed input stays unanswered; no user entry was added. + expect(await first.status(context)).toMatchObject({ status: "unanswered" }); + expect(await userEntries(root)).toBe(1); + expect((await harness.snapshot(LiveDoc, root.id, context))?.runFailure).toBeUndefined(); + await harness.close(context); + }); + + // #11043 + it("answers the tool results of a round aborted midway", async () => { + const setup = chatSetup(); + const started = deferred(); + addTool( + setup.registry, + defineTool({ + name: "wait", + description: "Waits until aborted", + parameters: Type.Object({}), + execute: async (_args, _api, callContext) => { + started.resolve(); + return aborted(callContext.abortSignal!); + }, + }), + ); + const seen: Message["role"][][] = []; + setup.faux.setResponses([ + fauxAssistantMessage([fauxToolCall("wait", {}, { id: "c1" })], { stopReason: "toolUse" }), + answering("after the abort", seen), + ]); + const { harness, root } = await openChat(new MemoryStorage(), setup); + const first = await root.submit({ type: "input", content: "go" }, context); + await started.promise; + await root.abort(context); + expect(await first.wait(context)).toMatchObject({ status: "unanswered", reason: "aborted" }); + const settled = await (await root.submit({ type: "continue" }, context)).wait(context); + expect(await answerText(root, settled)).toBe("after the abort"); + expect(seen.at(-1)?.at(-1)).toBe("toolResult"); + await harness.close(context); + }); + + // #11043 + it("skips system messages when deciding there is something to answer", async () => { + const setup = chatSetup(); + const seen: Message["role"][][] = []; + setup.faux.setResponses([fauxAssistantMessage("first answer"), failure, answering("resumed", seen)]); + const { harness, root } = await openChat(new MemoryStorage(), setup); + await root.configure({ instructions: "Be thorough." }, context); + await (await root.submit({ type: "input", content: "one" }, context)).wait(context); + // Changed instructions append a system delta after the next user message, before its request. + await root.configure({ instructions: "Be brief." }, context); + const second = await root.submit({ type: "input", content: "two" }, context); + expect(await second.wait(context)).toMatchObject({ status: "unanswered", reason: "model_error" }); + const roles = (await root.context(context)).messages.map((message) => message.role); + expect(roles.at(-1)).toBe("system"); + const settled = await (await root.submit({ type: "continue" }, context)).wait(context); + expect(await answerText(root, settled)).toBe("resumed"); + await harness.close(context); + }); + + // #11043 + it("is rejected while busy, after request ID deduplication", async () => { + const setup = chatSetup(); + const busy = unanswered(); + setup.faux.setResponses([failure, busy.step]); + const { harness, root } = await openChat(new MemoryStorage(), setup); + await (await root.submit({ type: "input", content: "go" }, context)).wait(context); + const resumed = await root.submit({ type: "continue", requestId: "resume-1" }, context); + await busy.reached; + await expect(root.submit({ type: "continue" }, context)).rejects.toBeInstanceOf(ConversationBusy); + // The same request ID returns the existing submission instead of rejecting. + expect((await root.submit({ type: "continue", requestId: "resume-1" }, context)).id).toBe(resumed.id); + await harness.close(context); + }); + + // #11043 + it("ends without a request when there is nothing to answer", async () => { + const setup = chatSetup(); + setup.faux.setResponses([fauxAssistantMessage("answered")]); + const { harness, root } = await openChat(new MemoryStorage(), setup); + const empty = await (await root.submit({ type: "continue" }, context)).wait(context); + expect(empty).toMatchObject({ type: "continue", status: "unanswered", reason: "nothing_to_answer" }); + await (await root.submit({ type: "input", content: "hi" }, context)).wait(context); + const calls = setup.faux.state.callCount; + // A context ending in an answer has nothing to continue. + const after = await (await root.submit({ type: "continue" }, context)).wait(context); + expect(after).toMatchObject({ status: "unanswered", reason: "nothing_to_answer" }); + expect(setup.faux.state.callCount).toBe(calls); + expect((await harness.snapshot(LiveDoc, root.id, context))?.runFailure?.reason).toBe("nothing_to_answer"); + await harness.close(context); + }); + + // #11043 + it("continues an answer with tool calls cut off at its output limit, which ends in their missing results", async () => { + const setup = chatSetup(); + const seen: Message["role"][][] = []; + setup.faux.setResponses([ + fauxAssistantMessage([fauxText("calling"), fauxToolCall("echo", {}, { id: "c1" })], { stopReason: "length" }), + answering("resumed", seen), + ]); + const { harness, root } = await openChat(new MemoryStorage(), setup); + // The run answered: a cut-off answer is an answer, and its calls never ran. + expect((await (await root.submit({ type: "input", content: "go" }, context)).wait(context)).status).toBe("done"); + const settled = await (await root.submit({ type: "continue" }, context)).wait(context); + expect(await answerText(root, settled)).toBe("resumed"); + expect(seen.at(-1)?.slice(-2)).toEqual(["assistant", "toolResult"]); + await harness.close(context); + }); + + // #11043 + it("does not continue a text answer cut off at its output limit", async () => { + const setup = chatSetup(); + setup.faux.setResponses([fauxAssistantMessage([fauxText("cut")], { stopReason: "length" })]); + const { harness, root } = await openChat(new MemoryStorage(), setup); + await (await root.submit({ type: "input", content: "write a lot" }, context)).wait(context); + const settled = await (await root.submit({ type: "continue" }, context)).wait(context); + expect(settled).toMatchObject({ status: "unanswered", reason: "nothing_to_answer" }); + expect(setup.faux.state.callCount).toBe(1); + await harness.close(context); + }); + + // #11043 + it("answers a run interrupted by a host that died while the model answered", async () => { + const path = await sqlitePath(); + const first = chatSetup({ tokensPerSecond: 20, tokenSize: { min: 1, max: 1 } }); + first.faux.setResponses([fauxAssistantMessage("z".repeat(400))]); + first.settings.progress = { persistWithinMs: 0 }; + let opened = await open(path, first); + const id = (await opened.root.submit({ type: "input", content: "hi" }, context)).id; + await waitFor( + async () => + (await opened.harness.snapshot(LiveDoc, opened.root.id, context))?.generation?.message !== undefined, + ); + await new Promise((resolve) => setTimeout(resolve, 50)); + opened.kill(); + + const setup = chatSetup(); + setup.settings.retry = { enabled: false }; + const seen: Message["role"][][] = []; + setup.faux.setResponses([answering("after the crash", seen)]); + opened = await open(path, setup); + const interrupted = await (await opened.harness.submission(id, context))!.wait(context); + expect(interrupted).toMatchObject({ status: "unanswered", reason: "interrupted" }); + const settled = await (await opened.root.submit({ type: "continue" }, context)).wait(context); + expect(await answerText(opened.root, settled)).toBe("after the crash"); + // The aborted partial is not model context: the request ends in the user message. + expect(seen.at(-1)?.at(-1)).toBe("user"); + await opened.harness.close(context); + }); + + // #11043 + it("places a reset a failed run kept, and finds nothing to answer in the context it starts", async () => { + const setup = chatSetup(); + const gate = deferred(); + const reached = deferred(); + setup.faux.setResponses([ + async () => { + reached.resolve(); + await gate.promise; + return failure(); + }, + ]); + const { harness, root } = await openChat(new MemoryStorage(), setup); + const first = await root.submit({ type: "input", content: "first" }, context); + await reached.promise; + await root.reset(undefined, context); + gate.resolve(); + expect(await first.wait(context)).toMatchObject({ status: "unanswered", reason: "model_error" }); + expect((await harness.snapshot(InboxDoc, root.id, context))?.items).toHaveLength(1); + + const settled = await (await root.submit({ type: "continue" }, context)).wait(context); + expect(settled).toMatchObject({ status: "unanswered", reason: "nothing_to_answer" }); + expect((await harness.snapshot(InboxDoc, root.id, context))?.items ?? []).toEqual([]); + expect((await root.context(context)).messages.filter((message) => message.role !== "system")).toEqual([]); + expect(setup.faux.state.callCount).toBe(1); + await harness.close(context); + }); + + // #11043 + it("starts the inputs queued behind a continue with nothing to answer under afterFailedRun continue", async () => { + const setup = chatSetup(); + setup.settings.afterFailedRun = "continue"; + setup.faux.setResponses([fauxAssistantMessage("answered"), fauxAssistantMessage("second answered")]); + const { harness, root } = await openChat(new MemoryStorage(), setup); + await (await root.submit({ type: "input", content: "one" }, context)).wait(context); + const resumed = await root.submit({ type: "continue" }, context); + const queued = await root.submit({ type: "input", content: "two", whenBusy: "followUp" }, context); + expect((await queued.status(context)).status).toBe("queued"); + expect(await resumed.wait(context)).toMatchObject({ status: "unanswered", reason: "nothing_to_answer" }); + const settled = await queued.wait(context); + expect(await answerText(root, settled)).toBe("second answered"); + await harness.close(context); + }); + + // #11043 + it("is submitted by a tool through the handle of a conversation it owns, and rejected there while busy", async () => { + const setup = chatSetup(); + const busy = deferred(); + const release = deferred(); + const outcome: { busy?: unknown; first?: string; resumed?: string } = {}; + addTool( + setup.registry, + defineTool({ + name: "delegate", + description: "Delegates", + parameters: Type.Object({}), + execute: async (_args, api, callContext) => { + const child = await api.commit(async (tx) => { + const created = await tx.createConversation({ ownership: { kind: "task", taskId: api.taskId } }); + await configure(tx, created.id, { model: { provider: "faux", modelId: "faux-1" } }); + return created.id; + }, callContext); + const handle = (await api.conversation(child, callContext))!; + const first = await (await handle.submit({ type: "input", content: "work" }, callContext)).wait( + callContext, + ); + outcome.first = first.status; + const resumed = await handle.submit({ type: "continue" }, callContext); + await busy.promise; + outcome.busy = await handle.submit({ type: "continue" }, callContext).catch((error: unknown) => error); + release.resolve(); + const settled = await resumed.wait(callContext); + outcome.resumed = settled.status; + return { output: [{ type: "text", text: settled.status }] }; + }, + }), + ); + setup.faux.setResponses([ + fauxAssistantMessage([fauxToolCall("delegate", {}, { id: "c1" })], { stopReason: "toolUse" }), + failure, + async () => { + busy.resolve(); + await release.promise; + return fauxAssistantMessage("child recovered"); + }, + fauxAssistantMessage("done"), + ]); + const { harness, root } = await openChat(new MemoryStorage(), setup); + await (await root.submit({ type: "input", content: "go" }, context)).wait(context); + expect(outcome).toMatchObject({ first: "unanswered", resumed: "done" }); + expect(outcome.busy).toBeInstanceOf(ConversationBusy); + await harness.close(context); + }); + + // #11043 + it("starts a root trace linked to its caller, as an input does", async () => { + const setup = chatSetup(); + const telemetry = new InMemoryTelemetryContext(); + setup.faux.setResponses([failure, fauxAssistantMessage("recovered")]); + const harness = await Harness.open( + new MemoryStorage(), + { models: setup.models, registry: setup.registry, settings: setup.settings, telemetry }, + context, + ); + const root = await harness.root(context, { agent: { model: { provider: "faux", modelId: "faux-1" } } }); + await (await root.submit({ type: "input", content: "go" }, context)).wait(context); + const id = await telemetry.startSpan({ name: "caller" }, async (caller) => { + const callContext = withTelemetry(caller, context); + const resumed = await root.submit({ type: "continue" }, callContext); + await resumed.wait(callContext); + return resumed.id; + }); + await harness.waitForIdle(context); + await waitFor(() => telemetry.getSpans().every((span) => span.settled)); + const spans = telemetry.getSpans(); + const caller = spans.find((span) => span.name === "caller")!; + const submit = spans.find( + (span) => span.name === "pi.durable.submit" && span.attributes["submission.id"] === id, + )!; + expect(submit).toMatchObject({ parentId: null, attributes: { result: "placed" } }); + expect(submit.links).toEqual([{ context: caller.context, attributes: {} }]); + expect((await (await harness.submission(id, context))!.status(context)).trace).toEqual(submit.context); + expect( + spans.some( + (span) => + span.name === "pi.durable.phase" && + span.attributes.phase === "prepare" && + JSON.stringify(span.parentContext) === JSON.stringify(submit.context), + ), + ).toBe(true); + await harness.close(context); + }); + + // #11043 + it("places inputs a failed run left queued first, and they join its run", async () => { + const setup = chatSetup(); + const gate = deferred(); + const reached = deferred(); + setup.faux.setResponses([ + async () => { + reached.resolve(); + await gate.promise; + return failure(); + }, + fauxAssistantMessage("both handled"), + ]); + const { harness, root } = await openChat(new MemoryStorage(), setup); + await root.submit({ type: "input", content: "first" }, context); + await reached.promise; + const queued = await root.submit({ type: "input", content: "second", whenBusy: "followUp" }, context); + gate.resolve(); + await harness.waitForIdle(context); + expect((await queued.status(context)).status).toBe("queued"); + + const resumed = await root.submit({ type: "continue" }, context); + const live = await harness.snapshot(LiveDoc, root.id, context); + expect(live?.run?.inputs).toEqual([queued.id, resumed.id]); + const [byQueued, byContinue] = [await queued.wait(context), await resumed.wait(context)]; + expect(await answerText(root, byContinue)).toBe("both handled"); + expect(byQueued).toMatchObject({ status: "done", answer: (byContinue as { answer: number }).answer }); + await harness.close(context); + }); +}); diff --git a/packages/durable/test/harness-submissions.test.ts b/packages/durable/test/harness-submissions.test.ts index bae654985..08630183a 100644 --- a/packages/durable/test/harness-submissions.test.ts +++ b/packages/durable/test/harness-submissions.test.ts @@ -64,7 +64,7 @@ describe("submissions", () => { const submission = await root.submit({ type: "input", content: "hi" }, context); await busy.reached; const record = await submission.status(context); - if (record.status !== "placed") throw new Error(`Unexpected ${record.status}`); + if (record.type !== "input" || record.status !== "placed") throw new Error(`Unexpected ${record.status}`); const entry = await root.commit((tx) => tx.entry(UserEntry, record.entry), context); expect(entry?.model).toEqual([{ role: "user", content: "hi", timestamp: 42 }]); const live = await harness.snapshot(LiveDoc, root.id, context);