diff options
Diffstat (limited to 'libpanto/src')
| -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 |
9 files changed, 2016 insertions, 75 deletions
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(); +} |