diff options
Diffstat (limited to 'libpanto')
| -rw-r--r-- | libpanto/src/agent.zig | 2 | ||||
| -rw-r--r-- | libpanto/src/provider.zig | 21 | ||||
| -rw-r--r-- | libpanto/src/provider_anthropic_messages.zig | 21 | ||||
| -rw-r--r-- | libpanto/src/provider_openai_chat.zig | 131 | ||||
| -rw-r--r-- | libpanto/src/root.zig | 12 |
5 files changed, 175 insertions, 12 deletions
diff --git a/libpanto/src/agent.zig b/libpanto/src/agent.zig index 6c970ad..bbcc54c 100644 --- a/libpanto/src/agent.zig +++ b/libpanto/src/agent.zig @@ -580,6 +580,7 @@ const NoopReceiver = struct { const vt: provider_mod.ReceiverVTable = .{ .onMessageStart = noop1, .onBlockStart = noop2, + .onToolDetails = noopToolDetails, .onContentDelta = noop3, .onBlockComplete = noop4, .onMessageComplete = noop5, @@ -587,6 +588,7 @@ const NoopReceiver = struct { }; fn noop1(_: *anyopaque, _: conversation.MessageRole) anyerror!void {} fn noop2(_: *anyopaque, _: provider_mod.ContentBlockType, _: usize) anyerror!void {} + fn noopToolDetails(_: *anyopaque, _: usize, _: []const u8, _: []const u8) 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 {} diff --git a/libpanto/src/provider.zig b/libpanto/src/provider.zig index fd170f0..bc39026 100644 --- a/libpanto/src/provider.zig +++ b/libpanto/src/provider.zig @@ -19,12 +19,16 @@ pub const ContentBlockType = enum { /// stops streaming, calls `onError(err)`, and propagates `err` out of /// `streamStep`. No partial assistant message is appended to the conversation. /// -/// Identity fields (tool id/name for ToolUse blocks) are intentionally *not* -/// surfaced at `onBlockStart`. They cannot be reliably known at start time -/// across all wire protocols — OpenAI streaming may deliver `id` and `name` -/// in fragments across multiple deltas — so any "meta" we passed at start -/// would be a phantom guarantee. Receivers that need identity get it from -/// the fully-assembled `ContentBlock` delivered to `onBlockComplete`. +/// Tool-use identity (`id`, `name`) is delivered via `onToolDetails`, fired +/// once per ToolUse block at the earliest moment both fields are known. For +/// Anthropic this is immediately after `onBlockStart`, before any deltas. For +/// OpenAI Chat Completions this may be partway through the arg-deltas, since +/// the wire protocol can split `id` and `name` across multiple streaming +/// chunks. The only guarantees are: it fires strictly after the block's +/// `onBlockStart`, strictly before its `onBlockComplete`, and at most once +/// per ToolUse block. It never fires for non-ToolUse blocks. If a tool_use +/// block is dropped because identity never fully arrived, `onToolDetails` +/// (and `onBlockComplete`) never fire for it. /// /// `onError` is the receiver's cleanup hook. It fires exactly once per failed /// turn, whether the error originated in the receiver itself (a write failure) @@ -34,6 +38,7 @@ pub const ContentBlockType = enum { pub const ReceiverVTable = struct { onMessageStart: *const fn (*anyopaque, conversation.MessageRole) anyerror!void, onBlockStart: *const fn (*anyopaque, ContentBlockType, usize) anyerror!void, + onToolDetails: *const fn (*anyopaque, usize, []const u8, []const u8) anyerror!void, onContentDelta: *const fn (*anyopaque, usize, []const u8) anyerror!void, onBlockComplete: *const fn (*anyopaque, usize, conversation.ContentBlock) anyerror!void, onMessageComplete: *const fn (*anyopaque, conversation.Message) anyerror!void, @@ -52,6 +57,10 @@ pub const Receiver = struct { try self.vtable.onBlockStart(self.ptr, block_type, index); } + pub fn onToolDetails(self: Receiver, block_index: usize, id: []const u8, name: []const u8) !void { + try self.vtable.onToolDetails(self.ptr, block_index, id, name); + } + pub fn onContentDelta(self: Receiver, block_index: usize, delta: []const u8) !void { try self.vtable.onContentDelta(self.ptr, block_index, delta); } diff --git a/libpanto/src/provider_anthropic_messages.zig b/libpanto/src/provider_anthropic_messages.zig index 03c1d90..96649b6 100644 --- a/libpanto/src/provider_anthropic_messages.zig +++ b/libpanto/src/provider_anthropic_messages.zig @@ -283,6 +283,15 @@ const StreamState = struct { }; if (block_type) |bt| { try receiver.onBlockStart(bt, wire_index); + // Anthropic delivers tool id+name whole on content_block_start, + // so we can fire onToolDetails immediately — before any arg + // deltas. If the wire was malformed and either field is + // missing, skip: closeBlock will drop the block defensively. + if (kind == .tool_use) { + if (ab.tool_id != null and ab.tool_name != null) { + try receiver.onToolDetails(wire_index, ab.tool_id.?, ab.tool_name.?); + } + } } } @@ -506,6 +515,7 @@ const RecordingReceiver = struct { const vt: provider_mod.ReceiverVTable = .{ .onMessageStart = onMessageStart, .onBlockStart = onBlockStart, + .onToolDetails = onToolDetails, .onContentDelta = onContentDelta, .onBlockComplete = onBlockComplete, .onMessageComplete = onMessageComplete, @@ -524,6 +534,17 @@ const RecordingReceiver = struct { const self: *RecordingReceiver = @ptrCast(@alignCast(ptr)); try self.events.append(self.allocator, .{ .block_start = .{ .kind = bt, .index = idx } }); } + fn onToolDetails( + _: *anyopaque, + _: usize, + _: []const u8, + _: []const u8, + ) anyerror!void { + // Anthropic delivers identity at block_start time; the existing + // tests assert tool-use blocks via the ContentBlock in conv after + // finalize, not via the event stream. We accept and drop these + // here to keep the test recorder schema stable. + } fn onContentDelta(ptr: *anyopaque, idx: usize, delta: []const u8) anyerror!void { const self: *RecordingReceiver = @ptrCast(@alignCast(ptr)); const copy = try self.allocator.dupe(u8, delta); diff --git a/libpanto/src/provider_openai_chat.zig b/libpanto/src/provider_openai_chat.zig index c4f19f5..ca8fa4c 100644 --- a/libpanto/src/provider_openai_chat.zig +++ b/libpanto/src/provider_openai_chat.zig @@ -286,6 +286,11 @@ const StreamState = struct { /// is dropped silently rather than producing an empty /// start/complete pair. started: bool = false, + /// Set once we've emitted `onToolDetails` for this block. Fired + /// as soon as both id and name are non-empty, which may be on + /// the first delta (the common case) or partway through arg + /// deltas (fragmented-identity providers). + details_emitted: bool = false, fn deinit(self: *ToolUseInProgress, allocator: Allocator) void { self.id_buf.deinit(allocator); @@ -407,18 +412,47 @@ const StreamState = struct { 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 the block closes before - // any args arrive (zero-arg tool), `closeActiveTool` emits the - // start there. + // Defer `onBlockStart` until args begin. The first argument + // fragment is our signal that identity is likely settled enough + // to render. If the block closes before any args arrive (zero-arg + // tool), `closeActiveTool` emits the start there. if (d.arguments) |a| { try self.emitStartIfNeeded(receiver, tu); + // Fire `onToolDetails` as soon as both id and name are + // known. We can't know identity is *final* until the block + // closes (a later delta could append more bytes), but in + // practice OpenAI sends each whole on the first delta. A + // pathological backend that streams id/name across many + // chunks would have us emit a truncated value here. We + // accept that trade-off: receivers that need the canonical + // value can read it from the assembled ContentBlock at + // onBlockComplete. + try self.emitDetailsIfReady(receiver, tu); try tu.arguments.appendSlice(self.allocator, a); try receiver.onContentDelta(tu.block_index, a); + } else { + // Identity-only chunk (no args yet). Still try to emit + // details, in case both fields are now populated. + if (tu.started) try self.emitDetailsIfReady(receiver, tu); } } + /// Fire `onToolDetails` once both id and name are non-empty. No-op if + /// already fired or if either field is still empty. Requires that + /// `onBlockStart` has already been emitted. + fn emitDetailsIfReady( + self: *StreamState, + receiver: *provider_mod.Receiver, + tu: *ToolUseInProgress, + ) !void { + _ = self; + if (tu.details_emitted) return; + if (!tu.started) return; + if (tu.id_buf.items.len == 0 or tu.name_buf.items.len == 0) return; + tu.details_emitted = true; + try receiver.onToolDetails(tu.block_index, tu.id_buf.items, tu.name_buf.items); + } + /// Close the currently-active tool_use (if any), emitting onBlockStart /// (if it wasn't already), onBlockComplete, and recording the wire /// index as closed. No-op if there's no active tool_use. @@ -451,6 +485,9 @@ const StreamState = struct { // 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); + // Last chance to fire details if a fragmented-identity provider + // only finished id/name accumulation at the very end. + try self.emitDetailsIfReady(receiver, &tu); const id_owned = try tu.id_buf.toOwnedSlice(self.allocator); const name_owned = try tu.name_buf.toOwnedSlice(self.allocator); @@ -583,6 +620,7 @@ const NoopReceiver = struct { const vt: provider_mod.ReceiverVTable = .{ .onMessageStart = noopMsgStart, .onBlockStart = noopBlockStart, + .onToolDetails = noopToolDetails, .onContentDelta = noopDelta, .onBlockComplete = noopBlockComplete, .onMessageComplete = noopMsgComplete, @@ -590,6 +628,7 @@ const NoopReceiver = struct { }; fn noopMsgStart(_: *anyopaque, _: conversation.MessageRole) anyerror!void {} fn noopBlockStart(_: *anyopaque, _: provider_mod.ContentBlockType, _: usize) anyerror!void {} + fn noopToolDetails(_: *anyopaque, _: usize, _: []const u8, _: []const u8) anyerror!void {} fn noopDelta(_: *anyopaque, _: usize, _: []const u8) anyerror!void {} fn noopBlockComplete(_: *anyopaque, _: usize, _: conversation.ContentBlock) anyerror!void {} fn noopMsgComplete(_: *anyopaque, _: conversation.Message) anyerror!void {} @@ -723,6 +762,7 @@ const RecordingReceiver = struct { const vt: provider_mod.ReceiverVTable = .{ .onMessageStart = onMessageStart, .onBlockStart = onBlockStart, + .onToolDetails = onToolDetails, .onContentDelta = onContentDelta, .onBlockComplete = onBlockComplete, .onMessageComplete = onMessageComplete, @@ -756,6 +796,15 @@ const RecordingReceiver = struct { const self: *RecordingReceiver = @ptrCast(@alignCast(ptr)); try self.recordFmt("block_start[{d}]:{s}", .{ idx, @tagName(bt) }); } + fn onToolDetails( + ptr: *anyopaque, + idx: usize, + id: []const u8, + name: []const u8, + ) anyerror!void { + const self: *RecordingReceiver = @ptrCast(@alignCast(ptr)); + try self.recordFmt("tool_details[{d}]:{s}:{s}", .{ idx, id, name }); + } fn onContentDelta(ptr: *anyopaque, idx: usize, delta: []const u8) anyerror!void { const self: *RecordingReceiver = @ptrCast(@alignCast(ptr)); try self.recordFmt("delta[{d}]:{s}", .{ idx, delta }); @@ -814,15 +863,19 @@ test "parallel tool_calls emit one complete start/delta/complete cycle per block const expected = [_][]const u8{ "msg_start", "block_start[0]:ToolUse", + "tool_details[0]:c0:ping", "delta[0]:{\"host\":\"a\"}", "block_complete[0]", "block_start[1]:ToolUse", + "tool_details[1]:c1:ping", "delta[1]:{\"host\":\"b\"}", "block_complete[1]", "block_start[2]:ToolUse", + "tool_details[2]:c2:ping", "delta[2]:{\"host\":\"c\"}", "block_complete[2]", "block_start[3]:ToolUse", + "tool_details[3]:c3:ping", "delta[3]:{\"host\":\"d\"}", "block_complete[3]", "msg_complete", @@ -888,6 +941,74 @@ test "non-contiguous tool_call deltas: re-emission of a closed index is dropped" try testing.expectEqual(@as(usize, 1), n_complete_0); } +test "onToolDetails fires after id+name complete, even mid-arg-stream" { + // Fragmented-identity provider: id arrives split across two chunks, + // and an arg fragment appears between them. `onToolDetails` must + // wait until both id and name are non-empty (i.e. on the chunk that + // completes id), and must fire exactly once, before block_complete. + const allocator = testing.allocator; + + var conv = conversation.Conversation.init(allocator); + defer conv.deinit(); + try conv.addUserMessage("go"); + + var rec: RecordingReceiver = .{ .allocator = allocator }; + defer rec.deinit(); + var recv = rec.receiver(); + + const events = [_][]const u8{ + \\{"choices":[{"delta":{"role":"assistant"}}]} + , + // Identity-only chunk: name arrives whole, id starts. No args yet, + // so onBlockStart hasn't fired and details can't either. + \\{"choices":[{"delta":{"tool_calls":[{"index":0,"id":"call_","type":"function","function":{"name":"ping"}}]}}]} + , + // First arg chunk: onBlockStart fires. id is still "call_" — not + // empty — and name is non-empty, so onToolDetails fires here + // with whatever id we have so far (`call_`). + \\{"choices":[{"delta":{"tool_calls":[{"index":0,"function":{"arguments":"{\"x\":"}}]}}]} + , + // Mid-stream id completion + second arg chunk. onToolDetails has + // already fired so it does NOT fire again, even though id grew. + // The final ContentBlock will carry the full "call_xyz" id. + \\{"choices":[{"delta":{"tool_calls":[{"index":0,"id":"xyz","function":{"arguments":"1}"}}]}}]} + , + \\{"choices":[{"delta":{},"finish_reason":"tool_calls"}]} + , + "[DONE]", + }; + + try runStreamedTurn(allocator, &conv, &recv, &events); + + // Exactly one tool_details event, fired with the id-prefix that was + // current at first-args-arrival, and ordered between block_start and + // block_complete. + var n_details: usize = 0; + var details_pos: ?usize = null; + var block_start_pos: ?usize = null; + var block_complete_pos: ?usize = null; + for (rec.events.items, 0..) |e, i| { + if (std.mem.startsWith(u8, e, "tool_details[")) { + n_details += 1; + details_pos = i; + try testing.expectEqualStrings("tool_details[0]:call_:ping", e); + } else if (std.mem.eql(u8, e, "block_start[0]:ToolUse")) { + block_start_pos = i; + } else if (std.mem.eql(u8, e, "block_complete[0]")) { + block_complete_pos = i; + } + } + try testing.expectEqual(@as(usize, 1), n_details); + try testing.expect(block_start_pos.? < details_pos.?); + try testing.expect(details_pos.? < block_complete_pos.?); + + // Final ContentBlock has the full id assembled from both fragments. + const asst = conv.messages.items[1]; + try testing.expectEqualStrings("call_xyz", asst.content.items[0].ToolUse.id); + try testing.expectEqualStrings("ping", asst.content.items[0].ToolUse.name); + try testing.expectEqualStrings("{\"x\":1}", asst.content.items[0].ToolUse.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 diff --git a/libpanto/src/root.zig b/libpanto/src/root.zig index 5dacfef..ff02d4d 100644 --- a/libpanto/src/root.zig +++ b/libpanto/src/root.zig @@ -1,3 +1,5 @@ +const std = @import("std"); + pub const conversation = @import("conversation.zig"); pub const provider = @import("provider.zig"); pub const agent = @import("agent.zig"); @@ -25,7 +27,15 @@ const anthropic_messages_json = @import("anthropic_messages_json.zig"); const provider_anthropic_messages = @import("provider_anthropic_messages.zig"); test { - const std = @import("std"); + // Test contract: deliberate error-path tests should not produce visible + // log output. Library code logs at `.err` for genuine production failures + // and `.warn` for expected-failure paths exercised by tests; the test + // runner's logger gates on `std.testing.log_level`, which defaults to + // `.warn`. Raising it to `.err` silences expected warnings without + // changing production behavior. Anything that *should* be visible in a + // passing test must use `std.debug.print` or assert via the testing API. + std.testing.log_level = .err; + std.testing.refAllDecls(@This()); std.testing.refAllDecls(openai_chat_json); std.testing.refAllDecls(provider_openai_chat); |