From 71f3ed3fb00cbee0a54a4dcca66dc942be37a155 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 11 Sep 2026 17:56:44 +0000 Subject: [PATCH] feat(task-graph,ai): a tool call's whole life on the wire MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A host drawing a card for one tool call could only reconstruct it by diffing successive copies of the conversation the turn publishes: 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. Builder did not hit this because its tools are closures it owns, so the adapter fills the card in from inside the tool. A host whose tools are task types it named has nothing to fill it in from, and neither does one relaying the turn to a protocol. So each call now reports where it has got to: `pending` when the model asks (carrying the arguments), `running` when the loop takes it up, and `completed` or `failed` when it settles, carrying the same string the model is about to read — read back off the result block rather than built a second time, so a card cannot drift from the answer or miss its clamp. Every call passes all three states, including the ones nothing executes for — an unknown tool, arguments its schema rejects, a call nobody approves — so a host draws one lifecycle rather than one per way a call can end. `running` covers waiting on a person for the same reason: an approval is part of making the call. Metadata on the same terms as `phase`: emitted on `stream_chunk`, never accumulated into a port, absent from `finish`, and no status flip. A task is not streaming its output because something it called started. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01JEFYAGb9D3mWfyhYkmeAvN --- .claude/CLAUDE.md | 11 +- packages/ai/src/task/AgentTask.ts | 41 ++++- .../task-graph/src/task/StreamProcessor.ts | 7 + packages/task-graph/src/task/StreamTypes.ts | 57 ++++++- packages/test/src/test/ai/AgentTask.test.ts | 155 +++++++++++++++++- 5 files changed, 260 insertions(+), 11 deletions(-) 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"); + }); + }); });