From 819190dc4690a83e0f3bc22b9b751846143214af Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 9 Sep 2026 20:06:31 +0000 Subject: [PATCH 1/6] =?UTF-8?q?feat(ai):=20AgentTask=20=E2=80=94=20the=20t?= =?UTF-8?q?ool-calling=20turn=20loop=20as=20a=20task?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ToolCallingTask is one round: it returns the model's text and the calls it wants, and stops. Every host driving a conversation has had to write the part that runs those calls, feeds the results back and goes round again — and the invariants there are easy to miss. AgentTask is that loop. `messages` goes in and comes back out, so the host still owns the conversation; what moves here is the machinery: - Every `tool_use` is answered. An unknown tool, arguments failing the tool's own schema, a throw inside it, a person declining it — each is an error result the model reads and can recover from. Dropping the call looks cheaper and orphans the `tool_use`, which the provider rejects on the NEXT round, one turn away from the mistake. - Tool-call ids are made unique against the whole conversation, so a model that restarts at `call_0` each turn cannot attach this turn's result to an earlier turn's call. - Tools run in order, because one may block on a person. - A round that produced neither text nor a usable call records nothing: an empty assistant message is not a reply and poisons a replayed prefix. - The turn's text is summed from each round's settled output rather than from the deltas forwarded, so a provider that reports text only on its finish event is not reported as an empty answer. Tools resolve the way ToolDefinition already documented but nothing yet read: an explicit `type`, else a supplied `execute`, else the task registry — by `taskType` when the tool is presented under another name. `taskType` moves onto ToolDefinition itself, since that is the field a runner needs and ToolDefinitionWithTaskType only narrowed it to required. Approval reuses the entitlement taxonomy rather than adding a second list: a tool whose backing class reaches beyond INFERENCE_ENTITLEMENTS is put to the IHumanConnector as a `confirm`, carrying what it reaches and with what arguments. `requiresApproval` overrides per tool in both directions, and a headless run says `approval: "never"`. With approval called for and no connector registered the call is refused, not run — a gate that fails open is not a gate, and the refusal reaches the model as an ordinary result. A host function tool defaults to no approval: it cannot arrive by name in graph JSON, so supplying one is already a host decision. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01JEFYAGb9D3mWfyhYkmeAvN --- .claude/CLAUDE.md | 11 + packages/ai/src/task/AgentTask.ts | 420 ++++++++++++++ packages/ai/src/task/AgentToolExecution.ts | 208 +++++++ packages/ai/src/task/ToolCallingTask.ts | 6 + packages/ai/src/task/ToolCallingUtils.ts | 15 + packages/ai/src/task/index.ts | 2 + packages/ai/src/task/registerAiTasks.ts | 2 + packages/test/src/test/ai/AgentTask.test.ts | 611 ++++++++++++++++++++ 8 files changed, 1275 insertions(+) create mode 100644 packages/ai/src/task/AgentTask.ts create mode 100644 packages/ai/src/task/AgentToolExecution.ts create mode 100644 packages/test/src/test/ai/AgentTask.test.ts diff --git a/.claude/CLAUDE.md b/.claude/CLAUDE.md index 8207c9fb1..3c82fb40b 100644 --- a/.claude/CLAUDE.md +++ b/.claude/CLAUDE.md @@ -206,6 +206,17 @@ RAG tasks: `ChunkVectorUpsertTask` (`knowledgeBase` + `chunks` + `vector`, optio `method: "similarity" | "hybrid"`), `HierarchyJoinTask`, `RerankerTask`, `QueryExpanderTask`, `TextChunkerTask`, `HierarchicalChunkerTask`. +`AgentTask` is the turn loop: one `ToolCallingTask` per round, then the tools the model asked +for, then their results back to it, until the model answers or `maxRounds` runs out. +`messages` goes in and comes back out, so the host owns the conversation. **Every `tool_use` +is answered** — an unknown tool, arguments failing the tool's schema, a throw, a person +declining — because dropping the call orphans it and the provider rejects the next round. A +tool is backed by a registered task (looked up by `taskType`, else `name`) or by a +`ToolDefinition.execute` function. 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. + Conversation helpers for a host keeping its own message list, none of which run inside a task: `normalizeHistoryForModel` / `trimHistoryForModel` (`ChatHistory.ts`), and `collectToolUseIds` / `uniquifyToolCallIds` / `repairDuplicateToolCallIds` (`ToolCallIds.ts`). diff --git a/packages/ai/src/task/AgentTask.ts b/packages/ai/src/task/AgentTask.ts new file mode 100644 index 000000000..8e7022c83 --- /dev/null +++ b/packages/ai/src/task/AgentTask.ts @@ -0,0 +1,420 @@ +/** + * @license + * Copyright 2026 Steven Roussey + * SPDX-License-Identifier: Apache-2.0 + */ + +import type { + CachePolicy, + IExecuteContext, + IRunConfig, + StreamEvent, + TaskConfig, + TaskEntitlements, +} from "@workglow/task-graph"; +import { CreateWorkflow, Entitlements, Task, Workflow } from "@workglow/task-graph"; +import type { DataPortSchema } from "@workglow/util/schema"; +import type { Capability } from "../capability/Capabilities"; +import { createEmitQueue } from "../capability/emitQueue"; +import type { ModelConfig } from "../model/ModelSchema"; +import type { AgentApprovalMode } from "./AgentToolExecution"; +import { runAgentTool } from "./AgentToolExecution"; +import { + DEFAULT_MAX_HISTORY_CHARS, + normalizeHistoryForModel, + trimHistoryForModel, +} from "./ChatHistory"; +import type { + ChatMessage, + ContentBlock, + ContentBlockInToolResultBody, + ContentBlockToolResult, +} from "./ChatMessage"; +import { ChatMessageSchema } from "./ChatMessage"; +import { promptToUserMessage } from "./base/CheckpointPorts"; +import { collectToolUseIds, uniquifyToolCallIds } from "./ToolCallIds"; +import type { ToolCallingTaskInput, ToolCallingTaskOutput } from "./ToolCallingTask"; +import { ToolCallingInputSchema, ToolCallingTask } from "./ToolCallingTask"; +import type { ToolCall, ToolDefinition } from "./ToolCallingUtils"; +import { compileToolValidators, sanitizeToolArgs } from "./ToolCallingUtils"; + +/** Rounds before the loop gives up on the model reaching an answer. */ +const DEFAULT_MAX_ROUNDS = 8; + +/** + * Characters of one tool's output the model is shown. + * + * A single unbounded result — a fetched page, a table dump — is the ordinary + * way an agent turn dies: it fills the window, and every later round is spent + * re-sending it. Truncation is marked in the text so the model can tell it is + * reading a prefix and ask for less next time. + */ +const DEFAULT_MAX_TOOL_RESULT_CHARS = 20_000; + +export const AgentInputSchema = { + type: "object", + properties: { + model: ToolCallingInputSchema.properties.model, + prompt: ToolCallingInputSchema.properties.prompt, + systemPrompt: ToolCallingInputSchema.properties.systemPrompt, + messages: { + ...ToolCallingInputSchema.properties.messages, + description: + "Conversation history BEFORE this turn. The prompt is appended to it as the user message, and the whole thing comes back on the messages output.", + }, + tools: ToolCallingInputSchema.properties.tools, + temperature: ToolCallingInputSchema.properties.temperature, + maxTokens: ToolCallingInputSchema.properties.maxTokens, + maxRounds: { + type: "integer", + title: "Max Rounds", + description: + "Model calls before the turn ends unanswered. One round is one model call plus every tool it asked for.", + minimum: 1, + "x-ui-group": "Configuration", + }, + maxToolResultChars: { + type: "integer", + title: "Max Tool Result Characters", + description: "Characters of a single tool's output the model is shown before truncation", + minimum: 1, + "x-ui-group": "Configuration", + }, + maxHistoryChars: { + type: "integer", + title: "Max History Characters", + description: + "Character budget for the history sent to the model; older turns are dropped whole", + minimum: 1, + "x-ui-group": "Configuration", + }, + approval: { + type: "string", + title: "Approval", + description: + 'When a tool call is put to a person first: "beyond-inference" confirms any tool reaching past running a model, "never" confirms nothing', + enum: ["beyond-inference", "never"], + "x-ui-group": "Configuration", + }, + }, + required: ["model", "prompt", "tools"], + additionalProperties: false, +} as const satisfies DataPortSchema; + +export const AgentOutputSchema = { + type: "object", + properties: { + text: { + type: "string", + title: "Text", + description: + "Everything the assistant said this turn, including what it narrated between tool calls", + "x-stream": "append", + }, + messages: { + type: "array", + items: ChatMessageSchema, + title: "Messages", + description: + "The input history plus this turn: the user message, each assistant reply, and each round's tool results", + }, + rounds: { + type: "integer", + title: "Rounds", + description: "Model calls this turn took", + }, + stopReason: { + type: "string", + title: "Stop Reason", + description: + '"answered" when the model replied without asking for a tool, "max-rounds" when it ran out of rounds first', + enum: ["answered", "max-rounds"], + }, + }, + required: ["text", "messages", "rounds", "stopReason"], + additionalProperties: false, +} as const satisfies DataPortSchema; + +/** + * Written out rather than derived from {@link AgentInputSchema}: the shared + * ports carry `ToolCallingTask`'s `prompt`, whose `oneOf` costs more + * type-instantiation budget through `FromSchema` than the whole task is worth. + */ +export type AgentTaskInput = { + readonly model: string | ModelConfig; + /** Taken from the round task rather than restated, so the two cannot drift. */ + readonly prompt: ToolCallingTaskInput["prompt"]; + readonly systemPrompt?: string | undefined; + readonly messages?: ReadonlyArray | undefined; + readonly tools: ToolDefinition[]; + readonly temperature?: number | undefined; + readonly maxTokens?: number | undefined; + readonly maxRounds?: number | undefined; + readonly maxToolResultChars?: number | undefined; + readonly maxHistoryChars?: number | undefined; + readonly approval?: AgentApprovalMode | undefined; +}; + +export type AgentTaskOutput = { + text: string; + messages: ChatMessage[]; + rounds: number; + stopReason: "answered" | "max-rounds"; +}; + +export type AgentTaskConfig = TaskConfig; + +/** A call the model made that cannot be answered, so must not be committed. */ +function isAnswerable(call: ToolCall): boolean { + return typeof call.id === "string" && call.id.length > 0 && typeof call.name === "string"; +} + +function assistantMessage(text: string, calls: readonly ToolCall[]): ChatMessage { + const content: ContentBlock[] = []; + if (text.length > 0) content.push({ type: "text", text }); + for (const call of calls) { + content.push({ + type: "tool_use", + id: call.id, + name: call.name, + input: call.input, + ...(call.providerSignature === undefined + ? {} + : { providerSignature: call.providerSignature }), + }); + } + return { role: "assistant", content }; +} + +function toolResult(call: ToolCall, text: string, isError: boolean): ContentBlockToolResult { + const body: ContentBlockInToolResultBody[] = [{ type: "text", text }]; + return { + type: "tool_result", + tool_use_id: call.id, + content: body, + is_error: isError ? true : undefined, + }; +} + +/** + * One conversational turn: call the model, run every tool it asks for, feed the + * results back, and repeat until it answers without tools. + * + * The loop is the part hosts get wrong, so it lives here rather than in each of + * them. Three invariants it keeps: + * + * - **Every `tool_use` gets a `tool_result`.** An unknown tool, arguments that + * fail the tool's own schema, a throw inside the tool, a person declining it + * — each becomes an error result the model reads and can recover from. + * Dropping the call instead leaves an unanswered `tool_use` in the history, + * which providers reject on the next round, so the cheap-looking filter + * breaks the conversation one turn later. + * - **Tool-call ids are unique across the whole conversation.** A model that + * restarts its numbering at `call_0` each turn would otherwise attach this + * turn's result to an earlier turn's call. + * - **Tools run in order.** One may block on a person; the next must not race + * ahead of the answer. + * + * What it deliberately does NOT do is keep the conversation. `messages` comes + * in and goes out, so the host owns the record — which is what lets the same + * loop serve a chat transcript, a CLI session, and a graph node. + */ +export class AgentTask extends Task { + public static override type = "AgentTask"; + /** + * A statement about the model port, not a gate this class enforces: the model + * is handed to a {@link ToolCallingTask} each round, and that is where the + * capability is checked. Declared so a model picker filters the same way. + */ + public static readonly requires = ["tool-use"] as const satisfies Capability[]; + public static override category = "AI Text"; + public static override title = "Agent"; + public static override description = + "Runs a conversational turn to completion: calls a model, runs the tools it asks for, and feeds the results back until it answers"; + /** Tools have side effects and a person may have approved one; never replay a turn from cache. */ + public static override cachePolicy: CachePolicy = { kind: "none" }; + /** + * The reach is in the tools, which arrive as an input and so are unknown + * until the run. Saying so is what puts an approval in front of an agent + * used as somebody else's tool. + */ + public static override entitlementsFromChildren: boolean = true; + + public static override entitlements(): TaskEntitlements { + return { + entitlements: [{ id: Entitlements.AI_INFERENCE, reason: "Runs a model once per round" }], + }; + } + + public static override inputSchema(): DataPortSchema { + return AgentInputSchema as DataPortSchema; + } + + public static override outputSchema(): DataPortSchema { + return AgentOutputSchema as DataPortSchema; + } + + async *executeStream( + input: AgentTaskInput, + context: IExecuteContext + ): AsyncIterable> { + const maxRounds = input.maxRounds ?? DEFAULT_MAX_ROUNDS; + const maxToolResultChars = input.maxToolResultChars ?? DEFAULT_MAX_TOOL_RESULT_CHARS; + const maxHistoryChars = input.maxHistoryChars ?? DEFAULT_MAX_HISTORY_CHARS; + const approval: AgentApprovalMode = input.approval ?? "beyond-inference"; + const validators = compileToolValidators(input.tools); + const byName = new Map(input.tools.map((tool) => [tool.name, tool])); + + const messages: ChatMessage[] = [...(input.messages ?? []), promptToUserMessage(input.prompt)]; + let text = ""; + let rounds = 0; + + for (let round = 0; round < maxRounds; round++) { + context.signal.throwIfAborted(); + rounds = round + 1; + await context.updateProgress(undefined, "Thinking"); + + const turn = new ToolCallingTask({ title: `Round ${rounds}` }); + const captured: { output: ToolCallingTaskOutput | undefined } = { output: undefined }; + for await (const event of this.streamRound( + turn, + captured, + input, + messages, + maxHistoryChars, + context + )) { + yield event; + } + const output = captured.output; + // Summed from the round's settled output rather than from the deltas just + // forwarded: a provider that reports its text only on the finish event + // streams nothing, and counting deltas would report an empty answer while + // `messages` carried the real one. + text += output?.text ?? ""; + + const calls = uniquifyToolCallIds( + (output?.toolCalls ?? []).filter(isAnswerable), + collectToolUseIds(messages) + ); + // A turn with neither text nor a usable call records nothing: an empty + // assistant message is not a reply, and providers reject a replayed + // prefix containing one. + const reply = assistantMessage(output?.text ?? "", calls); + if (reply.content.length > 0) messages.push(reply); + if (calls.length === 0) { + yield { type: "finish", data: { text, messages, rounds, stopReason: "answered" } }; + return; + } + + const results: ContentBlockToolResult[] = []; + for (const call of calls) { + context.signal.throwIfAborted(); + await context.updateProgress(undefined, `Running ${call.name}`); + results.push( + await this.runCall(call, byName, validators, context, { + approval, + maxResultChars: maxToolResultChars, + }) + ); + } + messages.push({ role: "tool", content: results }); + } + + yield { type: "finish", data: { text, messages, rounds, stopReason: "max-rounds" } }; + } + + /** + * Runs one round's model call and re-yields its text as this task's own. + * + * Owned, so the round shows up under this task and inherits the registry, + * the abort signal and the run's usage accounting. The child's `toolCalls` + * deltas are deliberately NOT forwarded: this task has no such port, and a + * delta naming one would accumulate onto an output that does not exist. + */ + private async *streamRound( + turn: ToolCallingTask, + captured: { output: ToolCallingTaskOutput | undefined }, + input: AgentTaskInput, + messages: readonly ChatMessage[], + maxHistoryChars: number, + context: IExecuteContext + ): AsyncIterable> { + context.own(turn); + const queue = createEmitQueue>(); + const off = turn.subscribe("stream_chunk", (event: StreamEvent) => { + if (event.type !== "text-delta") return; + if ((event.port ?? "text") !== "text") return; + queue.push({ type: "text-delta", port: "text", textDelta: event.textDelta }); + }); + const run = (async () => { + try { + captured.output = await turn.run({ + model: input.model, + prompt: input.prompt, + systemPrompt: input.systemPrompt, + messages: normalizeHistoryForModel(trimHistoryForModel(messages, maxHistoryChars)), + tools: input.tools, + temperature: input.temperature, + maxTokens: input.maxTokens, + }); + } finally { + off(); + queue.close(); + } + })(); + // Held so a rejection while the queue is still draining is not reported as + // unhandled; it is re-thrown below, once the events already produced have + // reached the caller. + run.catch(() => {}); + for await (const event of queue.iterable) yield event; + await run; + } + + /** + * One tool call, from the model's arguments to what it reads back. + * + * Sanitising before validating is the order that matters: a `__proto__` key + * would otherwise pass a schema that allows additional properties. + */ + private async runCall( + call: ToolCall, + byName: ReadonlyMap, + validators: ReturnType, + context: IExecuteContext, + options: { readonly approval: AgentApprovalMode; readonly maxResultChars: number } + ): Promise { + const tool = byName.get(call.name); + if (!tool) { + const known = [...byName.keys()].join(", "); + return toolResult(call, `Unknown tool "${call.name}". Available: ${known}`, true); + } + const sanitized = sanitizeToolArgs(call.input) as Record; + const validator = validators.get(call.name); + if (validator) { + const check = validator.validate(sanitized); + if (!check.valid) { + const detail = check.errors.map((error) => error.message).join("; ") || "invalid arguments"; + return toolResult(call, `Invalid arguments for ${call.name}: ${detail}`, true); + } + } + const result = await runAgentTool(tool, { ...call, input: sanitized }, context, options); + return toolResult(call, result.text, result.isError); + } +} + +export const agent = ( + input: AgentTaskInput, + config?: AgentTaskConfig, + runConfig?: Partial +) => { + return new AgentTask(config).run(input, runConfig); +}; + +declare module "@workglow/task-graph" { + interface Workflow { + agent: CreateWorkflow; + } +} + +Workflow.prototype.agent = CreateWorkflow(AgentTask); diff --git a/packages/ai/src/task/AgentToolExecution.ts b/packages/ai/src/task/AgentToolExecution.ts new file mode 100644 index 000000000..7380f2edd --- /dev/null +++ b/packages/ai/src/task/AgentToolExecution.ts @@ -0,0 +1,208 @@ +/** + * @license + * Copyright 2026 Steven Roussey + * SPDX-License-Identifier: Apache-2.0 + */ + +import type { IExecuteContext, TaskConfig } from "@workglow/task-graph"; +import { + describeTaskClassReach, + getTaskConstructors, + taskClassNeedsApproval, +} from "@workglow/task-graph"; +import { HUMAN_CONNECTOR, uuid4 } from "@workglow/util"; +import type { DataPortSchema } from "@workglow/util/schema"; +import type { ToolCall, ToolDefinition } from "./ToolCallingUtils"; + +/** + * When a tool call is put to a human before it runs. + * + * - `"beyond-inference"`: a tool whose reach exceeds what a caller running a + * model already holds is confirmed first. Which tools those are is read from + * the backing task class, so the answer tracks the entitlement taxonomy + * rather than a second list kept in step by hand. + * - `"never"`: no tool is ever confirmed. For a headless run, where there is + * nobody to ask and blocking on an answer that cannot come is worse than the + * reach. + */ +export type AgentApprovalMode = "beyond-inference" | "never"; + +/** What a tool call produced, as the model will read it back. */ +export interface AgentToolResult { + readonly text: string; + readonly isError: boolean; +} + +/** + * The card a person approves. Titles are the labels a connector renders, so + * they are written for a reader rather than for a schema. + */ +const APPROVAL_SCHEMA = { + type: "object", + properties: { + tool: { type: "string", title: "Tool" }, + reach: { type: "string", title: "Reaches" }, + arguments: { type: "string", title: "Arguments" }, + }, + additionalProperties: false, +} as const satisfies DataPortSchema; + +/** + * The task type behind a tool, which is not always what the tool is called. + * `taskTypesToTools` carries the type separately for exactly this reason: a + * host may present `search_the_web` for a task registered under another name, + * and resolving on the presented name would then find nothing. + */ +function backingTaskType(tool: ToolDefinition): string { + return tool.taskType && tool.taskType.length > 0 ? tool.taskType : tool.name; +} + +/** Bound on a value shown for approval — a person reads a line, not a payload. */ +const MAX_APPROVAL_ARGUMENT_CHARS = 400; + +function clamp(text: string, max: number): string { + if (text.length <= max) return text; + return `${text.slice(0, max)}… [truncated, ${text.length} chars total]`; +} + +function stringifyForModel(value: unknown): string { + if (typeof value === "string") return value; + try { + return JSON.stringify(value) ?? String(value); + } catch { + return String(value); + } +} + +function errorMessage(error: unknown): string { + return error instanceof Error ? error.message : String(error); +} + +/** + * Whether this tool call is put to a human first. + * + * A **task-backed** tool is answered from its class, because a task name can + * arrive in graph JSON the host did not author — so the reach has to be read + * rather than trusted. A **function-backed** tool can only have been handed in + * by host code, which is a decision already made, so it defaults to no + * approval. Either can say so outright with `requiresApproval`, which wins in + * both directions. + */ +export function toolCallNeedsApproval( + tool: ToolDefinition, + mode: AgentApprovalMode, + registry: Parameters[0] +): boolean { + if (mode === "never") return false; + if (typeof tool.requiresApproval === "boolean") return tool.requiresApproval; + const ctor = getTaskConstructors(registry).get(backingTaskType(tool)); + if (!ctor) return false; + return taskClassNeedsApproval(ctor); +} + +/** + * Puts one tool call to a human and reports whether it may run. + * + * Fails **closed** when approval is called for and no connector is registered: + * running unapproved would make the gate advisory, and the refusal reaches the + * model as an ordinary tool result, so it can say why rather than stalling. + */ +async function approveToolCall( + tool: ToolDefinition, + call: ToolCall, + context: IExecuteContext +): Promise { + if (!context.registry.has(HUMAN_CONNECTOR)) { + return { + text: + `Running "${tool.name}" needs a person's approval, and this run has no way to ask for one. ` + + `Tell the user that, and do not try this tool again.`, + isError: true, + }; + } + const ctor = getTaskConstructors(context.registry).get(backingTaskType(tool)); + const response = await context.registry.get(HUMAN_CONNECTOR).send( + { + requestId: uuid4(), + targetHumanId: "default", + kind: "confirm", + message: `Run "${tool.name}"?`, + contentSchema: APPROVAL_SCHEMA as DataPortSchema, + contentData: { + tool: tool.name, + reach: ctor ? describeTaskClassReach(ctor) : "not declared — this tool is a host function", + arguments: clamp(stringifyForModel(call.input), MAX_APPROVAL_ARGUMENT_CHARS), + }, + expectsResponse: true, + mode: "single", + metadata: { toolUseId: call.id, toolName: tool.name }, + }, + context.signal + ); + if (response.action === "accept") return undefined; + // Declined and cancelled read the same way to a model — it did not happen and + // repeating the request is not what the person wants next. + return { + text: + `The user did not approve running "${tool.name}". Do not retry it; ` + + `ask what they would rather do.`, + isError: true, + }; +} + +/** + * Runs one tool call and returns what the model reads back. + * + * Every failure here is a **result**, never a throw: the loop has already + * committed the model's `tool_use` to the conversation, and a provider rejects + * a turn whose `tool_use` has no matching `tool_result`. An abort is the one + * exception — the run is over, so there is no next turn to keep well-formed. + */ +export async function runAgentTool( + tool: ToolDefinition, + call: ToolCall, + context: IExecuteContext, + options: { readonly approval: AgentApprovalMode; readonly maxResultChars: number } +): Promise { + const refusal = toolCallNeedsApproval(tool, options.approval, context.registry) + ? await approveToolCall(tool, call, context) + : undefined; + if (refusal) return refusal; + + try { + const output = await invokeTool(tool, call, context); + return { text: clamp(stringifyForModel(output), options.maxResultChars), isError: false }; + } catch (error) { + if (context.signal.aborted) throw error; + return { text: `${tool.name} failed: ${errorMessage(error)}`, isError: true }; + } +} + +/** + * Resolves a tool to the thing that runs it, in the order + * {@link ToolDefinition} documents: an explicit `type` decides, and otherwise a + * supplied `execute` wins over a registry lookup on the name. + */ +async function invokeTool( + tool: ToolDefinition, + call: ToolCall, + context: IExecuteContext +): Promise { + if (tool.type === "function" || (tool.type === undefined && tool.execute)) { + if (!tool.execute) { + throw new Error(`Tool "${tool.name}" is declared type "function" but supplies no execute()`); + } + return await tool.execute(call.input); + } + const ctor = getTaskConstructors(context.registry).get(backingTaskType(tool)); + if (!ctor) { + throw new Error( + `Tool "${tool.name}" is backed by no registered task type and supplies no execute()` + ); + } + // A fresh id per call: two calls to one tool would otherwise both carry the + // id from `config` and collide as siblings in this task's subgraph. + const task = new ctor({ ...tool.config, id: uuid4() } as TaskConfig); + context.own(task); + return await task.run(call.input); +} diff --git a/packages/ai/src/task/ToolCallingTask.ts b/packages/ai/src/task/ToolCallingTask.ts index 0cefd1ba5..3d0a5706b 100644 --- a/packages/ai/src/task/ToolCallingTask.ts +++ b/packages/ai/src/task/ToolCallingTask.ts @@ -106,6 +106,12 @@ export const ToolDefinitionSchema = { description: "JSON Schema describing what the tool returns", additionalProperties: true, }, + taskType: { + type: "string", + title: "Task Type", + description: + "The registered task type backing this tool, when it differs from the name the model sees", + }, configSchema: { type: "object", title: "Config Schema", diff --git a/packages/ai/src/task/ToolCallingUtils.ts b/packages/ai/src/task/ToolCallingUtils.ts index 13c3fb9bb..8920ef8cc 100644 --- a/packages/ai/src/task/ToolCallingUtils.ts +++ b/packages/ai/src/task/ToolCallingUtils.ts @@ -35,6 +35,13 @@ export interface ToolDefinition { * (check `execute`, then registry lookup, then stub). */ type?: "function" | "task"; + /** + * The task type backing this tool, when it is not the name the model sees. + * A host is free to present `search_the_web` for a task registered under + * another name, and a runner resolving on the presented name would then find + * nothing. `taskTypesToTools` always sets it. + */ + taskType?: string; /** JSON Schema describing the task's configuration options. */ configSchema?: JsonSchema; /** Concrete configuration values matching {@link configSchema}. */ @@ -44,6 +51,14 @@ export interface ToolDefinition { * by calling this function directly instead of instantiating a Task. */ execute?: (input: Record) => Promise>; + /** + * Whether a person approves each call before it runs, overriding whatever a + * runner would otherwise decide for this tool. Set it in both directions: a + * host function that spends money says `true`, and a task whose reach a host + * has already scoped away says `false`. Absent leaves the decision to the + * runner — see `AgentTask`, which reads it from the backing task class. + */ + requiresApproval?: boolean; } /** diff --git a/packages/ai/src/task/index.ts b/packages/ai/src/task/index.ts index 0dda1f80e..58cb29bc2 100644 --- a/packages/ai/src/task/index.ts +++ b/packages/ai/src/task/index.ts @@ -8,6 +8,8 @@ export { registerAiTasks } from "./registerAiTasks"; +export * from "./AgentTask"; +export * from "./AgentToolExecution"; export * from "./AiChatTask"; export * from "./AiChatWithKbTask"; export * from "./BackgroundRemovalTask"; diff --git a/packages/ai/src/task/registerAiTasks.ts b/packages/ai/src/task/registerAiTasks.ts index 0b28f179f..73794026a 100644 --- a/packages/ai/src/task/registerAiTasks.ts +++ b/packages/ai/src/task/registerAiTasks.ts @@ -5,6 +5,7 @@ */ import { TaskRegistry } from "@workglow/task-graph"; +import { AgentTask } from "./AgentTask"; import { AiChatTask } from "./AiChatTask"; import { AiChatWithKbTask } from "./AiChatWithKbTask"; import { BackgroundRemovalTask } from "./BackgroundRemovalTask"; @@ -65,6 +66,7 @@ import { VectorSimilarityTask } from "./VectorSimilarityTask"; */ export const registerAiTasks = () => { const tasks = [ + AgentTask, AiChatTask, AiChatWithKbTask, BackgroundRemovalTask, diff --git a/packages/test/src/test/ai/AgentTask.test.ts b/packages/test/src/test/ai/AgentTask.test.ts new file mode 100644 index 000000000..efaee60f8 --- /dev/null +++ b/packages/test/src/test/ai/AgentTask.test.ts @@ -0,0 +1,611 @@ +/** + * @license + * Copyright 2026 Steven Roussey + * SPDX-License-Identifier: Apache-2.0 + */ + +import type { AiProviderRunFn, ModelConfig, ToolDefinition } from "@workglow/ai"; +import { + AgentTask, + AiProviderRegistry, + DirectExecutionStrategy, + getAiProviderRegistry, + setAiProviderRegistry, +} from "@workglow/ai"; +import type { TaskEntitlements } from "@workglow/task-graph"; +import { Entitlements, Task, TaskRegistry } from "@workglow/task-graph"; +import { HumanInputTask } from "@workglow/tasks"; +import type { IHumanConnector, IHumanRequest, IHumanResponse } from "@workglow/util"; +import { Container, HUMAN_CONNECTOR, ServiceRegistry } from "@workglow/util"; +import type { DataPortSchema } from "@workglow/util/schema"; +import { afterEach, beforeEach, describe, expect, it } from "vitest"; + +const MOCK_PROVIDER = "mock-agent-provider"; + +const MODEL: ModelConfig = { + model_id: "mock/agent-1", + provider: MOCK_PROVIDER, + title: "", + description: "", + capabilities: ["tool-use"], + provider_config: {}, + metadata: {}, +}; + +// ======================================================================== +// A scripted model: one entry per round, the last repeating forever +// ======================================================================== + +interface ScriptedRound { + readonly text?: string; + /** Report the text on the finish event only, as a non-streaming provider does. */ + readonly withoutDeltas?: boolean; + readonly calls?: ReadonlyArray<{ + readonly id: string; + readonly name: string; + readonly input: Record; + }>; +} + +function scriptModel(rounds: readonly ScriptedRound[]): () => number { + let called = 0; + const runFn: AiProviderRunFn = async (_input, _model, _signal, emit) => { + const round = rounds[Math.min(called, rounds.length - 1)]!; + called++; + if (round.text && !round.withoutDeltas) { + emit({ type: "text-delta", port: "text", textDelta: round.text }); + } + if (round.calls?.length) { + emit({ type: "object-delta", port: "toolCalls", objectDelta: [...round.calls] }); + } + emit({ type: "finish", data: round.withoutDeltas ? { text: round.text ?? "" } : {} }); + }; + getAiProviderRegistry().registerRunFn(MOCK_PROVIDER, { serves: ["tool-use"], runFn }); + return () => called; +} + +// ======================================================================== +// Two task-backed tools: one that reaches nothing, one that reaches the network +// ======================================================================== + +let echoRuns = 0; +let fetchRuns = 0; + +class AgentTest_EchoTask extends Task<{ text: string }, { echoed: string }> { + public static override type = "AgentTest_EchoTask"; + public static override description = "Echoes text back"; + public static override inputSchema(): DataPortSchema { + return { + type: "object", + properties: { text: { type: "string" } }, + required: ["text"], + additionalProperties: false, + } as const satisfies DataPortSchema; + } + public static override outputSchema(): DataPortSchema { + return { + type: "object", + properties: { echoed: { type: "string" } }, + additionalProperties: false, + } as const satisfies DataPortSchema; + } + override async execute(input: { text: string }) { + echoRuns++; + return { echoed: input.text.toUpperCase() }; + } +} + +class AgentTest_FetchTask extends Task<{ url: string }, { body: string }> { + public static override type = "AgentTest_FetchTask"; + public static override description = "Fetches a URL"; + public static override entitlements(): TaskEntitlements { + return { + entitlements: [{ id: Entitlements.NETWORK_HTTP, reason: "Fetches data from URLs" }], + }; + } + public static override inputSchema(): DataPortSchema { + return { + type: "object", + properties: { url: { type: "string" } }, + required: ["url"], + additionalProperties: false, + } as const satisfies DataPortSchema; + } + public static override outputSchema(): DataPortSchema { + return { + type: "object", + properties: { body: { type: "string" } }, + additionalProperties: false, + } as const satisfies DataPortSchema; + } + override async execute(input: { url: string }) { + fetchRuns++; + return { body: `body of ${input.url}` }; + } +} + +const ECHO_TOOL: ToolDefinition = { + name: "AgentTest_EchoTask", + description: "Echoes text back", + inputSchema: { + type: "object", + properties: { text: { type: "string" } }, + required: ["text"], + }, +}; + +const FETCH_TOOL: ToolDefinition = { + name: "AgentTest_FetchTask", + description: "Fetches a URL", + inputSchema: { + type: "object", + properties: { url: { type: "string" } }, + required: ["url"], + }, +}; + +function connector(handler: (request: IHumanRequest) => IHumanResponse): { + readonly connector: IHumanConnector; + readonly requests: IHumanRequest[]; +} { + const requests: IHumanRequest[] = []; + return { + requests, + connector: { + async send(request: IHumanRequest): Promise { + requests.push(request); + return handler(request); + }, + }, + }; +} + +function toolResults(messages: readonly { role: string; content: readonly unknown[] }[]) { + return messages + .filter((message) => message.role === "tool") + .flatMap((message) => message.content as ReadonlyArray>); +} + +describe("AgentTask", () => { + let registry: ServiceRegistry; + + beforeEach(() => { + echoRuns = 0; + fetchRuns = 0; + setAiProviderRegistry(new AiProviderRegistry()); + getAiProviderRegistry().setDefaultStrategy(new DirectExecutionStrategy()); + registry = new ServiceRegistry(new Container()); + TaskRegistry.registerTask(AgentTest_EchoTask); + TaskRegistry.registerTask(AgentTest_FetchTask); + }); + + afterEach(() => { + TaskRegistry.unregisterTask(AgentTest_EchoTask.type); + TaskRegistry.unregisterTask(AgentTest_FetchTask.type); + getAiProviderRegistry().unregisterProvider(MOCK_PROVIDER); + }); + + it("runs the tool the model asks for and feeds the result back", async () => { + const called = scriptModel([ + { + text: "Looking. ", + calls: [{ id: "c1", name: "AgentTest_EchoTask", input: { text: "hi" } }], + }, + { text: "It said HI." }, + ]); + + const output = await new AgentTask().run( + { model: MODEL, prompt: "echo hi", tools: [ECHO_TOOL], approval: "never" }, + { registry } + ); + + expect(called()).toBe(2); + expect(echoRuns).toBe(1); + expect(output.rounds).toBe(2); + expect(output.stopReason).toBe("answered"); + // The turn's whole narration, not only its last round. + expect(output.text).toBe("Looking. It said HI."); + expect(output.messages.map((message) => message.role)).toEqual([ + "user", + "assistant", + "tool", + "assistant", + ]); + const [result] = toolResults(output.messages); + expect(result).toMatchObject({ type: "tool_result", tool_use_id: "c1", is_error: undefined }); + expect(JSON.stringify(result)).toContain("HI"); + }); + + it("streams the model's text to a subscriber as it arrives, across rounds", async () => { + scriptModel([ + { + text: "Looking. ", + calls: [{ id: "c1", name: "AgentTest_EchoTask", input: { text: "hi" } }], + }, + { text: "It said HI." }, + ]); + + const task = new AgentTask(); + const deltas: string[] = []; + task.subscribe("stream_chunk", (event) => { + if (event.type === "text-delta") deltas.push(event.textDelta); + }); + await task.run( + { model: MODEL, prompt: "echo hi", tools: [ECHO_TOOL], approval: "never" }, + { registry } + ); + + // Both rounds reach the caller, in order — not one payload at the end. + expect(deltas).toEqual(["Looking. ", "It said HI."]); + }); + + it("reports the answer of a provider that streams nothing", async () => { + scriptModel([{ text: "All at once.", withoutDeltas: true }]); + + const output = await new AgentTask().run( + { model: MODEL, prompt: "hi", tools: [ECHO_TOOL], approval: "never" }, + { registry } + ); + + expect(output.text).toBe("All at once."); + expect(output.messages.at(-1)).toMatchObject({ role: "assistant" }); + }); + + it("answers an unknown tool rather than dropping the call", async () => { + scriptModel([ + { calls: [{ id: "c1", name: "no_such_tool", input: {} }] }, + { text: "Sorry about that." }, + ]); + + const output = await new AgentTask().run( + { model: MODEL, prompt: "go", tools: [ECHO_TOOL], approval: "never" }, + { registry } + ); + + // Every tool_use is answered: a provider rejects the next round otherwise. + const uses = output.messages.flatMap((message) => + message.content.filter((block) => block.type === "tool_use") + ); + const results = toolResults(output.messages); + expect(uses).toHaveLength(1); + expect(results).toHaveLength(1); + expect(results[0]).toMatchObject({ tool_use_id: "c1", is_error: true }); + expect(JSON.stringify(results[0])).toContain("Unknown tool"); + }); + + it("answers arguments that fail the tool's own schema", async () => { + scriptModel([ + { calls: [{ id: "c1", name: "AgentTest_EchoTask", input: { text: 42 } }] }, + { text: "Fixed." }, + ]); + + const output = await new AgentTask().run( + { model: MODEL, prompt: "go", tools: [ECHO_TOOL], approval: "never" }, + { registry } + ); + + expect(echoRuns).toBe(0); + expect(toolResults(output.messages)[0]).toMatchObject({ tool_use_id: "c1", is_error: true }); + }); + + it("records no assistant message for a round that said nothing", async () => { + scriptModel([{}]); + + const output = await new AgentTask().run( + { model: MODEL, prompt: "silence", tools: [ECHO_TOOL], approval: "never" }, + { registry } + ); + + expect(output.stopReason).toBe("answered"); + expect(output.text).toBe(""); + // An empty assistant message is not a reply, and a provider replaying this + // history as a prefix rejects one. + expect(output.messages.map((message) => message.role)).toEqual(["user"]); + }); + + it("stops at maxRounds when the model never stops calling tools", async () => { + const called = scriptModel([ + { calls: [{ id: "c", name: "AgentTest_EchoTask", input: { text: "again" } }] }, + ]); + + const output = await new AgentTask().run( + { model: MODEL, prompt: "loop", tools: [ECHO_TOOL], maxRounds: 3, approval: "never" }, + { registry } + ); + + expect(called()).toBe(3); + expect(output.rounds).toBe(3); + expect(output.stopReason).toBe("max-rounds"); + }); + + it("renames a tool-call id the conversation already used", async () => { + scriptModel([ + { calls: [{ id: "call_0", name: "AgentTest_EchoTask", input: { text: "now" } }] }, + { text: "done" }, + ]); + + const output = await new AgentTask().run( + { + model: MODEL, + prompt: "again", + tools: [ECHO_TOOL], + approval: "never", + messages: [ + { role: "user", content: [{ type: "text", text: "earlier" }] }, + { + role: "assistant", + content: [{ type: "tool_use", id: "call_0", name: "AgentTest_EchoTask", input: {} }], + }, + { + role: "tool", + content: [ + { type: "tool_result", tool_use_id: "call_0", content: [], is_error: undefined }, + ], + }, + ], + }, + { registry } + ); + + const results = toolResults(output.messages); + expect(results).toHaveLength(2); + // The new call was renamed off the old one, and its result followed it. + expect(results[1]).not.toMatchObject({ tool_use_id: "call_0" }); + const renamed = (results[1] as { tool_use_id: string }).tool_use_id; + const uses = output.messages + .flatMap((message) => message.content) + .filter((block) => block.type === "tool_use") + .map((block) => (block as { id: string }).id); + expect(uses).toEqual(["call_0", renamed]); + }); + + describe("approval", () => { + it("confirms a tool that reaches beyond running a model", async () => { + scriptModel([ + { + calls: [ + { id: "c1", name: "AgentTest_FetchTask", input: { url: "https://example.test" } }, + ], + }, + { text: "got it" }, + ]); + const human = connector((request) => ({ + requestId: request.requestId, + action: "accept", + content: undefined, + done: true, + })); + registry.registerInstance(HUMAN_CONNECTOR, human.connector); + + await new AgentTask().run( + { model: MODEL, prompt: "fetch", tools: [FETCH_TOOL] }, + { registry } + ); + + expect(fetchRuns).toBe(1); + expect(human.requests).toHaveLength(1); + expect(human.requests[0]!.kind).toBe("confirm"); + // The card says where it reaches and with what, not just which tool. + expect(human.requests[0]!.contentData).toMatchObject({ tool: "AgentTest_FetchTask" }); + expect(String(human.requests[0]!.contentData?.reach)).toContain("network:http"); + expect(String(human.requests[0]!.contentData?.arguments)).toContain("example.test"); + }); + + it("does not confirm a tool that reaches nothing beyond it", async () => { + scriptModel([ + { calls: [{ id: "c1", name: "AgentTest_EchoTask", input: { text: "hi" } }] }, + { text: "done" }, + ]); + const human = connector((request) => ({ + requestId: request.requestId, + action: "accept", + content: undefined, + done: true, + })); + registry.registerInstance(HUMAN_CONNECTOR, human.connector); + + await new AgentTask().run({ model: MODEL, prompt: "echo", tools: [ECHO_TOOL] }, { registry }); + + expect(echoRuns).toBe(1); + expect(human.requests).toHaveLength(0); + }); + + it("does not run a declined tool, and tells the model why", async () => { + scriptModel([ + { + calls: [ + { id: "c1", name: "AgentTest_FetchTask", input: { url: "https://example.test" } }, + ], + }, + { text: "understood" }, + ]); + const human = connector((request) => ({ + requestId: request.requestId, + action: "decline", + content: undefined, + done: true, + })); + registry.registerInstance(HUMAN_CONNECTOR, human.connector); + + const output = await new AgentTask().run( + { model: MODEL, prompt: "fetch", tools: [FETCH_TOOL] }, + { registry } + ); + + expect(fetchRuns).toBe(0); + const result = toolResults(output.messages)[0]; + expect(result).toMatchObject({ tool_use_id: "c1", is_error: true }); + expect(JSON.stringify(result)).toContain("did not approve"); + }); + + it("refuses rather than runs when there is nobody to ask", async () => { + scriptModel([ + { + calls: [ + { id: "c1", name: "AgentTest_FetchTask", input: { url: "https://example.test" } }, + ], + }, + { text: "ok" }, + ]); + + const output = await new AgentTask().run( + { model: MODEL, prompt: "fetch", tools: [FETCH_TOOL] }, + { registry } + ); + + expect(fetchRuns).toBe(0); + expect(JSON.stringify(toolResults(output.messages)[0])).toContain("no way to ask"); + }); + + it('runs an unapproved-reach tool when approval is "never"', async () => { + scriptModel([ + { + calls: [ + { id: "c1", name: "AgentTest_FetchTask", input: { url: "https://example.test" } }, + ], + }, + { text: "ok" }, + ]); + + await new AgentTask().run( + { model: MODEL, prompt: "fetch", tools: [FETCH_TOOL], approval: "never" }, + { registry } + ); + + expect(fetchRuns).toBe(1); + }); + + it("honours requiresApproval in both directions", async () => { + scriptModel([ + { calls: [{ id: "c1", name: "AgentTest_EchoTask", input: { text: "hi" } }] }, + { text: "done" }, + ]); + const human = connector((request) => ({ + requestId: request.requestId, + action: "accept", + content: undefined, + done: true, + })); + registry.registerInstance(HUMAN_CONNECTOR, human.connector); + + await new AgentTask().run( + { + model: MODEL, + prompt: "echo", + tools: [{ ...ECHO_TOOL, requiresApproval: true }], + }, + { registry } + ); + expect(human.requests).toHaveLength(1); + + await new AgentTask().run( + { + model: MODEL, + prompt: "fetch", + tools: [{ ...FETCH_TOOL, requiresApproval: false }], + }, + { registry } + ); + expect(human.requests).toHaveLength(1); + }); + }); + + it("resolves a renamed tool through its taskType", async () => { + scriptModel([ + { calls: [{ id: "c1", name: "shout", input: { text: "hi" } }] }, + { text: "done" }, + ]); + + await new AgentTask().run( + { + model: MODEL, + prompt: "shout", + tools: [{ ...ECHO_TOOL, name: "shout", taskType: AgentTest_EchoTask.type }], + approval: "never", + }, + { registry } + ); + + expect(echoRuns).toBe(1); + }); + + it("reaches the human connector through a tool that asks a person", async () => { + scriptModel([ + { calls: [{ id: "c1", name: "HumanInputTask", input: { prompt: "What is your name?" } }] }, + { text: "Thanks, Alice." }, + ]); + const human = connector((request) => ({ + requestId: request.requestId, + action: "accept", + content: { name: "Alice" }, + done: true, + })); + registry.registerInstance(HUMAN_CONNECTOR, human.connector); + TaskRegistry.registerTask(HumanInputTask); + + try { + const output = await new AgentTask().run( + { + model: MODEL, + prompt: "ask my name", + tools: [ + { + name: "HumanInputTask", + description: "Asks the user something", + inputSchema: { type: "object", properties: { prompt: { type: "string" } } }, + config: { + contentSchema: { + type: "object", + properties: { name: { type: "string" } }, + additionalProperties: false, + }, + }, + }, + ], + }, + { registry } + ); + + // No new vocabulary: the tool asked through the same IHumanConnector a + // workflow would, and its answer came back as an ordinary tool result. + const elicits = human.requests.filter((request) => request.kind === "elicit"); + expect(elicits).toHaveLength(1); + expect(elicits[0]!.message).toBe("What is your name?"); + // The form came from the tool's `config`, which the model never sees. + expect(elicits[0]!.contentSchema).toMatchObject({ properties: { name: { type: "string" } } }); + expect(JSON.stringify(toolResults(output.messages)[0])).toContain("Alice"); + } finally { + TaskRegistry.unregisterTask(HumanInputTask.type); + } + }); + + it("runs a host function tool without a task behind it", async () => { + let seen: Record | undefined; + scriptModel([{ calls: [{ id: "c1", name: "add", input: { a: 2, b: 3 } }] }, { text: "five" }]); + + const output = await new AgentTask().run( + { + model: MODEL, + prompt: "add", + tools: [ + { + name: "add", + description: "Adds two numbers", + inputSchema: { + type: "object", + properties: { a: { type: "number" }, b: { type: "number" } }, + }, + execute: async (input) => { + seen = input; + return { sum: (input.a as number) + (input.b as number) }; + }, + }, + ], + }, + { registry } + ); + + expect(seen).toEqual({ a: 2, b: 3 }); + expect(JSON.stringify(toolResults(output.messages)[0])).toContain("5"); + }); +}); From 6e77914113f8d40a8a7bb79bda5c307a8d93399d Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 9 Sep 2026 20:31:39 +0000 Subject: [PATCH 2/6] feat(cli): `workglow agent chat`, the first consumer of AgentTask MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A line-based REPL: read a line, run one AgentTask turn, carry `messages` from that turn's output into the next turn's input. That is the whole of the loop's state here — nothing in the CLI knows what a tool result or a tool-call id looks like. Streamed to stdout rather than through the Ink run UI. `withCli` clears the frame on completion and calls process.exit on failure, both right for a command whose life is one graph run and both fatal to a conversation: the answer would vanish when the turn ended, and one bad turn would end the session. Text written to stdout stays in scrollback, which is what a transcript is. Three supporting pieces: - `PromptHumanConnector` — the connector for anything prompting BETWEEN runs rather than during one. `InkHumanConnector` hands a request to a mounted HumanInteractionHost and throws when there is none, so nothing outside a graph run could ask a person anything. This draws its own prompt and gives the screen back. It reads `humanPromptModel`, the same function the Ink panel and the web console read, so the three cannot disagree about whether a request is a form or an approval — the case that matters, since a confirm drawn as a form leaves no way to say no. No `followUp`: a modal prompt settles the question it asked, so no response is ever `done: false`. The conformance suite checks that a multiTurn:false connector does not carry one. - `--tools` has no default. What an agent may call decides what it can reach, and inheriting a set nobody chose is how a chat session ends up able to write files. - A turn's readline interface lives only as long as its question. A long-lived one keeps listeners on stdin while an approval prompt renders its own app over the same terminal, and two readers of one stdin drop keystrokes into whichever happens to be listening. Also here, both found by building on them: - `TaskRunApp` read `event.text` off a stream chunk; the field is `textDelta`, so the streaming-output panel has been rendering nothing. The local type was loose enough to hide it. - A connector that throws no longer ends an agent turn. Approval that cannot be asked for refuses the call and tells the model, rather than killing a conversation over one tool. The tool still does not run; an abort still propagates. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01JEFYAGb9D3mWfyhYkmeAvN --- .claude/CLAUDE.md | 12 ++ examples/cli/src/agent/chatCommands.ts | 47 +++++ examples/cli/src/agent/chatTranscript.ts | 47 +++++ examples/cli/src/agent/runAgentChat.ts | 163 ++++++++++++++++++ examples/cli/src/commands/agent.ts | 85 +++++++++ examples/cli/src/human.ts | 2 + examples/cli/src/test/agentChat.test.ts | 149 ++++++++++++++++ examples/cli/src/test/chatCommands.test.ts | 79 +++++++++ examples/cli/src/ui/PromptHumanConnector.ts | 119 +++++++++++++ examples/cli/src/ui/TaskRunApp.tsx | 9 +- packages/ai/src/task/AgentToolExecution.ts | 66 +++++-- packages/test/src/test/ai/AgentTask.test.ts | 27 +++ .../PromptHumanConnector.conformance.test.ts | 72 ++++++++ 13 files changed, 856 insertions(+), 21 deletions(-) create mode 100644 examples/cli/src/agent/chatCommands.ts create mode 100644 examples/cli/src/agent/chatTranscript.ts create mode 100644 examples/cli/src/agent/runAgentChat.ts create mode 100644 examples/cli/src/test/agentChat.test.ts create mode 100644 examples/cli/src/test/chatCommands.test.ts create mode 100644 examples/cli/src/ui/PromptHumanConnector.ts create mode 100644 packages/test/src/test/human/PromptHumanConnector.conformance.test.ts diff --git a/.claude/CLAUDE.md b/.claude/CLAUDE.md index 3c82fb40b..73a366306 100644 --- a/.claude/CLAUDE.md +++ b/.claude/CLAUDE.md @@ -413,6 +413,18 @@ terminal runs. Three load-bearing properties: `timeline`, `markdown`, `empty` and `error`; a status widget contributes meters **or** text lines, since most of what an operator checks has no denominator to draw a bar against. +`workglow agent chat` is the terminal consumer of `AgentTask`: a line-based REPL that +streams the model's reply straight to stdout — so the transcript stays in scrollback, which +the Ink run UI's clear-on-complete would erase — prints one row per tool call off the task's +own progress messages, and carries `messages` from one turn's output into the next's input. +`--tools` takes task type names and **has no default**: what an agent may call decides what it +can reach. Approvals go through `PromptHumanConnector`, the connector for anything prompting +BETWEEN runs rather than during one: `InkHumanConnector` needs a mounted +`HumanInteractionHost` and throws without one, while this draws its own prompt and gives the +screen back. It has no `followUp` — a modal prompt settles the question it asked — and +decides form-vs-approval through the same `humanPromptModel` the Ink panel and the console +read. + `workglow mcp serve` is the second server the CLI hosts: the registered tasks offered to MCP clients as tools, one per task type, named for the registered type itself (`task list` prints the same types with the `Task` suffix trimmed) and carrying the diff --git a/examples/cli/src/agent/chatCommands.ts b/examples/cli/src/agent/chatCommands.ts new file mode 100644 index 000000000..aed914d2d --- /dev/null +++ b/examples/cli/src/agent/chatCommands.ts @@ -0,0 +1,47 @@ +/** + * @license + * Copyright 2026 Steven Roussey + * SPDX-License-Identifier: Apache-2.0 + */ + +/** + * What a typed line means before the model sees it. + * + * Only lines that are exactly a known command are one: a message beginning + * "/exit is the command that..." is a message, and a model that never receives + * it because of a prefix match is a chat that silently eats input. + */ +export type ChatLineIntent = + | { readonly kind: "blank" } + | { readonly kind: "quit" } + | { readonly kind: "reset" } + | { readonly kind: "help" } + | { readonly kind: "unknown-command"; readonly typed: string } + | { readonly kind: "message"; readonly text: string }; + +const COMMANDS: ReadonlyMap = new Map([ + ["/exit", "quit"], + ["/quit", "quit"], + ["/reset", "reset"], + ["/help", "help"], +]); + +export function classifyChatLine(line: string): ChatLineIntent { + const text = line.trim(); + if (text.length === 0) return { kind: "blank" }; + if (!text.startsWith("/")) return { kind: "message", text }; + const known = COMMANDS.get(text.toLowerCase()); + if (known === "quit") return { kind: "quit" }; + if (known === "reset") return { kind: "reset" }; + if (known === "help") return { kind: "help" }; + // A lone word starting with "/" is a mistyped command far more often than a + // message; anything longer is prose that happens to open with a slash. + if (/^\/\S*$/.test(text)) return { kind: "unknown-command", typed: text }; + return { kind: "message", text }; +} + +export const CHAT_HELP_LINES: readonly string[] = [ + "/exit, /quit end the session", + "/reset forget the conversation so far", + "/help this list", +]; diff --git a/examples/cli/src/agent/chatTranscript.ts b/examples/cli/src/agent/chatTranscript.ts new file mode 100644 index 000000000..0d02478fc --- /dev/null +++ b/examples/cli/src/agent/chatTranscript.ts @@ -0,0 +1,47 @@ +/** + * @license + * Copyright 2026 Steven Roussey + * SPDX-License-Identifier: Apache-2.0 + */ + +/** + * Where a chat turn is written, and the one thing that is easy to get wrong + * about it: a model's text arrives as deltas that rarely end on a line break, + * and everything else the turn prints — a tool line, an approval, the next + * prompt — has to start at column 0 or it lands in the middle of a sentence. + * + * Kept apart from the loop so the rule can be read, and tested, without a + * terminal. + */ +export interface ChatTranscript { + /** Model text, exactly as it arrived. */ + delta(text: string): void; + /** A line about the run rather than from the model. */ + note(line: string): void; + /** Close the turn, leaving the cursor at column 0. */ + endTurn(): void; +} + +export function createChatTranscript(write: (text: string) => void): ChatTranscript { + let atLineStart = true; + const breakLine = (): void => { + if (!atLineStart) { + write("\n"); + atLineStart = true; + } + }; + return { + delta(text) { + if (text.length === 0) return; + write(text); + atLineStart = text.endsWith("\n"); + }, + note(line) { + breakLine(); + write(`${line}\n`); + }, + endTurn() { + breakLine(); + }, + }; +} diff --git a/examples/cli/src/agent/runAgentChat.ts b/examples/cli/src/agent/runAgentChat.ts new file mode 100644 index 000000000..4ed7531cf --- /dev/null +++ b/examples/cli/src/agent/runAgentChat.ts @@ -0,0 +1,163 @@ +/** + * @license + * Copyright 2026 Steven Roussey + * SPDX-License-Identifier: Apache-2.0 + */ + +import type { AgentApprovalMode, ChatMessage, ToolDefinition } from "@workglow/ai"; +import { AgentTask } from "@workglow/ai"; +import type { StreamEvent } from "@workglow/task-graph"; +import { globalServiceRegistry, HUMAN_CONNECTOR, ServiceRegistry } from "@workglow/util"; +import { createInterface } from "node:readline/promises"; +import { formatError } from "../util"; +import { PromptHumanConnector } from "../ui/PromptHumanConnector"; +import { CHAT_HELP_LINES, classifyChatLine } from "./chatCommands"; +import { createChatTranscript } from "./chatTranscript"; + +/** + * The session's two ends, injectable so the loop can be driven without a + * terminal. Defaults are readline and stdout. + */ +export interface AgentChatIo { + /** One line from the person, or `undefined` at end of input. */ + readonly ask: (prompt: string) => Promise; + readonly write: (text: string) => void; +} + +export interface AgentChatOptions { + readonly model: string; + readonly tools: readonly ToolDefinition[]; + readonly systemPrompt: string | undefined; + readonly maxRounds: number | undefined; + readonly approval: AgentApprovalMode; +} + +/** + * One question, on a readline interface that exists only for the length of it. + * + * A long-lived interface keeps listeners on stdin, and the approval prompts + * this session raises render their own Ink app over the same terminal — two + * readers of one stdin is a session that drops keystrokes into whichever + * happens to be listening. Creating and closing per question means nothing but + * the prompt of the moment ever holds it. + */ +async function askLine(prompt: string): Promise { + const rl = createInterface({ input: process.stdin, output: process.stdout }); + try { + return await rl.question(prompt); + } catch { + // The interface closes on EOF (Ctrl-D), which rejects the pending question. + return undefined; + } finally { + rl.close(); + } +} + +/** A child of the host's registry, so the chat prompts without the Ink run UI. */ +function chatRegistry(parent: ServiceRegistry): ServiceRegistry { + const registry = new ServiceRegistry(parent.container.createChildContainer()); + registry.registerInstance(HUMAN_CONNECTOR, new PromptHumanConnector()); + return registry; +} + +/** + * The chat loop: read a line, run one {@link AgentTask} turn, carry the + * conversation forward. + * + * `messages` is this loop's only state — the task takes the history in and + * hands it back with the turn appended, so nothing here has to know what a + * tool result or a tool-call id looks like. + */ +export async function runAgentChat(options: AgentChatOptions, io?: AgentChatIo): Promise { + const ask = io?.ask ?? askLine; + const transcript = createChatTranscript(io?.write ?? ((text) => void process.stdout.write(text))); + const registry = chatRegistry(globalServiceRegistry); + let messages: ChatMessage[] = []; + + transcript.note( + options.tools.length === 0 + ? "No tools — this agent can talk, and nothing else. Pass --tools to give it some." + : `Tools: ${options.tools.map((tool) => tool.name).join(", ")}` + ); + transcript.note("/help for commands, /exit to leave."); + + for (;;) { + transcript.endTurn(); + const line = await ask("\n› "); + if (line === undefined) return; + const intent = classifyChatLine(line); + if (intent.kind === "quit") return; + if (intent.kind === "blank") continue; + if (intent.kind === "reset") { + messages = []; + transcript.note("Conversation cleared."); + continue; + } + if (intent.kind === "help") { + for (const help of CHAT_HELP_LINES) transcript.note(help); + continue; + } + if (intent.kind === "unknown-command") { + transcript.note(`Unknown command ${intent.typed}. /help for the list.`); + continue; + } + messages = await runTurn(intent.text, messages, options, registry, transcript); + } +} + +async function runTurn( + text: string, + messages: ChatMessage[], + options: AgentChatOptions, + registry: ServiceRegistry, + transcript: ReturnType +): Promise { + const task = new AgentTask(); + const controller = new AbortController(); + const onInterrupt = (): void => controller.abort(); + process.on("SIGINT", onInterrupt); + + let phase: string | undefined; + const offStream = task.subscribe("stream_chunk", (event: StreamEvent) => { + if (event.type === "text-delta") transcript.delta(event.textDelta); + }); + const offProgress = task.subscribe("progress", (_progress, message) => { + // Only the tool lines are worth a row: "Thinking" is what the blank space + // between the prompt and the first token already says. + if (!message || message === phase) return; + phase = message; + if (message.startsWith("Running ")) transcript.note(` · ${message.slice("Running ".length)}`); + }); + + try { + const output = await task.run( + { + model: options.model, + prompt: text, + messages, + tools: [...options.tools], + systemPrompt: options.systemPrompt, + maxRounds: options.maxRounds, + approval: options.approval, + }, + { registry, signal: controller.signal } + ); + if (output.stopReason === "max-rounds") { + transcript.note(` · stopped after ${output.rounds} rounds without an answer`); + } + return output.messages; + } catch (error) { + if (controller.signal.aborted) { + // The turn is gone but the conversation is not: an interrupted turn wrote + // nothing to `messages`, so the next one continues from where it was. + transcript.note(" · interrupted"); + return messages; + } + transcript.note(` · ${formatError(error)}`); + return messages; + } finally { + process.off("SIGINT", onInterrupt); + offStream(); + offProgress(); + } +} diff --git a/examples/cli/src/commands/agent.ts b/examples/cli/src/commands/agent.ts index e53ca0f8b..2aa48306d 100644 --- a/examples/cli/src/commands/agent.ts +++ b/examples/cli/src/commands/agent.ts @@ -4,6 +4,8 @@ * SPDX-License-Identifier: Apache-2.0 */ +import type { ToolDefinition } from "@workglow/ai"; +import { AgentTask, taskTypesToTools } from "@workglow/ai"; import { computeGraphInputSchema, createGraphFromGraphJSON, @@ -28,6 +30,7 @@ import { ensureCredentialStoreUnlocked } from "../keyring"; import { createAgentRepository } from "../storage"; import { renderSelectPrompt, renderWorkflowRun } from "../ui/render"; import { formatError, formatTable, outputResult } from "../util"; +import { runAgentChat } from "../agent/runAgentChat"; export function registerAgentCommand(program: Command): void { const agent = program.command("agent").description("Manage and run agents"); @@ -351,4 +354,86 @@ export function registerAgentCommand(program: Command): void { process.exit(1); } }); + + agent + .command("chat") + .description("Talk to a model that can run registered tasks as tools") + .option("-m, --model ", "Model to use; prompted for when omitted") + .option( + "-t, --tools ", + "Comma-separated task types the model may call. None by default — an agent reaches only what it is given." + ) + .option("-s, --system ", "System prompt") + .option("--max-rounds ", "Model calls per turn before it gives up", parseRounds) + .option( + "--no-approval", + "Run every tool without asking. For a session you are not watching; the default confirms anything reaching past the model." + ) + .action(async (opts: Record) => { + if (!process.stdin.isTTY) { + console.error( + "agent chat needs a terminal. Use `workglow task run AgentTask` for a scripted turn." + ); + process.exit(1); + } + let tools: ToolDefinition[]; + try { + tools = toolsFromTypes(opts.tools as string | undefined); + } catch (err) { + console.error(`Error: ${formatError(err)}`); + process.exit(1); + } + const model = await resolveChatModel(opts.model as string | undefined); + await ensureCredentialStoreUnlocked(); + await runAgentChat({ + model, + tools, + systemPrompt: opts.system as string | undefined, + maxRounds: opts.maxRounds as number | undefined, + // Commander gives `--no-approval` as `approval: false`. + approval: opts.approval === false ? "never" : "beyond-inference", + }); + }); +} + +function parseRounds(raw: string): number { + const rounds = Number.parseInt(raw, 10); + if (!Number.isFinite(rounds) || rounds < 1) { + throw new Error(`--max-rounds must be a positive integer, got "${raw}"`); + } + return rounds; +} + +/** + * The task types named on `--tools`, as tool definitions. + * + * No default, for the reason a pinned search provider has none: which tools an + * agent holds decides what it can reach, and inheriting a set nobody chose is + * how a chat session ends up able to write files. + */ +function toolsFromTypes(raw: string | undefined): ToolDefinition[] { + const names = (raw ?? "") + .split(",") + .map((name) => name.trim()) + .filter((name) => name.length > 0); + return names.length === 0 ? [] : taskTypesToTools(names); +} + +/** `--model`, or the same picker every other model port gets. */ +async function resolveChatModel(named: string | undefined): Promise { + if (named) return named; + // Only the model port, so the picker asks the one thing a session needs and + // does not walk the rest of the task's inputs. + const full = AgentTask.inputSchema() as DataPortSchemaObject; + const schema: DataPortSchemaObject = { + type: "object", + properties: { model: full.properties.model! }, + required: ["model"], + }; + const filled = await promptMissingInput({}, schema); + const model = filled.model; + if (typeof model !== "string" || model.length === 0) { + throw new Error("No model chosen"); + } + return model; } diff --git a/examples/cli/src/human.ts b/examples/cli/src/human.ts index 7bc212064..a040e669a 100644 --- a/examples/cli/src/human.ts +++ b/examples/cli/src/human.ts @@ -17,6 +17,8 @@ export type { AnswerReaderFactory } from "./run-events/RunEventHumanConnector"; export type { RunEvent } from "./run-events/RunEventTypes"; export type { RunEventSink } from "./run-events/runEventChannel"; export { InkHumanConnector } from "./ui/InkHumanConnector"; +export { PromptHumanConnector } from "./ui/PromptHumanConnector"; +export type { PromptHumanRenderers } from "./ui/PromptHumanConnector"; export { humanPromptModel } from "./ui/model/humanPrompt"; export type { HumanPromptDetail, diff --git a/examples/cli/src/test/agentChat.test.ts b/examples/cli/src/test/agentChat.test.ts new file mode 100644 index 000000000..c17de8ee3 --- /dev/null +++ b/examples/cli/src/test/agentChat.test.ts @@ -0,0 +1,149 @@ +/** + * @license + * Copyright 2026 Steven Roussey + * SPDX-License-Identifier: Apache-2.0 + */ + +import type { AiProviderRunFn, ModelConfig } from "@workglow/ai"; +import { + AiProviderRegistry, + DirectExecutionStrategy, + getAiProviderRegistry, + setAiProviderRegistry, +} from "@workglow/ai"; +import { afterEach, beforeEach, describe, expect, it } from "vitest"; +import { runAgentChat, type AgentChatIo } from "../agent/runAgentChat"; + +const PROVIDER = "mock-chat-provider"; + +const MODEL: ModelConfig = { + model_id: "mock/chat-1", + provider: PROVIDER, + title: "", + description: "", + capabilities: ["tool-use"], + provider_config: {}, + metadata: {}, +}; + +/** Answers each turn with the next scripted reply; the last repeats. */ +function scriptModel(replies: readonly string[]): () => number { + let called = 0; + const runFn: AiProviderRunFn = async (_input, _model, _signal, emit) => { + const reply = replies[Math.min(called, replies.length - 1)]!; + called++; + emit({ type: "text-delta", port: "text", textDelta: reply }); + emit({ type: "finish", data: {} }); + }; + getAiProviderRegistry().registerRunFn(PROVIDER, { serves: ["tool-use"], runFn }); + return () => called; +} + +/** How many messages the model was handed, per turn. */ +function historySizes(): { readonly sizes: number[]; readonly runFn: AiProviderRunFn } { + const sizes: number[] = []; + const runFn: AiProviderRunFn = async (input, _model, _signal, emit) => { + sizes.push(((input as { messages?: unknown[] }).messages ?? []).length); + emit({ type: "text-delta", port: "text", textDelta: "ok" }); + emit({ type: "finish", data: {} }); + }; + return { sizes, runFn }; +} + +function scriptedIo(lines: readonly string[]): { io: AgentChatIo; out: () => string } { + let index = 0; + const written: string[] = []; + return { + io: { + ask: async () => lines[index++], + write: (text) => written.push(text), + }, + out: () => written.join(""), + }; +} + +describe("agent chat loop", () => { + beforeEach(() => { + setAiProviderRegistry(new AiProviderRegistry()); + getAiProviderRegistry().setDefaultStrategy(new DirectExecutionStrategy()); + }); + + afterEach(() => { + getAiProviderRegistry().unregisterProvider(PROVIDER); + }); + + const baseOptions = { + model: MODEL as unknown as string, + tools: [], + systemPrompt: undefined, + maxRounds: undefined, + approval: "never" as const, + }; + + it("runs a turn per message and prints the reply", async () => { + const called = scriptModel(["Hello back."]); + const { io, out } = scriptedIo(["hello", "/exit"]); + + await runAgentChat(baseOptions, io); + + expect(called()).toBe(1); + expect(out()).toContain("Hello back."); + }); + + it("ends at end of input as well as on /exit", async () => { + scriptModel(["…"]); + const { io } = scriptedIo(["hello", undefined as unknown as string]); + await expect(runAgentChat(baseOptions, io)).resolves.toBeUndefined(); + }); + + it("carries the conversation forward, and /reset drops it", async () => { + const { sizes, runFn } = historySizes(); + getAiProviderRegistry().registerRunFn(PROVIDER, { serves: ["tool-use"], runFn }); + const { io } = scriptedIo(["one", "two", "/reset", "three", "/exit"]); + + await runAgentChat(baseOptions, io); + + // Turn 1 sends just the user message; turn 2 sends that turn plus this + // one's; after /reset turn 3 is back to a single message. + expect(sizes).toEqual([1, 3, 1]); + }); + + it("says what it can reach, and that it reaches nothing without --tools", async () => { + scriptModel(["hi"]); + const { io, out } = scriptedIo(["/exit"]); + + await runAgentChat(baseOptions, io); + + expect(out()).toContain("No tools"); + }); + + it("keeps the session alive when a turn fails, and keeps the history", async () => { + let calls = 0; + const runFn: AiProviderRunFn = async (input, _model, _signal, emit) => { + calls++; + if (calls === 1) throw new Error("provider exploded"); + const messages = (input as { messages?: unknown[] }).messages ?? []; + emit({ type: "text-delta", port: "text", textDelta: `saw ${messages.length}` }); + emit({ type: "finish", data: {} }); + }; + getAiProviderRegistry().registerRunFn(PROVIDER, { serves: ["tool-use"], runFn }); + const { io, out } = scriptedIo(["boom", "again", "/exit"]); + + await runAgentChat(baseOptions, io); + + expect(out()).toContain("provider exploded"); + // A failed turn recorded nothing, so the next one is still the first + // message the model ever sees. + expect(out()).toContain("saw 1"); + }); + + it("answers /help without calling the model", async () => { + const called = scriptModel(["never"]); + const { io, out } = scriptedIo(["/help", "/exit"]); + + await runAgentChat(baseOptions, io); + + expect(called()).toBe(0); + expect(out()).toContain("/reset"); + }); +}); diff --git a/examples/cli/src/test/chatCommands.test.ts b/examples/cli/src/test/chatCommands.test.ts new file mode 100644 index 000000000..b76b292a4 --- /dev/null +++ b/examples/cli/src/test/chatCommands.test.ts @@ -0,0 +1,79 @@ +/** + * @license + * Copyright 2026 Steven Roussey + * SPDX-License-Identifier: Apache-2.0 + */ + +import { describe, expect, it } from "vitest"; +import { classifyChatLine } from "../agent/chatCommands"; +import { createChatTranscript } from "../agent/chatTranscript"; + +describe("classifyChatLine", () => { + it("reads the commands it knows, in any case, around whitespace", () => { + expect(classifyChatLine("/exit").kind).toBe("quit"); + expect(classifyChatLine(" /QUIT ").kind).toBe("quit"); + expect(classifyChatLine("/reset").kind).toBe("reset"); + expect(classifyChatLine("/help").kind).toBe("help"); + }); + + it("sends prose that merely starts with a command to the model", () => { + // A prefix match here is a chat that silently eats a message. + const intent = classifyChatLine("/exit is the command that ends this, right?"); + expect(intent).toEqual({ + kind: "message", + text: "/exit is the command that ends this, right?", + }); + }); + + it("calls a lone unknown slash-word a mistyped command, not a message", () => { + expect(classifyChatLine("/resett")).toEqual({ kind: "unknown-command", typed: "/resett" }); + }); + + it("treats an empty line as nothing to send", () => { + expect(classifyChatLine(" ").kind).toBe("blank"); + }); +}); + +describe("createChatTranscript", () => { + function capture() { + const written: string[] = []; + return { written, transcript: createChatTranscript((text) => written.push(text)) }; + } + + it("starts a note on its own line when the model stopped mid-sentence", () => { + const { written, transcript } = capture(); + transcript.delta("Looking that up"); + transcript.note(" · FetchUrlTask"); + expect(written.join("")).toBe("Looking that up\n · FetchUrlTask\n"); + }); + + it("does not open a blank line when the model already ended one", () => { + const { written, transcript } = capture(); + transcript.delta("Done.\n"); + transcript.note(" · one round"); + expect(written.join("")).toBe("Done.\n · one round\n"); + }); + + it("writes deltas through untouched", () => { + const { written, transcript } = capture(); + transcript.delta("Hel"); + transcript.delta("lo"); + expect(written).toEqual(["Hel", "lo"]); + }); + + it("closes a turn onto column 0, and only when it is not already there", () => { + const { written, transcript } = capture(); + transcript.delta("tail"); + transcript.endTurn(); + transcript.endTurn(); + expect(written.join("")).toBe("tail\n"); + }); + + it("ignores an empty delta rather than counting it as content", () => { + const { written, transcript } = capture(); + transcript.delta("x\n"); + transcript.delta(""); + transcript.note("after"); + expect(written.join("")).toBe("x\nafter\n"); + }); +}); diff --git a/examples/cli/src/ui/PromptHumanConnector.ts b/examples/cli/src/ui/PromptHumanConnector.ts new file mode 100644 index 000000000..d7a84bd48 --- /dev/null +++ b/examples/cli/src/ui/PromptHumanConnector.ts @@ -0,0 +1,119 @@ +/** + * @license + * Copyright 2026 Steven Roussey + * SPDX-License-Identifier: Apache-2.0 + */ + +import type { IHumanConnector, IHumanRequest, IHumanResponse } from "@workglow/util"; +import { prepareSchemaFormFields, type PromptFieldDescriptor } from "../input/prompt"; +import { asDataPortSchemaObject } from "./humanSchema"; +import { humanPromptModel } from "./model/humanPrompt"; +import { renderSchemaPrompt, renderSelectPrompt } from "./render"; + +/** + * The prompts this connector draws, injectable so the mapping can be exercised + * without a terminal — and so a host with its own prompt stack can supply it. + */ +export interface PromptHumanRenderers { + readonly select: ( + options: ReadonlyArray<{ label: string; value: string }>, + message?: string + ) => Promise; + readonly form: ( + fields: readonly PromptFieldDescriptor[] + ) => Promise | undefined>; + readonly notice: (lines: readonly string[]) => void; +} + +const defaultRenderers: PromptHumanRenderers = { + select: (options, message) => renderSelectPrompt([...options], message), + form: (fields) => renderSchemaPrompt(fields), + notice: (lines) => { + for (const line of lines) console.log(line); + }, +}; + +/** Labels for the two decisions an approval offers, in the order it offers them. */ +const APPROVAL_OPTIONS = [ + { label: "Approve", value: "accept" }, + { label: "Decline", value: "decline" }, +] as const; + +/** + * The terminal connector for anything that is NOT inside the Ink run UI. + * + * {@link InkHumanConnector} hands a request to a mounted + * {@link HumanInteractionHost} and throws when there is none — right for a + * command whose whole life is one graph run, and useless to anything that + * prompts between runs rather than during one. This draws its own prompt for + * the length of the question and takes the screen back afterwards, so a + * transcript printed around it survives. + * + * What each request is asking is decided by {@link humanPromptModel}, the same + * function the Ink panel and the web console read, so the three cannot disagree + * about whether something is a form or an approval — the case that matters, + * since a confirm drawn as a form leaves a person no way to say no. + * + * There is deliberately no `followUp`: a modal prompt settles the question it + * asked, so no response is ever `done: false` and there is no partial answer for + * a follow-up to continue. Declaring one that just asked again would report a + * multi-turn conversation this surface cannot hold. + */ +export class PromptHumanConnector implements IHumanConnector { + private readonly renderers: PromptHumanRenderers; + + constructor(renderers: PromptHumanRenderers = defaultRenderers) { + this.renderers = renderers; + } + + async send(request: IHumanRequest, signal: AbortSignal): Promise { + signal.throwIfAborted(); + const model = humanPromptModel({ + kind: request.kind, + message: request.message, + schema: request.contentSchema, + data: request.contentData, + }); + const settle = ( + action: IHumanResponse["action"], + content?: Record + ): IHumanResponse => ({ + requestId: request.requestId, + action, + content: action === "accept" && model.carriesContent ? content : undefined, + done: true, + }); + + if (model.shape === "acknowledge") { + this.renderers.notice([ + `${model.title}: ${model.message}`, + ...model.details.map((detail) => ` ${detail.label}: ${detail.value}`), + ]); + return settle("accept"); + } + + if (model.shape === "approval") { + // Printed rather than offered as fields: the schema of a confirm + // describes the action awaiting approval, and drawing it as a form would + // turn a description into something to edit. + this.renderers.notice([ + model.title, + ...model.details.map((detail) => ` ${detail.label}: ${detail.value}`), + ]); + const chosen = await this.renderers.select(APPROVAL_OPTIONS, model.message); + // Walking away is not the same answer as refusing, and the caller acts + // differently on each. + if (chosen === undefined) return settle("cancel"); + return settle(chosen === "decline" ? "decline" : "accept"); + } + + const schema = asDataPortSchemaObject(request.contentSchema); + const fields = await prepareSchemaFormFields( + (request.contentData as Record | undefined) ?? {}, + schema + ); + const values = await this.renderers.form(fields); + if (values === undefined) return settle("cancel"); + return settle("accept", values); + } +} diff --git a/examples/cli/src/ui/TaskRunApp.tsx b/examples/cli/src/ui/TaskRunApp.tsx index 4911ac62b..d93e7711d 100644 --- a/examples/cli/src/ui/TaskRunApp.tsx +++ b/examples/cli/src/ui/TaskRunApp.tsx @@ -152,9 +152,12 @@ export function TaskRunApp({ if (msg) progressRef.current.msg = msg; }); - task.events.on("stream_chunk", (event: { type: string; text?: string }) => { - if (event.type === "text-delta" && event.text) { - setStreamText((prev) => prev + event.text); + // `textDelta`, not `text`: that is the field name on StreamTextDelta, and + // the loose local type here is why reading the wrong one showed as an + // empty panel rather than as a compile error. + task.events.on("stream_chunk", (event: { type: string; textDelta?: string }) => { + if (event.type === "text-delta" && event.textDelta) { + setStreamText((prev) => prev + event.textDelta); } }); diff --git a/packages/ai/src/task/AgentToolExecution.ts b/packages/ai/src/task/AgentToolExecution.ts index 7380f2edd..4b042d07c 100644 --- a/packages/ai/src/task/AgentToolExecution.ts +++ b/packages/ai/src/task/AgentToolExecution.ts @@ -10,7 +10,8 @@ import { getTaskConstructors, taskClassNeedsApproval, } from "@workglow/task-graph"; -import { HUMAN_CONNECTOR, uuid4 } from "@workglow/util"; +import type { IHumanRequest, IHumanResponse } from "@workglow/util"; +import { getLogger, HUMAN_CONNECTOR, uuid4 } from "@workglow/util"; import type { DataPortSchema } from "@workglow/util/schema"; import type { ToolCall, ToolDefinition } from "./ToolCallingUtils"; @@ -100,6 +101,30 @@ export function toolCallNeedsApproval( return taskClassNeedsApproval(ctor); } +/** + * Puts one request to the connector, reporting a failure to ask as `undefined` + * rather than letting it out. + * + * A connector that throws is a host wired wrong — the terminal one raises when + * no run UI is mounted, for instance — and letting that end the turn kills a + * whole conversation over one tool call. The caller refuses the call instead: + * the tool still does not run, and the model can say why. An abort is the one + * exception, since the run is over either way. + */ +async function askToApprove( + context: IExecuteContext, + toolName: string, + request: IHumanRequest +): Promise { + try { + return await context.registry.get(HUMAN_CONNECTOR).send(request, context.signal); + } catch (error) { + if (context.signal.aborted) throw error; + getLogger().warn(`Could not ask for approval of "${toolName}"`, { error }); + return undefined; + } +} + /** * Puts one tool call to a human and reports whether it may run. * @@ -121,24 +146,29 @@ async function approveToolCall( }; } const ctor = getTaskConstructors(context.registry).get(backingTaskType(tool)); - const response = await context.registry.get(HUMAN_CONNECTOR).send( - { - requestId: uuid4(), - targetHumanId: "default", - kind: "confirm", - message: `Run "${tool.name}"?`, - contentSchema: APPROVAL_SCHEMA as DataPortSchema, - contentData: { - tool: tool.name, - reach: ctor ? describeTaskClassReach(ctor) : "not declared — this tool is a host function", - arguments: clamp(stringifyForModel(call.input), MAX_APPROVAL_ARGUMENT_CHARS), - }, - expectsResponse: true, - mode: "single", - metadata: { toolUseId: call.id, toolName: tool.name }, + const response = await askToApprove(context, tool.name, { + requestId: uuid4(), + targetHumanId: "default", + kind: "confirm", + message: `Run "${tool.name}"?`, + contentSchema: APPROVAL_SCHEMA as DataPortSchema, + contentData: { + tool: tool.name, + reach: ctor ? describeTaskClassReach(ctor) : "not declared — this tool is a host function", + arguments: clamp(stringifyForModel(call.input), MAX_APPROVAL_ARGUMENT_CHARS), }, - context.signal - ); + expectsResponse: true, + mode: "single", + metadata: { toolUseId: call.id, toolName: tool.name }, + }); + if (response === undefined) { + return { + text: + `Running "${tool.name}" needs a person's approval and asking for one failed. ` + + `Tell the user, and do not try this tool again.`, + isError: true, + }; + } if (response.action === "accept") return undefined; // Declined and cancelled read the same way to a model — it did not happen and // repeating the request is not what the person wants next. diff --git a/packages/test/src/test/ai/AgentTask.test.ts b/packages/test/src/test/ai/AgentTask.test.ts index efaee60f8..cd5ea6d6f 100644 --- a/packages/test/src/test/ai/AgentTask.test.ts +++ b/packages/test/src/test/ai/AgentTask.test.ts @@ -438,6 +438,33 @@ describe("AgentTask", () => { expect(JSON.stringify(result)).toContain("did not approve"); }); + it("refuses rather than runs when asking itself fails", async () => { + scriptModel([ + { + calls: [ + { id: "c1", name: "AgentTest_FetchTask", input: { url: "https://example.test" } }, + ], + }, + { text: "ok" }, + ]); + // The terminal connector raises exactly like this when no run UI is + // mounted; one badly wired host must not end the conversation. + registry.registerInstance(HUMAN_CONNECTOR, { + send: async () => { + throw new Error("no run UI is mounted"); + }, + }); + + const output = await new AgentTask().run( + { model: MODEL, prompt: "fetch", tools: [FETCH_TOOL] }, + { registry } + ); + + expect(fetchRuns).toBe(0); + expect(output.stopReason).toBe("answered"); + expect(JSON.stringify(toolResults(output.messages)[0])).toContain("asking for one failed"); + }); + it("refuses rather than runs when there is nobody to ask", async () => { scriptModel([ { diff --git a/packages/test/src/test/human/PromptHumanConnector.conformance.test.ts b/packages/test/src/test/human/PromptHumanConnector.conformance.test.ts new file mode 100644 index 000000000..a481a391e --- /dev/null +++ b/packages/test/src/test/human/PromptHumanConnector.conformance.test.ts @@ -0,0 +1,72 @@ +/** + * @license + * Copyright 2026 Steven Roussey + * SPDX-License-Identifier: Apache-2.0 + */ + +import { PromptHumanConnector } from "@workglow/cli/human"; +import type { PromptHumanRenderers } from "@workglow/cli/human"; +import type { IHumanConnector, IHumanRequest, IHumanResponse } from "@workglow/util"; + +import { runHumanConnectorConformance } from "../../contract/human-connector/runHumanConnectorConformance"; +import { createCliHumanSurface } from "./cliHumanSurface"; + +/** + * The connector a command uses when it prompts BETWEEN runs rather than during + * one, answered through what its prompts actually offer. + * + * The renderers are bound per request rather than per connector: the suite runs + * several requests at once, and a single "current request" would answer them in + * whatever order they happened to interleave. The connector holds no state, so + * one per request costs nothing. + */ +function connectorFor( + surface: ReturnType, + request: IHumanRequest, + signal: AbortSignal +): PromptHumanConnector { + const answer = (): Promise => surface.answer(request, signal); + const renderers: PromptHumanRenderers = { + select: async () => { + const wanted = await answer(); + // Esc on the picker is how a person walks away; the two decisions it + // offers are the other two answers. + if (wanted.action === "cancel") return undefined; + return wanted.action; + }, + form: async () => { + const wanted = await answer(); + if (wanted.action !== "accept") return undefined; + return wanted.content ?? {}; + }, + notice: () => {}, + }; + return new PromptHumanConnector(renderers); +} + +runHumanConnectorConformance({ + name: "PromptHumanConnector", + timeout: 10_000, + factory: async () => { + const surface = createCliHumanSurface(); + // No `followUp`, matching the connector: the suite checks that a + // multiTurn:false connector does not carry one. + const connector: IHumanConnector = { + send: (request, signal) => connectorFor(surface, request, signal).send(request, signal), + }; + return { connector, script: surface.script, dispose: async () => {} }; + }, + capabilities: { + elicit: true, + confirm: true, + notify: true, + display: true, + multiTurn: false, + concurrent: true, + abortMidElicit: true, + }, + // The form offers submit and Esc, so a person can walk away from an elicit + // but cannot refuse it — the same gap the Ink panel has, and for the same + // reason: nothing yet turns on the difference for an elicit's caller. + expectedFailures: ["roundtrip.decline"], +}); From e89e686e846c3682dad5d32caee5959a22a4d632 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 9 Sep 2026 21:20:23 +0000 Subject: [PATCH 3/6] =?UTF-8?q?fix:=20three=20review=20findings=20?= =?UTF-8?q?=E2=80=94=20abort,=20an=20unbounded=20answer,=20a=20wrong=20lab?= =?UTF-8?q?el?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit All three were real. **PromptHumanConnector did not honour AbortSignal past its first line.** `throwIfAborted()` at the top of `send`, then an await on a prompt that has no way to be cancelled: Ctrl-C during an approval left the Ink app on screen and the promise pending forever. The connector now races the prompt against the signal and rejects, which is what `IHumanConnector` requires, and hands the signal down so the prompt tears its own app down rather than only being abandoned. A renderer resolving `undefined` on abort would not do: that is indistinguishable from Esc, which is a `cancel` a caller may act on. The conformance suite already claimed `abortMidElicit` and passed — on the stand-in, not on the connector. Its fake renderer answered through the mock connector, which honours abort itself, so the assertion never reached the code under test. The stand-in now gets a signal that never aborts, and removing the race fails the test at a 20-second timeout. **The unknown-tool answer was unbounded.** `Unknown tool "x". Available: …` names every registered tool and bypassed `maxToolResultChars`, so a large registry could fill the window — and re-fill it on every round the model kept guessing. Every path into a `tool_result` now goes through the same clamp, not just the one carrying a tool's output. **An unregistered task type was labelled a host function.** The approval card's reach fell back to "this tool is a host function" whenever no backing class was found, which is also what a misspelled or unregistered `taskType` looks like — a different situation, described wrongly, on the one card a person reads before approving. Three answers now, and the third names the type that is missing. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01JEFYAGb9D3mWfyhYkmeAvN --- examples/cli/src/ui/PromptHumanConnector.ts | 46 +++++++++++-- examples/cli/src/ui/render.ts | 33 +++++++++- packages/ai/src/task/AgentTask.ts | 36 +++++++++-- packages/ai/src/task/AgentToolExecution.ts | 40 ++++++++++-- packages/test/src/test/ai/AgentTask.test.ts | 64 +++++++++++++++++++ .../PromptHumanConnector.conformance.test.ts | 12 ++-- 6 files changed, 208 insertions(+), 23 deletions(-) diff --git a/examples/cli/src/ui/PromptHumanConnector.ts b/examples/cli/src/ui/PromptHumanConnector.ts index d7a84bd48..0586f4847 100644 --- a/examples/cli/src/ui/PromptHumanConnector.ts +++ b/examples/cli/src/ui/PromptHumanConnector.ts @@ -17,22 +17,53 @@ import { renderSchemaPrompt, renderSelectPrompt } from "./render"; export interface PromptHumanRenderers { readonly select: ( options: ReadonlyArray<{ label: string; value: string }>, - message?: string + message: string | undefined, + signal: AbortSignal ) => Promise; readonly form: ( - fields: readonly PromptFieldDescriptor[] + fields: readonly PromptFieldDescriptor[], + signal: AbortSignal ) => Promise | undefined>; readonly notice: (lines: readonly string[]) => void; } const defaultRenderers: PromptHumanRenderers = { - select: (options, message) => renderSelectPrompt([...options], message), - form: (fields) => renderSchemaPrompt(fields), + select: (options, message, signal) => renderSelectPrompt([...options], message, signal), + form: (fields, signal) => renderSchemaPrompt(fields, undefined, signal), notice: (lines) => { for (const line of lines) console.log(line); }, }; +/** + * Rejects when `signal` aborts, so a pending prompt does not outlive the run + * that asked the question. + * + * The renderer is handed the same signal and tears its own app down, but the + * rejection is raised here rather than left to it: `IHumanConnector` requires + * an aborted `send` to reject, and a renderer that resolves `undefined` on + * abort would be indistinguishable from a person pressing Esc — which is a + * `cancel` a caller may act on. + */ +function whenAborted(signal: AbortSignal): Promise { + return new Promise((_resolve, reject) => { + const fail = (): void => + reject(signal.reason ?? new DOMException("The operation was aborted", "AbortError")); + if (signal.aborted) { + fail(); + return; + } + signal.addEventListener("abort", fail, { once: true }); + }); +} + +async function untilAborted(work: Promise, signal: AbortSignal): Promise { + const aborted = whenAborted(signal); + // Held so the loser of the race is not reported as an unhandled rejection. + aborted.catch(() => {}); + return await Promise.race([work, aborted]); +} + /** Labels for the two decisions an approval offers, in the order it offers them. */ const APPROVAL_OPTIONS = [ { label: "Approve", value: "accept" }, @@ -100,7 +131,10 @@ export class PromptHumanConnector implements IHumanConnector { model.title, ...model.details.map((detail) => ` ${detail.label}: ${detail.value}`), ]); - const chosen = await this.renderers.select(APPROVAL_OPTIONS, model.message); + const chosen = await untilAborted( + this.renderers.select(APPROVAL_OPTIONS, model.message, signal), + signal + ); // Walking away is not the same answer as refusing, and the caller acts // differently on each. if (chosen === undefined) return settle("cancel"); @@ -112,7 +146,7 @@ export class PromptHumanConnector implements IHumanConnector { (request.contentData as Record | undefined) ?? {}, schema ); - const values = await this.renderers.form(fields); + const values = await untilAborted(this.renderers.form(fields, signal), signal); if (values === undefined) return settle("cancel"); return settle("accept", values); } diff --git a/examples/cli/src/ui/render.ts b/examples/cli/src/ui/render.ts index 70dad6205..588b2e05b 100644 --- a/examples/cli/src/ui/render.ts +++ b/examples/cli/src/ui/render.ts @@ -131,9 +131,35 @@ export interface SchemaPromptRenderOptions { readonly initialFocusedFieldKey?: string; } +/** + * Gives the terminal back when the caller stops waiting. + * + * A prompt mounts an Ink app and resolves when a person answers it; nothing + * about that is reachable from outside, so an aborted run would otherwise leave + * the app on screen and the promise pending forever. + */ +function releaseOnAbort( + signal: AbortSignal | undefined, + instance: { clear(): void; unmount(): void }, + resolve: (value: undefined) => void +): void { + if (!signal) return; + const release = (): void => { + instance.clear(); + instance.unmount(); + resolve(undefined); + }; + if (signal.aborted) { + release(); + return; + } + signal.addEventListener("abort", release, { once: true }); +} + export async function renderSchemaPrompt( fields: readonly PromptFieldDescriptor[], - options?: SchemaPromptRenderOptions + options?: SchemaPromptRenderOptions, + signal?: AbortSignal ): Promise | undefined> { return new Promise | undefined>((resolve) => { const onComplete = (values: Record) => { @@ -159,6 +185,7 @@ export async function renderSchemaPrompt( }) ) ); + releaseOnAbort(signal, instance, resolve); }); } @@ -195,7 +222,8 @@ export async function renderSearchSelect( export async function renderSelectPrompt( options: Array<{ label: string; value: string }>, - message?: string + message?: string, + signal?: AbortSignal ): Promise { return new Promise((resolve) => { const onSelect = (value: string) => { @@ -224,5 +252,6 @@ export async function renderSelectPrompt( }) ) ); + releaseOnAbort(signal, instance, resolve); }); } diff --git a/packages/ai/src/task/AgentTask.ts b/packages/ai/src/task/AgentTask.ts index 8e7022c83..c0271a551 100644 --- a/packages/ai/src/task/AgentTask.ts +++ b/packages/ai/src/task/AgentTask.ts @@ -18,7 +18,7 @@ import type { Capability } from "../capability/Capabilities"; import { createEmitQueue } from "../capability/emitQueue"; import type { ModelConfig } from "../model/ModelSchema"; import type { AgentApprovalMode } from "./AgentToolExecution"; -import { runAgentTool } from "./AgentToolExecution"; +import { clampToolText, runAgentTool } from "./AgentToolExecution"; import { DEFAULT_MAX_HISTORY_CHARS, normalizeHistoryForModel, @@ -186,8 +186,22 @@ function assistantMessage(text: string, calls: readonly ToolCall[]): ChatMessage return { role: "assistant", content }; } -function toolResult(call: ToolCall, text: string, isError: boolean): ContentBlockToolResult { - const body: ContentBlockInToolResultBody[] = [{ type: "text", text }]; +/** + * Every path into a `tool_result` is bounded by the same budget, not just the + * one carrying a tool's output. "Unknown tool X. Available: …" names every tool + * the caller registered, which on a large registry is as capable of filling a + * context window as a fetched page — and it would then do it on every round the + * model keeps guessing. + */ +function toolResult( + call: ToolCall, + text: string, + isError: boolean, + maxChars: number +): ContentBlockToolResult { + const body: ContentBlockInToolResultBody[] = [ + { type: "text", text: clampToolText(text, maxChars) }, + ]; return { type: "tool_result", tool_use_id: call.id, @@ -387,7 +401,12 @@ export class AgentTask extends Task; const validator = validators.get(call.name); @@ -395,11 +414,16 @@ export class AgentTask extends Task error.message).join("; ") || "invalid arguments"; - return toolResult(call, `Invalid arguments for ${call.name}: ${detail}`, true); + return toolResult( + call, + `Invalid arguments for ${call.name}: ${detail}`, + true, + options.maxResultChars + ); } } const result = await runAgentTool(tool, { ...call, input: sanitized }, context, options); - return toolResult(call, result.text, result.isError); + return toolResult(call, result.text, result.isError, options.maxResultChars); } } diff --git a/packages/ai/src/task/AgentToolExecution.ts b/packages/ai/src/task/AgentToolExecution.ts index 4b042d07c..5f979c78d 100644 --- a/packages/ai/src/task/AgentToolExecution.ts +++ b/packages/ai/src/task/AgentToolExecution.ts @@ -4,7 +4,11 @@ * SPDX-License-Identifier: Apache-2.0 */ -import type { IExecuteContext, TaskConfig } from "@workglow/task-graph"; +import type { + EntitlementDeclaringTaskClass, + IExecuteContext, + TaskConfig, +} from "@workglow/task-graph"; import { describeTaskClassReach, getTaskConstructors, @@ -61,7 +65,13 @@ function backingTaskType(tool: ToolDefinition): string { /** Bound on a value shown for approval — a person reads a line, not a payload. */ const MAX_APPROVAL_ARGUMENT_CHARS = 400; -function clamp(text: string, max: number): string { +/** + * Bounds anything sent back to a model, marking the cut so it can tell it is + * reading a prefix. Every path into a `tool_result` goes through this: an error + * message naming every tool the caller registered is as capable of filling a + * context window as a fetched page is. + */ +export function clampToolText(text: string, max: number): string { if (text.length <= max) return text; return `${text.slice(0, max)}… [truncated, ${text.length} chars total]`; } @@ -125,6 +135,23 @@ async function askToApprove( } } +/** + * What the card says this tool reaches. + * + * Three answers, and the third is why "host function" cannot be the fallback: + * a task-backed tool whose type is misspelled or unregistered also finds no + * class, and telling a person approving it that it is a host function + * describes something else entirely. + */ +function describeToolReach( + tool: ToolDefinition, + ctor: EntitlementDeclaringTaskClass | undefined +): string { + if (ctor) return describeTaskClassReach(ctor); + if (tool.execute) return "not declared — this tool is a host function"; + return `unknown — no task type "${backingTaskType(tool)}" is registered, so this call will fail`; +} + /** * Puts one tool call to a human and reports whether it may run. * @@ -154,8 +181,8 @@ async function approveToolCall( contentSchema: APPROVAL_SCHEMA as DataPortSchema, contentData: { tool: tool.name, - reach: ctor ? describeTaskClassReach(ctor) : "not declared — this tool is a host function", - arguments: clamp(stringifyForModel(call.input), MAX_APPROVAL_ARGUMENT_CHARS), + reach: describeToolReach(tool, ctor), + arguments: clampToolText(stringifyForModel(call.input), MAX_APPROVAL_ARGUMENT_CHARS), }, expectsResponse: true, mode: "single", @@ -201,7 +228,10 @@ export async function runAgentTool( try { const output = await invokeTool(tool, call, context); - return { text: clamp(stringifyForModel(output), options.maxResultChars), isError: false }; + return { + text: clampToolText(stringifyForModel(output), options.maxResultChars), + isError: false, + }; } catch (error) { if (context.signal.aborted) throw error; return { text: `${tool.name} failed: ${errorMessage(error)}`, isError: true }; diff --git a/packages/test/src/test/ai/AgentTask.test.ts b/packages/test/src/test/ai/AgentTask.test.ts index cd5ea6d6f..8be2f6383 100644 --- a/packages/test/src/test/ai/AgentTask.test.ts +++ b/packages/test/src/test/ai/AgentTask.test.ts @@ -273,6 +273,36 @@ describe("AgentTask", () => { expect(JSON.stringify(results[0])).toContain("Unknown tool"); }); + it("bounds the unknown-tool answer by the same budget as a tool's output", async () => { + scriptModel([{ calls: [{ id: "c1", name: "nope", input: {} }] }, { text: "sorry" }]); + // Naming every registered tool is what makes this message unbounded, and it + // would be re-sent on every round the model keeps guessing. + const many: ToolDefinition[] = Array.from({ length: 40 }, (_, index) => ({ + ...ECHO_TOOL, + name: `a_tool_with_a_fairly_long_name_${index}`, + taskType: AgentTest_EchoTask.type, + })); + + const output = await new AgentTask().run( + { + model: MODEL, + prompt: "go", + tools: many, + approval: "never", + maxToolResultChars: 80, + }, + { registry } + ); + + const text = ( + (toolResults(output.messages)[0] as { content: { text: string }[] }).content[0] as { + text: string; + } + ).text; + expect(text).toContain("truncated"); + expect(text.length).toBeLessThan(200); + }); + it("answers arguments that fail the tool's own schema", async () => { scriptModel([ { calls: [{ id: "c1", name: "AgentTest_EchoTask", input: { text: 42 } }] }, @@ -484,6 +514,40 @@ describe("AgentTask", () => { expect(JSON.stringify(toolResults(output.messages)[0])).toContain("no way to ask"); }); + it("does not call an unregistered task type a host function on the card", async () => { + scriptModel([ + { calls: [{ id: "c1", name: "AgentTest_MisspelledTask", input: {} }] }, + { text: "ok" }, + ]); + const human = connector((request) => ({ + requestId: request.requestId, + action: "accept", + content: undefined, + done: true, + })); + registry.registerInstance(HUMAN_CONNECTOR, human.connector); + + await new AgentTask().run( + { + model: MODEL, + prompt: "go", + tools: [ + { + name: "AgentTest_MisspelledTask", + description: "A task type nobody registered", + inputSchema: { type: "object", properties: {} }, + requiresApproval: true, + }, + ], + }, + { registry } + ); + + const reach = String(human.requests[0]!.contentData?.reach); + expect(reach).toContain("AgentTest_MisspelledTask"); + expect(reach).not.toContain("host function"); + }); + it('runs an unapproved-reach tool when approval is "never"', async () => { scriptModel([ { diff --git a/packages/test/src/test/human/PromptHumanConnector.conformance.test.ts b/packages/test/src/test/human/PromptHumanConnector.conformance.test.ts index a481a391e..5f4675bcb 100644 --- a/packages/test/src/test/human/PromptHumanConnector.conformance.test.ts +++ b/packages/test/src/test/human/PromptHumanConnector.conformance.test.ts @@ -22,10 +22,14 @@ import { createCliHumanSurface } from "./cliHumanSurface"; */ function connectorFor( surface: ReturnType, - request: IHumanRequest, - signal: AbortSignal + request: IHumanRequest ): PromptHumanConnector { - const answer = (): Promise => surface.answer(request, signal); + // The scripted person is deliberately NOT given the caller's signal: a real + // Ink prompt cannot reject, it can only be torn down, so letting the stand-in + // reject would make the abort assertions pass on the mock's behaviour instead + // of the connector's own race. + const patient = new AbortController().signal; + const answer = (): Promise => surface.answer(request, patient); const renderers: PromptHumanRenderers = { select: async () => { const wanted = await answer(); @@ -52,7 +56,7 @@ runHumanConnectorConformance({ // No `followUp`, matching the connector: the suite checks that a // multiTurn:false connector does not carry one. const connector: IHumanConnector = { - send: (request, signal) => connectorFor(surface, request, signal).send(request, signal), + send: (request, signal) => connectorFor(surface, request).send(request, signal), }; return { connector, script: surface.script, dispose: async () => {} }; }, From 2d51e7a7d9cc8cf167e2e1af10142784bc0ef025 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 9 Sep 2026 22:03:44 +0000 Subject: [PATCH 4/6] fix(cli): detach a prompt's abort listener when the prompt settles MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `releaseOnAbort` attached an abort listener and never removed it. The harmless-looking half is `resolve(undefined)` on an already-settled promise; the half that costs is `clear()`, which writes an erase sequence to stdout. Left attached past a normal answer, it fires the next time the run aborts and wipes whatever is on screen then — in agent chat, where one AbortController spans the turn and the transcript lives in scrollback, that is the conversation, erased by a Ctrl-C that arrives after an approval was answered. A run holding one signal across several prompts also stacked a listener per prompt, so an abort erased once for each question it had asked, and Node warns past ten. `releaseOnAbort` now returns its detach, every settle path calls it, and it moves to its own module so the rule can be tested without a terminal: five tests over a fake instance, two of which fail if the detach goes. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01JEFYAGb9D3mWfyhYkmeAvN --- examples/cli/src/test/promptAbort.test.ts | 80 +++++++++++++++++++++++ examples/cli/src/ui/promptAbort.ts | 46 +++++++++++++ examples/cli/src/ui/render.ts | 36 +++------- 3 files changed, 135 insertions(+), 27 deletions(-) create mode 100644 examples/cli/src/test/promptAbort.test.ts create mode 100644 examples/cli/src/ui/promptAbort.ts diff --git a/examples/cli/src/test/promptAbort.test.ts b/examples/cli/src/test/promptAbort.test.ts new file mode 100644 index 000000000..690962524 --- /dev/null +++ b/examples/cli/src/test/promptAbort.test.ts @@ -0,0 +1,80 @@ +/** + * @license + * Copyright 2026 Steven Roussey + * SPDX-License-Identifier: Apache-2.0 + */ + +import { describe, expect, it } from "vitest"; +import { releaseOnAbort, type PromptInstance } from "../ui/promptAbort"; + +function fakeInstance(): PromptInstance & { readonly calls: string[] } { + const calls: string[] = []; + return { + calls, + clear: () => calls.push("clear"), + unmount: () => calls.push("unmount"), + }; +} + +describe("releaseOnAbort", () => { + it("takes the terminal back when the signal aborts", () => { + const controller = new AbortController(); + const instance = fakeInstance(); + let settled: unknown = "unset"; + + releaseOnAbort(controller.signal, instance, (value) => { + settled = value; + }); + controller.abort(); + + expect(instance.calls).toEqual(["clear", "unmount"]); + expect(settled).toBeUndefined(); + }); + + it("releases immediately when the signal is already aborted", () => { + const controller = new AbortController(); + controller.abort(); + const instance = fakeInstance(); + + releaseOnAbort(controller.signal, instance, () => {}); + + expect(instance.calls).toEqual(["clear", "unmount"]); + }); + + it("does nothing after the prompt settled and detached", () => { + const controller = new AbortController(); + const instance = fakeInstance(); + + const detach = releaseOnAbort(controller.signal, instance, () => {}); + detach(); + controller.abort(); + + // `clear()` writes an erase sequence, so a listener left attached past a + // normal answer wipes whatever is on screen when the run later aborts. + expect(instance.calls).toEqual([]); + }); + + it("leaves no listener behind per prompt on one long-lived signal", () => { + const controller = new AbortController(); + const instances = [fakeInstance(), fakeInstance(), fakeInstance()]; + + for (const instance of instances) { + releaseOnAbort(controller.signal, instance, () => {})(); + } + controller.abort(); + + // A turn holding one signal across several prompts would otherwise erase + // the screen once per prompt it had asked. + expect(instances.flatMap((instance) => instance.calls)).toEqual([]); + }); + + it("detaches idempotently, and with no signal at all", () => { + const instance = fakeInstance(); + const detach = releaseOnAbort(undefined, instance, () => {}); + expect(() => { + detach(); + detach(); + }).not.toThrow(); + expect(instance.calls).toEqual([]); + }); +}); diff --git a/examples/cli/src/ui/promptAbort.ts b/examples/cli/src/ui/promptAbort.ts new file mode 100644 index 000000000..17abd7909 --- /dev/null +++ b/examples/cli/src/ui/promptAbort.ts @@ -0,0 +1,46 @@ +/** + * @license + * Copyright 2026 Steven Roussey + * SPDX-License-Identifier: Apache-2.0 + */ + +/** The part of an Ink instance a prompt has to take back. */ +export interface PromptInstance { + clear(): void; + unmount(): void; +} + +/** + * Gives the terminal back when the caller stops waiting. + * + * A prompt mounts an Ink app and resolves when a person answers it; neither is + * reachable from outside, so an aborted run would otherwise leave the app on + * screen and the promise pending forever. + * + * **Detach when the prompt settles.** `clear()` writes an erase sequence to + * stdout, so a listener left attached past a normal answer wipes whatever is on + * screen the next time the run aborts — in a session that keeps its transcript + * in scrollback, that is the conversation. A run holding one signal across + * several prompts would also stack a listener per prompt, and erase once each. + * + * @returns the detach, which is safe to call more than once and when no signal + * was given. + */ +export function releaseOnAbort( + signal: AbortSignal | undefined, + instance: PromptInstance, + resolve: (value: undefined) => void +): () => void { + if (!signal) return () => {}; + const release = (): void => { + instance.clear(); + instance.unmount(); + resolve(undefined); + }; + if (signal.aborted) { + release(); + return () => {}; + } + signal.addEventListener("abort", release, { once: true }); + return () => signal.removeEventListener("abort", release); +} diff --git a/examples/cli/src/ui/render.ts b/examples/cli/src/ui/render.ts index 588b2e05b..58d8cd11b 100644 --- a/examples/cli/src/ui/render.ts +++ b/examples/cli/src/ui/render.ts @@ -11,6 +11,7 @@ import React from "react"; import type { PromptFieldDescriptor } from "../input/prompt"; import { getCliTheme } from "../terminal/detectTerminalTheme"; import { formatError, outputResult } from "../util"; +import { releaseOnAbort } from "./promptAbort"; import { CliThemeProvider } from "./CliThemeContext"; import { SchemaPromptApp } from "./SchemaPromptApp"; import type { SearchSelectAppProps, SearchSelectItem } from "./SearchSelectApp"; @@ -131,44 +132,22 @@ export interface SchemaPromptRenderOptions { readonly initialFocusedFieldKey?: string; } -/** - * Gives the terminal back when the caller stops waiting. - * - * A prompt mounts an Ink app and resolves when a person answers it; nothing - * about that is reachable from outside, so an aborted run would otherwise leave - * the app on screen and the promise pending forever. - */ -function releaseOnAbort( - signal: AbortSignal | undefined, - instance: { clear(): void; unmount(): void }, - resolve: (value: undefined) => void -): void { - if (!signal) return; - const release = (): void => { - instance.clear(); - instance.unmount(); - resolve(undefined); - }; - if (signal.aborted) { - release(); - return; - } - signal.addEventListener("abort", release, { once: true }); -} - export async function renderSchemaPrompt( fields: readonly PromptFieldDescriptor[], options?: SchemaPromptRenderOptions, signal?: AbortSignal ): Promise | undefined> { return new Promise | undefined>((resolve) => { + let detachAbort = (): void => {}; const onComplete = (values: Record) => { + detachAbort(); instance.clear(); instance.unmount(); resolve(values); }; const onCancel = () => { + detachAbort(); instance.clear(); instance.unmount(); console.log("Cancelled."); @@ -185,7 +164,7 @@ export async function renderSchemaPrompt( }) ) ); - releaseOnAbort(signal, instance, resolve); + detachAbort = releaseOnAbort(signal, instance, resolve); }); } @@ -226,7 +205,9 @@ export async function renderSelectPrompt( signal?: AbortSignal ): Promise { return new Promise((resolve) => { + let detachAbort = (): void => {}; const onSelect = (value: string) => { + detachAbort(); instance.clear(); instance.unmount(); const label = message?.replace(/:$/, "") ?? "Selected"; @@ -236,6 +217,7 @@ export async function renderSelectPrompt( }; const onCancel = () => { + detachAbort(); instance.clear(); instance.unmount(); console.log("Cancelled."); @@ -252,6 +234,6 @@ export async function renderSelectPrompt( }) ) ); - releaseOnAbort(signal, instance, resolve); + detachAbort = releaseOnAbort(signal, instance, resolve); }); } From c494622564fd0fc17df507414253a9e407f4abc0 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 9 Sep 2026 22:24:08 +0000 Subject: [PATCH 5/6] feat(cli): agent chat in the web console MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The console needed no channel it did not already have. A chat is a run that keeps asking a person something, and asking a person something is what the run-event channel already carries — up as a `human_request`, back down fd 4. So the same command serves both surfaces: - Each turn runs through `withCli` instead of `task.run`, so a session the console started reports its rows and its text like every other command. `interactive: false` keeps the Ink run UI out of it — on a terminal that UI clears its frame when a run completes, which is the transcript this session is writing. - The next message is asked for as an ordinary `elicit` whose field carries `format: "chat-message"`. That marker is how a renderer knows to draw a composer rather than the one-line field every string port gets, and to fold the answer into a transcript rather than leave it showing as a form somebody once filled in. - A reported session does NOT install `PromptHumanConnector`. The channel installs its own with itself, and overriding it would point a console session's approvals at an Ink prompt on a process whose stdout is a pipe. Console side: `chatTranscript` zips what the console sent against the `AgentTask` rows it saw — one turn per message, since a turn is only ever started by a message, so the pairing cannot slip. Derived rather than a second copy of the run: a turn still running shows what it has said so far, one that finished silently contributes no empty bubble. `WithCliTaskHandle.run` also gains the `runConfig` argument its implementation has always threaded through to `task.run` and to the Ink renderer. The caller that needs it is one running a task inside something longer than a command, where the registry and the abort signal belong to the session rather than to the process. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01JEFYAGb9D3mWfyhYkmeAvN --- .claude/CLAUDE.md | 11 ++ examples/cli/src/agent/runAgentChat.ts | 91 +++++++++++++-- examples/cli/src/commands/agent.ts | 8 +- examples/cli/src/run-interactive.ts | 8 +- examples/cli/src/test/agentChat.test.ts | 74 +++++++++++- examples/cli/src/web/client/main.tsx | 36 +++++- examples/cli/src/web/client/state.test.ts | 110 +++++++++++++++++- examples/cli/src/web/client/state.ts | 71 +++++++++++ .../src/web/client/views/ChatTranscript.tsx | 92 +++++++++++++++ 9 files changed, 487 insertions(+), 14 deletions(-) create mode 100644 examples/cli/src/web/client/views/ChatTranscript.tsx diff --git a/.claude/CLAUDE.md b/.claude/CLAUDE.md index 73a366306..832905e96 100644 --- a/.claude/CLAUDE.md +++ b/.claude/CLAUDE.md @@ -425,6 +425,17 @@ screen back. It has no `followUp` — a modal prompt settles the question it ask decides form-vs-approval through the same `humanPromptModel` the Ink panel and the console read. +**The same command serves the web console**, because a chat needs no channel the console did +not already have: each turn runs through `withCli` so its rows and text report up the event +stream, and the next message is asked for as an ordinary `elicit` whose field carries +`format: "chat-message"` — the marker the console keys on to draw a composer instead of a +one-line field, and to fold the answer into a transcript instead of a form somebody once +filled in. A reported session does NOT install `PromptHumanConnector`: the channel installs +its own connector, and overriding it would point a console session's approvals at an Ink +prompt on a process whose stdout is a pipe. `chatTranscript` (`web/client/state.ts`) zips +what the console sent against the `AgentTask` rows it saw, one turn per message, so the +conversation is derived rather than a second copy of the run. + `workglow mcp serve` is the second server the CLI hosts: the registered tasks offered to MCP clients as tools, one per task type, named for the registered type itself (`task list` prints the same types with the `Task` suffix trimmed) and carrying the diff --git a/examples/cli/src/agent/runAgentChat.ts b/examples/cli/src/agent/runAgentChat.ts index 4ed7531cf..f6277eafa 100644 --- a/examples/cli/src/agent/runAgentChat.ts +++ b/examples/cli/src/agent/runAgentChat.ts @@ -4,10 +4,19 @@ * SPDX-License-Identifier: Apache-2.0 */ -import type { AgentApprovalMode, ChatMessage, ToolDefinition } from "@workglow/ai"; +import type { AgentApprovalMode, AgentTaskOutput, ChatMessage, ToolDefinition } from "@workglow/ai"; import { AgentTask } from "@workglow/ai"; import type { StreamEvent } from "@workglow/task-graph"; -import { globalServiceRegistry, HUMAN_CONNECTOR, ServiceRegistry } from "@workglow/util"; +import { + globalServiceRegistry, + HUMAN_CONNECTOR, + resolveHumanConnector, + ServiceRegistry, + uuid4, +} from "@workglow/util"; +import type { DataPortSchema } from "@workglow/util/schema"; +import { ensureRunReporting } from "../run-events/runReporting"; +import { withCli } from "../run-interactive"; import { createInterface } from "node:readline/promises"; import { formatError } from "../util"; import { PromptHumanConnector } from "../ui/PromptHumanConnector"; @@ -53,13 +62,72 @@ async function askLine(prompt: string): Promise { } } -/** A child of the host's registry, so the chat prompts without the Ink run UI. */ -function chatRegistry(parent: ServiceRegistry): ServiceRegistry { +/** + * The schema of the one thing this loop asks a person for. + * + * `format` is the marker a renderer keys on: the console draws a chat composer + * for it rather than the one-line text field every other string port gets, and + * folds the answer into the transcript instead of showing it as a form it once + * filled in. + */ +export const CHAT_MESSAGE_SCHEMA: DataPortSchema = { + type: "object", + properties: { + message: { type: "string", title: "Message", format: "chat-message" }, + }, + required: ["message"], + additionalProperties: false, +}; + +/** + * A child of the host's registry carrying the connector this session prompts + * through — but ONLY when the session owns a terminal. + * + * A run reporting to a parent process already has a connector wired to that + * channel, installed with the channel itself. Overriding it here would point a + * console session's approvals at an Ink prompt nobody can see, on a process + * whose stdout is a pipe. + */ +export function chatRegistry(parent: ServiceRegistry, reported: boolean): ServiceRegistry { + if (reported) return parent; const registry = new ServiceRegistry(parent.container.createChildContainer()); registry.registerInstance(HUMAN_CONNECTOR, new PromptHumanConnector()); return registry; } +/** + * The next message, asked through whoever is listening. + * + * A reported run has no terminal to read a line from — its stdin is not a + * person — so the question goes up the same channel every other question does + * and the answer comes back down it. Declining or dismissing ends the session, + * which is what closing the composer means. + */ +export function askThroughConnector( + registry: ServiceRegistry, + signal: AbortSignal +): () => Promise { + return async (): Promise => { + const response = await resolveHumanConnector({ registry }).send( + { + requestId: uuid4(), + targetHumanId: "default", + kind: "elicit", + message: "Your message", + contentSchema: CHAT_MESSAGE_SCHEMA, + contentData: undefined, + expectsResponse: true, + mode: "single", + metadata: undefined, + }, + signal + ); + if (response.action !== "accept") return undefined; + const message = response.content?.message; + return typeof message === "string" ? message : undefined; + }; +} + /** * The chat loop: read a line, run one {@link AgentTask} turn, carry the * conversation forward. @@ -69,9 +137,11 @@ function chatRegistry(parent: ServiceRegistry): ServiceRegistry { * tool result or a tool-call id looks like. */ export async function runAgentChat(options: AgentChatOptions, io?: AgentChatIo): Promise { - const ask = io?.ask ?? askLine; + const reported = ensureRunReporting() !== undefined; + const registry = chatRegistry(globalServiceRegistry, reported); + const sessionAbort = new AbortController(); + const ask = io?.ask ?? (reported ? askThroughConnector(registry, sessionAbort.signal) : askLine); const transcript = createChatTranscript(io?.write ?? ((text) => void process.stdout.write(text))); - const registry = chatRegistry(globalServiceRegistry); let messages: ChatMessage[] = []; transcript.note( @@ -130,7 +200,12 @@ async function runTurn( }); try { - const output = await task.run( + // Through `withCli` rather than `task.run` so a session the console + // started reports its rows and its text up the event channel like every + // other command. `interactive: false` keeps the Ink run UI out of it: on a + // terminal that UI clears its frame when the run completes, which is the + // transcript this session is writing. + const output = (await withCli(task, { interactive: false, suppressResultOutput: true }).run( { model: options.model, prompt: text, @@ -141,7 +216,7 @@ async function runTurn( approval: options.approval, }, { registry, signal: controller.signal } - ); + )) as AgentTaskOutput; if (output.stopReason === "max-rounds") { transcript.note(` · stopped after ${output.rounds} rounds without an answer`); } diff --git a/examples/cli/src/commands/agent.ts b/examples/cli/src/commands/agent.ts index 2aa48306d..1f47bc950 100644 --- a/examples/cli/src/commands/agent.ts +++ b/examples/cli/src/commands/agent.ts @@ -31,6 +31,7 @@ import { createAgentRepository } from "../storage"; import { renderSelectPrompt, renderWorkflowRun } from "../ui/render"; import { formatError, formatTable, outputResult } from "../util"; import { runAgentChat } from "../agent/runAgentChat"; +import { ensureRunReporting } from "../run-events/runReporting"; export function registerAgentCommand(program: Command): void { const agent = program.command("agent").description("Manage and run agents"); @@ -370,9 +371,12 @@ export function registerAgentCommand(program: Command): void { "Run every tool without asking. For a session you are not watching; the default confirms anything reaching past the model." ) .action(async (opts: Record) => { - if (!process.stdin.isTTY) { + // A session started from the web console has no terminal and does not + // need one: it asks and answers over the run's event channel. What it + // cannot do is read a line from a pipe nobody is typing into. + if (!process.stdin.isTTY && !ensureRunReporting()) { console.error( - "agent chat needs a terminal. Use `workglow task run AgentTask` for a scripted turn." + "agent chat needs a terminal, or the web console. Use `workglow task run AgentTask` for a scripted turn." ); process.exit(1); } diff --git a/examples/cli/src/run-interactive.ts b/examples/cli/src/run-interactive.ts index 8a4c3a2ea..5394de7df 100644 --- a/examples/cli/src/run-interactive.ts +++ b/examples/cli/src/run-interactive.ts @@ -142,7 +142,13 @@ export function tuiDisabledByEnv(): boolean { export interface WithCliTaskHandle { readonly kind: "task"; - run(overrides?: Record): Promise; + /** + * `runConfig` reaches `task.run` and the Ink renderer alike — the + * implementation has always threaded it, and the caller that needs it is one + * running a task inside something longer than a command, where the registry + * and the abort signal are the session's rather than the process's. + */ + run(overrides?: Record, runConfig?: Partial): Promise; abort(): void; } diff --git a/examples/cli/src/test/agentChat.test.ts b/examples/cli/src/test/agentChat.test.ts index c17de8ee3..f6b5f8ccf 100644 --- a/examples/cli/src/test/agentChat.test.ts +++ b/examples/cli/src/test/agentChat.test.ts @@ -12,7 +12,15 @@ import { setAiProviderRegistry, } from "@workglow/ai"; import { afterEach, beforeEach, describe, expect, it } from "vitest"; -import { runAgentChat, type AgentChatIo } from "../agent/runAgentChat"; +import type { IHumanRequest } from "@workglow/util"; +import { Container, HUMAN_CONNECTOR, ServiceRegistry } from "@workglow/util"; +import { + askThroughConnector, + chatRegistry, + CHAT_MESSAGE_SCHEMA, + runAgentChat, + type AgentChatIo, +} from "../agent/runAgentChat"; const PROVIDER = "mock-chat-provider"; @@ -137,6 +145,70 @@ describe("agent chat loop", () => { expect(out()).toContain("saw 1"); }); + describe("under a run that reports to a parent process", () => { + it("asks for the next message through the human connector", async () => { + const asked: IHumanRequest[] = []; + const registry = new ServiceRegistry(new Container()); + registry.registerInstance(HUMAN_CONNECTOR, { + send: async (request: IHumanRequest) => { + asked.push(request); + return { + requestId: request.requestId, + action: "accept" as const, + content: { message: "hello" }, + done: true, + }; + }, + }); + + const ask = askThroughConnector(registry, new AbortController().signal); + + expect(await ask()).toBe("hello"); + expect(asked).toHaveLength(1); + expect(asked[0]!.kind).toBe("elicit"); + // The marker a renderer keys on to draw a composer rather than a field. + expect(asked[0]!.contentSchema).toEqual(CHAT_MESSAGE_SCHEMA); + }); + + it("ends the session when the person closes the composer", async () => { + const registry = new ServiceRegistry(new Container()); + registry.registerInstance(HUMAN_CONNECTOR, { + // Carrying content on a decline is a connector misbehaving — libs says + // only an accepted elicit answers with data — and the session must end + // on the action rather than on whether anything came back with it. + send: async (request: IHumanRequest) => ({ + requestId: request.requestId, + action: "decline" as const, + content: { message: "typed, then dismissed" }, + done: true, + }), + }); + + // `undefined` is how the loop reads end-of-input, the same as Ctrl-D. + expect(await askThroughConnector(registry, new AbortController().signal)()).toBeUndefined(); + }); + + it("leaves the channel's own connector in place", () => { + const parent = new ServiceRegistry(new Container()); + const installed = { + send: async () => ({ + requestId: "x", + action: "decline" as const, + content: undefined, + done: true, + }), + }; + parent.registerInstance(HUMAN_CONNECTOR, installed); + + // A console session's approvals have to reach the channel; a child + // registry carrying an Ink prompt would answer nobody. + expect(chatRegistry(parent, true)).toBe(parent); + expect(chatRegistry(parent, true).get(HUMAN_CONNECTOR)).toBe(installed); + // On a terminal it is the other way round: the session prompts itself. + expect(chatRegistry(parent, false).get(HUMAN_CONNECTOR)).not.toBe(installed); + }); + }); + it("answers /help without calling the model", async () => { const called = scriptModel(["never"]); const { io, out } = scriptedIo(["/help", "/exit"]); diff --git a/examples/cli/src/web/client/main.tsx b/examples/cli/src/web/client/main.tsx index a15291c08..034d7a2ac 100644 --- a/examples/cli/src/web/client/main.tsx +++ b/examples/cli/src/web/client/main.tsx @@ -43,14 +43,18 @@ import { } from "./heartbeat"; import { loadRailWidths, saveRailWidths, type RailSide, type RailWidths } from "./railWidths"; import { + appendChatAsk, applyRecord, + chatTranscript, emptyRunView, filterCommandTree, + isChatRequest, openPathsFor, stackedPane, type RunViewState, type StackedPane, } from "./state"; +import { ChatTranscript } from "./views/ChatTranscript"; import { CommandTree } from "./views/CommandTree"; import { GroupView } from "./views/GroupView"; import { HumanPrompt } from "./views/HumanPrompt"; @@ -610,7 +614,37 @@ function App(): JSX.Element { {error} ) : null} - {view.humanRequest && run ? ( + {view.humanRequest && run && isChatRequest(view.humanRequest) ? ( + { + if (!cli.online) return; + void answerHuman(run.id, { + requestId: view.humanRequest!.requestId, + action: "accept", + content: { message: text }, + done: true, + }); + setView((current) => ({ + ...appendChatAsk(current, text), + humanRequest: undefined, + })); + }} + onEnd={() => { + if (!cli.online) return; + void answerHuman(run.id, { + requestId: view.humanRequest!.requestId, + action: "decline", + content: undefined, + done: true, + }); + setView((current) => ({ ...current, humanRequest: undefined })); + }} + /> + ) : null} + {view.humanRequest && run && !isChatRequest(view.humanRequest) ? ( { expect(stackedPane("back")).toBe("list"); }); }); + +describe("the console's view of a conversation", () => { + const chatSchema = { + type: "object", + properties: { message: { type: "string", format: "chat-message" } }, + }; + + function turnRow(state: RunViewState, id: string): RunViewState { + return reduceRunEvent(state, { + k: "task_added", + id, + type: "AgentTask", + label: "Agent", + depth: 0, + }); + } + + it("tells a chat request apart from a form", () => { + expect( + isChatRequest({ + requestId: "r1", + kind: "elicit", + message: "Your message", + schema: chatSchema, + data: undefined, + }) + ).toBe(true); + expect( + isChatRequest({ + requestId: "r2", + kind: "elicit", + message: "Your name", + schema: { type: "object", properties: { name: { type: "string" } } }, + data: undefined, + }) + ).toBe(false); + // An approval carrying the same field is still an approval — drawing it as + // a chat box would leave nowhere to say no. + expect( + isChatRequest({ + requestId: "r3", + kind: "confirm", + message: "Run it?", + schema: chatSchema, + data: undefined, + }) + ).toBe(false); + expect(isChatRequest(undefined)).toBe(false); + }); + + it("pairs each answer with the turn it started", () => { + let state = appendChatAsk(emptyRunView(), "first"); + state = turnRow(state, "t1"); + state = reduceRunEvent(state, { k: "text", id: "t1", delta: "one" }); + state = reduceRunEvent(state, { k: "status", id: "t1", status: "COMPLETED" }); + state = appendChatAsk(state, "second"); + state = turnRow(state, "t2"); + state = reduceRunEvent(state, { k: "text", id: "t2", delta: "tw" }); + + expect(chatTranscript(state)).toEqual([ + { role: "user", text: "first", pending: false }, + { role: "assistant", text: "one", pending: false }, + { role: "user", text: "second", pending: false }, + // Still running, and showing what it has said so far rather than nothing. + { role: "assistant", text: "tw", pending: true }, + ]); + }); + + it("shows a message whose turn has not written anything yet", () => { + let state = appendChatAsk(emptyRunView(), "hello"); + state = turnRow(state, "t1"); + + expect(chatTranscript(state)).toEqual([ + { role: "user", text: "hello", pending: false }, + { role: "assistant", text: "", pending: true }, + ]); + }); + + it("shows a message whose turn has not started at all", () => { + const state = appendChatAsk(emptyRunView(), "hello"); + expect(chatTranscript(state)).toEqual([{ role: "user", text: "hello", pending: false }]); + }); + + it("leaves a turn that finished silently out rather than showing an empty reply", () => { + let state = appendChatAsk(emptyRunView(), "hello"); + state = turnRow(state, "t1"); + state = reduceRunEvent(state, { k: "status", id: "t1", status: "COMPLETED" }); + + expect(chatTranscript(state)).toEqual([{ role: "user", text: "hello", pending: false }]); + }); + + it("ignores rows that are not turns", () => { + let state = appendChatAsk(emptyRunView(), "hello"); + state = reduceRunEvent(state, { + k: "task_added", + id: "tool", + type: "FetchUrlTask", + label: "Fetch", + depth: 1, + }); + state = reduceRunEvent(state, { k: "text", id: "tool", delta: "not the answer" }); + + expect(chatTranscript(state)).toEqual([{ role: "user", text: "hello", pending: false }]); + }); +}); diff --git a/examples/cli/src/web/client/state.ts b/examples/cli/src/web/client/state.ts index 98189f158..1f878bfa1 100644 --- a/examples/cli/src/web/client/state.ts +++ b/examples/cli/src/web/client/state.ts @@ -72,10 +72,80 @@ export interface RunViewState { readonly data: unknown; } | undefined; + /** + * What the person has sent this session, in order. + * + * Held rather than derived because it never crosses the event stream: the + * console typed it and answered with it, and what comes back up is the run + * getting on with the turn. + */ + readonly chatAsks: readonly string[]; readonly lastSeq: number; readonly nextOrder: number; } +/** One side of a conversation, as the console draws it. */ +export interface ChatEntry { + readonly role: "user" | "assistant"; + readonly text: string; + /** Whether this turn is still being written. */ + readonly pending: boolean; +} + +/** The task whose rows are turns of a conversation. */ +const CHAT_TURN_TYPE = "AgentTask"; + +/** + * Whether a run is asking for the next message of a conversation rather than + * filling in a form. + * + * Keyed on the `format` the asking side puts on the field, which is the same + * way every other port says what it means. A chat drawn as a one-line text + * input is answerable, but it is not a conversation, and the answer it takes + * shows up afterwards as a form somebody once filled in. + */ +export function isChatRequest(request: RunViewState["humanRequest"]): boolean { + if (!request || request.kind !== "elicit") return false; + const properties = (request.schema as { properties?: Record } | null) + ?.properties; + return Object.values(properties ?? {}).some((field) => field?.format === "chat-message"); +} + +/** + * The conversation, zipped from the two halves the console holds: what it sent, + * and what each turn wrote back. + * + * One turn is one {@link CHAT_TURN_TYPE} row, in the order the rows arrived, so + * the Nth answer belongs under the Nth message — a turn is only ever started by + * a message, so the pairing cannot slip. A turn still running shows what it has + * said so far rather than nothing. + */ +export function chatTranscript(state: RunViewState): readonly ChatEntry[] { + const turns = [...state.rows.values()] + .filter((row) => row.type === CHAT_TURN_TYPE) + .sort((left, right) => left.order - right.order); + const entries: ChatEntry[] = []; + for (let index = 0; index < state.chatAsks.length; index++) { + entries.push({ role: "user", text: state.chatAsks[index]!, pending: false }); + const turn = turns[index]; + if (!turn) continue; + const settled = SETTLED.has(turn.status); + if (turn.streamText.length === 0 && !settled) { + entries.push({ role: "assistant", text: "", pending: true }); + continue; + } + if (turn.streamText.length > 0) { + entries.push({ role: "assistant", text: turn.streamText, pending: !settled }); + } + } + return entries; +} + +/** Records a message the console just sent, so the transcript can show it. */ +export function appendChatAsk(state: RunViewState, text: string): RunViewState { + return { ...state, chatAsks: [...state.chatAsks, text] }; +} + export function emptyRunView(): RunViewState { return { rows: new Map(), @@ -87,6 +157,7 @@ export function emptyRunView(): RunViewState { error: undefined, output: undefined, humanRequest: undefined, + chatAsks: [], lastSeq: 0, nextOrder: 0, }; diff --git a/examples/cli/src/web/client/views/ChatTranscript.tsx b/examples/cli/src/web/client/views/ChatTranscript.tsx new file mode 100644 index 000000000..718d96088 --- /dev/null +++ b/examples/cli/src/web/client/views/ChatTranscript.tsx @@ -0,0 +1,92 @@ +/** @jsxImportSource preact */ +/** + * @license + * Copyright 2026 Steven Roussey + * SPDX-License-Identifier: Apache-2.0 + */ + +import type { JSX } from "preact"; +import { useState } from "preact/hooks"; +import type { ChatEntry } from "../state"; + +/** + * A conversation, and the box to add to it. + * + * The run asks for each message the same way it asks for anything else, so this + * is a rendering of a human request rather than a channel of its own — what + * makes it a conversation rather than a form is that the answers stay on + * screen underneath the question. + */ +export function ChatTranscript({ + entries, + message, + canAnswer, + onSend, + onEnd, +}: { + readonly entries: readonly ChatEntry[]; + /** What the run called the thing it is asking for. */ + readonly message: string; + /** False while the CLI is not answering; the run cannot receive a reply. */ + readonly canAnswer: boolean; + readonly onSend: (text: string) => void; + readonly onEnd: () => void; +}): JSX.Element { + const [draft, setDraft] = useState(""); + const trimmed = draft.trim(); + const send = (): void => { + if (!canAnswer || trimmed.length === 0) return; + setDraft(""); + onSend(trimmed); + }; + + return ( +
+
+ {entries.length === 0 ? ( +
+ Nothing said yet. +
+ ) : ( + entries.map((entry, index) => ( +
+
+ {entry.role === "user" ? "You" : "Agent"} + {entry.pending ? · still writing : null} +
+
+ {entry.text} +
+
+ )) + )} +
+
{message}
+