diff options
Diffstat (limited to 'libpanto/src/provider_anthropic_messages.zig')
| -rw-r--r-- | libpanto/src/provider_anthropic_messages.zig | 99 |
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" { |