From 6a282511362bb61097a7d07e56f4864e7536c933 Mon Sep 17 00:00:00 2001 From: Mario Zechner Date: Wed, 28 Oct 2026 13:08:18 +0000 Subject: [PATCH] feat(durable): reject input on idle conversations with whenIdle An input with whenIdle: "reject" is refused with the new ConversationIdle when the conversation has no run, before any write, so it never starts a run. With whenBusy: "steer" it steers the current run or is refused. Request-ID deduplication runs first. --- packages/durable/CHANGELOG.md | 1 + packages/durable/README.md | 2 + packages/durable/docs/spec.md | 13 +- packages/durable/src/errors.ts | 11 ++ packages/durable/src/harness/submissions.ts | 12 +- packages/durable/src/harness/types.ts | 10 +- packages/durable/src/index.ts | 9 +- .../test/harness-submit-in-commit.test.ts | 122 ++++++++++++++++++ 8 files changed, 171 insertions(+), 9 deletions(-) create mode 100644 packages/durable/test/harness-submit-in-commit.test.ts diff --git a/packages/durable/CHANGELOG.md b/packages/durable/CHANGELOG.md index e63340b27..f5faab47b 100644 --- a/packages/durable/CHANGELOG.md +++ b/packages/durable/CHANGELOG.md @@ -109,6 +109,7 @@ - `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)) - Per-conversation turn limits: `AgentState.maxTurns` (`configure({ maxTurns })`, `null` clears; a `RangeError` for anything but an integer 1 or more) bounds the tool rounds one run takes. The round that reaches it runs its tools, then the run ends `unanswered` with the new `RunFailureReason` `turn_limit`, like a failed run, so `settings.afterFailedRun` decides the queued inputs and a continue resumes it on the round's results. The count is `pi.live.run.turns`; every run starts at 0, and an `onYield` continuation starts it over. Task-owned children and forks inherit the limit; `Agent.maxTurns` exposes it ([#11043](https://github.com/earendil-works/pi/issues/11043)) +- `whenIdle: "reject"` on input submissions: an idle conversation rejects the input with the new `ConversationIdle` and writes nothing, so it never starts a run. With `whenBusy: "steer"` it steers the current run or is refused, in one admission instead of a host's check and a separate submit that can race the run's end. A known request ID still returns its submission ([#11149](https://github.com/earendil-works/pi/issues/11149)) ### Changed diff --git a/packages/durable/README.md b/packages/durable/README.md index a60ee6b17..283ccafbb 100644 --- a/packages/durable/README.md +++ b/packages/durable/README.md @@ -534,11 +534,13 @@ A conversation is busy while a run is working on an input. Submitting to a busy await root.submit({ type: "input", content: "Also run the tests" }, context); // follow-up (default) await root.submit({ type: "input", content: "Use pnpm, not npm", whenBusy: "steer" }, context); await root.submit({ type: "input", content: "Only if idle", whenBusy: "reject" }, context); // throws ConversationBusy +await root.submit({ type: "input", content: "Only if busy", whenBusy: "steer", whenIdle: "reject" }, context); // throws ConversationIdle when idle await root.submit({ type: "write", entry: { kind: "app.note", data: "user opened a file" } }, context); ``` - **Steers** are placed after the current tool round and join the running work. - **Follow-ups** are placed when the run answers, and start the next run. +- **`whenIdle: "reject"`** refuses an input when no run is working, so it never starts a run of its own. With `whenBusy: "steer"`, it steers the current run or throws `ConversationIdle`. - **Writes** append an entry without asking the model anything. A write of only `kind` and `data` changes nothing the model sees, so it is appended at once, even mid-run, and is never queued: a chat message shows in the transcript as soon as it is sent. It can therefore appear before items submitted earlier that are still queued, except behind a queued reset, where it waits so the reset does not hide it. A write with `model`, `head`, or `edits` waits for the next boundary like a steer. - `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. diff --git a/packages/durable/docs/spec.md b/packages/durable/docs/spec.md index 284009b86..dbe7cb96a 100644 --- a/packages/durable/docs/spec.md +++ b/packages/durable/docs/spec.md @@ -367,6 +367,8 @@ type SubmissionDraft = { readonly type: "input"; readonly content: UserInput; readonly whenBusy?: "steer" | "followUp" | "reject"; + /** `reject`: an idle conversation rejects the input with `ConversationIdle` (section 6). */ + readonly whenIdle?: "reject"; readonly entry?: never; } | { @@ -374,6 +376,7 @@ type SubmissionDraft = { readonly entry: EntryDraft; readonly content?: never; readonly whenBusy?: never; + readonly whenIdle?: never; } /** Start a run on the context as it is, without a user message; idle only (section 6). */ | { @@ -381,6 +384,7 @@ type SubmissionDraft = { readonly content?: never; readonly entry?: never; readonly whenBusy?: never; + readonly whenIdle?: never; } ); @@ -2835,6 +2839,7 @@ Admission and terminal transitions: | 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` | +| input with `whenIdle: "reject"`, idle | none | reject with `ConversationIdle` | | 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 | @@ -2848,8 +2853,12 @@ Admission and terminal transitions: `requestId` deduplicates within one conversation before any write; reusing one 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. +writes no record and reports `ConversationBusy`. An idle input with +`whenIdle: "reject"` writes no record and reports `ConversationIdle`, also when +the inbox holds queued items, so the input never starts a run: with +`whenBusy: "steer"` it steers the current run or is refused. Idle means +`run === undefined`, the inverse of `busy` above. 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 diff --git a/packages/durable/src/errors.ts b/packages/durable/src/errors.ts index e2a36dfc1..796ae7e1f 100644 --- a/packages/durable/src/errors.ts +++ b/packages/durable/src/errors.ts @@ -66,3 +66,14 @@ export class ConversationBusy extends Error { this.conversationId = conversationId; } } + +/** An input with `whenIdle: "reject"` reached an idle conversation and was not admitted. */ +export class ConversationIdle extends Error { + readonly conversationId: ConversationId; + + constructor(conversationId: ConversationId) { + super(`Conversation ${conversationId} is idle`); + this.name = "ConversationIdle"; + this.conversationId = conversationId; + } +} diff --git a/packages/durable/src/harness/submissions.ts b/packages/durable/src/harness/submissions.ts index 7de0054b8..76379d1b2 100644 --- a/packages/durable/src/harness/submissions.ts +++ b/packages/durable/src/harness/submissions.ts @@ -6,7 +6,7 @@ import { type TelemetryContext, type TelemetrySpan, } from "@earendil-works/pi-telemetry"; -import { ConversationBusy } from "../errors.ts"; +import { ConversationBusy, ConversationIdle } from "../errors.ts"; import type { ObservedCommit, SessionImpl } from "../session/session.ts"; import type { ConversationId, JsonObject, Storage, SubmissionId, SubmissionRecord, Tx } from "../types.ts"; import { startRun } from "./generation.ts"; @@ -188,10 +188,11 @@ function checkInputEntry(entry: InputEntry): void { * Admit a submission inside a commit (spec ยง6); `Conversation.submit()` and conversation-owned compactions share it. A * 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 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 + * 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 and any continue with `ConversationBusy`; an idle + * one rejects `whenIdle: "reject"` input with `ConversationIdle`. 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. */ @@ -235,6 +236,7 @@ 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 (!busy && draft.type === "input" && draft.whenIdle === "reject") throw new ConversationIdle(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); diff --git a/packages/durable/src/harness/types.ts b/packages/durable/src/harness/types.ts index 742edb085..8ec9b6c8b 100644 --- a/packages/durable/src/harness/types.ts +++ b/packages/durable/src/harness/types.ts @@ -75,6 +75,11 @@ export type SubmissionDraft = { readonly type: "input"; readonly content: UserInput; readonly whenBusy?: "steer" | "followUp" | "reject"; + /** + * `reject`: an idle conversation (no run) rejects the input with `ConversationIdle` and writes nothing, so the input + * never starts a run. With `whenBusy: "steer"` it steers the current run or is refused. + */ + readonly whenIdle?: "reject"; /** Place the input as this entry; default `pi.user` without data. The kind must not start with `pi.`. */ readonly entry?: InputEntry; } @@ -83,6 +88,7 @@ export type SubmissionDraft = { readonly entry: EntryDraft; readonly content?: never; readonly whenBusy?: never; + readonly whenIdle?: never; } | { /** @@ -96,6 +102,7 @@ export type SubmissionDraft = { readonly content?: never; readonly entry?: never; readonly whenBusy?: never; + readonly whenIdle?: never; } ); @@ -788,7 +795,8 @@ export interface Conversation { /** * 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`) + * `ConversationBusy` instead and writes nothing. An idle conversation rejects `whenIdle: "reject"` input with + * `ConversationIdle` 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. */ diff --git a/packages/durable/src/index.ts b/packages/durable/src/index.ts index be0ea2317..a08c8ceb5 100644 --- a/packages/durable/src/index.ts +++ b/packages/durable/src/index.ts @@ -9,7 +9,14 @@ export { ToolResultEntry, UserEntry, } from "./entries.ts"; -export { ConversationBusy, ReadAfterWrite, ReadOnlySession, SessionFailed, StorageRequestError } from "./errors.ts"; +export { + ConversationBusy, + ConversationIdle, + ReadAfterWrite, + ReadOnlySession, + SessionFailed, + StorageRequestError, +} from "./errors.ts"; export { AgentDoc, configure, diff --git a/packages/durable/test/harness-submit-in-commit.test.ts b/packages/durable/test/harness-submit-in-commit.test.ts new file mode 100644 index 000000000..8cd6997da --- /dev/null +++ b/packages/durable/test/harness-submit-in-commit.test.ts @@ -0,0 +1,122 @@ +// Admission for hosts that must not start runs: `whenIdle: "reject"` (#11149). +import { type AssistantMessage, fauxAssistantMessage, fauxText, fauxToolCall, Type } from "@earendil-works/pi-ai"; +import { type Conversation, ConversationIdle, defineTool, InboxDoc, LiveDoc } from "@earendil-works/pi-durable"; +import { describe, expect, it } from "vitest"; +import { chatSetup, openChat, unanswered } from "./chat-support.ts"; +import { addTool } from "./harness-support.ts"; +import { ControlledStorage, context } from "./session-support.ts"; +import { deferred } from "./task-support.ts"; + +const MODEL = { provider: "faux", modelId: "faux-1" } as const; + +const failure = (): AssistantMessage => + fauxAssistantMessage([], { stopReason: "error", errorMessage: "Invalid request" }); + +/** Open a chat whose run hangs until aborted, and wait until it works on its first input. */ +async function openBusy(storage = new ControlledStorage()) { + const setup = chatSetup(); + const busy = unanswered(); + setup.faux.setResponses([busy.step]); + const opened = await openChat(storage, setup); + const first = await opened.root.submit({ type: "input", content: "work" }, context); + await busy.reached; + return { ...opened, storage, first }; +} + +describe("whenIdle reject", () => { + // #11149 + it("rejects input to an idle conversation without writing", async () => { + const storage = new ControlledStorage(); + const { harness, root } = await openChat(storage, chatSetup()); + const commits = storage.commits.length; + const rejected = root.submit({ type: "input", content: "hi", whenBusy: "steer", whenIdle: "reject" }, context); + await expect(rejected).rejects.toBeInstanceOf(ConversationIdle); + await expect(rejected).rejects.toMatchObject({ conversationId: root.id }); + expect(storage.commits.length).toBe(commits); + expect((await harness.snapshot(LiveDoc, root.id, context))?.run).toBeUndefined(); + await harness.close(context); + }); + + // #11149 + it("rejects input to an idle conversation that keeps queued items after a failed run", async () => { + const setup = chatSetup(); + const failing = deferred(); + const requested = deferred(); + setup.faux.setResponses([ + async () => { + requested.resolve(); + await failing.promise; + return failure(); + }, + ]); + const { harness, root } = await openChat(new ControlledStorage(), setup); + const first = await root.submit({ type: "input", content: "first" }, context); + await requested.promise; + const queued = await root.submit({ type: "input", content: "queued" }, context); + failing.resolve(); + expect(await first.wait(context)).toMatchObject({ status: "unanswered", reason: "model_error" }); + await harness.waitForIdle(context); + const rejected = root.submit({ type: "input", content: "steer", whenBusy: "steer", whenIdle: "reject" }, context); + await expect(rejected).rejects.toBeInstanceOf(ConversationIdle); + expect(await queued.status(context)).toMatchObject({ status: "queued" }); + expect((await harness.snapshot(LiveDoc, root.id, context))?.run).toBeUndefined(); + await harness.close(context); + }); + + // #11149 + it("steers a working run", async () => { + const { harness, root, first } = await openBusy(); + const steer = await root.submit( + { type: "input", content: "also this", whenBusy: "steer", whenIdle: "reject" }, + context, + ); + expect(await steer.status(context)).toMatchObject({ type: "input", status: "queued" }); + const inbox = await harness.snapshot(InboxDoc, root.id, context); + expect(inbox?.items).toEqual([{ id: steer.id, mode: "steer", content: "also this" }]); + expect((await harness.snapshot(LiveDoc, root.id, context))?.run?.inputs).toEqual([first.id]); + await harness.close(context); + }); + + // #11149 + it("returns the submission of a known request ID, also once idle", async () => { + const setup = chatSetup(); + setup.faux.setResponses([fauxAssistantMessage([fauxText("done")])]); + const { harness, root } = await openChat(new ControlledStorage(), setup); + const first = await root.submit({ type: "input", content: "work", requestId: "r1" }, context); + await first.wait(context); + const draft = { type: "input", content: "work", requestId: "r1", whenBusy: "steer", whenIdle: "reject" } as const; + expect((await root.submit(draft, context)).id).toBe(first.id); + await harness.close(context); + }); + + // #11149 + it("rejects through a tool's conversation handle", async () => { + const setup = chatSetup(); + let other: Conversation | undefined; + let rejection: unknown; + addTool( + setup.registry, + defineTool({ + name: "nudge", + description: "Steers another conversation", + parameters: Type.Object({}), + execute: async (_args, api, callContext) => { + const handle = (await api.conversation(other!.id, callContext))!; + const draft = { type: "input", content: "nudge", whenBusy: "steer", whenIdle: "reject" } as const; + rejection = await handle.submit(draft, callContext).catch((error: unknown) => error); + return { output: [{ type: "text", text: "nudged" }] }; + }, + }), + ); + setup.faux.setResponses([ + fauxAssistantMessage([fauxToolCall("nudge", {}, { id: "c1" })], { stopReason: "toolUse" }), + fauxAssistantMessage([fauxText("ok")]), + ]); + const { harness, root } = await openChat(new ControlledStorage(), setup); + other = await harness.createConversation({ ownership: { kind: "ownerless" }, agent: { model: MODEL } }, context); + await (await root.submit({ type: "input", content: "go" }, context)).wait(context); + expect(rejection).toBeInstanceOf(ConversationIdle); + expect((await harness.snapshot(LiveDoc, other.id, context))?.run).toBeUndefined(); + await harness.close(context); + }); +});