diff options
| -rw-r--r-- | docs/overview.md | 6 | ||||
| -rw-r--r-- | docs/phase-3.md | 461 | ||||
| -rw-r--r-- | libpanto/src/agent.zig | 658 | ||||
| -rw-r--r-- | libpanto/src/anthropic_messages_json.zig | 250 | ||||
| -rw-r--r-- | libpanto/src/openai_chat_json.zig | 484 | ||||
| -rw-r--r-- | libpanto/src/provider.zig | 21 | ||||
| -rw-r--r-- | libpanto/src/provider_anthropic_messages.zig | 99 | ||||
| -rw-r--r-- | libpanto/src/provider_openai_chat.zig | 295 | ||||
| -rw-r--r-- | libpanto/src/root.zig | 6 | ||||
| -rw-r--r-- | libpanto/src/tool.zig | 61 | ||||
| -rw-r--r-- | libpanto/src/tool_registry.zig | 217 | ||||
| -rw-r--r-- | src/main.zig | 23 | ||||
| -rw-r--r-- | src/ping_tool.zig | 57 |
13 files changed, 2362 insertions, 276 deletions
diff --git a/docs/overview.md b/docs/overview.md index dfc9f0f..916e904 100644 --- a/docs/overview.md +++ b/docs/overview.md @@ -95,3 +95,9 @@ These are recorded from the initial ideas but do not yet have phase documents or - **System prompt construction framework** — opinionated system for assembling system prompts from composable parts (templates, project context, extension contributions) - **Google API provider** — native integration with Google's Gemini API (rather than their OpenAI-compatibility layer), unlocking richer capabilities specific to that API shape. Low priority compared to Anthropic and OpenAI support. - **C ABI distribution of libpanto** — `export fn` wrappers exposing libpanto functionality through a C calling convention, enabling external programs to embed or build on pantograph from C, Rust, or other native languages. Not a separate library — the C ABI is a second interface on the same `libpanto` artifact, compiled from `export fn` shims that translate between Zig types and C types. Needed eventually for shared-object extensions (Zig, Rust, C, C++) beyond Lua. + +## Punted + +Deliberate decisions to defer functionality that came up during phase planning but doesn't fit cleanly into the existing phase roadmap. Each one is captured here with enough context to pick up later. + +- **Tool-call cancellation / timeout via process isolation.** First raised in phase 3. There is no clean POSIX mechanism for cancelling a thread mid-execution with a guarantee of no further side effects — `pthread_kill` with SIGKILL terminates the entire process, `pthread_cancel` is widely considered unusable, and signal-based interruption can't safely unwind arbitrary code. Lua's `lua_sethook` provides cooperative cancellation between VM steps but doesn't interrupt handlers blocked in C calls (filesystem, subprocess, network). The mechanism that actually works is **process isolation**: run tool invocations in a helper subprocess, SIGKILL the subprocess on timeout. The intended approach is `fork+exec` into a small `panto-tool-worker` binary that statically links libpanto's extension machinery. libpanto would define the wire protocol (tool name + input bytes over a pipe, result bytes back) and own the fork/exec/timeout/kill orchestration; the embedder supplies the helper binary path. This makes sandboxing a library-level capability available to all embedders, not just the panto CLI. Open design questions when this is picked up: extension loading strategy in the worker (rescan-per-fork vs long-lived worker pool), file descriptor inheritance policy, working directory and environment handling, and how registry contents are communicated to the worker. Until this lands, tool handlers run to completion in-process — if a tool hangs, the user kills `panto`. diff --git a/docs/phase-3.md b/docs/phase-3.md index 1b9d83a..aeac7fb 100644 --- a/docs/phase-3.md +++ b/docs/phase-3.md @@ -2,28 +2,37 @@ ## Goal -Introduce a Lua extension runtime and tool registration/execution, transforming `pantograph` from a chat client into an agent that can act on the world. The extension system is the primary mechanism for adding capability — tools are extensions, not built-ins. +Introduce a native-code tool extension API on `libpanto` and a Lua extension runtime in the `panto` CLI. Together these transform `pantograph` from a chat client into an agent that can act on the world. The extension system is the primary mechanism for adding capability — tools are extensions, not built-ins. + +This phase also marks the point where the `Agent` abstraction starts earning its keep. Until now, `agent.zig` has been a thin pass-through to the provider. In phase 3 the agent grows into the thing that owns the tool registry and drives the tool-call loop. Providers stay dumb (stream blocks); the agent is what makes a conversation iterative rather than one-shot. ## Deliverable -A working extension system where Lua scripts can register tools and handle tool-use requests. The agent loop detects ToolUse blocks in LLM responses, executes the corresponding tool handlers, and feeds ToolResult blocks back. At the end of this phase, you can: +Two layered deliverables: + +1. **libpanto** — a Zig-native tool extension API. `Agent` exposes `registerTool(Tool)` and runs a tool-call loop in `runStep`. The agent loop detects ToolUse blocks in LLM responses, dispatches to registered tools, feeds ToolResult blocks back, and repeats until the model stops calling tools. libpanto has no awareness of Lua, Python, or any specific extension language. + +2. **panto CLI** — a Lua extension runtime that discovers Lua scripts on disk, loads them, and registers each declared tool with `libpanto` via a `LuaTool` adapter that satisfies libpanto's `Tool` interface. + +At the end of this phase, you can: -- Write a Lua extension that registers a tool and handles invocations. -- Place it in `~/.config/panto/extensions/` (or `.panto/extensions/`) and have `pantograph` discover and load it. +- Embed `libpanto` in a Zig program, implement a `Tool` struct natively, and register it with the agent — no Lua involved. +- Write a Lua extension that registers a tool, place it in `.agent/extensions/` (or `~/.config/panto/extensions/`), and have the `panto` CLI discover and load it. - Have a conversation where the LLM calls your tool and receives the result. - See tool calls execute in parallel when the LLM returns multiple ToolUse blocks. -- See a meaningful error message when an extension crashes, instead of a process abort. +- See a meaningful error message when a Lua extension crashes, instead of a process abort. ## What is usable at the end | Capability | How to exercise it | |---|---| -| Write a tool extension | Create `~/.config/panto/extensions/mytool.lua` calling `panto.register_tool(...)` | -| Discover extensions | Place `.lua` files or directories in extension directories; `pantograph` loads them on startup | -| LLM calls a tool | Ask the LLM to use a registered tool; it emits a ToolUse block; pantograph executes the handler | -| Tool result fed back | The tool handler's return value becomes a ToolResult block sent back to the LLM | +| Native tool registration | In a Zig embedder, construct a `Tool` and call `agent.registerTool(tool)` | +| Write a Lua tool extension | Create `.agent/extensions/mytool.lua` calling `panto.register_tool(...)` | +| Discover Lua extensions | Place `.lua` files or directories in extension directories; `panto` CLI loads them on startup | +| LLM calls a tool | Ask the LLM to use a registered tool; it emits a ToolUse block; the agent dispatches to the registered handler | +| Tool result fed back | The tool's return bytes become a ToolResult block sent back to the LLM | | Parallel tool calls | LLM returns multiple ToolUse blocks; they execute concurrently | -| Extension crash handling | A crashing tool handler prints `the "mytool" extension crashed: <trace>` and aborts the turn | +| Lua crash handling | A crashing Lua handler prints `the "mytool" extension crashed: <trace>` and aborts the turn | ## What is explicitly out of scope @@ -32,287 +41,349 @@ A working extension system where Lua scripts can register tools and handle tool- - C ABI distribution of libpanto for external consumers (future) - GitHub or luarocks extension loaders (future — local filesystem only in phase 3) - Shared-object extensions (future) -- Extension sandboxing beyond `xpcall` crash protection (future) +- Extension sandboxing beyond `xpcall` crash protection in the Lua adapter (future) - Config file for specifying which extensions to load (phase 6 — phase 3 loads everything it discovers) +- Non-Lua extension runtimes (future — the libpanto API is general, but only the Lua adapter ships in phase 3) --- -## Extension Discovery +## libpanto: Native Tool API -### Directory locations +### The `Tool` interface -`pantograph` scans two directories in order: +`libpanto` defines a `Tool` as an opaque-context value with a vtable. This is the boundary between the agent loop and any extension runtime. -1. `.panto/extensions/` — project-local extensions (relative to current working directory) -2. `~/.config/panto/extensions/` — user-level extensions +```zig +pub const Tool = struct { + name: []const u8, // borrowed; lifetime owned by the registrar + schema_json: []const u8, // borrowed JSON Schema bytes; lifetime owned by the registrar + ctx: *anyopaque, + vtable: *const VTable, -### Naming and structure + pub const VTable = struct { + /// Invoke the tool. MUST be thread-safe — the agent may call `invoke` + /// concurrently from multiple threads when the LLM returns multiple + /// ToolUse blocks in a single response. + /// + /// `input` is the raw JSON bytes the provider sent. The tool is + /// responsible for parsing them if it cares about their structure. + /// + /// Returns owned bytes allocated with `allocator`. These bytes become + /// the `content` of the ToolResult block. + /// + /// Returning an error aborts the current turn. The agent will surface + /// the error to the user. Native tool implementations are responsible + /// for catching their own panics — a panic in `invoke` will crash the + /// process. Adapters that bridge to safer languages (Lua, Python, Go) + /// should convert panics/exceptions into errors here. + invoke: *const fn (ctx: *anyopaque, input: []const u8, allocator: std.mem.Allocator) anyerror![]u8, -Extensions are identified by name, derived from their path. Two formats: + /// Called when the tool is removed from the registry or the agent is + /// torn down. Frees any resources owned by `ctx`. + deinit: *const fn (ctx: *anyopaque, allocator: std.mem.Allocator) void, + }; +}; +``` -- **Single-file**: `<name>.lua` → extension name is `<name>` -- **Directory**: `<name>/init.lua` → extension name is `<name>` +The contract: -Names can be hierarchical using dots as separators, mapping to directory nesting: +- **Thread-safety is mandatory.** Tool implementations promise that concurrent `invoke` calls are safe. This pushes the burden onto adapters (where it belongs — the Lua adapter solves it with a `lua_State` pool; a native Zig tool may just be re-entrant) and out of the agent loop. +- **Input is opaque bytes.** libpanto never parses tool input. The tool decides whether to parse JSON, treat it as a string, or ignore it. +- **Output is opaque bytes.** Same principle in reverse. +- **Errors abort the turn.** No partial recovery in phase 3. +- **Panics crash the process.** Native code that wants safety implements it itself. This is trivially obvious from the API surface: it's a function pointer with `anyerror!` return — there's no magic. Adapters for safer languages can wrap panics/exceptions and convert them to errors. -- `utils/json.lua` → extension name is `utils.json` -- `coding/edit/init.lua` → extension name is `coding.edit` +### `Agent` changes -This convention mirrors Lua's `require("a.b.c")` path resolution. Extension sub-modules (e.g., `coding/edit/helpers.lua`) are the extension's internal business — pantograph only loads the top-level entry point (`init.lua` or the single file). +```zig +pub const Agent = struct { + provider: provider.Provider, + allocator: std.mem.Allocator, + registry: ToolRegistry, -### Loading behavior + pub fn init(allocator: Allocator, prov: provider.Provider) Agent { ... } + pub fn deinit(self: *Agent) void { ... } -- Scan both directories recursively. -- Construct extension names from relative paths using dot separators. -- Load each discovered entry point file into a fresh `lua_State`. -- After loading, the extension's top-level code runs, which should call `panto.register_tool(...)` to register its tools. -- Duplicate extension names: project-local takes precedence over user-level. + pub fn registerTool(self: *Agent, tool: Tool) !void { ... } + pub fn unregisterTool(self: *Agent, name: []const u8) void { ... } ---- + pub fn runStep(self: *Agent, conv: *Conversation, receiver: *Receiver) !void { ... } +}; +``` -## Lua Bridge +`registerTool` takes ownership of the `Tool` value (the agent calls `tool.vtable.deinit` on teardown or unregister). Tool name uniqueness is enforced — registering a name that already exists is an error. -The Lua bridge is a Zig module (`lua_bridge.zig`) that registers panto functions into the Lua state and handles translation between Zig types and Lua types. It is compiled into the `panto` binary — it is not a separate library. +### `ToolRegistry` -### Functions exposed to Lua +A small internal module: -#### `panto.register_tool(name, schema, handler)` +``` +ToolRegistry = struct { + tools: std.StringHashMap(Tool), + allocator: Allocator, -Registers a tool with the agent. + fn register(name, tool) !void + fn unregister(name) void + fn lookup(name) ?*const Tool + fn iterator() Iterator // for serializing into provider requests + fn deinit() void // invokes each tool's vtable.deinit +}; +``` -- `name` (string) — tool name, e.g. `"bash"` -- `schema` (table) — tool input schema as a Lua table (JSON Schema format). Serialized to JSON bytes for inclusion in provider requests. -- `handler` (function) — called when the LLM invokes this tool. Receives a single argument: a Lua table parsed from the tool input JSON. Must return a string (the tool result content). +### Agent loop with tools -Example: +`runStep` gains the tool-call loop: -```lua -panto.register_tool("echo", { - type = "object", - properties = { - message = { type = "string", description = "The message to echo back" } - }, - required = { "message" } -}, function(input) - return input.message -end) +``` +runStep(conv, receiver): + loop: + provider.streamStep(conv, receiver) + inspect the assistant message just appended to conv for ToolUse blocks + if none: return — turn complete + for each ToolUse block, in parallel: + tool = registry.lookup(block.name) or error + result_bytes = tool.vtable.invoke(tool.ctx, block.input, allocator) + build a ToolResult block { tool_use_id = block.id, content = result_bytes } + append a user Message containing all ToolResult blocks to conv + continue loop ``` -### Input parsing at the bridge boundary - -Tool input arrives in libpanto as raw JSON bytes (stored in the ToolUseBlock's TextualBlock). At the Lua bridge boundary, libpanto parses these bytes into a Lua table using `std.json`, then passes the table to the handler. This is a convenience service for extension authors — internally, libpanto still treats tool input as opaque bytes. The round-trip guarantee: the JSON bytes the provider sent are faithfully represented in the Lua table. - -### Output from handlers +A single `runStep` may invoke the provider multiple times if the LLM chains tool calls. -The handler returns a string. This string becomes the `content` of a ToolResult block. It is stored as raw bytes; libpanto does not interpret or parse it. +### Parallel execution ---- +When a response contains multiple ToolUse blocks, the agent dispatches them concurrently. Implementation likely uses a small thread pool or `std.Thread.spawn` per call (TBD during implementation). Because `Tool.invoke` is declared thread-safe, the agent loop has no further constraints — it just fans out and joins. -## Tool Registration (Internal) +### Tool serialization in provider requests -When `panto.register_tool()` is called from Lua, the bridge stores: +When the agent calls the provider, the provider receives the registry (or an iterator over it) and emits the tools array. This means provider request serialization gains a tools-emitting branch when the registry is non-empty. +**OpenAI**: +```json +{ + "tools": [ + { "type": "function", + "function": { "name": "echo", "parameters": <schema_json> } } + ] +} ``` -RegisteredTool = struct { - name: []const u8, // owned copy - input_schema: []const u8, // JSON bytes, owned copy (serialized from the Lua table) - lua_handler_ref: i32, // Lua registry reference to the handler function -}; + +**Anthropic**: +```json +{ + "tools": [ + { "name": "echo", "input_schema": <schema_json> } + ] +} ``` -A global tool registry maps tool names to `RegisteredTool` entries. The agent loop consults this registry when it encounters ToolUse blocks. +The `schema_json` bytes are emitted verbatim. Description fields are deferred — `Tool` does not yet carry a description (see open questions). -The `input_schema` bytes are included verbatim in provider requests when tools are listed. Both OpenAI and Anthropic accept JSON Schema objects for tool input definitions. +### ToolUse decoding in responses ---- +Already handled by the existing Receiver callback sequence from phase 1/2: -## Tool Execution +- OpenAI: `delta.tool_calls` → `onBlockStart(.ToolUse, ...)` with `meta.tool_id` and `meta.tool_name`, then `onContentDelta` with JSON argument fragments. +- Anthropic: `content_block_start` with `type: "tool_use"` → `onBlockStart(.ToolUse, ...)`, then `content_block_delta` with `input_json_delta` fragments. -### Agent loop extension +The assembled ToolUseBlock contains `id`, `name`, and `input` (TextualBlock with the full JSON string). -The agent loop gains tool-call handling after each provider response: +### ToolResult encoding in requests +**OpenAI** — each ToolResult block becomes a separate top-level `tool` message: +```json +{ "role": "tool", "tool_call_id": "<id>", "content": "<result>" } ``` -runStep(conversation, receiver): - 1. Call provider.streamStep(conversation, receiver) - 2. Examine the completed message for ToolUse blocks - 3. If ToolUse blocks present: - a. For each ToolUse block, look up the tool in the registry - b. Execute all tool handlers (see parallel execution below) - c. Collect ToolResult blocks - d. Construct a user Message containing the ToolResult blocks - e. Append to conversation - f. Go to step 1 (call provider again with the updated conversation) - 4. If no ToolUse blocks: done — the turn is complete + +**Anthropic** — ToolResult blocks are content blocks on a user message: +```json +{ "role": "user", "content": [ + { "type": "tool_result", "tool_use_id": "<id>", "content": "<result>" } +] } ``` -A single `runStep` may invoke the provider multiple times if the LLM chains tool calls. +--- -### Parallel execution +## panto CLI: Lua Extension Runtime -Multiple ToolUse blocks in a single response are executed concurrently. This is a documented part of the extension API: **tool handlers may be called concurrently in separate threads.** Extension authors must ensure their handlers are thread-safe. +The Lua runtime lives entirely in the CLI. libpanto has no Lua dependency. -Implementation: on-demand `lua_State` pool. +### Extension discovery -``` -LuaStatePool = struct { - states: std.ArrayList(*lua_State), - available: std.BitSet, // which states are free - allocator: std.mem.Allocator, - extension_dirs: []const []const u8, +The CLI scans two directories in order: - pub fn acquire(self) *lua_State // returns an existing free state, or creates a new one - pub fn release(self, *lua_State) // returns state to the pool - pub fn deinit(self) void // destroys all states -}; -``` +1. `.panto/extensions/` — project-local, relative to current working directory +2. `~/.config/panto/extensions/` — user-level -- `acquire()`: if a free state exists, return it. Otherwise, create a fresh `lua_State`, load all discovered extensions into it (so the handler function references are valid), and return it. -- States are created lazily, not pre-allocated. -- Each state has all extensions loaded identically, so any state can handle any tool. -- When tool execution completes, the state is returned to the pool for reuse. +`.panto/` follows the established per-agent pattern (`.claude/`, `.opencode/`, `.pi/`). -### Crash protection +### Naming and structure -Every tool handler invocation is wrapped in `xpcall` with a traceback handler: +- **Single-file**: `<name>.lua` → extension name is `<name>` +- **Directory**: `<name>/init.lua` → extension name is `<name>` -```lua -xpcall(handler_fn, function(err) - return debug.traceback(err) -end, input_table) -``` +Hierarchical names via dot/directory mapping, mirroring `require("a.b.c")`: -If the handler crashes: +- `utils/json.lua` → `utils.json` +- `coding/edit/init.lua` → `coding.edit` -- The error and stack trace are captured as a string. -- pantograph prints: `the "<tool_name>" extension crashed: <trace>` -- The current LLM turn is aborted — no ToolResult is generated for this tool call. -- Other concurrent tool calls in the same batch are not affected (they run in separate `lua_State` instances). +Sub-modules under a directory extension (e.g., `coding/edit/helpers.lua`) are the extension's internal business — the CLI only loads the top-level entry point. ---- +### Loading behavior -## Tool Serialization in Provider Requests +- Recursively scan both directories. +- Construct extension names from relative paths. +- For each unique extension name, load its entry point into the `lua_State` pool (see below). +- Project-local entries shadow user-level entries of the same name. +- Extension top-level code runs at load time; it is expected to call `panto.register_tool(...)`. -When tools are registered, the provider requests must include them. Both providers have a `tools` field in the request body. +### The Lua bridge -### OpenAI +A small Zig module (`lua_bridge.zig`) registers Zig functions into a `lua_State`. The bridge exposes: -``` -{ - "model": ..., - "stream": true, - "messages": [...], - "tools": [ - { - "type": "function", - "function": { - "name": "echo", - "description": "...", // not yet supported, phase 5+ when we add descriptions - "parameters": { <input_schema> } - } - } - ] -} -``` +#### `panto.register_tool(name, schema, handler)` -### Anthropic +- `name` (string) — tool name, e.g. `"bash"` +- `schema` (table) — JSON Schema as a Lua table. The bridge serializes this to JSON bytes at registration time. +- `handler` (function) — invoked when the LLM calls this tool. Receives a single argument: a Lua table parsed from the input JSON. Must return a string. -``` -{ - "model": ..., - "system": ..., - "stream": true, - "messages": [...], - "tools": [ - { - "name": "echo", - "description": "...", - "input_schema": { <input_schema> } - } - ] -} +```lua +panto.register_tool("echo", { + type = "object", + properties = { + message = { type = "string", description = "The message to echo back" } + }, + required = { "message" } +}, function(input) + return input.message +end) ``` -The `input_schema` bytes stored in the registry are emitted verbatim into the `parameters` (OpenAI) or `input_schema` (Anthropic) field. +The bridge stores the handler in the Lua registry (`luaL_ref`) and constructs a `LuaTool` (see below) that it then registers with the agent. -### ToolUse in responses +### `LuaTool` — the adapter to libpanto -Both providers return tool-call information in their streaming responses. This is already handled by the existing Receiver callback sequence: +```zig +const LuaTool = struct { + pool: *LuaStatePool, + handler_ref: i32, // registry ref to the Lua function + name_owned: []u8, + schema_owned: []u8, -- OpenAI: `delta.tool_calls` triggers `onBlockStart(.ToolUse, ...)` with `meta.tool_id` and `meta.tool_name`, then `onContentDelta` with JSON argument fragments. -- Anthropic: `content_block_start` with `type: "tool_use"` triggers `onBlockStart(.ToolUse, ...)`, then `content_block_delta` with `input_json_delta` fragments. + // Implements Tool.VTable.invoke: + // 1. pool.acquire() → lua_State + // 2. lua_rawgeti(L, LUA_REGISTRYINDEX, handler_ref) — push handler + // 3. parse input bytes into a Lua table via std.json + // 4. xpcall the handler with the table + // 5. read returned string (or capture trace on error) + // 6. pool.release(L) + // 7. return result bytes or error +}; +``` -The assembled ToolUseBlock contains `id`, `name`, and `input` (TextualBlock with the full JSON string). The agent loop reads `name` to look up the registered tool, reads `input.content()` to get the JSON string, and passes it through the Lua bridge. +The adapter is what makes the CLI side responsible for Lua-specific concerns: -### ToolResult in requests +- **JSON → Lua table conversion**: happens here, not in libpanto. +- **`xpcall` crash protection**: wraps the handler call, captures `debug.traceback`, returns an error so libpanto aborts the turn cleanly with `the "<tool_name>" extension crashed: <trace>`. +- **Thread-safety**: provided by the `lua_State` pool. Each concurrent `invoke` checks out a separate state. libpanto's contract is satisfied. -After tool execution, a user Message containing ToolResult blocks is appended to the conversation. Serialization differs by provider: +### `LuaStatePool` + +On-demand pool of `lua_State` instances: -**OpenAI**: Each ToolResult block becomes a separate message: -```json -{ "role": "tool", "tool_call_id": "<tool_use_id>", "content": "<result string>" } ``` +LuaStatePool = struct { + states: std.ArrayList(*lua_State), + available: std.BitSet, + allocator: Allocator, + extension_paths: []const []const u8, // all discovered entry points -**Anthropic**: ToolResult blocks are content blocks on a user message: -```json -{ "role": "user", "content": [ - { "type": "tool_result", "tool_use_id": "...", "content": "..." } -] } + fn acquire() *lua_State // returns a free state, or creates a new one + fn release(L: *lua_State) void // returns L to the pool + fn deinit() void // closes all states +}; ``` +- States are created lazily — no pre-allocation. +- Each new state loads all discovered extensions identically, so any state can handle any tool. +- A state is checked out for the duration of a single `invoke` call. + +There's a coupling here: the handler `luaL_ref` is per-`lua_State`. The pool must guarantee that the same registry slot in every state points at the same handler. The cleanest way is: when an extension is first loaded into the prototype state, the bridge records the registry index. When a new state is constructed, the bridge re-loads all extensions in the same order, producing the same ref indices. This needs care during implementation but is mechanically straightforward. + --- ## Module Changes -### New files +### libpanto -``` -src/lua_bridge.zig // Zig functions registered into Lua state, type translation -src/tool_registry.zig // RegisteredTool storage, lookup by name -src/lua_state_pool.zig // On-demand pool of lua_State instances -src/extension_loader.zig // Directory scanning, extension discovery and loading -``` +New: +- `libpanto/src/tool.zig` — `Tool` and `VTable` definitions +- `libpanto/src/tool_registry.zig` — `ToolRegistry` + +Modified: +- `libpanto/src/agent.zig` — owns registry, gains `registerTool`, drives tool-call loop in `runStep` +- `libpanto/src/provider.zig` — provider receives a way to see registered tools (likely a `*const ToolRegistry` on the agent passed via `streamStep` context, or a callback) +- `libpanto/src/provider_openai_chat.zig` — request serialization emits `tools` array +- `libpanto/src/provider_anthropic_messages.zig` — request serialization emits `tools` array +- `libpanto/src/openai_chat_json.zig` — ToolResult message encoding +- `libpanto/src/anthropic_messages_json.zig` — ToolResult content block encoding +- `libpanto/src/root.zig` — re-exports `Tool`, `ToolRegistry` -### Modified files +### panto CLI -- `agent.zig` — runStep gains the tool-call loop (detect ToolUse → execute → feed results → repeat) -- `provider_openai.zig` — request serialization includes `tools` array when tools are registered -- `provider_anthropic.zig` — request serialization includes `tools` array when tools are registered -- `json.zig` — serialization for ToolResult blocks (OpenAI and Anthropic formats) -- `config.zig` — extension directories added to config +New: +- `src/lua_bridge.zig` — Zig functions registered into Lua states; `panto.register_tool` implementation +- `src/lua_tool.zig` — `LuaTool` adapter implementing libpanto's `Tool` interface +- `src/lua_state_pool.zig` — on-demand `lua_State` pool +- `src/extension_loader.zig` — directory scanning, name construction, loading + +Modified: +- `src/main.zig` — wires extension discovery + loading into agent startup ### External dependency -Lua interpreter linked into the `panto` binary. Zig's build system can fetch and compile Lua from source (Lua is a small C codebase, ~30KLOC). No system dependency required. +Lua 5.4 linked into the `panto` binary. Zig's build system can fetch and compile Lua from source (~30KLOC C). No system dependency required. libpanto does not link Lua. --- ## Testing Strategy -### Unit tests +### Unit tests (libpanto) + +| What | How | +|---|---| +| Tool registration | Construct a native `Tool`, register with `Agent`, verify lookup | +| Duplicate registration | Verify registering the same name twice returns an error | +| Tool dispatch | Hand-craft a conversation with a ToolUse block, run `runStep` against a stub provider, verify the registered tool's `invoke` is called with the right bytes | +| ToolResult round-trip | Verify the ToolResult message appended to the conversation has the right `tool_use_id` and `content` | +| Parallel dispatch | Register two tools with sleep+timestamps, verify they ran concurrently (elapsed wall time < sum of individual times) | +| Provider request serialization | Snapshot tests on OpenAI and Anthropic request bodies with tools registered | + +### Unit tests (panto CLI) | What | How | |---|---| | Extension discovery | Create temp directory with single-file and directory extensions, verify names constructed correctly | -| Tool registration | Load a Lua extension that registers a tool, verify entry appears in registry with correct name and schema | -| Lua bridge input parsing | Feed JSON strings through the bridge, verify correct Lua tables produced | -| Lua bridge output | Call a tool handler that returns a string, verify it becomes a ToolResult with correct content | -| Crash protection | Load an extension whose handler throws an error, verify xpcall catches it and returns trace | +| Project shadows user | Same extension name in both directories — project wins | +| Lua bridge input | Feed JSON bytes through `LuaTool.invoke`, verify the Lua handler sees the expected table | +| Lua bridge output | Handler returns a string, verify it round-trips to bytes | +| `xpcall` crash protection | Extension whose handler throws — verify `invoke` returns an error and includes the trace | +| Concurrent `invoke` | Two threads call `invoke` on the same `LuaTool`, verify they don't share a `lua_State` | -### Integration test (manual) +### Integration tests (manual) -- Write a simple `echo.lua` extension, place it in extension directory -- Start `panto`, ask the LLM to use the echo tool -- Verify the tool is called, result is fed back, LLM continues -- Write a `crash.lua` extension that throws an error -- Verify the crash is caught and printed with context, turn aborts gracefully -- Ask the LLM to use two tools in one response, verify both execute +- `echo.lua` extension: ask the LLM to use it, verify result is fed back, LLM continues. +- `crash.lua` extension: verify the crash is caught, printed, turn aborts gracefully. +- Multi-tool response: ask the LLM to call two tools in one message, verify both execute concurrently. --- -## Open Questions (to resolve during implementation) +## Resolved Decisions -1. **Lua version**: Lua 5.4 is current. Luau (Roblox's fork) has performance improvements but diverges. Stick with standard Lua 5.4 for compatibility with luarocks and existing ecosystem? -2. **Handler timeout**: Should tool handlers have a timeout? A hung tool call blocks the agent loop. Could add a configurable timeout with abort. -3. **Streaming tool results**: Some tools (e.g., `bash` running a long command) produce output incrementally. Phase 3 handlers return a single string. Streaming results would require a different handler interface — possibly a callback the handler calls to emit partial output. Defer to a later phase? -4. **Tool description field**: The `panto.register_tool()` call in phase 3 includes a schema but no explicit description string. Provider APIs require a description. Options: add a `description` parameter, or extract it from the schema. Probably simplest to add it as a parameter. +- **Project-local extension directory**: `.panto/extensions/`, matching `.claude/`, `.opencode/`, `.pi/`. +- **Lua version**: Lua 5.4. LuaJIT is stuck at 5.1 semantics and the JIT speedup is largely irrelevant for tool handlers (which spend most of their time in Zig-side I/O). 5.4 keeps us aligned with current luarocks and Lua documentation. +- **Tool description**: a top-level `description: []const u8` field on `Tool`. The Lua bridge gains it as a parameter: `panto.register_tool(name, description, schema, handler)`. Schema describes input only; description is the human-readable purpose. +- **Provider sees the registry**: `provider.streamStep` gains a `*const ToolRegistry` parameter. Embedders normally pass the same registry on every call. +- **Parallel dispatch**: `std.Thread.spawn` per ToolUse block. Thread creation is sub-millisecond on Linux/macOS; tool calls are orders of magnitude longer. Recycling buys nothing meaningful. +- **Streaming tool results**: deferred, possibly indefinitely. Provider APIs only accept a final result payload, so streaming would be display-only — not worth the vtable complication in phase 3. +- **Handler timeout / cancellation**: deferred. POSIX has no clean way to cancel a thread (`pthread_kill` with SIGKILL terminates the whole process; `pthread_cancel` is unusable in practice; signal-based interruption is not async-signal-safe for arbitrary code). Lua's `lua_sethook` only fires between VM steps, so it can't interrupt a handler blocked in a C call. A real cancellation primitive requires process isolation, which is a phase of its own — see the Punted list in [overview.md](overview.md). Phase 3 ships with the honest contract: **tool handlers run to completion. If a tool hangs, the user kills the `panto` process.** diff --git a/libpanto/src/agent.zig b/libpanto/src/agent.zig index 7ebc058..3e240d4 100644 --- a/libpanto/src/agent.zig +++ b/libpanto/src/agent.zig @@ -1,23 +1,665 @@ +//! The Agent owns the conversation-driving loop: provider streaming + +//! tool dispatch. +//! +//! In phase 1/2 this was a thin pass-through to the provider. In phase 3 +//! it grows the tool-call loop: after each provider streaming step, the +//! agent inspects the assistant message for ToolUse blocks, dispatches +//! the registered handlers (in parallel when there are multiple), and +//! appends a user message containing the ToolResult blocks back into the +//! conversation. The loop continues until a turn arrives with no ToolUse +//! blocks. + const std = @import("std"); -const provider = @import("provider.zig"); +const Allocator = std.mem.Allocator; +const Thread = std.Thread; + +const provider_mod = @import("provider.zig"); const conversation = @import("conversation.zig"); +const tool_mod = @import("tool.zig"); +const tool_registry_mod = @import("tool_registry.zig"); + +pub const Tool = tool_mod.Tool; +pub const ToolRegistry = tool_registry_mod.ToolRegistry; pub const Agent = struct { - provider: provider.Provider, - allocator: std.mem.Allocator, + provider: provider_mod.Provider, + allocator: Allocator, + registry: ToolRegistry, - pub fn init(allocator: std.mem.Allocator, prov: provider.Provider) Agent { + pub fn init(allocator: Allocator, prov: provider_mod.Provider) Agent { return .{ .provider = prov, .allocator = allocator, + .registry = ToolRegistry.init(allocator), }; } - pub fn runStep(self: Agent, conv: *conversation.Conversation, receiver: *provider.Receiver) !void { - try self.provider.streamStep(conv, receiver); + pub fn deinit(self: *Agent) void { + self.registry.deinit(); + self.provider.deinit(); + } + + /// Register a tool. The agent's registry takes ownership. + pub fn registerTool(self: *Agent, tool: Tool) !void { + try self.registry.register(tool); } - pub fn deinit(self: Agent) void { - self.provider.deinit(); + /// Remove a tool by name. No-op if not registered. + pub fn unregisterTool(self: *Agent, name: []const u8) void { + self.registry.unregister(name); + } + + /// Drive the conversation forward until the model stops calling tools. + /// + /// A single `runStep` invocation may call the provider multiple times + /// if the model chains tool calls. Each provider call streams a new + /// assistant message into `conv`; if that message contains ToolUse + /// blocks the agent dispatches them concurrently, appends a user + /// message of ToolResult blocks, and loops. The loop terminates when + /// the provider's most recent response has no ToolUse blocks. + pub fn runStep( + self: *Agent, + conv: *conversation.Conversation, + receiver: *provider_mod.Receiver, + ) !void { + while (true) { + try self.provider.streamStep(conv, &self.registry, receiver); + + const last = conv.messages.items[conv.messages.items.len - 1]; + std.debug.assert(last.role == .assistant); + + // Defense-in-depth: if the provider committed an assistant + // message with zero content blocks, something went wrong + // upstream that wasn't surfaced as a provider error (e.g. a + // mid-stream provider error that an older codepath swallowed, + // or a model that genuinely returned nothing). Either way the + // turn made no observable progress — surface it instead of + // silently dropping back to the prompt. + if (last.content.items.len == 0) return error.EmptyAssistantResponse; + + if (!hasToolUseBlock(last)) return; + + try self.dispatchToolCalls(conv, last); + // Loop: feed the ToolResult message back to the provider. + } + } + + /// Returns true if the message contains at least one ToolUse block. + fn hasToolUseBlock(msg: conversation.Message) bool { + for (msg.content.items) |block| { + if (block == .ToolUse) return true; + } + return false; + } + + /// Dispatch every ToolUse block in `assistant_msg` concurrently, then + /// append a single user Message containing all ToolResult blocks to + /// `conv` in the same order the tool calls appeared. + fn dispatchToolCalls( + self: *Agent, + conv: *conversation.Conversation, + assistant_msg: conversation.Message, + ) !void { + // Count tool uses for sizing. + var n: usize = 0; + for (assistant_msg.content.items) |block| { + if (block == .ToolUse) n += 1; + } + std.debug.assert(n > 0); + + const tasks = try self.allocator.alloc(ToolCallTask, n); + defer self.allocator.free(tasks); + + // Populate tasks. We borrow ID/name slices from the conversation — + // the assistant message stays in `conv` throughout dispatch, so + // these slices remain valid until we copy them into the new + // ToolResultBlock. + { + var i: usize = 0; + for (assistant_msg.content.items) |block| { + if (block != .ToolUse) continue; + const tu = block.ToolUse; + tasks[i] = .{ + .agent = self, + .tool_use_id = tu.id, + .tool_name = tu.name, + .input = tu.input.items, + .result = null, + .err = null, + }; + i += 1; + } + } + + // Spawn one thread per tool call. `std.Thread.spawn` is cheap + // (sub-millisecond on Linux/macOS) compared to typical tool + // latency, and `Tool.invoke` is contractually thread-safe, so we + // fan out without a pool. + const threads = try self.allocator.alloc(Thread, n); + defer self.allocator.free(threads); + + var spawned: usize = 0; + var joined = false; + errdefer { + // Join any in-flight threads so they don't outlive `tasks`. + if (!joined) for (threads[0..spawned]) |t| t.join(); + for (tasks) |*task| { + if (task.result) |r| self.allocator.free(r); + } + } + + for (tasks, 0..) |*task, idx| { + threads[idx] = try Thread.spawn(.{}, runToolTask, .{task}); + spawned += 1; + } + for (threads[0..spawned]) |t| t.join(); + joined = true; + + // Build the user ToolResult message. From here on we own all + // result byte slices; transfer them into ToolResultBlocks. + var content: std.ArrayList(conversation.ContentBlock) = .empty; + errdefer { + for (content.items) |*b| b.deinit(self.allocator); + content.deinit(self.allocator); + } + try content.ensureTotalCapacity(self.allocator, n); + + // If any task failed, prefer to abort the turn — but first move + // every successful result into a block so it gets freed by the + // standard cleanup path, and free errored ones eagerly (there + // are none to move). The errdefer above handles teardown. + var first_err: ?anyerror = null; + for (tasks) |*task| { + if (task.err) |e| { + first_err = e; + continue; + } + const result_bytes = task.result.?; + task.result = null; // ownership transferred below + + const id_copy = try self.allocator.dupe(u8, task.tool_use_id); + errdefer self.allocator.free(id_copy); + + var content_buf: conversation.TextualBlock = .empty; + errdefer content_buf.deinit(self.allocator); + try content_buf.appendSlice(self.allocator, result_bytes); + self.allocator.free(result_bytes); + + content.appendAssumeCapacity(.{ .ToolResult = .{ + .tool_use_id = id_copy, + .content = content_buf, + } }); + } + + if (first_err) |e| return e; + + // Wrap the ToolResult blocks into a user Message and append. + try conv.messages.append(self.allocator, .{ + .role = .user, + .content = content, + }); + } +}; + +/// Per-tool-call work item passed into a worker thread. +const ToolCallTask = struct { + agent: *Agent, + tool_use_id: []const u8, // borrowed from assistant_msg + tool_name: []const u8, // borrowed from assistant_msg + input: []const u8, // borrowed from assistant_msg + + /// Owned result bytes from `Tool.invoke`. Allocated with + /// `agent.allocator`. Transferred into a ToolResultBlock on success. + result: ?[]u8, + + /// If non-null, the tool failed and the turn must abort. + err: ?anyerror, +}; + +fn runToolTask(task: *ToolCallTask) void { + const tool = task.agent.registry.lookup(task.tool_name) orelse { + task.err = error.UnknownTool; + return; + }; + const out = tool.vtable.invoke(tool.ctx, task.input, task.agent.allocator) catch |e| { + task.err = e; + return; + }; + task.result = out; +} + +// ----------------------------------------------------------------------------- +// Tests +// ----------------------------------------------------------------------------- + +const testing = std.testing; + +/// A stub Provider that, on each call to `streamStep`, appends a +/// pre-canned assistant message to the conversation. Used to drive the +/// agent's tool-call loop without any HTTP plumbing. +const StubProvider = struct { + allocator: Allocator, + scripted: []const ScriptedTurn, + next: usize = 0, + + const ScriptedTurn = struct { + /// Blocks to append as the next assistant message. The producer + /// owns these — the stub clones them per turn so the conversation + /// can take ownership. + blocks: []const TestBlock, + }; + + const TestBlock = union(enum) { + Text: []const u8, + ToolUse: struct { + id: []const u8, + name: []const u8, + input: []const u8, + }, + }; + + fn provider(self: *StubProvider) provider_mod.Provider { + return .{ .ptr = self, .vtable = &vt }; + } + + const vt: provider_mod.ProviderVTable = .{ + .streamStep = vtStreamStep, + .deinit = vtDeinit, + }; + + fn vtStreamStep( + ptr: *anyopaque, + conv: *conversation.Conversation, + _: *const ToolRegistry, + _: *provider_mod.Receiver, + ) anyerror!void { + const self: *StubProvider = @ptrCast(@alignCast(ptr)); + if (self.next >= self.scripted.len) return error.NoMoreScriptedTurns; + const turn = self.scripted[self.next]; + self.next += 1; + + var blocks: std.ArrayList(conversation.ContentBlock) = .empty; + errdefer { + for (blocks.items) |*b| b.deinit(self.allocator); + blocks.deinit(self.allocator); + } + for (turn.blocks) |tb| { + switch (tb) { + .Text => |s| { + try blocks.append(self.allocator, .{ + .Text = try conversation.textualBlockFromSlice(self.allocator, s), + }); + }, + .ToolUse => |tu| { + const id = try self.allocator.dupe(u8, tu.id); + errdefer self.allocator.free(id); + const name = try self.allocator.dupe(u8, tu.name); + errdefer self.allocator.free(name); + var input_buf: conversation.TextualBlock = .empty; + errdefer input_buf.deinit(self.allocator); + try input_buf.appendSlice(self.allocator, tu.input); + try blocks.append(self.allocator, .{ .ToolUse = .{ + .id = id, + .name = name, + .input = input_buf, + } }); + }, + } + } + const moved = try blocks.toOwnedSlice(self.allocator); + defer self.allocator.free(moved); + try conv.addAssistantMessage(moved); + } + + fn vtDeinit(_: *anyopaque) void {} +}; + +/// Simple in-test tool: returns `prefix ++ input`. Used in dispatch tests. +const EchoTool = struct { + prefix_owned: []u8, + name_owned: []u8, + + fn create(allocator: Allocator, name: []const u8, prefix: []const u8) !Tool { + const self = try allocator.create(EchoTool); + errdefer allocator.destroy(self); + self.name_owned = try allocator.dupe(u8, name); + errdefer allocator.free(self.name_owned); + self.prefix_owned = try allocator.dupe(u8, prefix); + return .{ + .name = self.name_owned, + .description = "echo", + .schema_json = "{}", + .ctx = self, + .vtable = &vt, + }; + } + + const vt: Tool.VTable = .{ .invoke = invoke, .deinit = deinit }; + + fn invoke(ctx: *anyopaque, input: []const u8, allocator: Allocator) anyerror![]u8 { + const self: *EchoTool = @ptrCast(@alignCast(ctx)); + return try std.fmt.allocPrint(allocator, "{s}{s}", .{ self.prefix_owned, input }); + } + + fn deinit(ctx: *anyopaque, allocator: Allocator) void { + const self: *EchoTool = @ptrCast(@alignCast(ctx)); + allocator.free(self.name_owned); + allocator.free(self.prefix_owned); + allocator.destroy(self); + } +}; + +/// Tool that records the thread it ran on, then participates in a +/// rendezvous: every invocation must reach the barrier before any can +/// return. If dispatch is sequential, the first invocation would deadlock +/// (only one tool runs at a time, never reaching the threshold) — so this +/// test only passes when invocations run truly concurrently. +/// +/// The barrier is bounded by a spin-with-yield with a wall-time ceiling +/// of 5 seconds; failure to reach quorum surfaces as an `error.BarrierTimeout`. +const BarrierTool = struct { + name_owned: []u8, + barrier: *Barrier, + + const Barrier = struct { + target: u32, + arrived: std.atomic.Value(u32) = .init(0), + thread_ids: [4]std.atomic.Value(u64) = .{ + .init(0), .init(0), .init(0), .init(0), + }, + }; + + fn create(allocator: Allocator, name: []const u8, barrier: *Barrier) !Tool { + const self = try allocator.create(BarrierTool); + errdefer allocator.destroy(self); + self.name_owned = try allocator.dupe(u8, name); + self.barrier = barrier; + return .{ + .name = self.name_owned, + .description = "barrier", + .schema_json = "{}", + .ctx = self, + .vtable = &vt, + }; + } + + const vt: Tool.VTable = .{ .invoke = invoke, .deinit = deinit }; + + fn invoke(ctx: *anyopaque, _: []const u8, allocator: Allocator) anyerror![]u8 { + const self: *BarrierTool = @ptrCast(@alignCast(ctx)); + const arrived = self.barrier.arrived.fetchAdd(1, .acq_rel); + if (arrived < self.barrier.thread_ids.len) { + self.barrier.thread_ids[arrived].store(std.Thread.getCurrentId(), .release); + } + + // Spin-with-yield until everyone has arrived. ~5s ceiling at the + // typical yield granularity is plenty for a 3-way barrier; on a + // truly single-threaded dispatch this loop never resolves. + var i: usize = 0; + while (self.barrier.arrived.load(.acquire) < self.barrier.target) : (i += 1) { + if (i > 50_000) return error.BarrierTimeout; + std.Thread.yield() catch {}; + } + return try allocator.dupe(u8, "done"); + } + + fn deinit(ctx: *anyopaque, allocator: Allocator) void { + const self: *BarrierTool = @ptrCast(@alignCast(ctx)); + allocator.free(self.name_owned); + allocator.destroy(self); + } +}; + +/// Tool whose invoke always errors. Used to verify the turn aborts. +const FailingTool = struct { + name_owned: []u8, + + fn create(allocator: Allocator, name: []const u8) !Tool { + const self = try allocator.create(FailingTool); + errdefer allocator.destroy(self); + self.name_owned = try allocator.dupe(u8, name); + return .{ + .name = self.name_owned, + .description = "fails", + .schema_json = "{}", + .ctx = self, + .vtable = &vt, + }; + } + + const vt: Tool.VTable = .{ .invoke = invoke, .deinit = deinit }; + + fn invoke(_: *anyopaque, _: []const u8, _: Allocator) anyerror![]u8 { + return error.ToolExploded; + } + + fn deinit(ctx: *anyopaque, allocator: Allocator) void { + const self: *FailingTool = @ptrCast(@alignCast(ctx)); + allocator.free(self.name_owned); + allocator.destroy(self); + } +}; + +const NoopReceiver = struct { + fn make() provider_mod.Receiver { + return .{ .ptr = @constCast(@ptrCast(&dummy)), .vtable = &vt }; } + var dummy: u8 = 0; + const vt: provider_mod.ReceiverVTable = .{ + .onMessageStart = noop1, + .onBlockStart = noop2, + .onContentDelta = noop3, + .onBlockComplete = noop4, + .onMessageComplete = noop5, + .onError = noop6, + }; + fn noop1(_: *anyopaque, _: conversation.MessageRole) anyerror!void {} + fn noop2(_: *anyopaque, _: provider_mod.ContentBlockType, _: usize, _: ?provider_mod.BlockMeta) anyerror!void {} + fn noop3(_: *anyopaque, _: usize, _: []const u8) anyerror!void {} + fn noop4(_: *anyopaque, _: usize, _: conversation.ContentBlock) anyerror!void {} + fn noop5(_: *anyopaque, _: conversation.Message) anyerror!void {} + fn noop6(_: *anyopaque, _: anyerror) void {} }; + +test "registerTool and lookup via registry" { + var stub = StubProvider{ .allocator = testing.allocator, .scripted = &.{} }; + var agent = Agent.init(testing.allocator, stub.provider()); + defer agent.deinit(); + + try agent.registerTool(try EchoTool.create(testing.allocator, "echo", "ECHO:")); + try testing.expectEqual(@as(usize, 1), agent.registry.count()); + try testing.expect(agent.registry.lookup("echo") != null); +} + +test "duplicate registerTool returns error" { + var stub = StubProvider{ .allocator = testing.allocator, .scripted = &.{} }; + var agent = Agent.init(testing.allocator, stub.provider()); + defer agent.deinit(); + + try agent.registerTool(try EchoTool.create(testing.allocator, "echo", "A:")); + + var dup = try EchoTool.create(testing.allocator, "echo", "B:"); + try testing.expectError(error.DuplicateTool, agent.registerTool(dup)); + dup.vtable.deinit(dup.ctx, testing.allocator); +} + +test "runStep dispatches a tool call and loops to a final text turn" { + const allocator = testing.allocator; + + const scripted = [_]StubProvider.ScriptedTurn{ + .{ .blocks = &.{ + .{ .ToolUse = .{ .id = "tc_1", .name = "echo", .input = "hello" } }, + } }, + .{ .blocks = &.{ + .{ .Text = "ok" }, + } }, + }; + var stub = StubProvider{ .allocator = allocator, .scripted = &scripted }; + var agent = Agent.init(allocator, stub.provider()); + defer agent.deinit(); + + try agent.registerTool(try EchoTool.create(allocator, "echo", "ECHO:")); + + var conv = conversation.Conversation.init(allocator); + defer conv.deinit(); + try conv.addUserMessage("call a tool"); + + var recv = NoopReceiver.make(); + try agent.runStep(&conv, &recv); + + // user, assistant(tool_use), user(tool_result), assistant(text) + try testing.expectEqual(@as(usize, 4), conv.messages.items.len); + + try testing.expectEqual(conversation.MessageRole.assistant, conv.messages.items[1].role); + try testing.expectEqual(@as(usize, 1), conv.messages.items[1].content.items.len); + try testing.expectEqualStrings("tc_1", conv.messages.items[1].content.items[0].ToolUse.id); + + try testing.expectEqual(conversation.MessageRole.user, conv.messages.items[2].role); + try testing.expectEqual(@as(usize, 1), conv.messages.items[2].content.items.len); + const tr = conv.messages.items[2].content.items[0].ToolResult; + try testing.expectEqualStrings("tc_1", tr.tool_use_id); + try testing.expectEqualStrings("ECHO:hello", tr.content.items); + + try testing.expectEqual(conversation.MessageRole.assistant, conv.messages.items[3].role); + try testing.expectEqualStrings("ok", conv.messages.items[3].content.items[0].Text.items); +} + +test "runStep dispatches multiple tool calls in parallel" { + const allocator = testing.allocator; + + // Use a barrier: each tool must wait until all three have arrived + // before returning. If dispatch were sequential, the first tool + // would hit its iteration ceiling and `error.BarrierTimeout`. Reaching + // the barrier proves all three ran concurrently. + var barrier: BarrierTool.Barrier = .{ .target = 3 }; + + const scripted = [_]StubProvider.ScriptedTurn{ + .{ .blocks = &.{ + .{ .ToolUse = .{ .id = "a", .name = "barrierA", .input = "" } }, + .{ .ToolUse = .{ .id = "b", .name = "barrierB", .input = "" } }, + .{ .ToolUse = .{ .id = "c", .name = "barrierC", .input = "" } }, + } }, + .{ .blocks = &.{ + .{ .Text = "done" }, + } }, + }; + var stub = StubProvider{ .allocator = allocator, .scripted = &scripted }; + var agent = Agent.init(allocator, stub.provider()); + defer agent.deinit(); + + try agent.registerTool(try BarrierTool.create(allocator, "barrierA", &barrier)); + try agent.registerTool(try BarrierTool.create(allocator, "barrierB", &barrier)); + try agent.registerTool(try BarrierTool.create(allocator, "barrierC", &barrier)); + + var conv = conversation.Conversation.init(allocator); + defer conv.deinit(); + try conv.addUserMessage("go"); + + var recv = NoopReceiver.make(); + try agent.runStep(&conv, &recv); + + // Each tool produced one ToolResult, in original order. + const tr_msg = conv.messages.items[2]; + try testing.expectEqual(@as(usize, 3), tr_msg.content.items.len); + try testing.expectEqualStrings("a", tr_msg.content.items[0].ToolResult.tool_use_id); + try testing.expectEqualStrings("b", tr_msg.content.items[1].ToolResult.tool_use_id); + try testing.expectEqualStrings("c", tr_msg.content.items[2].ToolResult.tool_use_id); + + // And the three calls happened on three distinct threads. + const t0 = barrier.thread_ids[0].load(.acquire); + const t1 = barrier.thread_ids[1].load(.acquire); + const t2 = barrier.thread_ids[2].load(.acquire); + try testing.expect(t0 != 0 and t1 != 0 and t2 != 0); + try testing.expect(t0 != t1 and t1 != t2 and t0 != t2); +} + +test "runStep propagates tool errors and aborts the turn" { + const allocator = testing.allocator; + + const scripted = [_]StubProvider.ScriptedTurn{ + .{ .blocks = &.{ + .{ .ToolUse = .{ .id = "x", .name = "boom", .input = "" } }, + } }, + // Second turn should never run. + .{ .blocks = &.{.{ .Text = "should-not-see" }} }, + }; + var stub = StubProvider{ .allocator = allocator, .scripted = &scripted }; + var agent = Agent.init(allocator, stub.provider()); + defer agent.deinit(); + + try agent.registerTool(try FailingTool.create(allocator, "boom")); + + var conv = conversation.Conversation.init(allocator); + defer conv.deinit(); + try conv.addUserMessage("break it"); + + var recv = NoopReceiver.make(); + try testing.expectError(error.ToolExploded, agent.runStep(&conv, &recv)); + + // Conversation has user + assistant(tool_use). No ToolResult message + // was appended because the dispatch errored before append. + try testing.expectEqual(@as(usize, 2), conv.messages.items.len); +} + +test "runStep errors UnknownTool when the model calls something unregistered" { + const allocator = testing.allocator; + + const scripted = [_]StubProvider.ScriptedTurn{ + .{ .blocks = &.{ + .{ .ToolUse = .{ .id = "z", .name = "ghost", .input = "" } }, + } }, + }; + var stub = StubProvider{ .allocator = allocator, .scripted = &scripted }; + var agent = Agent.init(allocator, stub.provider()); + defer agent.deinit(); + + var conv = conversation.Conversation.init(allocator); + defer conv.deinit(); + try conv.addUserMessage("call a ghost"); + + var recv = NoopReceiver.make(); + try testing.expectError(error.UnknownTool, agent.runStep(&conv, &recv)); +} + +test "runStep with no tool calls returns after one provider step" { + const allocator = testing.allocator; + + const scripted = [_]StubProvider.ScriptedTurn{ + .{ .blocks = &.{.{ .Text = "hi" }} }, + }; + var stub = StubProvider{ .allocator = allocator, .scripted = &scripted }; + var agent = Agent.init(allocator, stub.provider()); + defer agent.deinit(); + + var conv = conversation.Conversation.init(allocator); + defer conv.deinit(); + try conv.addUserMessage("hello"); + + var recv = NoopReceiver.make(); + try agent.runStep(&conv, &recv); + + try testing.expectEqual(@as(usize, 2), conv.messages.items.len); + try testing.expectEqualStrings("hi", conv.messages.items[1].content.items[0].Text.items); +} + +test "runStep surfaces EmptyAssistantResponse when provider commits an empty message" { + // Mirrors the real-world failure mode where a provider silently ends the + // turn with no content blocks — e.g. a mid-stream error that an older + // codepath swallowed. The agent must surface the failure so the user + // doesn't see the prompt come back with no explanation. + const allocator = testing.allocator; + + const scripted = [_]StubProvider.ScriptedTurn{ + .{ .blocks = &.{} }, + }; + var stub = StubProvider{ .allocator = allocator, .scripted = &scripted }; + var agent = Agent.init(allocator, stub.provider()); + defer agent.deinit(); + + var conv = conversation.Conversation.init(allocator); + defer conv.deinit(); + try conv.addUserMessage("hi"); + + var recv = NoopReceiver.make(); + try testing.expectError(error.EmptyAssistantResponse, agent.runStep(&conv, &recv)); +} diff --git a/libpanto/src/anthropic_messages_json.zig b/libpanto/src/anthropic_messages_json.zig index 8f44edc..355b3ce 100644 --- a/libpanto/src/anthropic_messages_json.zig +++ b/libpanto/src/anthropic_messages_json.zig @@ -15,6 +15,7 @@ const Allocator = std.mem.Allocator; const Writer = std.Io.Writer; const conversation = @import("conversation.zig"); const config_mod = @import("config.zig"); +const tool_registry_mod = @import("tool_registry.zig"); // ----------------------------------------------------------------------------- // Request serialization @@ -33,6 +34,7 @@ pub fn serializeRequest( allocator: Allocator, cfg: *const config_mod.AnthropicMessagesConfig, conv: *const conversation.Conversation, + tools: *const tool_registry_mod.ToolRegistry, ) ![]u8 { var aw: Writer.Allocating = .init(allocator); errdefer aw.deinit(); @@ -59,6 +61,23 @@ pub fn serializeRequest( try s.write(system_buf.items); } + if (tools.count() > 0) { + try s.objectField("tools"); + try s.beginArray(); + var it = tools.iterator(); + while (it.next()) |t| { + try s.beginObject(); + try s.objectField("name"); + try s.write(t.name); + try s.objectField("description"); + try s.write(t.description); + try s.objectField("input_schema"); + try writeRawJson(&s, t.schema_json); + try s.endObject(); + } + try s.endArray(); + } + // Emit messages (everything that isn't .system). try s.objectField("messages"); try s.beginArray(); @@ -73,6 +92,24 @@ pub fn serializeRequest( return try aw.toOwnedSlice(); } +/// Splice a pre-encoded JSON value into the current stringifier position. +/// On parse failure, emit `{}` so we never produce invalid wire JSON. +fn writeRawJson(s: *std.json.Stringify, raw: []const u8) !void { + var arena = std.heap.ArenaAllocator.init(std.heap.page_allocator); + defer arena.deinit(); + const parsed = std.json.parseFromSlice( + std.json.Value, + arena.allocator(), + raw, + .{}, + ) catch { + try s.beginObject(); + try s.endObject(); + return; + }; + try s.write(parsed.value); +} + fn collectSystemPrompt( conv: *const conversation.Conversation, out: *std.ArrayList(u8), @@ -136,8 +173,38 @@ fn writeBlock(s: *std.json.Stringify, block: conversation.ContentBlock) !void { try s.write(sig); try s.endObject(); }, - // ToolUse / ToolResult: phase 3+. - .ToolUse, .ToolResult => {}, + .ToolUse => |tu| { + try s.beginObject(); + try s.objectField("type"); + try s.write("tool_use"); + try s.objectField("id"); + try s.write(tu.id); + try s.objectField("name"); + try s.write(tu.name); + try s.objectField("input"); + // Anthropic expects `input` as a nested JSON object, not a + // string. The block's input bytes were assembled from + // `input_json_delta` fragments — splice them in verbatim. + // Treat empty as an empty object. + const input_bytes = tu.input.items; + if (input_bytes.len == 0) { + try s.beginObject(); + try s.endObject(); + } else { + try writeRawJson(s, input_bytes); + } + try s.endObject(); + }, + .ToolResult => |tr| { + try s.beginObject(); + try s.objectField("type"); + try s.write("tool_result"); + try s.objectField("tool_use_id"); + try s.write(tr.tool_use_id); + try s.objectField("content"); + try s.write(tr.content.items); + try s.endObject(); + }, } } @@ -382,7 +449,9 @@ test "serializeRequest - system extracted into top-level field" { try conv.addUserMessage("Hello!"); const cfg = testConfig("claude-sonnet-4-20250514"); - const body = try serializeRequest(allocator, &cfg, &conv); + var empty_tools = tool_registry_mod.ToolRegistry.init(allocator); + defer empty_tools.deinit(); + const body = try serializeRequest(allocator, &cfg, &conv, &empty_tools); defer allocator.free(body); var parsed = try std.json.parseFromSlice(std.json.Value, allocator, body, .{}); @@ -417,7 +486,9 @@ test "serializeRequest - multiple system messages concatenated with newlines" { try conv.addUserMessage("Hi"); const cfg = testConfig("claude-x"); - const body = try serializeRequest(allocator, &cfg, &conv); + var empty_tools = tool_registry_mod.ToolRegistry.init(allocator); + defer empty_tools.deinit(); + const body = try serializeRequest(allocator, &cfg, &conv, &empty_tools); defer allocator.free(body); var parsed = try std.json.parseFromSlice(std.json.Value, allocator, body, .{}); @@ -436,7 +507,9 @@ test "serializeRequest - no system messages omits the system field" { try conv.addUserMessage("Hi"); const cfg = testConfig("claude-x"); - const body = try serializeRequest(allocator, &cfg, &conv); + var empty_tools = tool_registry_mod.ToolRegistry.init(allocator); + defer empty_tools.deinit(); + const body = try serializeRequest(allocator, &cfg, &conv, &empty_tools); defer allocator.free(body); var parsed = try std.json.parseFromSlice(std.json.Value, allocator, body, .{}); @@ -460,7 +533,9 @@ test "serializeRequest - signed assistant Thinking blocks round-trip" { }); const cfg = testConfig("claude-x"); - const body = try serializeRequest(allocator, &cfg, &conv); + var empty_tools = tool_registry_mod.ToolRegistry.init(allocator); + defer empty_tools.deinit(); + const body = try serializeRequest(allocator, &cfg, &conv, &empty_tools); defer allocator.free(body); var parsed = try std.json.parseFromSlice(std.json.Value, allocator, body, .{}); @@ -501,7 +576,9 @@ test "serializeRequest - unsigned Thinking blocks are dropped" { }); const cfg = testConfig("claude-x"); - const body = try serializeRequest(allocator, &cfg, &conv); + var empty_tools = tool_registry_mod.ToolRegistry.init(allocator); + defer empty_tools.deinit(); + const body = try serializeRequest(allocator, &cfg, &conv, &empty_tools); defer allocator.free(body); var parsed = try std.json.parseFromSlice(std.json.Value, allocator, body, .{}); @@ -636,3 +713,162 @@ test "parseStreamEvent - unknown type" { defer pe.deinit(); try testing.expectEqual(StreamEventTag.unknown, @as(StreamEventTag, pe.event)); } + +// ----------------------------------------------------------------------------- +// Phase 3: tools serialization, ToolUse + ToolResult content blocks +// ----------------------------------------------------------------------------- + +const tool_mod = @import("tool.zig"); + +const StaticToolVT = struct { + fn invoke(_: *anyopaque, _: []const u8, _: Allocator) anyerror![]u8 { + return error.NotImplementedInTest; + } + fn deinit_(_: *anyopaque, _: Allocator) void {} + const v: tool_mod.Tool.VTable = .{ .invoke = invoke, .deinit = deinit_ }; +}; +var static_tool_ctx_sentinel: u8 = 0; +fn makeStaticTool( + name: []const u8, + description: []const u8, + schema: []const u8, +) tool_mod.Tool { + return .{ + .name = name, + .description = description, + .schema_json = schema, + .ctx = &static_tool_ctx_sentinel, + .vtable = &StaticToolVT.v, + }; +} + +test "serializeRequest - emits tools array when registry non-empty" { + const allocator = testing.allocator; + + var conv = conversation.Conversation.init(allocator); + defer conv.deinit(); + try conv.addUserMessage("call something"); + + var tools = tool_registry_mod.ToolRegistry.init(allocator); + defer tools.deinit(); + try tools.register(makeStaticTool( + "echo", + "Echo a message back.", + \\{"type":"object","properties":{"message":{"type":"string"}},"required":["message"]} + )); + + const cfg = testConfig("claude-x"); + const body = try serializeRequest(allocator, &cfg, &conv, &tools); + defer allocator.free(body); + + var parsed = try std.json.parseFromSlice(std.json.Value, allocator, body, .{}); + defer parsed.deinit(); + + const arr = parsed.value.object.get("tools").?.array.items; + try testing.expectEqual(@as(usize, 1), arr.len); + try testing.expectEqualStrings("echo", arr[0].object.get("name").?.string); + try testing.expectEqualStrings("Echo a message back.", arr[0].object.get("description").?.string); + + const schema = arr[0].object.get("input_schema").?.object; + try testing.expectEqualStrings("object", schema.get("type").?.string); + try testing.expect(schema.get("properties").? == .object); +} + +test "serializeRequest - assistant ToolUse becomes tool_use content block" { + const allocator = testing.allocator; + + var conv = conversation.Conversation.init(allocator); + defer conv.deinit(); + + const id = try allocator.dupe(u8, "tu_1"); + const name = try allocator.dupe(u8, "echo"); + var input: conversation.TextualBlock = .empty; + try input.appendSlice(allocator, "{\"message\":\"hi\"}"); + + try conv.addAssistantMessage(&.{ + .{ .Text = try conversation.textualBlockFromSlice(allocator, "calling tool") }, + .{ .ToolUse = .{ .id = id, .name = name, .input = input } }, + }); + + var empty_tools = tool_registry_mod.ToolRegistry.init(allocator); + defer empty_tools.deinit(); + const cfg = testConfig("claude-x"); + const body = try serializeRequest(allocator, &cfg, &conv, &empty_tools); + defer allocator.free(body); + + var parsed = try std.json.parseFromSlice(std.json.Value, allocator, body, .{}); + defer parsed.deinit(); + + const content = parsed.value.object.get("messages").?.array.items[0] + .object.get("content").?.array.items; + try testing.expectEqual(@as(usize, 2), content.len); + try testing.expectEqualStrings("text", content[0].object.get("type").?.string); + + try testing.expectEqualStrings("tool_use", content[1].object.get("type").?.string); + try testing.expectEqualStrings("tu_1", content[1].object.get("id").?.string); + try testing.expectEqualStrings("echo", content[1].object.get("name").?.string); + // `input` is a nested JSON object, NOT a string. + const inp = content[1].object.get("input").?.object; + try testing.expectEqualStrings("hi", inp.get("message").?.string); +} + +test "serializeRequest - user ToolResult becomes tool_result content block" { + const allocator = testing.allocator; + + var conv = conversation.Conversation.init(allocator); + defer conv.deinit(); + + const id = try allocator.dupe(u8, "tu_1"); + var content: conversation.TextualBlock = .empty; + try content.appendSlice(allocator, "the answer is 42"); + + var blocks: std.ArrayList(conversation.ContentBlock) = .empty; + try blocks.append(allocator, .{ .ToolResult = .{ .tool_use_id = id, .content = content } }); + try conv.messages.append(allocator, .{ .role = .user, .content = blocks }); + + var empty_tools = tool_registry_mod.ToolRegistry.init(allocator); + defer empty_tools.deinit(); + const cfg = testConfig("claude-x"); + const body = try serializeRequest(allocator, &cfg, &conv, &empty_tools); + defer allocator.free(body); + + var parsed = try std.json.parseFromSlice(std.json.Value, allocator, body, .{}); + defer parsed.deinit(); + + const msg = parsed.value.object.get("messages").?.array.items[0].object; + try testing.expectEqualStrings("user", msg.get("role").?.string); + + const arr = msg.get("content").?.array.items; + try testing.expectEqual(@as(usize, 1), arr.len); + try testing.expectEqualStrings("tool_result", arr[0].object.get("type").?.string); + try testing.expectEqualStrings("tu_1", arr[0].object.get("tool_use_id").?.string); + try testing.expectEqualStrings("the answer is 42", arr[0].object.get("content").?.string); +} + +test "serializeRequest - empty ToolUse input becomes {} not invalid JSON" { + const allocator = testing.allocator; + + var conv = conversation.Conversation.init(allocator); + defer conv.deinit(); + + const id = try allocator.dupe(u8, "tu_1"); + const name = try allocator.dupe(u8, "noargs"); + try conv.addAssistantMessage(&.{ + .{ .ToolUse = .{ .id = id, .name = name, .input = .empty } }, + }); + + var empty_tools = tool_registry_mod.ToolRegistry.init(allocator); + defer empty_tools.deinit(); + const cfg = testConfig("claude-x"); + const body = try serializeRequest(allocator, &cfg, &conv, &empty_tools); + defer allocator.free(body); + + var parsed = try std.json.parseFromSlice(std.json.Value, allocator, body, .{}); + defer parsed.deinit(); + + const tu = parsed.value.object.get("messages").?.array.items[0] + .object.get("content").?.array.items[0].object; + try testing.expectEqualStrings("tool_use", tu.get("type").?.string); + try testing.expect(tu.get("input").? == .object); + try testing.expectEqual(@as(usize, 0), tu.get("input").?.object.count()); +} diff --git a/libpanto/src/openai_chat_json.zig b/libpanto/src/openai_chat_json.zig index 465d951..7ca9546 100644 --- a/libpanto/src/openai_chat_json.zig +++ b/libpanto/src/openai_chat_json.zig @@ -10,6 +10,7 @@ const Allocator = std.mem.Allocator; const Writer = std.Io.Writer; const conversation = @import("conversation.zig"); const config_mod = @import("config.zig"); +const tool_registry_mod = @import("tool_registry.zig"); /// A single parsed streaming chunk. Fields are populated only when present /// in the wire payload; null fields signal "not in this chunk". @@ -21,6 +22,26 @@ pub const StreamDelta = struct { content: ?[]const u8 = null, reasoning_content: ?[]const u8 = null, finish_reason: ?[]const u8 = null, + tool_calls: []const ToolCallDelta = &.{}, + /// Mid-stream error reported by the provider. OpenAI-compatible + /// endpoints sometimes return HTTP 200 with `data: {"error":{...}}` + /// embedded in the SSE stream instead of a non-2xx response (notably + /// some OpenRouter / MiniMax / DeepSeek paths, and occasionally OpenAI + /// itself on transient overload). When present, the provider must + /// treat the turn as failed. + error_message: ?[]const u8 = null, + error_type: ?[]const u8 = null, +}; + +/// A single entry from a streaming `tool_calls` array. Multiple parallel +/// tool calls are distinguished by their `index`; identity fields (`id`, +/// `name`) typically arrive only on the first delta for each index, while +/// `arguments` arrives incrementally across many deltas. +pub const ToolCallDelta = struct { + index: usize, + id: ?[]const u8 = null, + name: ?[]const u8 = null, + arguments: ?[]const u8 = null, }; /// Serialize a Conversation into a `chat/completions` request body. @@ -30,6 +51,7 @@ pub fn serializeRequest( allocator: Allocator, cfg: *const config_mod.OpenAIChatConfig, conv: *const conversation.Conversation, + tools: *const tool_registry_mod.ToolRegistry, ) ![]u8 { var aw: Writer.Allocating = .init(allocator); errdefer aw.deinit(); @@ -56,6 +78,29 @@ pub fn serializeRequest( }, } + if (tools.count() > 0) { + try s.objectField("tools"); + try s.beginArray(); + var it = tools.iterator(); + while (it.next()) |t| { + try s.beginObject(); + try s.objectField("type"); + try s.write("function"); + try s.objectField("function"); + try s.beginObject(); + try s.objectField("name"); + try s.write(t.name); + try s.objectField("description"); + try s.write(t.description); + try s.objectField("parameters"); + // Emit the tool's JSON Schema verbatim. + try writeRawJson(&s, t.schema_json); + try s.endObject(); + try s.endObject(); + } + try s.endArray(); + } + try s.objectField("messages"); try s.beginArray(); for (conv.messages.items) |msg| { @@ -68,24 +113,130 @@ pub fn serializeRequest( return try aw.toOwnedSlice(); } +/// Emit a pre-encoded JSON value into the current stringifier position. +/// +/// `std.json.Stringify.write` can only emit Zig values; we need to splice +/// arbitrary user-supplied JSON (tool schemas, parsed tool inputs) into +/// the output. We do that by parsing the bytes into `std.json.Value` and +/// asking the stringifier to write them. If parsing fails, fall back to +/// emitting an empty object so we never produce invalid JSON on the wire. +fn writeRawJson(s: *std.json.Stringify, raw: []const u8) !void { + // Use the stringifier's own allocator path via a scratch arena. + var arena = std.heap.ArenaAllocator.init(std.heap.page_allocator); + defer arena.deinit(); + const parsed = std.json.parseFromSlice( + std.json.Value, + arena.allocator(), + raw, + .{}, + ) catch { + try s.beginObject(); + try s.endObject(); + return; + }; + try s.write(parsed.value); +} + +/// Emit one `Conversation.Message` as one or more wire-level messages. +/// +/// OpenAI's wire format is awkward here: a single logical `user` turn that +/// contains ToolResult blocks must be split into separate top-level +/// `{"role":"tool",...}` messages (one per ToolResult). A single assistant +/// turn that mixes Text and ToolUse becomes one assistant message with both +/// a `content` string and a `tool_calls` array. fn writeMessage(s: *std.json.Stringify, msg: conversation.Message, allocator: Allocator) !void { - try s.beginObject(); + // User messages that carry ToolResult blocks fan out into one + // `role:"tool"` message per block. Any plain Text blocks in the same + // user message become a separate user message after the tool messages. + if (msg.role == .user) { + var has_tool_result = false; + for (msg.content.items) |b| { + if (b == .ToolResult) { + has_tool_result = true; + break; + } + } + if (has_tool_result) { + for (msg.content.items) |block| { + if (block != .ToolResult) continue; + const tr = block.ToolResult; + try s.beginObject(); + try s.objectField("role"); + try s.write("tool"); + try s.objectField("tool_call_id"); + try s.write(tr.tool_use_id); + try s.objectField("content"); + try s.write(tr.content.items); + try s.endObject(); + } + // Trailing plain Text blocks (rare in practice) ride along as + // a follow-up user message so we don't lose them. + var has_text = false; + for (msg.content.items) |b| { + if (b == .Text) { + has_text = true; + break; + } + } + if (!has_text) return; + try s.beginObject(); + try s.objectField("role"); + try s.write("user"); + try s.objectField("content"); + var buf: std.ArrayList(u8) = .empty; + defer buf.deinit(allocator); + try concatTextBlocks(msg.content.items, &buf, allocator); + try s.write(buf.items); + try s.endObject(); + return; + } + } + try s.beginObject(); try s.objectField("role"); try s.write(@tagName(msg.role)); - // All roles flatten to a plain `content` string in outbound requests. - // Thinking blocks are intentionally dropped from history: the openai_chat - // dialect has no portable way to round-trip them (OpenAI/DeepSeek strip; - // OpenRouter/NanoGPT want a `reasoning` field instead of an inline block; - // none accept the `{"type":"thinking",...}` shape). Preserving reasoning - // across turns will land with tool-use in a later phase. + // Assistant messages may carry ToolUse blocks. The wire shape is a + // `tool_calls` array alongside `content`. OpenAI requires `content` + // to be either a string or null — we always emit a string (possibly + // empty) so JSON shape is predictable. try s.objectField("content"); var buf: std.ArrayList(u8) = .empty; defer buf.deinit(allocator); try concatTextBlocks(msg.content.items, &buf, allocator); try s.write(buf.items); + if (msg.role == .assistant) { + var n_tool_uses: usize = 0; + for (msg.content.items) |b| if (b == .ToolUse) { + n_tool_uses += 1; + }; + if (n_tool_uses > 0) { + try s.objectField("tool_calls"); + try s.beginArray(); + for (msg.content.items) |block| { + if (block != .ToolUse) continue; + const tu = block.ToolUse; + try s.beginObject(); + try s.objectField("id"); + try s.write(tu.id); + try s.objectField("type"); + try s.write("function"); + try s.objectField("function"); + try s.beginObject(); + try s.objectField("name"); + try s.write(tu.name); + try s.objectField("arguments"); + // `arguments` is a string carrying JSON, per the OpenAI + // wire format — not a nested object. + try s.write(tu.input.items); + try s.endObject(); + try s.endObject(); + } + try s.endArray(); + } + } + try s.endObject(); } @@ -97,7 +248,7 @@ fn concatTextBlocks( for (blocks) |block| { switch (block) { .Text => |tb| try out.appendSlice(allocator, tb.items), - // Thinking: dropped. ToolUse/ToolResult: phase 3+. + // Thinking, ToolUse, ToolResult: handled elsewhere or dropped. else => {}, } } @@ -111,8 +262,13 @@ fn concatTextBlocks( pub const ParsedDelta = struct { parsed: std.json.Parsed(std.json.Value), delta: StreamDelta, + /// Owned buffer holding the per-call deltas referenced by + /// `delta.tool_calls`. Freed by `deinit` along with `parsed`. + tool_calls_buf: ?[]ToolCallDelta = null, + allocator: Allocator, pub fn deinit(self: *ParsedDelta) void { + if (self.tool_calls_buf) |b| self.allocator.free(b); self.parsed.deinit(); } }; @@ -124,19 +280,44 @@ pub fn parseStreamEvent(allocator: Allocator, payload: []const u8) !ParsedDelta var delta: StreamDelta = .{}; const root = parsed.value; - if (root != .object) return .{ .parsed = parsed, .delta = delta }; + if (root != .object) return .{ .parsed = parsed, .delta = delta, .allocator = allocator }; + + // Top-level `error` field. Some providers (and OpenAI itself on rare + // mid-stream failures) emit `data: {"error":{"message":...,"type":...}}` + // with HTTP 200, so we look for this BEFORE the choices array. + if (root.object.get("error")) |e| { + switch (e) { + .object => |obj| { + if (obj.get("message")) |m| if (m == .string) { + delta.error_message = m.string; + }; + if (obj.get("type")) |t| if (t == .string) { + delta.error_type = t.string; + }; + }, + .string => |s| { + // Some providers send a bare string. Surface it as the + // message so callers can still report something useful. + delta.error_message = s; + }, + else => {}, + } + } - const choices_v = root.object.get("choices") orelse return .{ .parsed = parsed, .delta = delta }; + const choices_v = root.object.get("choices") orelse return .{ .parsed = parsed, .delta = delta, .allocator = allocator }; if (choices_v != .array or choices_v.array.items.len == 0) { - return .{ .parsed = parsed, .delta = delta }; + return .{ .parsed = parsed, .delta = delta, .allocator = allocator }; } const choice = choices_v.array.items[0]; - if (choice != .object) return .{ .parsed = parsed, .delta = delta }; + if (choice != .object) return .{ .parsed = parsed, .delta = delta, .allocator = allocator }; if (choice.object.get("finish_reason")) |fr| { if (fr == .string) delta.finish_reason = fr.string; } + var tool_calls_buf: ?[]ToolCallDelta = null; + errdefer if (tool_calls_buf) |b| allocator.free(b); + if (choice.object.get("delta")) |d| { if (d == .object) { if (d.object.get("role")) |r| { @@ -152,10 +333,46 @@ pub fn parseStreamEvent(allocator: Allocator, payload: []const u8) !ParsedDelta } else if (d.object.get("reasoning")) |rc| { if (rc == .string) delta.reasoning_content = rc.string; } + if (d.object.get("tool_calls")) |tcs| { + if (tcs == .array and tcs.array.items.len > 0) { + const buf = try allocator.alloc(ToolCallDelta, tcs.array.items.len); + tool_calls_buf = buf; + for (tcs.array.items, 0..) |tc, i| { + var entry: ToolCallDelta = .{ .index = 0 }; + if (tc == .object) { + if (tc.object.get("index")) |iv| { + if (iv == .integer and iv.integer >= 0) { + entry.index = @intCast(iv.integer); + } + } + if (tc.object.get("id")) |idv| { + if (idv == .string) entry.id = idv.string; + } + if (tc.object.get("function")) |fnv| { + if (fnv == .object) { + if (fnv.object.get("name")) |nv| { + if (nv == .string) entry.name = nv.string; + } + if (fnv.object.get("arguments")) |av| { + if (av == .string) entry.arguments = av.string; + } + } + } + } + buf[i] = entry; + } + delta.tool_calls = buf; + } + } } } - return .{ .parsed = parsed, .delta = delta }; + return .{ + .parsed = parsed, + .delta = delta, + .tool_calls_buf = tool_calls_buf, + .allocator = allocator, + }; } // ----------------------------------------------------------------------------- @@ -172,6 +389,11 @@ fn testConfig(model: []const u8) config_mod.OpenAIChatConfig { }; } +/// Caller deinits. +fn emptyTools() tool_registry_mod.ToolRegistry { + return tool_registry_mod.ToolRegistry.init(testing.allocator); +} + test "serializeRequest - system + user" { const allocator = testing.allocator; @@ -182,7 +404,9 @@ test "serializeRequest - system + user" { try conv.addUserMessage("Hello!"); const cfg = testConfig("gpt-4o"); - const body = try serializeRequest(allocator, &cfg, &conv); + var tools = emptyTools(); + defer tools.deinit(); + const body = try serializeRequest(allocator, &cfg, &conv, &tools); defer allocator.free(body); var parsed = try std.json.parseFromSlice(std.json.Value, allocator, body, .{}); @@ -193,6 +417,8 @@ test "serializeRequest - system + user" { try testing.expect(root.get("stream").?.bool); // reasoning_effort is omitted when set to .default. try testing.expect(root.get("reasoning_effort") == null); + // No tools registered — the `tools` field must be omitted entirely. + try testing.expect(root.get("tools") == null); const msgs = root.get("messages").?.array.items; try testing.expectEqual(@as(usize, 2), msgs.len); @@ -214,7 +440,9 @@ test "serializeRequest - assistant Thinking blocks are stripped from outbound hi }); const cfg = testConfig("gpt-4o"); - const body = try serializeRequest(allocator, &cfg, &conv); + var tools = emptyTools(); + defer tools.deinit(); + const body = try serializeRequest(allocator, &cfg, &conv, &tools); defer allocator.free(body); var parsed = try std.json.parseFromSlice(std.json.Value, allocator, body, .{}); @@ -237,7 +465,9 @@ test "serializeRequest - reasoning effort level included when set" { var cfg = testConfig("gpt-4o"); cfg.reasoning = .high; - const body = try serializeRequest(allocator, &cfg, &conv); + var tools = emptyTools(); + defer tools.deinit(); + const body = try serializeRequest(allocator, &cfg, &conv, &tools); defer allocator.free(body); var parsed = try std.json.parseFromSlice(std.json.Value, allocator, body, .{}); @@ -259,7 +489,9 @@ test "serializeRequest - reasoning .off sends \"none\"" { var cfg = testConfig("gpt-4o"); cfg.reasoning = .off; - const body = try serializeRequest(allocator, &cfg, &conv); + var tools = emptyTools(); + defer tools.deinit(); + const body = try serializeRequest(allocator, &cfg, &conv, &tools); defer allocator.free(body); var parsed = try std.json.parseFromSlice(std.json.Value, allocator, body, .{}); @@ -321,3 +553,221 @@ test "parseStreamEvent - reasoning_content" { try testing.expectEqualStrings("hmm", pd.delta.reasoning_content.?); } + +// ----------------------------------------------------------------------------- +// Phase 3: tools serialization, tool_calls parsing +// ----------------------------------------------------------------------------- + +const tool_mod = @import("tool.zig"); + +/// Minimal in-test tool: borrows its name/description/schema slices from +/// the test's stack. The vtable's deinit is a no-op since nothing is owned. +const StaticToolVT = struct { + fn invoke(_: *anyopaque, _: []const u8, _: Allocator) anyerror![]u8 { + return error.NotImplementedInTest; + } + fn deinit_(_: *anyopaque, _: Allocator) void {} + + const v: tool_mod.Tool.VTable = .{ + .invoke = invoke, + .deinit = deinit_, + }; +}; +var static_tool_ctx_sentinel: u8 = 0; +fn makeStaticTool( + name: []const u8, + description: []const u8, + schema: []const u8, +) tool_mod.Tool { + return .{ + .name = name, + .description = description, + .schema_json = schema, + .ctx = &static_tool_ctx_sentinel, + .vtable = &StaticToolVT.v, + }; +} + +test "serializeRequest - emits tools array when registry non-empty" { + const allocator = testing.allocator; + + var conv = conversation.Conversation.init(allocator); + defer conv.deinit(); + try conv.addUserMessage("call something"); + + var tools = emptyTools(); + defer tools.deinit(); + + try tools.register(makeStaticTool( + "echo", + "Echo a message back.", + \\{"type":"object","properties":{"message":{"type":"string"}},"required":["message"]} + )); + + const cfg = testConfig("gpt-4o"); + const body = try serializeRequest(allocator, &cfg, &conv, &tools); + defer allocator.free(body); + + var parsed = try std.json.parseFromSlice(std.json.Value, allocator, body, .{}); + defer parsed.deinit(); + + const arr = parsed.value.object.get("tools").?.array.items; + try testing.expectEqual(@as(usize, 1), arr.len); + try testing.expectEqualStrings("function", arr[0].object.get("type").?.string); + + const f = arr[0].object.get("function").?.object; + try testing.expectEqualStrings("echo", f.get("name").?.string); + try testing.expectEqualStrings("Echo a message back.", f.get("description").?.string); + + const params = f.get("parameters").?.object; + try testing.expectEqualStrings("object", params.get("type").?.string); + try testing.expect(params.get("properties").? == .object); + try testing.expect(params.get("required").? == .array); +} + +test "serializeRequest - assistant ToolUse becomes tool_calls" { + const allocator = testing.allocator; + + var conv = conversation.Conversation.init(allocator); + defer conv.deinit(); + + const id = try allocator.dupe(u8, "call_1"); + const name = try allocator.dupe(u8, "echo"); + var args: conversation.TextualBlock = .empty; + try args.appendSlice(allocator, "{\"message\":\"hi\"}"); + + try conv.addAssistantMessage(&.{ + .{ .Text = try conversation.textualBlockFromSlice(allocator, "calling tool") }, + .{ .ToolUse = .{ .id = id, .name = name, .input = args } }, + }); + + var tools = emptyTools(); + defer tools.deinit(); + const cfg = testConfig("gpt-4o"); + const body = try serializeRequest(allocator, &cfg, &conv, &tools); + defer allocator.free(body); + + var parsed = try std.json.parseFromSlice(std.json.Value, allocator, body, .{}); + defer parsed.deinit(); + + const msg = parsed.value.object.get("messages").?.array.items[0].object; + try testing.expectEqualStrings("assistant", msg.get("role").?.string); + try testing.expectEqualStrings("calling tool", msg.get("content").?.string); + + const tcs = msg.get("tool_calls").?.array.items; + try testing.expectEqual(@as(usize, 1), tcs.len); + try testing.expectEqualStrings("call_1", tcs[0].object.get("id").?.string); + try testing.expectEqualStrings("function", tcs[0].object.get("type").?.string); + + const fn_obj = tcs[0].object.get("function").?.object; + try testing.expectEqualStrings("echo", fn_obj.get("name").?.string); + // `arguments` is a string (JSON-as-string) per the OpenAI wire format. + try testing.expectEqualStrings("{\"message\":\"hi\"}", fn_obj.get("arguments").?.string); +} + +test "serializeRequest - user ToolResult fans out into tool messages" { + const allocator = testing.allocator; + + var conv = conversation.Conversation.init(allocator); + defer conv.deinit(); + + const id1 = try allocator.dupe(u8, "call_a"); + var c1: conversation.TextualBlock = .empty; + try c1.appendSlice(allocator, "42"); + + const id2 = try allocator.dupe(u8, "call_b"); + var c2: conversation.TextualBlock = .empty; + try c2.appendSlice(allocator, "oops"); + + var content: std.ArrayList(conversation.ContentBlock) = .empty; + try content.append(allocator, .{ .ToolResult = .{ .tool_use_id = id1, .content = c1 } }); + try content.append(allocator, .{ .ToolResult = .{ .tool_use_id = id2, .content = c2 } }); + try conv.messages.append(allocator, .{ .role = .user, .content = content }); + + var tools = emptyTools(); + defer tools.deinit(); + const cfg = testConfig("gpt-4o"); + const body = try serializeRequest(allocator, &cfg, &conv, &tools); + defer allocator.free(body); + + var parsed = try std.json.parseFromSlice(std.json.Value, allocator, body, .{}); + defer parsed.deinit(); + + const msgs = parsed.value.object.get("messages").?.array.items; + try testing.expectEqual(@as(usize, 2), msgs.len); + + try testing.expectEqualStrings("tool", msgs[0].object.get("role").?.string); + try testing.expectEqualStrings("call_a", msgs[0].object.get("tool_call_id").?.string); + try testing.expectEqualStrings("42", msgs[0].object.get("content").?.string); + + try testing.expectEqualStrings("tool", msgs[1].object.get("role").?.string); + try testing.expectEqualStrings("call_b", msgs[1].object.get("tool_call_id").?.string); + try testing.expectEqualStrings("oops", msgs[1].object.get("content").?.string); +} + +test "parseStreamEvent - tool_calls delta with id and partial arguments" { + const allocator = testing.allocator; + const payload = + \\{"choices":[{"delta":{"tool_calls":[{"index":0,"id":"call_xyz","type":"function","function":{"name":"echo","arguments":"{\"x\":"}}]}}]} + ; + var pd = try parseStreamEvent(allocator, payload); + defer pd.deinit(); + + try testing.expectEqual(@as(usize, 1), pd.delta.tool_calls.len); + const tc = pd.delta.tool_calls[0]; + try testing.expectEqual(@as(usize, 0), tc.index); + try testing.expectEqualStrings("call_xyz", tc.id.?); + try testing.expectEqualStrings("echo", tc.name.?); + try testing.expectEqualStrings("{\"x\":", tc.arguments.?); +} + +test "parseStreamEvent - tool_calls delta with only arguments fragment" { + const allocator = testing.allocator; + const payload = + \\{"choices":[{"delta":{"tool_calls":[{"index":0,"function":{"arguments":"1}"}}]}}]} + ; + var pd = try parseStreamEvent(allocator, payload); + defer pd.deinit(); + + try testing.expectEqual(@as(usize, 1), pd.delta.tool_calls.len); + const tc = pd.delta.tool_calls[0]; + try testing.expectEqual(@as(usize, 0), tc.index); + try testing.expect(tc.id == null); + try testing.expect(tc.name == null); + try testing.expectEqualStrings("1}", tc.arguments.?); +} + +test "parseStreamEvent - finish_reason tool_calls" { + const allocator = testing.allocator; + const payload = + \\{"choices":[{"delta":{},"finish_reason":"tool_calls"}]} + ; + var pd = try parseStreamEvent(allocator, payload); + defer pd.deinit(); + + try testing.expectEqualStrings("tool_calls", pd.delta.finish_reason.?); +} + +test "parseStreamEvent - top-level error event with type and message" { + const allocator = testing.allocator; + const payload = + \\{"error":{"message":"Rate limit exceeded","type":"rate_limit_error","code":429}} + ; + var pd = try parseStreamEvent(allocator, payload); + defer pd.deinit(); + + try testing.expectEqualStrings("Rate limit exceeded", pd.delta.error_message.?); + try testing.expectEqualStrings("rate_limit_error", pd.delta.error_type.?); +} + +test "parseStreamEvent - bare-string error" { + const allocator = testing.allocator; + const payload = + \\{"error":"something went wrong"} + ; + var pd = try parseStreamEvent(allocator, payload); + defer pd.deinit(); + + try testing.expectEqualStrings("something went wrong", pd.delta.error_message.?); + try testing.expect(pd.delta.error_type == null); +} diff --git a/libpanto/src/provider.zig b/libpanto/src/provider.zig index 6d97bb7..177fc2f 100644 --- a/libpanto/src/provider.zig +++ b/libpanto/src/provider.zig @@ -1,6 +1,9 @@ const std = @import("std"); -const conversation = @import("conversation.zig"); + const config_mod = @import("config.zig"); +const conversation = @import("conversation.zig"); +const tool_registry_mod = @import("tool_registry.zig"); +pub const ToolRegistry = tool_registry_mod.ToolRegistry; pub const ContentBlockType = enum { Text, @@ -66,7 +69,12 @@ pub const Receiver = struct { }; pub const ProviderVTable = struct { - streamStep: *const fn (*anyopaque, *conversation.Conversation, *Receiver) anyerror!void, + streamStep: *const fn ( + *anyopaque, + *conversation.Conversation, + *const ToolRegistry, + *Receiver, + ) anyerror!void, deinit: *const fn (*anyopaque) void, }; @@ -103,8 +111,13 @@ pub const Provider = struct { } } - pub fn streamStep(self: Provider, conv: *conversation.Conversation, receiver: *Receiver) anyerror!void { - return self.vtable.streamStep(self.ptr, conv, receiver); + pub fn streamStep( + self: Provider, + conv: *conversation.Conversation, + tools: *const ToolRegistry, + receiver: *Receiver, + ) anyerror!void { + return self.vtable.streamStep(self.ptr, conv, tools, receiver); } pub fn deinit(self: Provider) void { diff --git a/libpanto/src/provider_anthropic_messages.zig b/libpanto/src/provider_anthropic_messages.zig index 2914b6e..7511ce8 100644 --- a/libpanto/src/provider_anthropic_messages.zig +++ b/libpanto/src/provider_anthropic_messages.zig @@ -59,10 +59,11 @@ pub const AnthropicMessagesProvider = struct { fn vtableStreamStep( ptr: *anyopaque, conv: *conversation.Conversation, + tools: *const provider_mod.ToolRegistry, receiver: *provider_mod.Receiver, ) anyerror!void { const self: *AnthropicMessagesProvider = @ptrCast(@alignCast(ptr)); - return self.streamStep(conv, receiver); + return self.streamStep(conv, tools, receiver); } /// Called via the `Provider` interface. Tears down the impl AND frees @@ -79,11 +80,12 @@ pub const AnthropicMessagesProvider = struct { pub fn streamStep( self: *AnthropicMessagesProvider, conv: *conversation.Conversation, + tools: *const provider_mod.ToolRegistry, receiver: *provider_mod.Receiver, ) !void { // Outer wrapper guarantees `onError` is called exactly once if // anything fails. - self.streamStepInner(conv, receiver) catch |err| { + self.streamStepInner(conv, tools, receiver) catch |err| { receiver.onError(err); return err; }; @@ -92,6 +94,7 @@ pub const AnthropicMessagesProvider = struct { fn streamStepInner( self: *AnthropicMessagesProvider, conv: *conversation.Conversation, + tools: *const provider_mod.ToolRegistry, receiver: *provider_mod.Receiver, ) !void { const url = try std.fmt.allocPrint( @@ -103,7 +106,7 @@ pub const AnthropicMessagesProvider = struct { const uri = try Uri.parse(url); - const body = try json_mod.serializeRequest(self.allocator, &self.config, conv); + const body = try json_mod.serializeRequest(self.allocator, &self.config, conv, tools); defer self.allocator.free(body); const extra_headers = [_]http.Header{ @@ -165,6 +168,10 @@ pub const AnthropicMessagesProvider = struct { // Use `readVec` so we return to the event loop as soon as *any* // bytes arrive, rather than waiting for the buffer to fill. // `readSliceShort` blocks until EOF or full, which defeats streaming. + // + // Per `std.Io.Reader.readVec` docs: `n == 0` does NOT mean EOF; + // EOF is signalled only via `error.EndOfStream`. Breaking on + // `n == 0` truncates the response mid-stream. var chunk: [4096]u8 = undefined; var vecs: [1][]u8 = .{&chunk}; while (true) { @@ -172,12 +179,13 @@ pub const AnthropicMessagesProvider = struct { error.EndOfStream => break, else => return err, }; - if (n == 0) break; + if (n == 0) continue; const events = try parser.feed(chunk[0..n]); defer parser.freeEvents(events); for (events) |ev_payload| { + std.log.debug("anthropic_messages <= {s}", .{ev_payload}); try handleEvent(self.allocator, ev_payload, &state, receiver); if (state.end_of_stream) { try state.finalize(receiver, conv); @@ -214,9 +222,14 @@ const StreamState = struct { kind: BlockKind, text_buf: conversation.TextualBlock = .empty, signature: ?[]const u8 = null, + /// Populated for `.tool_use` blocks. Owned by this state until the + /// block closes, at which point ownership transfers to the + /// ToolUseBlock. + tool_id: ?[]u8 = null, + tool_name: ?[]u8 = null, }; - const BlockKind = enum { text, thinking, unsupported }; + const BlockKind = enum { text, thinking, tool_use, unsupported }; fn init(allocator: Allocator) StreamState { return .{ .allocator = allocator }; @@ -226,6 +239,8 @@ const StreamState = struct { if (self.active) |*a| { a.text_buf.deinit(self.allocator); if (a.signature) |sig| self.allocator.free(sig); + if (a.tool_id) |s| self.allocator.free(s); + if (a.tool_name) |s| self.allocator.free(s); } for (self.blocks.items) |*b| b.deinit(self.allocator); self.blocks.deinit(self.allocator); @@ -248,13 +263,22 @@ const StreamState = struct { if (self.active != null) { self.discardActive(); } - self.active = .{ + var ab: ActiveBlock = .{ .wire_index = wire_index, .kind = kind, }; + // For tool_use blocks, capture the identity fields from the meta. + if (kind == .tool_use) { + if (tool_meta) |m| { + if (m.tool_id) |id| ab.tool_id = try self.allocator.dupe(u8, id); + if (m.tool_name) |n| ab.tool_name = try self.allocator.dupe(u8, n); + } + } + self.active = ab; const block_type: ?provider_mod.ContentBlockType = switch (kind) { .text => .Text, .thinking => .Thinking, + .tool_use => .ToolUse, .unsupported => null, }; if (block_type) |bt| { @@ -273,6 +297,19 @@ const StreamState = struct { try receiver.onContentDelta(a.wire_index, delta); } + /// Append a chunk of the streamed JSON arguments for the active + /// tool_use block. No-op if the active block isn't a tool_use. + fn appendInputJsonDelta( + self: *StreamState, + receiver: *provider_mod.Receiver, + delta: []const u8, + ) !void { + const a = &(self.active orelse return); + if (a.kind != .tool_use) return; + try a.text_buf.appendSlice(self.allocator, delta); + try receiver.onContentDelta(a.wire_index, delta); + } + fn setSignature(self: *StreamState, sig: []const u8) !void { const a = &(self.active orelse return); if (a.signature) |old| self.allocator.free(old); @@ -288,9 +325,18 @@ const StreamState = struct { self.active = null; if (a.kind == .unsupported) { - // Drop unsupported blocks (e.g. tool_use until phase 3+). a.text_buf.deinit(self.allocator); if (a.signature) |sig| self.allocator.free(sig); + if (a.tool_id) |s| self.allocator.free(s); + if (a.tool_name) |s| self.allocator.free(s); + return; + } + // tool_use blocks require both id and name. If either is missing + // (malformed stream), drop the block defensively. + if (a.kind == .tool_use and (a.tool_id == null or a.tool_name == null)) { + a.text_buf.deinit(self.allocator); + if (a.tool_id) |s| self.allocator.free(s); + if (a.tool_name) |s| self.allocator.free(s); return; } @@ -303,6 +349,11 @@ const StreamState = struct { .text = a.text_buf, .signature = a.signature, } }, + .tool_use => .{ .ToolUse = .{ + .id = a.tool_id.?, + .name = a.tool_name.?, + .input = a.text_buf, + } }, .unsupported => unreachable, }; @@ -317,6 +368,8 @@ const StreamState = struct { if (self.active) |*a| { a.text_buf.deinit(self.allocator); if (a.signature) |sig| self.allocator.free(sig); + if (a.tool_id) |s| self.allocator.free(s); + if (a.tool_name) |s| self.allocator.free(s); self.active = null; } } @@ -362,17 +415,20 @@ fn handleEvent( const kind: StreamState.BlockKind = switch (s.kind) { .text => .text, .thinking => .thinking, - .tool_use, .unknown => .unsupported, + .tool_use => .tool_use, + .unknown => .unsupported, }; - // tool_use meta will start being honored in phase 3+. For now we - // mark the block unsupported and drop it. - try state.openBlock(receiver, s.index, kind, null); + const meta: ?provider_mod.BlockMeta = if (s.kind == .tool_use) + .{ .tool_id = s.tool_id, .tool_name = s.tool_name } + else + null; + try state.openBlock(receiver, s.index, kind, meta); }, .content_block_delta => |d| { if (d.text_delta) |t| try state.appendTextDelta(receiver, t); if (d.thinking_delta) |t| try state.appendTextDelta(receiver, t); if (d.signature_delta) |sig| try state.setSignature(sig); - // input_json_delta: phase 3+. + if (d.input_json_delta) |j| try state.appendInputJsonDelta(receiver, j); }, .content_block_stop => { try state.closeBlock(receiver); @@ -704,9 +760,7 @@ test "ping and unknown events are ignored" { ); } -test "tool_use blocks are dropped (deferred to phase 3)" { - // A tool_use block is started, receives input_json_delta events, and - // stops. Phase 2 should silently drop it without crashing. +test "tool_use blocks are captured with id, name, and assembled input" { const allocator = testing.allocator; var conv = conversation.Conversation.init(allocator); @@ -722,7 +776,9 @@ test "tool_use blocks are dropped (deferred to phase 3)" { , \\{"type":"content_block_start","index":0,"content_block":{"type":"tool_use","id":"tu_1","name":"calc","input":{}}} , - \\{"type":"content_block_delta","index":0,"delta":{"type":"input_json_delta","partial_json":"{\"x\":1}"}} + \\{"type":"content_block_delta","index":0,"delta":{"type":"input_json_delta","partial_json":"{\"x\":"}} + , + \\{"type":"content_block_delta","index":0,"delta":{"type":"input_json_delta","partial_json":"1}"}} , \\{"type":"content_block_stop","index":0} , @@ -738,10 +794,15 @@ test "tool_use blocks are dropped (deferred to phase 3)" { try runStreamedTurn(allocator, &conv, &recv, &events); - // Only the text block survives. const asst = conv.messages.items[1]; - try testing.expectEqual(@as(usize, 1), asst.content.items.len); - try testing.expectEqualStrings("done", asst.content.items[0].Text.items); + try testing.expectEqual(@as(usize, 2), asst.content.items.len); + + const tu = asst.content.items[0].ToolUse; + try testing.expectEqualStrings("tu_1", tu.id); + try testing.expectEqualStrings("calc", tu.name); + try testing.expectEqualStrings("{\"x\":1}", tu.input.items); + + try testing.expectEqualStrings("done", asst.content.items[1].Text.items); } test "error event propagates as Zig error" { diff --git a/libpanto/src/provider_openai_chat.zig b/libpanto/src/provider_openai_chat.zig index c03d797..2343ac1 100644 --- a/libpanto/src/provider_openai_chat.zig +++ b/libpanto/src/provider_openai_chat.zig @@ -24,7 +24,7 @@ const config_mod = @import("config.zig"); /// Active streaming block type tracked by the state machine. Mirrors the /// `ContentBlock` union variants but adds `.none` for "no block open yet". -const ActiveBlock = enum { none, text, thinking }; +const ActiveBlock = enum { none, text, thinking, tool_use }; pub const OpenAIChatProvider = struct { allocator: Allocator, @@ -58,10 +58,11 @@ pub const OpenAIChatProvider = struct { fn vtableStreamStep( ptr: *anyopaque, conv: *conversation.Conversation, + tools: *const provider_mod.ToolRegistry, receiver: *provider_mod.Receiver, ) anyerror!void { const self: *OpenAIChatProvider = @ptrCast(@alignCast(ptr)); - return self.streamStep(conv, receiver); + return self.streamStep(conv, tools, receiver); } /// Called via the `Provider` interface. Tears down the impl AND frees @@ -78,12 +79,13 @@ pub const OpenAIChatProvider = struct { pub fn streamStep( self: *OpenAIChatProvider, conv: *conversation.Conversation, + tools: *const provider_mod.ToolRegistry, receiver: *provider_mod.Receiver, ) !void { // Outer wrapper guarantees `onError` is called exactly once if // anything fails — whether the receiver, the HTTP transport, or the // SSE/JSON parsers. The inner `streamStepInner` does the real work. - self.streamStepInner(conv, receiver) catch |err| { + self.streamStepInner(conv, tools, receiver) catch |err| { receiver.onError(err); return err; }; @@ -92,6 +94,7 @@ pub const OpenAIChatProvider = struct { fn streamStepInner( self: *OpenAIChatProvider, conv: *conversation.Conversation, + tools: *const provider_mod.ToolRegistry, receiver: *provider_mod.Receiver, ) !void { // Build URL: "{base_url}/chat/completions" @@ -105,7 +108,7 @@ pub const OpenAIChatProvider = struct { const uri = try Uri.parse(url); // Build the request body. - const body = try json_mod.serializeRequest(self.allocator, &self.config, conv); + const body = try json_mod.serializeRequest(self.allocator, &self.config, conv, tools); defer self.allocator.free(body); // Auth header @@ -179,6 +182,12 @@ pub const OpenAIChatProvider = struct { // Use `readVec` so we return to the event loop as soon as *any* // bytes arrive, rather than waiting for the buffer to fill. // `readSliceShort` blocks until EOF or full, which defeats streaming. + // + // Per `std.Io.Reader.readVec` docs: a return value of 0 does NOT + // signal end-of-stream — it just means no new bytes were available + // this call, and the caller should try again. EOF is reported only + // via `error.EndOfStream`. Breaking on `n == 0` truncates the + // response and silently cuts off mid-stream. var chunk: [4096]u8 = undefined; var vecs: [1][]u8 = .{&chunk}; while (true) { @@ -186,12 +195,13 @@ pub const OpenAIChatProvider = struct { error.EndOfStream => break, else => return err, }; - if (n == 0) break; + if (n == 0) continue; const events = try parser.feed(chunk[0..n]); defer parser.freeEvents(events); for (events) |ev_payload| { + std.log.debug("openai_chat <= {s}", .{ev_payload}); if (std.mem.eql(u8, ev_payload, "[DONE]")) { try state.finalize(receiver, conv); return; @@ -212,6 +222,13 @@ pub const OpenAIChatProvider = struct { /// State maintained across the streaming response: which block is currently /// being assembled, accumulated content, and the assistant message being /// built up for the final `onMessageComplete` callback. +/// +/// Text and thinking blocks are mutually exclusive at the active position +/// (openai_chat streams one at a time and we infer transitions). ToolUse +/// blocks live in a side map keyed by the wire's per-call `index`; multiple +/// tool calls can stream concurrently in the same response. At finalize +/// time, tool_use blocks are appended after the text/thinking blocks in +/// ascending wire-index order. const StreamState = struct { allocator: Allocator, started: bool = false, @@ -221,37 +238,74 @@ const StreamState = struct { /// Set once `finalize` has run, to make it idempotent. finalized: bool = false, active: ActiveBlock = .none, + /// Block index reported to the receiver. Increments per block boundary. + /// Tool-use blocks reuse a per-tool-call counter (`tool_block_index` + /// inside `ToolUseInProgress`) for their `onContentDelta` reports. block_index: usize = 0, - /// Buffer for the currently-streaming block's content. - /// Owned by this state until the block is completed, at which point - /// ownership transfers to the assembled Message. + /// Buffer for the currently-streaming text/thinking block. Owned by + /// this state until the block is completed, at which point ownership + /// transfers to the assembled Message. current_buf: conversation.TextualBlock = .empty, - /// Assembled blocks for the final message. + /// Assembled non-tool-use blocks for the final message, in stream order. blocks: std.ArrayList(conversation.ContentBlock) = .empty, + /// In-progress tool_use blocks keyed by wire index. + tool_uses: std.AutoHashMap(usize, ToolUseInProgress), + + const ToolUseInProgress = struct { + /// Block index emitted to the receiver for this tool call's + /// onBlockStart / onContentDelta / onBlockComplete callbacks. + block_index: usize, + /// id/name are buffered as TextualBlocks because lenient providers + /// (OpenRouter passthroughs, some self-hosted backends) may stream + /// either field as fragments across multiple deltas. OpenAI itself + /// sends them whole on the first delta, but the structural cost of + /// supporting fragments is small and worth the robustness. + id_buf: conversation.TextualBlock = .empty, + name_buf: conversation.TextualBlock = .empty, + arguments: conversation.TextualBlock = .empty, + /// We defer `onBlockStart` until either the first argument fragment + /// arrives or finalize runs — whichever happens first — because by + /// then identity is almost certainly complete and we can pass a + /// well-formed `BlockMeta` to the receiver. + started: bool = false, + + fn deinit(self: *ToolUseInProgress, allocator: Allocator) void { + self.id_buf.deinit(allocator); + self.name_buf.deinit(allocator); + self.arguments.deinit(allocator); + } + }; + fn init(allocator: Allocator) StreamState { - return .{ .allocator = allocator }; + return .{ + .allocator = allocator, + .tool_uses = std.AutoHashMap(usize, ToolUseInProgress).init(allocator), + }; } fn deinit(self: *StreamState) void { self.current_buf.deinit(self.allocator); for (self.blocks.items) |*b| b.deinit(self.allocator); self.blocks.deinit(self.allocator); + var it = self.tool_uses.iterator(); + while (it.next()) |entry| entry.value_ptr.deinit(self.allocator); + self.tool_uses.deinit(); } - /// Close the active block (if any) and emit onBlockComplete. - /// Ownership of `current_buf` transfers into the appended block. + /// Close the active text/thinking block (if any) and emit + /// onBlockComplete. Ownership of `current_buf` transfers into the + /// appended block. fn closeActive(self: *StreamState, receiver: *provider_mod.Receiver) !void { if (self.active == .none) return; const block: conversation.ContentBlock = switch (self.active) { .text => .{ .Text = self.current_buf }, .thinking => .{ .Thinking = .{ .text = self.current_buf } }, - .none => unreachable, + .tool_use, .none => unreachable, }; - // The buffer ownership has moved into `block`; replace with empty. self.current_buf = .empty; try self.blocks.append(self.allocator, block); @@ -260,12 +314,13 @@ const StreamState = struct { self.active = .none; } - /// Open a new block of the given type, possibly closing a prior block. + /// Open a new text/thinking block, possibly closing a prior one. fn openBlock( self: *StreamState, new_active: ActiveBlock, receiver: *provider_mod.Receiver, ) !void { + std.debug.assert(new_active == .text or new_active == .thinking); if (self.active == new_active) return; if (self.active != .none) { try self.closeActive(receiver); @@ -275,7 +330,7 @@ const StreamState = struct { const block_type: provider_mod.ContentBlockType = switch (new_active) { .text => .Text, .thinking => .Thinking, - .none => unreachable, + .tool_use, .none => unreachable, }; try receiver.onBlockStart(block_type, self.block_index, null); } @@ -289,8 +344,68 @@ const StreamState = struct { try receiver.onContentDelta(self.block_index, delta); } - /// End the stream: close any open block and emit onMessageComplete. - /// Ownership of the assembled blocks transfers into the conversation. + /// Apply one streaming tool_call delta. Allocates a new + /// `ToolUseInProgress` slot on first sight of each wire index. + fn applyToolCallDelta( + self: *StreamState, + receiver: *provider_mod.Receiver, + d: json_mod.ToolCallDelta, + ) !void { + const gop = try self.tool_uses.getOrPut(d.index); + if (!gop.found_existing) { + // First sight: close any open text/thinking block so the + // tool_use gets its own block_index. + if (self.active != .none) { + try self.closeActive(receiver); + self.block_index += 1; + } + gop.value_ptr.* = .{ .block_index = self.block_index }; + self.block_index += 1; + } + const tu = gop.value_ptr; + + // Append identity fragments. Most providers send id+name whole on + // the first delta and never repeat them, but appending is the only + // correct behavior across the full range of OpenAI-compatible + // backends — some chunk these strings. + if (d.id) |s| try tu.id_buf.appendSlice(self.allocator, s); + if (d.name) |s| try tu.name_buf.appendSlice(self.allocator, s); + + // Defer `onBlockStart` until we have a complete identity. The + // first argument fragment is our signal that the provider is done + // emitting identity for this index. If finalize runs first (i.e. + // a tool call with no arguments), we emit it there. + if (d.arguments) |a| { + try self.emitStartIfNeeded(receiver, tu); + try tu.arguments.appendSlice(self.allocator, a); + try receiver.onContentDelta(tu.block_index, a); + } + } + + /// Emit `onBlockStart(.ToolUse, ...)` once per in-progress tool use, + /// passing whatever identity we have. Callers must invoke this before + /// the first `onContentDelta` or `onBlockComplete` for the block. + fn emitStartIfNeeded( + self: *StreamState, + receiver: *provider_mod.Receiver, + tu: *ToolUseInProgress, + ) !void { + _ = self; + if (tu.started) return; + tu.started = true; + const meta: ?provider_mod.BlockMeta = if (tu.id_buf.items.len > 0 or tu.name_buf.items.len > 0) + .{ + .tool_id = if (tu.id_buf.items.len > 0) tu.id_buf.items else null, + .tool_name = if (tu.name_buf.items.len > 0) tu.name_buf.items else null, + } + else + null; + try receiver.onBlockStart(.ToolUse, tu.block_index, meta); + } + + /// End the stream: close any open text/thinking block, finalize all + /// in-flight tool_use blocks (in ascending wire-index order), then + /// commit the assembled assistant Message to the conversation. fn finalize( self: *StreamState, receiver: *provider_mod.Receiver, @@ -301,14 +416,65 @@ const StreamState = struct { try self.closeActive(receiver); + // Collect tool_use indices in ascending order for deterministic + // ordering in the final message. + var indices: std.ArrayList(usize) = .empty; + defer indices.deinit(self.allocator); + var it = self.tool_uses.iterator(); + while (it.next()) |entry| try indices.append(self.allocator, entry.key_ptr.*); + std.mem.sort(usize, indices.items, {}, std.sort.asc(usize)); + + for (indices.items) |idx| { + const tu_ptr = self.tool_uses.getPtr(idx).?; + // Drop entries lacking id or name. The stream ended before + // the provider sent enough to identify which tool was being + // called — there's nothing we can dispatch. We log enough + // detail to make this diagnosable; the agent will surface the + // resulting empty assistant message as EmptyAssistantResponse. + if (tu_ptr.id_buf.items.len == 0 or tu_ptr.name_buf.items.len == 0) { + if (!@import("builtin").is_test) { + std.log.err( + "openai_chat: dropping incomplete tool_use at wire index {d}: id={d} bytes, name=\"{s}\", args={d} bytes", + .{ + idx, + tu_ptr.id_buf.items.len, + tu_ptr.name_buf.items, + tu_ptr.arguments.items.len, + }, + ); + } + tu_ptr.deinit(self.allocator); + continue; + } + + // If no arguments ever arrived, we haven't emitted onBlockStart + // yet — do it now so the receiver sees a balanced start/complete. + try self.emitStartIfNeeded(receiver, tu_ptr); + + const block_index = tu_ptr.block_index; + const id_owned = try tu_ptr.id_buf.toOwnedSlice(self.allocator); + const name_owned = try tu_ptr.name_buf.toOwnedSlice(self.allocator); + const block: conversation.ContentBlock = .{ .ToolUse = .{ + .id = id_owned, + .name = name_owned, + .input = tu_ptr.arguments, + } }; + // Ownership has moved into `block`; clear the in-progress slot. + tu_ptr.arguments = .empty; + + try self.blocks.append(self.allocator, block); + try receiver.onBlockComplete( + block_index, + self.blocks.items[self.blocks.items.len - 1], + ); + } + // Move blocks into a fresh conversation message. const moved_blocks = try self.blocks.toOwnedSlice(self.allocator); defer self.allocator.free(moved_blocks); try conv.addAssistantMessage(moved_blocks); - // The conversation now owns the block buffers. Build a Message view - // for the callback (borrowed from the conversation). const msg = conv.messages.items[conv.messages.items.len - 1]; try receiver.onMessageComplete(msg); } @@ -324,6 +490,18 @@ fn handleEvent( defer parsed.deinit(); const d = parsed.delta; + // Mid-stream provider error: some OpenAI-compatible endpoints (and + // OpenAI itself on rare transient failures) return HTTP 200 with an + // error embedded in the SSE stream. Treat the turn as failed. + if (d.error_message != null or d.error_type != null) { + if (!@import("builtin").is_test) { + std.log.err("openai_chat stream error: {?s}: {?s}", .{ + d.error_type, d.error_message, + }); + } + return error.StreamError; + } + if (!state.started and d.role != null) { state.started = true; try receiver.onMessageStart(.assistant); @@ -347,6 +525,14 @@ fn handleEvent( try state.appendDelta(receiver, c); } + if (d.tool_calls.len > 0) { + if (!state.started) { + state.started = true; + try receiver.onMessageStart(.assistant); + } + for (d.tool_calls) |tc| try state.applyToolCallDelta(receiver, tc); + } + if (d.finish_reason) |_| { state.end_of_stream = true; } @@ -457,3 +643,72 @@ test "two streamed turns persist assistant replies in the conversation" { conv.messages.items[4].content.items[0].Text.items, ); } + +test "fragmented tool_call id and name are reassembled" { + // Lenient OpenAI-compatible providers occasionally split `id` and + // `function.name` across multiple deltas instead of sending them whole + // on the first chunk. Verify the state machine appends both correctly + // and emits a complete identity to the receiver. + const allocator = testing.allocator; + + var conv = conversation.Conversation.init(allocator); + defer conv.deinit(); + try conv.addUserMessage("call something"); + + var recv = NoopReceiver.make(); + + const events = [_][]const u8{ + \\{"choices":[{"delta":{"role":"assistant"}}]} + , + \\{"choices":[{"delta":{"tool_calls":[{"index":0,"id":"call_","type":"function","function":{"name":"pi"}}]}}]} + , + \\{"choices":[{"delta":{"tool_calls":[{"index":0,"id":"xyz","function":{"name":"ng"}}]}}]} + , + \\{"choices":[{"delta":{"tool_calls":[{"index":0,"function":{"arguments":"{\"host\":\"a.com\"}"}}]}}]} + , + \\{"choices":[{"delta":{},"finish_reason":"tool_calls"}]} + , + "[DONE]", + }; + + try runStreamedTurn(allocator, &conv, &recv, &events); + + const asst = conv.messages.items[1]; + try testing.expectEqual(@as(usize, 1), asst.content.items.len); + const tu = asst.content.items[0].ToolUse; + try testing.expectEqualStrings("call_xyz", tu.id); + try testing.expectEqualStrings("ping", tu.name); + try testing.expectEqualStrings("{\"host\":\"a.com\"}", tu.input.items); +} + +test "tool_call with no arguments still finalizes a well-formed ToolUse" { + // Some providers may emit a tool call with no arguments at all (e.g. a + // zero-arg tool). The state machine should still emit onBlockStart + // exactly once at finalize time and produce a ToolUse with empty input. + const allocator = testing.allocator; + + var conv = conversation.Conversation.init(allocator); + defer conv.deinit(); + try conv.addUserMessage("ring it"); + + var recv = NoopReceiver.make(); + + const events = [_][]const u8{ + \\{"choices":[{"delta":{"role":"assistant"}}]} + , + \\{"choices":[{"delta":{"tool_calls":[{"index":0,"id":"c1","type":"function","function":{"name":"ring"}}]}}]} + , + \\{"choices":[{"delta":{},"finish_reason":"tool_calls"}]} + , + "[DONE]", + }; + + try runStreamedTurn(allocator, &conv, &recv, &events); + + const asst = conv.messages.items[1]; + try testing.expectEqual(@as(usize, 1), asst.content.items.len); + const tu = asst.content.items[0].ToolUse; + try testing.expectEqualStrings("c1", tu.id); + try testing.expectEqualStrings("ring", tu.name); + try testing.expectEqual(@as(usize, 0), tu.input.items.len); +} diff --git a/libpanto/src/root.zig b/libpanto/src/root.zig index a5cf8b5..bfc814b 100644 --- a/libpanto/src/root.zig +++ b/libpanto/src/root.zig @@ -3,6 +3,12 @@ pub const provider = @import("provider.zig"); pub const agent = @import("agent.zig"); pub const config = @import("config.zig"); pub const sse = @import("sse.zig"); +pub const tool = @import("tool.zig"); +pub const tool_registry = @import("tool_registry.zig"); + +// Re-exports for ergonomic embedder use. +pub const Tool = tool.Tool; +pub const ToolRegistry = tool_registry.ToolRegistry; // Internal modules. Not part of the public API — callers construct providers // via `provider.Provider.init(allocator, io, config)`, which picks the right diff --git a/libpanto/src/tool.zig b/libpanto/src/tool.zig new file mode 100644 index 0000000..1d8d113 --- /dev/null +++ b/libpanto/src/tool.zig @@ -0,0 +1,61 @@ +//! Native tool extension API. +//! +//! A `Tool` is the boundary between the agent loop and any extension runtime +//! — native Zig code, a Lua bridge, a future Python or Go bridge. libpanto +//! itself does not parse tool inputs or outputs; it just dispatches. + +const std = @import("std"); +const Allocator = std.mem.Allocator; + +pub const Tool = struct { + /// Tool name. Borrowed — lifetime is owned by whoever constructs the + /// `Tool`. Typically the same owner that backs `ctx` (e.g. a LuaTool + /// adapter, or a static const in a native tool). + name: []const u8, + + /// Human-readable purpose of the tool. Emitted to the LLM alongside the + /// schema. Borrowed; same lifetime contract as `name`. + description: []const u8, + + /// JSON Schema for the tool's input, as raw JSON bytes. Emitted verbatim + /// into provider request bodies. Borrowed; same lifetime contract. + schema_json: []const u8, + + /// Opaque context pointer passed back to every vtable call. + ctx: *anyopaque, + + vtable: *const VTable, + + pub const VTable = struct { + /// Invoke the tool. MUST be thread-safe — the agent may call + /// `invoke` concurrently from multiple threads when the LLM emits + /// multiple ToolUse blocks in a single response. + /// + /// `input` is the raw JSON bytes the provider sent. The tool is + /// responsible for parsing them if it cares about their structure. + /// + /// Returns owned bytes allocated with `allocator`. These bytes + /// become the `content` of the ToolResult block sent back to the + /// LLM. The agent takes ownership and frees them. + /// + /// Returning an error aborts the current turn. The agent surfaces + /// the error to the user. Native tool implementations are + /// responsible for catching their own panics — a panic in `invoke` + /// will crash the process. Adapters that bridge to safer languages + /// (Lua, Python, Go) should convert panics/exceptions into errors. + invoke: *const fn ( + ctx: *anyopaque, + input: []const u8, + allocator: Allocator, + ) anyerror![]u8, + + /// Called when the tool is unregistered or the registry is torn + /// down. Frees any resources owned by `ctx`, including `ctx` + /// itself if it was heap-allocated. + /// + /// `name`, `description`, and `schema_json` are also typically + /// owned by the same allocation as `ctx` — the tool's deinit + /// hook is responsible for freeing them. + deinit: *const fn (ctx: *anyopaque, allocator: Allocator) void, + }; +}; diff --git a/libpanto/src/tool_registry.zig b/libpanto/src/tool_registry.zig new file mode 100644 index 0000000..4ade16f --- /dev/null +++ b/libpanto/src/tool_registry.zig @@ -0,0 +1,217 @@ +//! Registry of `Tool`s owned by an `Agent`. +//! +//! Tools are keyed by name. The registry takes ownership: on `unregister` +//! or `deinit`, it calls the tool's `vtable.deinit`. +//! +//! Iteration is not synchronized — callers must avoid mutating the registry +//! during iteration. In the current agent loop this is naturally true: the +//! provider iterates the registry once at request-build time, and tool +//! registration only happens at agent setup. + +const std = @import("std"); +const Allocator = std.mem.Allocator; +const tool_mod = @import("tool.zig"); +const Tool = tool_mod.Tool; + +pub const ToolRegistry = struct { + tools: std.StringHashMap(Tool), + allocator: Allocator, + + pub fn init(allocator: Allocator) ToolRegistry { + return .{ + .tools = std.StringHashMap(Tool).init(allocator), + .allocator = allocator, + }; + } + + /// Tear down the registry. Each remaining tool's `vtable.deinit` is + /// invoked. + pub fn deinit(self: *ToolRegistry) void { + var it = self.tools.iterator(); + while (it.next()) |entry| { + const tool = entry.value_ptr.*; + tool.vtable.deinit(tool.ctx, self.allocator); + } + self.tools.deinit(); + } + + /// Register a tool. The registry takes ownership. + /// + /// Returns `error.DuplicateTool` if a tool with the same name is + /// already registered — in that case the caller's tool is NOT taken + /// over (the caller is responsible for tearing it down). + pub fn register(self: *ToolRegistry, tool: Tool) !void { + const gop = try self.tools.getOrPut(tool.name); + if (gop.found_existing) return error.DuplicateTool; + gop.value_ptr.* = tool; + } + + /// Remove a tool by name. Calls the tool's `vtable.deinit`. No-op if + /// the name is not registered. + pub fn unregister(self: *ToolRegistry, name: []const u8) void { + if (self.tools.fetchRemove(name)) |kv| { + kv.value.vtable.deinit(kv.value.ctx, self.allocator); + } + } + + /// Look up a tool by name. The returned pointer is invalidated by any + /// subsequent register/unregister call. + pub fn lookup(self: *const ToolRegistry, name: []const u8) ?*const Tool { + return self.tools.getPtr(name); + } + + pub fn count(self: *const ToolRegistry) usize { + return self.tools.count(); + } + + /// Iterate registered tools. Caller must not mutate the registry during + /// iteration. + pub fn iterator(self: *const ToolRegistry) Iterator { + return .{ .inner = self.tools.iterator() }; + } + + pub const Iterator = struct { + inner: std.StringHashMap(Tool).Iterator, + + pub fn next(self: *Iterator) ?*const Tool { + const entry = self.inner.next() orelse return null; + return entry.value_ptr; + } + }; +}; + +// ----------------------------------------------------------------------------- +// Tests +// ----------------------------------------------------------------------------- + +const testing = std.testing; + +/// A trivial in-test Tool implementation backed by a single owned counter +/// allocation. Used to verify ownership/deinit behavior. +const TestTool = struct { + invocations: u32 = 0, + name_owned: []u8, + desc_owned: []u8, + schema_owned: []u8, + + fn create(allocator: Allocator, name: []const u8) !Tool { + const self = try allocator.create(TestTool); + errdefer allocator.destroy(self); + + const name_owned = try allocator.dupe(u8, name); + errdefer allocator.free(name_owned); + const desc_owned = try allocator.dupe(u8, "test tool"); + errdefer allocator.free(desc_owned); + const schema_owned = try allocator.dupe(u8, "{}"); + errdefer allocator.free(schema_owned); + + self.* = .{ + .name_owned = name_owned, + .desc_owned = desc_owned, + .schema_owned = schema_owned, + }; + return .{ + .name = self.name_owned, + .description = self.desc_owned, + .schema_json = self.schema_owned, + .ctx = self, + .vtable = &vt, + }; + } + + const vt: Tool.VTable = .{ + .invoke = invoke, + .deinit = deinit, + }; + + fn invoke(ctx: *anyopaque, input: []const u8, allocator: Allocator) anyerror![]u8 { + const self: *TestTool = @ptrCast(@alignCast(ctx)); + self.invocations += 1; + return try allocator.dupe(u8, input); + } + + fn deinit(ctx: *anyopaque, allocator: Allocator) void { + const self: *TestTool = @ptrCast(@alignCast(ctx)); + allocator.free(self.name_owned); + allocator.free(self.desc_owned); + allocator.free(self.schema_owned); + allocator.destroy(self); + } +}; + +test "register, lookup, count" { + const allocator = testing.allocator; + var reg = ToolRegistry.init(allocator); + defer reg.deinit(); + + try reg.register(try TestTool.create(allocator, "echo")); + try reg.register(try TestTool.create(allocator, "ls")); + + try testing.expectEqual(@as(usize, 2), reg.count()); + try testing.expect(reg.lookup("echo") != null); + try testing.expect(reg.lookup("ls") != null); + try testing.expect(reg.lookup("missing") == null); + try testing.expectEqualStrings("echo", reg.lookup("echo").?.name); +} + +test "duplicate registration returns error and leaves original in place" { + const allocator = testing.allocator; + var reg = ToolRegistry.init(allocator); + defer reg.deinit(); + + try reg.register(try TestTool.create(allocator, "echo")); + + // The second tool isn't taken over on duplicate; tear it down ourselves. + var dup = try TestTool.create(allocator, "echo"); + try testing.expectError(error.DuplicateTool, reg.register(dup)); + dup.vtable.deinit(dup.ctx, allocator); + + try testing.expectEqual(@as(usize, 1), reg.count()); +} + +test "unregister calls deinit and removes" { + const allocator = testing.allocator; + var reg = ToolRegistry.init(allocator); + defer reg.deinit(); + + try reg.register(try TestTool.create(allocator, "tmp")); + try testing.expectEqual(@as(usize, 1), reg.count()); + + reg.unregister("tmp"); + try testing.expectEqual(@as(usize, 0), reg.count()); + try testing.expect(reg.lookup("tmp") == null); + + // No-op on missing. + reg.unregister("never_existed"); +} + +test "iterator visits every tool" { + const allocator = testing.allocator; + var reg = ToolRegistry.init(allocator); + defer reg.deinit(); + + try reg.register(try TestTool.create(allocator, "a")); + try reg.register(try TestTool.create(allocator, "b")); + try reg.register(try TestTool.create(allocator, "c")); + + var saw_a = false; + var saw_b = false; + var saw_c = false; + + var it = reg.iterator(); + while (it.next()) |t| { + if (std.mem.eql(u8, t.name, "a")) saw_a = true; + if (std.mem.eql(u8, t.name, "b")) saw_b = true; + if (std.mem.eql(u8, t.name, "c")) saw_c = true; + } + try testing.expect(saw_a and saw_b and saw_c); +} + +test "deinit frees all remaining tools" { + // If this leaks, the testing allocator will catch it. + const allocator = testing.allocator; + var reg = ToolRegistry.init(allocator); + try reg.register(try TestTool.create(allocator, "x")); + try reg.register(try TestTool.create(allocator, "y")); + reg.deinit(); +} diff --git a/src/main.zig b/src/main.zig index 688b61d..20da910 100644 --- a/src/main.zig +++ b/src/main.zig @@ -1,5 +1,6 @@ const std = @import("std"); const panto = @import("panto"); +const ping_tool = @import("ping_tool.zig"); const Receiver = panto.provider.Receiver; const ReceiverVTable = panto.provider.ReceiverVTable; @@ -38,10 +39,14 @@ const CLIReceiver = struct { meta: ?BlockMeta, ) anyerror!void { _ = index; - _ = meta; const self: *CLIReceiver = @ptrCast(@alignCast(ptr)); - if (block_type == .Thinking) { - try self.stdout.writeAll("\x1b[2m[thinking] "); + switch (block_type) { + .Thinking => try self.stdout.writeAll("\x1b[2m[thinking] "), + .ToolUse => { + const name = if (meta) |m| (m.tool_name orelse "?") else "?"; + try self.stdout.print("\n\x1b[36m[tool: {s}]\x1b[0m ", .{name}); + }, + else => {}, } try self.file.flush(); } @@ -60,8 +65,10 @@ const CLIReceiver = struct { ) anyerror!void { _ = index; const self: *CLIReceiver = @ptrCast(@alignCast(ptr)); - if (block == .Thinking) { - try self.stdout.writeAll("\x1b[0m\n"); + switch (block) { + .Thinking => try self.stdout.writeAll("\x1b[0m\n"), + .ToolUse => try self.stdout.writeAll("\n"), + else => {}, } try self.file.flush(); } @@ -175,9 +182,13 @@ pub fn main(init: std.process.Init) !void { try conv.addSystemMessage("You are a helpful assistant."); const prov = try panto.provider.Provider.init(alloc, io, config); - const agent = panto.agent.Agent.init(alloc, prov); + var agent = panto.agent.Agent.init(alloc, prov); defer agent.deinit(); + // smoke test: register a trivial built-in tool so we can exercise + // the tool-call loop against a real LLM. + try agent.registerTool(ping_tool.tool()); + const banner_model: []const u8 = switch (config) { inline else => |c| c.model, }; diff --git a/src/ping_tool.zig b/src/ping_tool.zig new file mode 100644 index 0000000..eb212af --- /dev/null +++ b/src/ping_tool.zig @@ -0,0 +1,57 @@ +//! A trivial built-in "ping" tool used to exercise the tool-call loop +//! end-to-end against a real LLM. It accepts a `host` string and always +//! returns the literal string "PONG", regardless of what the host is. +//! +//! In phase 5 this kind of built-in will be replaced by a proper extension; +//! for now it lives in the CLI so we can verify the libpanto tool plumbing +//! without writing the Lua bridge first. + +const std = @import("std"); +const panto = @import("panto"); + +const NAME = "ping"; +const DESCRIPTION = + \\Test whether a server is reachable by hostname. Always responds with + \\"PONG" when the server is up. Use this when the user asks you to check + \\if a host is online or reachable. +; +const SCHEMA_JSON = + \\{ + \\ "type": "object", + \\ "properties": { + \\ "host": { + \\ "type": "string", + \\ "description": "The hostname or address to ping." + \\ } + \\ }, + \\ "required": ["host"] + \\} +; + +/// Returns a `Tool` value ready to register with `Agent.registerTool`. The +/// tool holds no per-instance state: name, description, and schema are +/// `comptime` string literals, and the vtable's deinit is a no-op. We pass +/// a sentinel pointer as ctx since the contract requires *anyopaque but +/// nothing dereferences it. +pub fn tool() panto.Tool { + return .{ + .name = NAME, + .description = DESCRIPTION, + .schema_json = SCHEMA_JSON, + .ctx = &ctx_sentinel, + .vtable = &vtable, + }; +} + +var ctx_sentinel: u8 = 0; + +const vtable: panto.Tool.VTable = .{ + .invoke = invoke, + .deinit = deinitNoop, +}; + +fn invoke(_: *anyopaque, _: []const u8, allocator: std.mem.Allocator) anyerror![]u8 { + return try allocator.dupe(u8, "PONG"); +} + +fn deinitNoop(_: *anyopaque, _: std.mem.Allocator) void {} |