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.
This commit is contained in:
Mario Zechner 2026-10-28 13:08:18 +00:00
parent c3d8ce44d3
commit 6a28251136
8 changed files with 171 additions and 9 deletions

View file

@ -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

View file

@ -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.

View file

@ -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

View file

@ -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;
}
}

View file

@ -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);

View file

@ -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.
*/

View file

@ -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,

View file

@ -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);
});
});