From 1c6928b3099a3c3f7d290dd07beda121df22e2a7 Mon Sep 17 00:00:00 2001 From: Mario Zechner Date: Wed, 28 Oct 2026 11:45:19 +0000 Subject: [PATCH] feat(durable): calls to tools not offered run as tool tasks --- packages/durable/CHANGELOG.md | 1 + packages/durable/docs/spec.md | 48 +++-- packages/durable/src/harness/generation.ts | 64 ++++--- packages/durable/src/harness/live.ts | 7 +- packages/durable/src/harness/tool.ts | 40 +++- .../durable/test/harness-generation.test.ts | 2 +- .../durable/test/harness-live-deltas.test.ts | 7 +- .../durable/test/harness-nested-tools.test.ts | 60 +++++- .../durable/test/harness-tool-hooks.test.ts | 178 ++++++++++++++++++ packages/durable/test/harness-tools.test.ts | 30 ++- .../durable/test/harness-view-run.test.ts | 15 +- 11 files changed, 370 insertions(+), 82 deletions(-) diff --git a/packages/durable/CHANGELOG.md b/packages/durable/CHANGELOG.md index 441b8644c..3ccc3662e 100644 --- a/packages/durable/CHANGELOG.md +++ b/packages/durable/CHANGELOG.md @@ -93,6 +93,7 @@ - `SessionEnd` (`Session.closed`, `Harness.closed`) is tagged by `kind`: `{ kind: "closed" }` or `{ kind: "failed", error }` (was `reason`). Before: `harness.closed.then((end) => { if (end.reason === "failed") reopen(end.error); })`. After: `harness.closed.then((end) => { if (end.kind === "failed") reopen(end.error); })`. - `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)) ### Added diff --git a/packages/durable/docs/spec.md b/packages/durable/docs/spec.md index 7fe639106..35330bce3 100644 --- a/packages/durable/docs/spec.md +++ b/packages/durable/docs/spec.md @@ -3140,7 +3140,7 @@ await conversation.configure({ extensions: { add: [Venv] } }, context); |---|---|---| | generation `prepare` | agent (tools, sections, model, thinking level), stream options, compaction thresholds for that model | once per request; model, stream options, and offered tools are fixed in the request checkpoint (section 8.3) | | every task phase | extension selection for hooks; `runtime.agent()` | at most once per phase handler, at its first hook dispatch or `agent()` call, from that phase's snapshot and the committed `pi.agent`, off the Session line, with the invocation's context; fixed for the rest of the phase | -| tool task | the implementation of an accepted call | at `call` and `execute`: the name among the phase agent's `tools`, the current selection after the tools filter; no such tool, for example one removed after preparation: `tool_unavailable`. Replay policy on reopen uses the same lookup (section 7.3) | +| tool task | the implementation of an accepted call | at `call` and `execute`: the name among the phase agent's `tools`, the current selection after the tools filter; no such tool, for example one removed after preparation, or a call flagged `unoffered` (section 8.3): `tool_unavailable`. Replay policy on reopen uses the same lookup (section 7.3) | | generation, when a tool round starts | tool execution mode | once per round (section 8.3) | | generation, compaction | retry policy | at each attempt's classification | | generation, on a context overflow | compaction policy for the request's model | at classification | @@ -3703,7 +3703,10 @@ A tool call is accepted only if its request offered the tool. Generation checks this against the tool set replayed from the committed model context through the request's `cutoff`; `beforeRequest` replacements do not change it. It answers a call to a tool it did not offer with the `tool_unavailable` result described in -section 2.2, without a tool task. An agent change after preparation does not +section 2.2, through a tool task whose input is flagged `unoffered`: the task +resolves nothing, so even a tool the agent has (a deferred one not loaded yet) +never runs, and `afterTool` and `control` apply to the result as to any call's. +An agent change after preparation does not affect this check (section 2.2). The tool task resolves the called name among the `tools` of the phase's agent resolution, the tools filter applied. It uses that implementation @@ -4256,7 +4259,7 @@ is exported as an `Entry` token. | `pi.user` | `[UserMessage]`, timestamp from the Harness clock at admission or placement | submissions, `onYield` continuations | | `pi.assistant` | `[AssistantMessage]` with any stop reason | generation | | `pi.system` | `[SystemMessage]` with `content: ""` (section 7.4) | generation preparation | -| `pi.tool-result` | `[ToolResultMessage]` | tool tasks; generation for calls to tools its request did not offer | +| `pi.tool-result` | `[ToolResultMessage]` | tool tasks; generation for calls an aborted round never started | | `pi.reset` | absent, or `[UserMessage]` with the handoff text; always `head: "self"` | `reset()`, generation `tools` phase for `handoff` | | `pi.compaction` | `[UserMessage]` with the wrapped summary; `head` is the first kept entry | compaction tasks (section 8.7) | @@ -4308,11 +4311,10 @@ type LiveState = { tools?: { callId: string; name: string; - /** - * Absent for a call not started yet (sequential round, `pending`) and for a call its request did not offer, - * which starts `done` with the `entry` generation wrote. - */ + /** Absent for a call not started yet (sequential round, `pending`). */ taskId?: TaskId; + /** A call to a tool its request did not offer: its task settles `tool_unavailable` without running anything. */ + unoffered?: true; status: "pending" | "running" | "done"; /** Retained running output and what the bounds dropped. */ output?: string; @@ -4396,8 +4398,8 @@ its 64 KiB scan, the commit writes the retained window as one set. Either way each commit writes at most one window, and the throttle bounds the rate. Generation creates `tools` with one `pending` slot per call in the commit that -appends the tool-calling answer; a call its request did not offer starts `done` -with its result entry. A tool task sets its slot `running` in its intent commit, +appends the tool-calling answer; a call its request did not offer is flagged +`unoffered`. A tool task sets its slot `running` in its intent commit, publishes throttled output and details into it with deferred commits (section 4.1), and in its terminal commit sets `done` and `entry` and removes `output`, `droppedBytes`, `droppedLines`, @@ -4506,7 +4508,11 @@ type GenerationCheckpoint = type GenerationResult = { entryId: EntryId }; ``` -`pi.generation` is version 1 and starts at `{ phase: "prepare", attempt: 1 }`. +`pi.generation` is version 2 and starts at `{ phase: "prepare", attempt: 1 }`; +`migrate` keeps a version 1 record. Version 2 keeps calls to tools not offered +in a sequential round's `pending`, flagged `unoffered` in their slots (section +8.5), which a version 1 Harness would start as tool tasks without the flag, so +it leaves version 2 blocked (`task_too_old`). The run's inputs live in `pi.live.run`, not in the task input. - `prepare` runs section 7.4 with the phase's agent resolution and resolves @@ -4630,11 +4636,12 @@ A tool round starts in the commit that appends the tool-calling answer: 1. The offered tools are replayed with pi-ai `getCurrentTools()` from the committed model context through `cutoff`: `request` already holds it, and - `poll` derives it again. A call to a tool not offered gets its - `tool_unavailable` result entry here, without a task. -2. Every other call gets a `pi.tool` task owned by the generation, with input - `{ assistant, callId }`. The round is sequential when the settings' - `toolExecution`, read as the round starts, is `sequential` or any called + `poll` derives it again. +2. Every call gets a `pi.tool` task owned by the generation, in call order, with + input `{ assistant, callId }`, plus `unoffered: true` (also on its slot) for a + call to a tool not offered, which settles `tool_unavailable` without resolving + the name (section 8.4). The round is sequential when the settings' + `toolExecution`, read as the round starts, is `sequential` or any offered called tool, resolved from the phase's agent as the tool task resolves it (section 7.3), has `executionMode: "sequential"`: then only the first call gets its task now and the rest wait in `pending`. Otherwise every call gets its task at @@ -4664,9 +4671,11 @@ type ToolTaskCheckpoint = type ToolTaskResult = { kind: "model"; entryId: EntryId; control?: ToolControl } | { kind: "nested" }; ``` -`pi.tool` is version 2 and starts at `{ phase: "call" }`; `migrate` gives a -version 1 input `kind: "model"`, and readers treat a terminal result without -`kind` as a model-issued call's. A model-issued call's input stays small +`pi.tool` is version 3 and starts at `{ phase: "call" }`; `migrate` gives a +version 1 input `kind: "model"` and keeps a version 2 input (version 3 added +`unoffered`), and readers treat a terminal result without `kind` as a +model-issued call's. A call flagged `unoffered` resolves no tool: `call` settles +it `tool_unavailable` and the abort handler does too. A model-issued call's input stays small because the terminal record keeps it; the call is read from the assistant entry. A nested call (section 7.3) carries its call, since no entry holds it, and settles differently: its result commit records the result's `usage` @@ -4767,7 +4776,8 @@ and then commits once: It completes with `{ entryId }` of the tool-calling answer. Its abort handler runs only after its tool tasks are terminal (section 5.5). For every call still -in `pending`, which never started, it appends an `aborted` error result; a +in `pending`, which never started, it appends an `aborted` error result +(`tool_unavailable` for an `unoffered` one), without `afterTool`; a started call whose task faulted or was orphaned keeps its `done` slot without an entry (section 8.2). It then settles the inputs `unanswered` with `aborted`, removes `run`, `generation`, and `tools`, and ends diff --git a/packages/durable/src/harness/generation.ts b/packages/durable/src/harness/generation.ts index 462f73e23..56d130082 100644 --- a/packages/durable/src/harness/generation.ts +++ b/packages/durable/src/harness/generation.ts @@ -143,8 +143,15 @@ const DEFAULT_POLL_AFTER_MS = 5000; */ export const GenerationTask = defineTask({ name: "pi.generation", - version: 1, + version: 2, initial: () => ({ phase: "prepare", attempt: 1 }), + // Version 2 leaves calls to tools not offered in a sequential round's `pending`, flagged `unoffered` in their slots. + // A version 1 Harness would start their tool tasks unflagged, resolving the name, so it leaves version 2 blocked + // instead. A version 1 checkpoint has no such calls and is kept as it is. + migrate: (input, checkpoint) => ({ + input: input as GenerationInput, + checkpoint: checkpoint as GenerationCheckpoint, + }), phases: { /** * Render the system prompt and tool loadout and append the positional `pi.system` entries they need, then move to @@ -307,8 +314,8 @@ export const GenerationTask = defineTask => { const live = await tx.doc(LiveDoc, runtime.conversationId); - const taskId = await createToolTask(tx, runtime, assistant, next); const slot = live.tools?.find((slot) => slot.callId === next && slot.taskId === undefined); + const taskId = await createToolTask(tx, runtime, assistant, next, slot?.unoffered); if (slot !== undefined) slot.taskId = taskId; const checkpoint: GenerationCheckpoint = { phase: "tools", @@ -325,7 +332,8 @@ export const GenerationTask = defineTask slot.callId === call.id)?.unoffered === true; + const result = unoffered + ? harnessError("tool_unavailable", `Tool ${call.name} is not available`) + : harnessError("aborted", `Tool ${call.name} was aborted`); await appendToolResult(tx, conversationId, call, result, { timestamp: runtime.now() }); } endRun(tx, live, runtime.taskId, { status: "unanswered", reason: "aborted" }); @@ -355,11 +367,17 @@ async function readCalls( return callIds.flatMap((id) => calls.filter((call) => call.id === id).slice(0, 1)); } -/** A tool task for call `callId`, owned by the generation. */ -function createToolTask(tx: Tx, runtime: Runtime, assistant: EntryId, callId: string): Promise> { +/** A tool task for call `callId`, owned by the generation; `unoffered` for a call to a tool its request did not offer. */ +function createToolTask( + tx: Tx, + runtime: Runtime, + assistant: EntryId, + callId: string, + unoffered: true | undefined, +): Promise> { return tx.createTask( ToolTask, - { kind: "model", assistant, callId }, + { kind: "model", assistant, callId, ...(unoffered === undefined ? {} : { unoffered }) }, { ownership: { kind: "task", taskId: runtime.taskId } }, ); } @@ -776,9 +794,10 @@ async function answer(runtime: Runtime, message: AssistantMessage, context: Cont } /** - * Append the tool-calling answer and start its tool round in one commit (spec §8.3). A call to a tool the request did - * not offer gets its `tool_unavailable` result here; every other call gets a tool task owned by the generation, only the - * first one now when the round is sequential. The generation then waits for them in its `tools` phase, keeping the run. + * Append the tool-calling answer and start its tool round in one commit (spec §8.3). Every call gets a tool task owned by + * the generation, in call order, only the first one now when the round is sequential; a call to a tool the request did + * not offer is flagged, and its task settles `tool_unavailable`. The generation then waits for them in its `tools` + * phase, keeping the run. */ async function startToolRound( runtime: Runtime, @@ -806,20 +825,15 @@ async function startToolRound( const tools: TaskId[] = []; const pending: string[] = []; for (const call of calls) { - if (!offered.has(call.name)) { - const unavailable = harnessError("tool_unavailable", `Tool ${call.name} is not available`); - const result = await appendToolResult(tx, conversationId, call, unavailable, { timestamp: runtime.now() }); - slots.push({ callId: call.id, name: call.name, status: "done", entry: result.id }); - continue; - } + const unoffered = offered.has(call.name) ? {} : ({ unoffered: true } as const); if (sequential && tools.length > 0) { pending.push(call.id); - slots.push({ callId: call.id, name: call.name, status: "pending" }); + slots.push({ callId: call.id, name: call.name, status: "pending", ...unoffered }); continue; } - const taskId = await createToolTask(tx, runtime, entry.id, call.id); + const taskId = await createToolTask(tx, runtime, entry.id, call.id, unoffered.unoffered); tools.push(taskId); - slots.push({ callId: call.id, name: call.name, taskId, status: "pending" }); + slots.push({ callId: call.id, name: call.name, taskId, status: "pending", ...unoffered }); } delete live.generation; live.tools = slots; @@ -853,13 +867,13 @@ async function finishToolRound( const slots = (await runtime.snapshot(LiveDoc, conversationId, context))?.tools ?? []; const results = slots.flatMap((slot) => (slot.entry === undefined ? [] : [slot.entry])); await runtime.hooks.each("afterTools", (hook) => hook(assistant, results, runtime, context)); - // Every call of the round, including those answered without a task, must ask to terminate. - const terminate = - slots.length > 0 && - slots.every((slot) => slot.taskId !== undefined && controls.get(slot.taskId)?.terminate === true); - const added = [...controls.values()].flatMap((control) => control?.addTools ?? []); + // In call order, the slots' order; a slot without a task (faulted or orphaned before it got one) has no control. + const ordered = slots.map((slot) => (slot.taskId === undefined ? undefined : controls.get(slot.taskId))); + // Every call of the round must ask to terminate. + const terminate = slots.length > 0 && ordered.every((control) => control?.terminate === true); + const added = ordered.flatMap((control) => control?.addTools ?? []); // The last handoff in call order wins. - const handoff = [...controls.values()].findLast((control) => control?.handoff !== undefined)?.handoff; + const handoff = ordered.findLast((control) => control?.handoff !== undefined)?.handoff; await runtime.commit(async (tx): Promise => { const boundary = await prepareBoundary(tx, conversationId, runtime.settings); if (added.length > 0) await addTools(tx, conversationId, added); diff --git a/packages/durable/src/harness/live.ts b/packages/durable/src/harness/live.ts index 837225a04..a6669e3cb 100644 --- a/packages/durable/src/harness/live.ts +++ b/packages/durable/src/harness/live.ts @@ -33,11 +33,10 @@ type SlotProgress = { /** Presentation of one tool call of the current round. */ export type ToolSlot = SlotProgress & { - /** - * Absent for a call not started yet (sequential round) and for a call its request did not offer, which starts `done` - * with the `entry` generation wrote. - */ + /** Absent for a call not started yet (sequential round). */ taskId?: TaskId; + /** A call to a tool its request did not offer: its task settles `tool_unavailable` without running anything. */ + unoffered?: true; /** Result entry once done; absent when the tool task faulted or was orphaned. */ entry?: EntryId; }; diff --git a/packages/durable/src/harness/tool.ts b/packages/durable/src/harness/tool.ts index 02179a817..bdc806aa9 100644 --- a/packages/durable/src/harness/tool.ts +++ b/packages/durable/src/harness/tool.ts @@ -64,7 +64,13 @@ import { optionalExtension } from "./util.ts"; * whose call the input carries because no entry holds it. */ export type ToolTaskInput = - | { kind: "model"; assistant: EntryId; callId: string } + | { + kind: "model"; + assistant: EntryId; + callId: string; + /** A call to a tool its request did not offer: settles `tool_unavailable` without resolving the name. */ + unoffered?: true; + } | { kind: "nested"; parent: TaskId; @@ -128,18 +134,21 @@ type Content = (TextContent | ImageContent)[]; */ export const ToolTask = defineTask({ name: "pi.tool", - version: 2, + version: 3, initial: () => ({ phase: "call" }), - // Version 1 had only model-issued calls. - migrate: (input, checkpoint) => ({ - input: { kind: "model", ...(input as { assistant: EntryId; callId: string }) }, + // Version 1 had only model-issued calls, without `kind`; version 3 added `unoffered`, absent in version 2 inputs. + migrate: (input, checkpoint, fromVersion) => ({ + input: + fromVersion === 1 + ? { kind: "model", ...(input as { assistant: EntryId; callId: string }) } + : (input as ToolTaskInput), checkpoint: checkpoint as ToolTaskCheckpoint, }), phases: { call: async (task, runtime, context) => { const call = await readCall(runtime, task.input, context); if (cutOffTooOften(task, runtime)) return endCutOff(runtime, task, call, context); - const tool = await resolveTool(runtime, task.input, call.name, context); + const tool = isUnoffered(task.input) ? undefined : await resolveTool(runtime, task.input, call.name, context); if (tool === undefined) { const error = harnessError("tool_unavailable", `Tool ${call.name} is not available`); return reject(runtime, task.input, call, undefined, error, context); @@ -205,9 +214,11 @@ export const ToolTask = defineTask { const call = await readCall(runtime, task.input, context); - // Abandoned after a restart (`TaskOptions.abandonOnRestart`): say whether the call may have run. - const [code, message] = - task.abortReason !== "restart" + // Abandoned after a restart (`TaskOptions.abandonOnRestart`): say whether the call may have run. An unoffered call + // could not run in any case, and its result says so. + const [code, message] = isUnoffered(task.input) + ? ["tool_unavailable", `Tool ${call.name} is not available`] + : task.abortReason !== "restart" ? ["aborted", `Tool ${call.name} was aborted`] : task.state.checkpoint.phase === "execute" ? ["interrupted", `Tool ${call.name} was interrupted by a restart and may have partially run`] @@ -288,16 +299,25 @@ async function settleFromSlot( await settle(runtime, input, call, ending, settled, context); } -/** The tool whose output limits bound a hooks' result: the called tool, or none when it, or the agent, fails to resolve. */ +/** + * The tool whose output limits bound a hooks' result: the called tool, or none when it, or the agent, fails to resolve, + * and for an unoffered call, which never resolves its name. + */ async function toolForLimits( runtime: Runtime, input: ToolTaskInput, name: string, context: Context, ): Promise { + if (isUnoffered(input)) return undefined; return resolveTool(runtime, input, name, context).catch(() => undefined); } +/** A model-issued call to a tool its request did not offer (spec §8.3). */ +function isUnoffered(input: ToolTaskInput): boolean { + return input.kind === "model" && input.unoffered === true; +} + /** The called tool: among the tools the model is offered for a model-issued call, else among those tools may call. */ async function resolveTool( runtime: Runtime, diff --git a/packages/durable/test/harness-generation.test.ts b/packages/durable/test/harness-generation.test.ts index fa6b0d21e..4fd7697f0 100644 --- a/packages/durable/test/harness-generation.test.ts +++ b/packages/durable/test/harness-generation.test.ts @@ -713,7 +713,7 @@ describe("generation", () => { entry: entry.id, }); const taskId = await tx.createTask( - { definition: { ...GenerationTask.definition, version: 2 } }, + { definition: { ...GenerationTask.definition, version: GenerationTask.definition.version + 1 } }, {}, { ownership: { kind: "conversation" }, conversationId: root.id }, ); diff --git a/packages/durable/test/harness-live-deltas.test.ts b/packages/durable/test/harness-live-deltas.test.ts index 0802a324e..737b577a8 100644 --- a/packages/durable/test/harness-live-deltas.test.ts +++ b/packages/durable/test/harness-live-deltas.test.ts @@ -347,7 +347,7 @@ describe("pi.live deltas", () => { await harness.close(context); }); - it("starts calls the request did not offer as done and marks a faulted tool's slot done without an entry", async () => { + it("starts calls the request did not offer as flagged tasks and marks a faulted tool's slot done without an entry", async () => { const setup = chatSetup(); addTool( setup.registry, @@ -376,14 +376,13 @@ describe("pi.live deltas", () => { end: () => {}, }); await (await root.submit({ type: "input", content: "go" }, context)).wait(context); - const entries = await root.entries({}, 10, undefined, context); - const ghostResult = entries.items.find((entry) => entry.kind === "pi.tool-result")!.id; const handover = commits.find((ops) => ops.some((op) => op[0] === "s" && op[1][0] === "tools"))!; + // #11043: an unoffered call gets a task too, flagged on its slot. expect(handover).toContainEqual([ "s", ["tools"], [ - { callId: "g", name: "ghost", status: "done", entry: ghostResult }, + { callId: "g", name: "ghost", taskId: expect.any(Number), status: "pending", unoffered: true }, { callId: "b", name: "bad", taskId: expect.any(Number), status: "pending" }, ], ]); diff --git a/packages/durable/test/harness-nested-tools.test.ts b/packages/durable/test/harness-nested-tools.test.ts index de61c3ec4..e1442b25f 100644 --- a/packages/durable/test/harness-nested-tools.test.ts +++ b/packages/durable/test/harness-nested-tools.test.ts @@ -1233,14 +1233,34 @@ describe("nested tool calls", () => { checkpoint: { phase: "call" }, }); }); + + // #11043 + it("keeps version 2 tool task input as it is", () => { + const migrate = ToolTask.definition.migrate!; + const model = { kind: "model", assistant: 3, callId: "c1" }; + const nested = { + kind: "nested", + parent: 7, + parentCallId: "c1", + key: "k", + call: { type: "toolCall", id: "c1/k", name: "echo", arguments: {} }, + }; + expect(migrate(model, { phase: "call" }, 2).input).toEqual(model); + expect(migrate(nested, { phase: "call" }, 2).input).toEqual(nested); + }); }); -describe("tool task version 1 records", () => { +describe("tool task version 1 and 2 records", () => { /** - * Store a live tool task for call `c1` of an assistant entry, close, and rewrite its record as version 1 stored it: - * `{ assistant, callId }` input at version 1, with `checkpoint`. Returns the storage path and the task ID. + * Store a live tool task for call `c1` of an assistant entry, close, and rewrite its record as `version` stored it, + * with `checkpoint`: version 1 had `{ assistant, callId }` input, version 2 added `kind`. Returns the storage path + * and the task ID. */ - async function storeVersion1(setup: ChatSetup, checkpoint: JsonValue): Promise<{ path: string; id: TaskId }> { + async function storeVersion( + setup: ChatSetup, + checkpoint: JsonValue, + version: 1 | 2 = 1, + ): Promise<{ path: string; id: TaskId }> { const directory = await mkdtemp(join(tmpdir(), "pi-durable-tool-v1-")); directories.add(directory); const path = join(directory, "session.sqlite"); @@ -1257,10 +1277,11 @@ describe("tool task version 1 records", () => { await harness.close(context); const storage = await openNodeSqliteStorage(path); const record = (await storage.task(id, context))!; - const { kind: _kind, ...input } = record.input as Extract; + const { kind: _kind, ...version1 } = record.input as Extract; + const input = version === 1 ? version1 : record.input; const state = { ...record.state, checkpoint } as typeof record.state; await storage.commit( - [{ type: "task", value: { ...record, version: 1, input, state } as typeof record }], + [{ type: "task", value: { ...record, version, input, state } as typeof record }], () => {}, context, ); @@ -1272,12 +1293,12 @@ describe("tool task version 1 records", () => { const setup = chatSetup(); const echo = echoTool(); addTool(setup.registry, echo.registration); - const { path, id } = await storeVersion1(setup, { phase: "call" }); + const { path, id } = await storeVersion(setup, { phase: "call" }); const { harness, root } = await openChat(await openNodeSqliteStorage(path), setup); expect((await harness.getTask(id, context))?.version).toBe(1); harness.resume(); const settled = await harness.waitForTask(id, context); - expect(settled.version).toBe(2); + expect(settled.version).toBe(3); expect(settled.input).toMatchObject({ kind: "model", callId: "c1" }); expect(settled.state.outcome).toMatchObject({ status: "completed", result: { kind: "model" } }); expect(echo.state.runs).toBe(1); @@ -1291,11 +1312,11 @@ describe("tool task version 1 records", () => { const echo = echoTool(); addTool(setup.registry, echo.registration); const intent = { phase: "execute", arguments: { text: "old" }, replay: "unsafe" }; - const { path, id } = await storeVersion1(setup, intent); + const { path, id } = await storeVersion(setup, intent); const { harness, root } = await openChat(await openNodeSqliteStorage(path), setup); harness.resume(); const settled = await harness.waitForTask(id, context); - expect(settled.version).toBe(2); + expect(settled.version).toBe(3); expect(settled.state.outcome).toMatchObject({ status: "failed", result: { kind: "model" } }); expect(echo.state.runs).toBe(0); const [result] = results(await allEntries(root)); @@ -1303,4 +1324,23 @@ describe("tool task version 1 records", () => { expect(text(result)).toContain("Tool echo was interrupted"); await harness.close(context); }); + + // #11043 + it("reserves a stored version 2 task under version 3 and runs it as a model-issued call", async () => { + const setup = chatSetup(); + const echo = echoTool(); + addTool(setup.registry, echo.registration); + const { path, id } = await storeVersion(setup, { phase: "call" }, 2); + const { harness, root } = await openChat(await openNodeSqliteStorage(path), setup); + expect((await harness.getTask(id, context))?.version).toBe(2); + harness.resume(); + const settled = await harness.waitForTask(id, context); + expect(settled.version).toBe(3); + expect(settled.input).toEqual({ kind: "model", assistant: expect.any(Number), callId: "c1" }); + expect(settled.state.outcome).toMatchObject({ status: "completed", result: { kind: "model" } }); + expect(echo.state.runs).toBe(1); + const [result] = results(await allEntries(root)); + expect([result!.toolCallId, text(result)]).toEqual(["c1", "echo old"]); + await harness.close(context); + }); }); diff --git a/packages/durable/test/harness-tool-hooks.test.ts b/packages/durable/test/harness-tool-hooks.test.ts index aeecb2e1f..1847c768b 100644 --- a/packages/durable/test/harness-tool-hooks.test.ts +++ b/packages/durable/test/harness-tool-hooks.test.ts @@ -19,6 +19,7 @@ import { LiveDoc, MemoryStorage, type NestedToolExecutionResult, + ResetEntry, ROOT_CONVERSATION_ID, type Storage, type TaskId, @@ -624,3 +625,180 @@ describe("afterTool on results rebuilt from the slot", () => { await opened.harness.close(context); }); }); + +describe("calls to tools the request did not offer", () => { + // #11043 + it("run afterTool inside their own tool task", async () => { + const setup = chatSetup(); + const seen: { readonly id: string; readonly taskId: TaskId }[] = []; + addHooks(setup.registry, ToolTask, { + afterTool: (call, result, api) => { + seen.push({ id: call.id, taskId: api.taskId }); + return { ...result, output: [{ type: "text", text: "try another tool" }] }; + }, + }); + setup.faux.setResponses([calls(["ghost", {}, "c1"]), DONE]); + const { harness, root } = await openChat(new MemoryStorage(), setup); + await (await root.submit({ type: "input", content: "go" }, context)).wait(context); + expect(seen.map((each) => each.id)).toEqual(["c1"]); + expect(await harness.getTask(seen[0]!.taskId, context)).toMatchObject({ + kind: "pi.tool", + input: { kind: "model", callId: "c1", unoffered: true }, + }); + const entries = await allEntries(root); + expect(text(results(entries).get("c1")).startsWith("try another tool|")).toBe(true); + expect(await diagnosticsOf(entries, "c1")).toEqual(["tool_unavailable"]); + await harness.close(context); + }); + + // #11043 + it("count toward a round's terminate", async () => { + const setup = chatSetup(); + addTool( + setup.registry, + tool("echo", async () => ({ output: [] })), + ); + addHooks(setup.registry, ToolTask, { + afterTool: (_call, result) => ({ ...result, control: { terminate: true } }), + }); + setup.faux.setResponses([calls(["echo", {}, "c1"], ["ghost", {}, "c2"]), DONE]); + const { harness, root } = await openChat(new MemoryStorage(), setup); + const settled = await (await root.submit({ type: "input", content: "go" }, context)).wait(context); + expect(settled.status).toBe("done"); + expect(setup.faux.state.callCount).toBe(1); + await harness.close(context); + }); + + // #11043 + it("take their turn in a sequential round, and the last handoff in call order wins", async () => { + const setup = chatSetup(); + setup.settings.toolExecution = "sequential"; + const order: string[] = []; + for (const name of ["a", "b"]) { + addTool( + setup.registry, + tool(name, async () => { + order.push(name); + return { output: [] }; + }), + ); + } + addHooks(setup.registry, ToolTask, { + afterTool: (call, result) => { + order.push(`hook ${call.id}`); + return call.id === "c1" ? undefined : { ...result, control: { handoff: `from ${call.id}` } }; + }, + }); + setup.faux.setResponses([calls(["a", {}, "c1"], ["ghost", {}, "c2"], ["b", {}, "c3"]), DONE]); + const { harness, root } = await openChat(new MemoryStorage(), setup); + await (await root.submit({ type: "input", content: "go" }, context)).wait(context); + expect(order).toEqual(["a", "hook c1", "hook c2", "b", "hook c3"]); + const resets = (await allEntries(root)).filter((entry) => ResetEntry.is(entry)); + expect(resets.map((entry) => (entry.model?.[0] as { content: string }).content)).toEqual(["from c3"]); + await harness.close(context); + }); + + // #11043 + it("each get a tool task in a parallel round of only such calls", async () => { + const setup = chatSetup(); + const seen: string[] = []; + addHooks(setup.registry, ToolTask, { + afterTool: (call) => { + seen.push(call.id); + return undefined; + }, + }); + setup.faux.setResponses([calls(["ghost", {}, "c1"], ["spirit", {}, "c2"], ["ghost", {}, "c3"]), DONE]); + const { harness, root } = await openChat(new MemoryStorage(), setup); + const settled = await (await root.submit({ type: "input", content: "go" }, context)).wait(context); + expect(settled.status).toBe("done"); + expect(seen.sort()).toEqual(["c1", "c2", "c3"]); + const entries = await allEntries(root); + for (const id of ["c1", "c2", "c3"]) expect(await diagnosticsOf(entries, id)).toEqual(["tool_unavailable"]); + // The round's results go to the next request. + expect(setup.faux.state.callCount).toBe(2); + await harness.close(context); + }); + + // #11043 + it("keep their flag in a version 2 generation's sequential round, also one migrated from version 1", async () => { + for (const version of [2, 1]) { + const path = await sqlitePath(); + const setup = chatSetup(); + setup.settings.toolExecution = "sequential"; + const started = deferred(); + let runs = 0; + const a = tool( + "a", + async (_args, _api, callContext) => { + runs++; + if (runs > 1) return { output: [] }; + started.resolve(); + return aborted(callContext.abortSignal!); + }, + { replay: "safe" }, + ); + addTool(setup.registry, a); + addTool( + setup.registry, + tool("ghost", async () => ({ output: [{ type: "text", text: "ran" }] })), + ); + setup.faux.setResponses([calls(["a", {}, "c1"], ["ghost", {}, "c2"]), DONE]); + let opened = await open(path, setup); + let root = opened.root; + // `ghost` is registered, but not offered to this request. + await root.configure({ tools: [a] }, context); + const id = (await root.submit({ type: "input", content: "go" }, context)).id; + await started.promise; + const live = (await opened.harness.snapshot(LiveDoc, ROOT_CONVERSATION_ID, context))!; + const generation = live.run!.taskId; + expect(live.tools?.map((slot) => [slot.callId, slot.unoffered])).toEqual([ + ["c1", undefined], + ["c2", true], + ]); + expect((await opened.harness.getTask(generation, context))?.version).toBe(2); + // The tools are offered when the round goes on: only the flag keeps `ghost` from running. + await root.configure({ tools: null }, context); + await opened.harness.close(context); + if (version === 1) { + const storage = await openNodeSqliteStorage(path); + const record = (await storage.task(generation, context))!; + await storage.commit([{ type: "task", value: { ...record, version: 1 } }], () => {}, context); + await storage.close(context); + } + + opened = await open(path, setup); + root = opened.root; + expect((await (await opened.harness.submission(id, context))!.wait(context)).status).toBe("done"); + expect((await opened.harness.getTask(generation, context))?.version).toBe(2); + const entries = await allEntries(root); + expect(await diagnosticsOf(entries, "c2")).toEqual(["tool_unavailable"]); + expect(text(results(entries).get("c2"))).not.toContain("ran"); + await opened.harness.close(context); + } + }); + + // #11043 + it("settle unavailable, not aborted, when their sequential round is aborted before they start", async () => { + const setup = chatSetup(); + setup.settings.toolExecution = "sequential"; + const started = deferred(); + addTool( + setup.registry, + tool("wait", async (_args, _api, callContext) => { + started.resolve(); + return aborted(callContext.abortSignal!); + }), + ); + setup.faux.setResponses([calls(["wait", {}, "c1"], ["ghost", {}, "c2"]), DONE]); + const { harness, root } = await openChat(new MemoryStorage(), setup); + const submission = await root.submit({ type: "input", content: "go" }, context); + await started.promise; + await root.abort(context); + expect((await submission.wait(context)).status).toBe("unanswered"); + const entries = await allEntries(root); + expect(await diagnosticsOf(entries, "c1")).toEqual(["aborted"]); + expect(await diagnosticsOf(entries, "c2")).toEqual(["tool_unavailable"]); + await harness.close(context); + }); +}); diff --git a/packages/durable/test/harness-tools.test.ts b/packages/durable/test/harness-tools.test.ts index 613975d52..6849861b8 100644 --- a/packages/durable/test/harness-tools.test.ts +++ b/packages/durable/test/harness-tools.test.ts @@ -138,7 +138,7 @@ describe("tool round", () => { await harness.close(context); }); - it("answers calls to tools the request did not offer without a task", async () => { + it("answers calls to tools the request did not offer through a flagged tool task", async () => { const setup = chatSetup(); addTool( setup.registry, @@ -157,8 +157,34 @@ describe("tool round", () => { expect(ghostEntry?.data).toEqual({ diagnostics: [{ severity: "error", code: "tool_unavailable", message: "Tool ghost is not available" }], }); + // #11043: the unoffered call has its own task, so `afterTool` and `control` apply to it as to any call. const tasks = await harness.commit((tx) => tx.scanTasks({ conversationId: root.id }, 20), context); - expect(tasks.items.filter((task) => task.kind === "pi.tool")).toHaveLength(1); + const inputs = tasks.items.filter((task) => task.kind === "pi.tool").map((task) => task.input); + expect(inputs).toHaveLength(2); + expect(inputs).toContainEqual(expect.objectContaining({ callId: "c1", unoffered: true })); + expect(inputs).toContainEqual(expect.not.objectContaining({ unoffered: true })); + await harness.close(context); + }); + + // #11043 + it("never runs a tool the request did not offer, even when the agent has it", async () => { + const setup = chatSetup(); + let ran = 0; + addTool( + setup.registry, + tool( + "hidden", + async () => { + ran++; + return { output: [] }; + }, + { deferred: true }, + ), + ); + const { harness, entries, status } = await run(setup, [calls(["hidden", {}, "c1"]), DONE]); + expect(status).toBe("done"); + expect(ran).toBe(0); + expect(resultText(results(entries)[0]!)).toBe("\n[error] Tool hidden is not available\n"); await harness.close(context); }); diff --git a/packages/durable/test/harness-view-run.test.ts b/packages/durable/test/harness-view-run.test.ts index 62c70b850..a4445ea91 100644 --- a/packages/durable/test/harness-view-run.test.ts +++ b/packages/durable/test/harness-view-run.test.ts @@ -327,7 +327,7 @@ describe("a run as the conversation view shows it", () => { await harness.close(context); }); - it("ends a call that never runs with its result at once, and a tool aborted with its generation", async () => { + it("ends a call that never runs through its own task, and a tool aborted with its generation", async () => { const setup = chatSetup(); addTool( setup.registry, @@ -352,13 +352,14 @@ describe("a run as the conversation view shows it", () => { await submission.wait(context); history.detach(); const states = history.live(); - // The call not offered is listed done, with its result entry, in the commit of its calling answer. + // #11043: the call not offered is listed pending with its task in the commit of its calling answer, and its task + // ends it done with its result entry. const listed = slotsOf(states, "c1").findIndex((slot) => slot !== undefined); - const ghost = slotsOf(states, "c1")[listed]!; - expect(ghost.status).toBe("done"); - const listing = history.views[listed]!.entries; - expect(listing.map((entry) => entry.kind).slice(-2)).toEqual(["pi.assistant", "pi.tool-result"]); - expect(ghost.entry).toBe(listing.at(-1)!.id); + expect(slotsOf(states, "c1")[listed]).toMatchObject({ status: "pending", unoffered: true }); + expect(slotsOf(states, "c1")[listed]!.taskId).toBeDefined(); + expect(history.views[listed]!.entries.at(-1)!.kind).toBe("pi.assistant"); + const ended = slotsOf(states, "c1").findIndex((slot) => slot?.status === "done"); + expect(slotsOf(states, "c1")[ended]!.entry).toBe(history.views[ended]!.entries.at(-1)!.id); // The aborted tool ends with its result entry, then the run ends without an answer. const done = slotsOf(states, "c2").findIndex((slot) => slot?.status === "done"); expect(slotsOf(states, "c2")[done]!.entry).toBe(history.views[done]!.entries.at(-1)!.id);