mirror of
https://github.com/badlogic/pi-mono.git
synced 2026-11-05 23:37:25 +00:00
feat(durable): calls to tools not offered run as tool tasks
This commit is contained in:
Mario Zechner
parent
87eff524f9
commit
1c6928b309
11 changed files with 370 additions and 82 deletions
|
|
@ -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
|
||||
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -143,8 +143,15 @@ const DEFAULT_POLL_AFTER_MS = 5000;
|
|||
*/
|
||||
export const GenerationTask = defineTask<GenerationInput, GenerationCheckpoint, GenerationResult, GenerationHooks>({
|
||||
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<GenerationInput, GenerationCheckpoint,
|
|||
// Sequential round: start the next call and wait for it.
|
||||
await runtime.commit(async (tx): Promise<Next> => {
|
||||
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<GenerationInput, GenerationCheckpoint,
|
|||
const polled = deferredOf(checkpoint);
|
||||
if (polled !== undefined) await cancelDeferred(runtime, polled, context);
|
||||
const conversationId = runtime.conversationId;
|
||||
// Runs after the round's tool tasks are terminal; calls never started get `aborted` results (spec §8.5).
|
||||
// Runs after the round's tool tasks are terminal; calls never started get `aborted` results (spec §8.5), unoffered
|
||||
// ones `tool_unavailable`.
|
||||
const unstarted =
|
||||
checkpoint.phase === "tools"
|
||||
? await readCalls(runtime, checkpoint.assistant, checkpoint.pending, context)
|
||||
|
|
@ -334,7 +342,11 @@ export const GenerationTask = defineTask<GenerationInput, GenerationCheckpoint,
|
|||
const live = await tx.doc(LiveDoc, conversationId);
|
||||
await convertPartial(tx, live, conversationId);
|
||||
for (const call of unstarted) {
|
||||
const result = harnessError("aborted", `Tool ${call.name} was aborted`);
|
||||
// A call its request did not offer could not have run either way.
|
||||
const unoffered = live.tools?.find((slot) => 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<TaskId<ToolTaskResult>> {
|
||||
/** 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<TaskId<ToolTaskResult>> {
|
||||
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<ToolTaskResult>[] = [];
|
||||
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<Next> => {
|
||||
const boundary = await prepareBoundary(tx, conversationId, runtime.settings);
|
||||
if (added.length > 0) await addTools(tx, conversationId, added);
|
||||
|
|
|
|||
|
|
@ -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;
|
||||
};
|
||||
|
|
|
|||
|
|
@ -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<ToolTaskInput, ToolTaskCheckpoint, ToolTaskResult, ToolHooks>({
|
||||
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<ToolTaskInput, ToolTaskCheckpoint, ToolTaskRe
|
|||
},
|
||||
abort: async (task, runtime, context) => {
|
||||
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<ToolRegistration | undefined> {
|
||||
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,
|
||||
|
|
|
|||
|
|
@ -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 },
|
||||
);
|
||||
|
|
|
|||
|
|
@ -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" },
|
||||
],
|
||||
]);
|
||||
|
|
|
|||
|
|
@ -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<ToolTaskInput, { kind: "model" }>;
|
||||
const { kind: _kind, ...version1 } = record.input as Extract<ToolTaskInput, { kind: "model" }>;
|
||||
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);
|
||||
});
|
||||
});
|
||||
|
|
|
|||
|
|
@ -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);
|
||||
});
|
||||
});
|
||||
|
|
|
|||
|
|
@ -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("<harness>\n[error] Tool hidden is not available\n</harness>");
|
||||
await harness.close(context);
|
||||
});
|
||||
|
||||
|
|
|
|||
|
|
@ -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);
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue