diff --git a/.claude/CLAUDE.md b/.claude/CLAUDE.md index b5f158179..9683a7729 100644 --- a/.claude/CLAUDE.md +++ b/.claude/CLAUDE.md @@ -215,10 +215,15 @@ tool is backed by a registered task (looked up by `taskType`, else `name`) or by `ToolDefinition.execute` function, which is handed a `ToolExecuteContext` — the tool-use id its answer belongs to, and the run's signal — and may throw a `ToolCallError` to report a failure in its own words rather than wrapped. The turn also emits a `snapshot` of `messages` -after every message it records, so a host can draw a tool card from the moment the model asks -for it; a `snapshot` rather than an object-delta because an array delta is folded as an upsert +after every message it records, so a host can mirror the conversation while the turn is still +running; a `snapshot` rather than an object-delta because an array delta is folded as an upsert list and successive whole-list snapshots would append into a transcript several times its -length. A tool reaching beyond `INFERENCE_ENTITLEMENTS` is put to +length. **A tool card reads `tool-call` events, not those snapshots** — one per call per state +(`pending` → `running` → `completed | failed`), so a card has its whole life on the wire +instead of being reconstructed by diffing transcripts, which reports outcomes only as one +batch after the last call and never says which call is running. Every call passes all three +states, the ones nothing executes for included (unknown tool, rejected arguments, refused +approval), so a host draws one lifecycle rather than one per way a call can end. A tool reaching beyond `INFERENCE_ENTITLEMENTS` is put to `IHumanConnector` as a `confirm` first: `requiresApproval` overrides that per tool, `approval: "never"` turns it off for a headless run, and with no connector registered such a call is refused rather than run. diff --git a/packages/ai/src/task/AgentTask.ts b/packages/ai/src/task/AgentTask.ts index 40b768a38..fc1043cf3 100644 --- a/packages/ai/src/task/AgentTask.ts +++ b/packages/ai/src/task/AgentTask.ts @@ -193,6 +193,11 @@ function assistantMessage(text: string, calls: readonly ToolCall[]): ChatMessage * context window as a fetched page — and it would then do it on every round the * model keeps guessing. */ +/** The text of a settled call, as the model will read it. */ +function toolResultText(result: ContentBlockToolResult): string { + return result.content.map((block) => (block.type === "text" ? block.text : "")).join(""); +} + function toolResult( call: ToolCall, text: string, @@ -337,16 +342,40 @@ export class AgentTask extends Task this.task.emit("stream_chunk", event as StreamEvent); break; } + case "tool-call": { + // Where one tool call has got to. Metadata only, like `phase`: no + // accumulator, no runOutputData, no status flip — a task is not + // streaming its output because something it called started. + this.task.emit("stream_chunk", event as StreamEvent); + break; + } case "finish": { sawFinish = true; // finish supersedes the snapshots it summarizes; `?? liveUsage` diff --git a/packages/task-graph/src/task/StreamTypes.ts b/packages/task-graph/src/task/StreamTypes.ts index 2f0434824..e810778e8 100644 --- a/packages/task-graph/src/task/StreamTypes.ts +++ b/packages/task-graph/src/task/StreamTypes.ts @@ -298,6 +298,60 @@ export type StreamPhase = { progress: number | undefined; }; +/** + * How far one tool call a task is running on a model's behalf has got. + * + * Emitted per call, so a host drawing a card for one has the card's whole + * life on the wire. Without it the only account of a tool call is the + * conversation the task publishes, and a card has to be reconstructed by + * diffing successive copies of that: the ask appears when the assistant + * message lands, and the outcome when the results land — as one batch, after + * the last of them, with nothing in between and no way to tell which call is + * running now. A host owning its own tools can fill that in from inside them, + * which is why this was not missed earlier; a host whose tools are task types + * it named cannot, and neither can one relaying to a protocol. + * + * Metadata, not data, on the same terms as {@link StreamPhase}: emitted on + * `stream_chunk`, never accumulated into a port, never part of a `finish` + * payload, and no status flip. A task that reports these still reports its + * conversation — this says where a call has got to, not what was said. + * + * `status` discriminates what else is known, rather than every field being + * optional on one shape: + * - `pending` — the model asked for it; nothing has run. Carries `input`. + * - `running` — the task has taken the call up. Every call passes through + * this, including the ones nothing ever executes for: a tool that does not + * exist, arguments its schema rejects, a call nobody approves. A host then + * draws one lifecycle rather than one per way a call can go, and `running` + * covers waiting on a person for the same reason — an approval is part of + * making the call, and a host drawing its own approval knows it asked. + * - `completed` / `failed` — settled, carrying the text the model reads + * back. `failed` is the call's own outcome — it threw, its arguments were + * rejected, nobody approved it — and not an error that ends the run. + */ +export type StreamToolCall = + | { + type: "tool-call"; + status: "pending"; + toolCallId: string; + name: string; + input: Record; + } + | { + type: "tool-call"; + status: "running"; + toolCallId: string; + name: string; + } + | { + type: "tool-call"; + status: "completed" | "failed"; + toolCallId: string; + name: string; + /** What the model reads back, after the same clamp the result carries. */ + result: string; + }; + /** * Discriminated union of all stream event types. * Used as the element type for `AsyncIterable` streams @@ -312,7 +366,8 @@ export type StreamEvent> = | StreamError | StreamRefusal | StreamUsage - | StreamPhase; + | StreamPhase + | StreamToolCall; // ======================================================================== // Port-level stream helpers diff --git a/packages/test/src/test/ai/AgentTask.test.ts b/packages/test/src/test/ai/AgentTask.test.ts index ab4a8390c..22ea91daa 100644 --- a/packages/test/src/test/ai/AgentTask.test.ts +++ b/packages/test/src/test/ai/AgentTask.test.ts @@ -14,7 +14,7 @@ import { setAiProviderRegistry, ToolCallError, } from "@workglow/ai"; -import type { TaskEntitlements, TaskGraphJson } from "@workglow/task-graph"; +import type { StreamEvent, TaskEntitlements, TaskGraphJson } from "@workglow/task-graph"; import { createGraphFromGraphJSON, Entitlements, Task, TaskRegistry } from "@workglow/task-graph"; import { HumanInputTask } from "@workglow/tasks"; import type { IHumanConnector, IHumanRequest, IHumanResponse } from "@workglow/util"; @@ -849,4 +849,157 @@ describe("AgentTask", () => { expect(seen).toEqual({ a: 2, b: 3 }); expect(JSON.stringify(toolResults(output.messages)[0])).toContain("5"); }); + + // ====================================================================== + // Per-tool lifecycle events + // ====================================================================== + + describe("tool-call events", () => { + /** Collects the lifecycle events a run reports, in the order they arrive. */ + function lifecycle(task: AgentTask): Array> { + const seen: Array> = []; + task.subscribe("stream_chunk", (event) => { + if (event.type === "tool-call") seen.push(event); + }); + return seen; + } + + /** + * The settled member, picked out by the field only it carries. Extracting + * on `status: "completed"` yields `never` instead — that member's status is + * the wider `"completed" | "failed"`, which no narrower constraint matches. + */ + type SettledToolCall = Extract; + + /** `status:id` for each event — the whole sequence in one readable line. */ + function trace(seen: ReadonlyArray>): string[] { + return seen.map((event) => `${event.status}:${event.toolCallId}`); + } + + it("reports a call pending, then running, then completed", async () => { + scriptModel([ + { calls: [{ id: "c1", name: "AgentTest_EchoTask", input: { text: "hi" } }] }, + { text: "It said HI." }, + ]); + const task = new AgentTask(); + const seen = lifecycle(task); + + const output = await task.run( + { model: MODEL, prompt: "echo hi", tools: [ECHO_TOOL], approval: "never" }, + { registry } + ); + + expect(trace(seen)).toEqual(["pending:c1", "running:c1", "completed:c1"]); + expect(seen.every((event) => event.name === "AgentTest_EchoTask")).toBe(true); + expect(seen[0]).toMatchObject({ status: "pending", input: { text: "hi" } }); + + // The settled text IS the string the model reads back, not a second copy + // of it that could drift from the result or miss its clamp. + const settled = seen[2] as SettledToolCall; + const block = toolResults(output.messages)[0] as { + readonly content: ReadonlyArray<{ readonly text?: string }>; + }; + expect(block.content[0]?.text).toBe(settled.result); + expect(settled.result).toContain("HI"); + }); + + it("announces every call the model asked for before running any of them", async () => { + scriptModel([ + { + calls: [ + { id: "c1", name: "AgentTest_EchoTask", input: { text: "a" } }, + { id: "c2", name: "AgentTest_EchoTask", input: { text: "b" } }, + ], + }, + { text: "done" }, + ]); + const task = new AgentTask(); + const seen = lifecycle(task); + + await task.run( + { model: MODEL, prompt: "echo twice", tools: [ECHO_TOOL], approval: "never" }, + { registry } + ); + + // The model asked for both at once, so both cards can be drawn at once; + // the calls then run one at a time, each settling before the next starts. + expect(trace(seen)).toEqual([ + "pending:c1", + "pending:c2", + "running:c1", + "completed:c1", + "running:c2", + "completed:c2", + ]); + }); + + it("settles a call the tool refused as failed, in the tool's own words", async () => { + scriptModel([{ calls: [{ id: "c1", name: "decline", input: {} }] }, { text: "ok" }]); + const task = new AgentTask(); + const seen = lifecycle(task); + + await task.run( + { + model: MODEL, + prompt: "go", + approval: "never", + tools: [ + { + name: "decline", + description: "Declines", + inputSchema: { type: "object", properties: {} }, + execute: async () => { + throw new ToolCallError("Not this time."); + }, + }, + ], + }, + { registry } + ); + + expect(trace(seen)).toEqual(["pending:c1", "running:c1", "failed:c1"]); + expect(seen[2]).toMatchObject({ status: "failed", result: "Not this time." }); + }); + + it("settles a call nobody approved as failed, without running it", async () => { + scriptModel([ + { + calls: [ + { id: "c1", name: "AgentTest_FetchTask", input: { url: "https://example.test" } }, + ], + }, + { text: "ok" }, + ]); + const human = connector((request) => ({ + requestId: request.requestId, + action: "decline", + content: undefined, + done: true, + })); + registry.registerInstance(HUMAN_CONNECTOR, human.connector); + const task = new AgentTask(); + const seen = lifecycle(task); + + await task.run({ model: MODEL, prompt: "fetch", tools: [FETCH_TOOL] }, { registry }); + + expect(trace(seen)).toEqual(["pending:c1", "running:c1", "failed:c1"]); + expect(fetchRuns).toBe(0); + }); + + it("settles a call for a tool that does not exist", async () => { + scriptModel([{ calls: [{ id: "c1", name: "nope", input: {} }] }, { text: "ok" }]); + const task = new AgentTask(); + const seen = lifecycle(task); + + await task.run( + { model: MODEL, prompt: "go", tools: [ECHO_TOOL], approval: "never" }, + { registry } + ); + + // Every call passes the same three states, the ones this loop answers + // itself included: a host draws one card lifecycle, not two. + expect(trace(seen)).toEqual(["pending:c1", "running:c1", "failed:c1"]); + expect((seen[2] as SettledToolCall).result).toContain("Unknown tool"); + }); + }); });