phase 3 part 1

1 files changed, 80 insertions, 19 deletions

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" {