diff options
Diffstat (limited to 'libpanto/src/provider_openai_chat.zig')
| -rw-r--r-- | libpanto/src/provider_openai_chat.zig | 403 |
1 files changed, 305 insertions, 98 deletions
diff --git a/libpanto/src/provider_openai_chat.zig b/libpanto/src/provider_openai_chat.zig index 2343ac1..c4f19f5 100644 --- a/libpanto/src/provider_openai_chat.zig +++ b/libpanto/src/provider_openai_chat.zig @@ -223,12 +223,19 @@ pub const OpenAIChatProvider = struct { /// 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. +/// We model the assistant message as a sequence of blocks, exactly one of +/// which is active at a time. Text/thinking transitions are inferred from +/// which field a delta carries. Tool_use blocks arrive as a per-call wire +/// `index`; the OpenAI Chat Completions streaming spec does not formally +/// promise that all fragments for a given index arrive contiguously, but in +/// practice every well-behaved backend (and the official Node SDK's own +/// reassembly logic) treats a delta for a new index as the implicit close +/// of the previous one. We do the same: seeing a delta for an index that +/// differs from `current_tool_index` closes the prior tool_use and opens a +/// new one. `finish_reason` closes the last still-open tool_use. A delta +/// arriving for an index that has already been closed is a degenerate +/// backend behavior (e.g. vLLM with speculative decoding under some +/// configurations) — we log an error and drop the fragment. const StreamState = struct { allocator: Allocator, started: bool = false, @@ -239,8 +246,6 @@ const StreamState = struct { 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 text/thinking block. Owned by @@ -248,11 +253,18 @@ const StreamState = struct { /// transfers to the assembled Message. current_buf: conversation.TextualBlock = .empty, - /// Assembled non-tool-use blocks for the final message, in stream order. + /// Assembled 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), + /// The currently-streaming tool_use, if any. Closed when a delta for + /// a different wire index arrives, or at finalize. + active_tool: ?ToolUseInProgress = null, + /// Wire index of `active_tool` (when non-null). + current_tool_index: ?usize = null, + /// Wire indices that have already been closed. Used solely to detect + /// (and report) the degenerate case of a delta arriving for an index + /// whose block we've already emitted. + closed_tool_indices: std.AutoHashMap(usize, void), const ToolUseInProgress = struct { /// Block index emitted to the receiver for this tool call's @@ -266,10 +278,13 @@ const StreamState = struct { 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. + /// Set once we've emitted `onBlockStart(.ToolUse, ...)` for this + /// block. We defer until either the first argument fragment + /// arrives or the block is closed — not for identity reasons + /// (identity is no longer passed at start) but to keep block + /// indices clean: a tool_call that turns out to lack id or name + /// is dropped silently rather than producing an empty + /// start/complete pair. started: bool = false, fn deinit(self: *ToolUseInProgress, allocator: Allocator) void { @@ -282,7 +297,7 @@ const StreamState = struct { fn init(allocator: Allocator) StreamState { return .{ .allocator = allocator, - .tool_uses = std.AutoHashMap(usize, ToolUseInProgress).init(allocator), + .closed_tool_indices = std.AutoHashMap(usize, void).init(allocator), }; } @@ -290,9 +305,8 @@ const StreamState = struct { 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(); + if (self.active_tool) |*tu| tu.deinit(self.allocator); + self.closed_tool_indices.deinit(); } /// Close the active text/thinking block (if any) and emit @@ -332,7 +346,7 @@ const StreamState = struct { .thinking => .Thinking, .tool_use, .none => unreachable, }; - try receiver.onBlockStart(block_type, self.block_index, null); + try receiver.onBlockStart(block_type, self.block_index); } fn appendDelta( @@ -344,25 +358,47 @@ const StreamState = struct { try receiver.onContentDelta(self.block_index, delta); } - /// Apply one streaming tool_call delta. Allocates a new - /// `ToolUseInProgress` slot on first sight of each wire index. + /// Apply one streaming tool_call delta. Opens a new tool_use on the + /// first sight of a wire index, closing any prior tool_use (or active + /// text/thinking block) first. A delta for an already-closed index is + /// a malformed stream — we log and drop it. 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. + // Degenerate backend: a delta arrived for an index whose block we + // already finalized. Drop the fragment so we don't reopen a closed + // block, but log loudly enough to make this diagnosable. + if (self.closed_tool_indices.contains(d.index)) { + if (!@import("builtin").is_test) { + std.log.err( + "openai_chat: dropping tool_call delta for already-closed wire index {d} (non-contiguous tool_call stream); id={?s} name={?s} args={?s}", + .{ d.index, d.id, d.name, d.arguments }, + ); + } + return; + } + + // Wire-index change closes the previously-active tool_use. This is + // the only signal openai_chat gives us for mid-stream tool_use + // boundaries; see the StreamState doc-comment for the rationale. + if (self.current_tool_index) |cur| { + if (cur != d.index) try self.closeActiveTool(receiver); + } + + if (self.active_tool == null) { + // Opening a new tool_use. First 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.active_tool = .{ .block_index = self.block_index }; + self.current_tool_index = d.index; self.block_index += 1; } - const tu = gop.value_ptr; + const tu = &self.active_tool.?; // Append identity fragments. Most providers send id+name whole on // the first delta and never repeat them, but appending is the only @@ -373,8 +409,9 @@ const StreamState = struct { // 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. + // emitting identity for this index. 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); try tu.arguments.appendSlice(self.allocator, a); @@ -382,9 +419,60 @@ const StreamState = struct { } } - /// 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. + /// 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. + fn closeActiveTool(self: *StreamState, receiver: *provider_mod.Receiver) !void { + var tu = self.active_tool orelse return; + self.active_tool = null; + const wire_index = self.current_tool_index.?; + self.current_tool_index = null; + try self.closed_tool_indices.put(wire_index, {}); + + // Drop entries lacking id or name. The stream closed the block + // before the provider sent enough to identify which tool was + // being called — there's nothing we can dispatch. + if (tu.id_buf.items.len == 0 or tu.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", + .{ + wire_index, + tu.id_buf.items.len, + tu.name_buf.items, + tu.arguments.items.len, + }, + ); + } + tu.deinit(self.allocator); + return; + } + + // 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); + + const id_owned = try tu.id_buf.toOwnedSlice(self.allocator); + const name_owned = try tu.name_buf.toOwnedSlice(self.allocator); + const block: conversation.ContentBlock = .{ .ToolUse = .{ + .id = id_owned, + .name = name_owned, + .input = tu.arguments, + } }; + // Ownership has moved into `block`; clear the local before it + // goes out of scope so deinit doesn't double-free. + tu.arguments = .empty; + + try self.blocks.append(self.allocator, block); + try receiver.onBlockComplete(tu.block_index, self.blocks.items[self.blocks.items.len - 1]); + } + + /// Emit `onBlockStart(.ToolUse, ...)` once per in-progress tool use. + /// Callers must invoke this before the first `onContentDelta` or + /// `onBlockComplete` for the block. Identity (id/name) is *not* + /// passed at start — see provider.zig's ReceiverVTable docs for the + /// rationale. Receivers get identity from the assembled ContentBlock + /// at onBlockComplete time. fn emitStartIfNeeded( self: *StreamState, receiver: *provider_mod.Receiver, @@ -393,19 +481,12 @@ const StreamState = struct { _ = 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); + try receiver.onBlockStart(.ToolUse, tu.block_index); } - /// 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. + /// End the stream: close any open text/thinking block, close the still- + /// active tool_use (if any), then commit the assembled assistant + /// Message to the conversation. fn finalize( self: *StreamState, receiver: *provider_mod.Receiver, @@ -415,59 +496,7 @@ const StreamState = struct { self.finalized = true; 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], - ); - } + try self.closeActiveTool(receiver); // Move blocks into a fresh conversation message. const moved_blocks = try self.blocks.toOwnedSlice(self.allocator); @@ -560,7 +589,7 @@ const NoopReceiver = struct { .onError = noopErr, }; fn noopMsgStart(_: *anyopaque, _: conversation.MessageRole) anyerror!void {} - fn noopBlockStart(_: *anyopaque, _: provider_mod.ContentBlockType, _: usize, _: ?provider_mod.BlockMeta) anyerror!void {} + fn noopBlockStart(_: *anyopaque, _: provider_mod.ContentBlockType, _: usize) 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 {} @@ -681,6 +710,184 @@ test "fragmented tool_call id and name are reassembled" { try testing.expectEqualStrings("{\"host\":\"a.com\"}", tu.input.items); } +/// A Receiver that records the sequence of callback events as compact +/// strings. Useful for asserting per-block start/complete ordering. +const RecordingReceiver = struct { + allocator: Allocator, + events: std.ArrayList([]const u8) = .empty, + + fn receiver(self: *RecordingReceiver) provider_mod.Receiver { + return .{ .ptr = self, .vtable = &vt }; + } + + const vt: provider_mod.ReceiverVTable = .{ + .onMessageStart = onMessageStart, + .onBlockStart = onBlockStart, + .onContentDelta = onContentDelta, + .onBlockComplete = onBlockComplete, + .onMessageComplete = onMessageComplete, + .onError = onError, + }; + + fn record(self: *RecordingReceiver, s: []const u8) !void { + const owned = try self.allocator.dupe(u8, s); + try self.events.append(self.allocator, owned); + } + + fn recordFmt(self: *RecordingReceiver, comptime fmt: []const u8, args: anytype) !void { + const owned = try std.fmt.allocPrint(self.allocator, fmt, args); + try self.events.append(self.allocator, owned); + } + + fn deinit(self: *RecordingReceiver) void { + for (self.events.items) |e| self.allocator.free(e); + self.events.deinit(self.allocator); + } + + fn onMessageStart(ptr: *anyopaque, _: conversation.MessageRole) anyerror!void { + const self: *RecordingReceiver = @ptrCast(@alignCast(ptr)); + try self.record("msg_start"); + } + fn onBlockStart( + ptr: *anyopaque, + bt: provider_mod.ContentBlockType, + idx: usize, + ) anyerror!void { + const self: *RecordingReceiver = @ptrCast(@alignCast(ptr)); + try self.recordFmt("block_start[{d}]:{s}", .{ idx, @tagName(bt) }); + } + 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 }); + } + fn onBlockComplete( + ptr: *anyopaque, + idx: usize, + _: conversation.ContentBlock, + ) anyerror!void { + const self: *RecordingReceiver = @ptrCast(@alignCast(ptr)); + try self.recordFmt("block_complete[{d}]", .{idx}); + } + fn onMessageComplete(ptr: *anyopaque, _: conversation.Message) anyerror!void { + const self: *RecordingReceiver = @ptrCast(@alignCast(ptr)); + try self.record("msg_complete"); + } + fn onError(_: *anyopaque, _: anyerror) void {} +}; + +test "parallel tool_calls emit one complete start/delta/complete cycle per block" { + // Regression test: previously, the OpenAI provider deferred ALL + // tool_use onBlockComplete callbacks to finalize, so a four-tool + // parallel batch produced start/start/start/start/delta*/complete/ + // complete/complete/complete — the receiver couldn't render each tool + // as its own discrete block. With the new contiguity-driven close-on- + // next-index logic, each tool_use should produce a contiguous + // start → delta(s) → complete trio. + const allocator = testing.allocator; + + var conv = conversation.Conversation.init(allocator); + defer conv.deinit(); + try conv.addUserMessage("ping four hosts"); + + var rec: RecordingReceiver = .{ .allocator = allocator }; + defer rec.deinit(); + var recv = rec.receiver(); + + const events = [_][]const u8{ + \\{"choices":[{"delta":{"role":"assistant"}}]} + , + \\{"choices":[{"delta":{"tool_calls":[{"index":0,"id":"c0","type":"function","function":{"name":"ping","arguments":"{\"host\":\"a\"}"}}]}}]} + , + \\{"choices":[{"delta":{"tool_calls":[{"index":1,"id":"c1","type":"function","function":{"name":"ping","arguments":"{\"host\":\"b\"}"}}]}}]} + , + \\{"choices":[{"delta":{"tool_calls":[{"index":2,"id":"c2","type":"function","function":{"name":"ping","arguments":"{\"host\":\"c\"}"}}]}}]} + , + \\{"choices":[{"delta":{"tool_calls":[{"index":3,"id":"c3","type":"function","function":{"name":"ping","arguments":"{\"host\":\"d\"}"}}]}}]} + , + \\{"choices":[{"delta":{},"finish_reason":"tool_calls"}]} + , + "[DONE]", + }; + + try runStreamedTurn(allocator, &conv, &recv, &events); + + const expected = [_][]const u8{ + "msg_start", + "block_start[0]:ToolUse", + "delta[0]:{\"host\":\"a\"}", + "block_complete[0]", + "block_start[1]:ToolUse", + "delta[1]:{\"host\":\"b\"}", + "block_complete[1]", + "block_start[2]:ToolUse", + "delta[2]:{\"host\":\"c\"}", + "block_complete[2]", + "block_start[3]:ToolUse", + "delta[3]:{\"host\":\"d\"}", + "block_complete[3]", + "msg_complete", + }; + + // Identity arrives in the assembled ContentBlock at completion time. + const asst = conv.messages.items[1]; + try testing.expectEqual(@as(usize, 4), asst.content.items.len); + for (asst.content.items) |b| { + try testing.expectEqualStrings("ping", b.ToolUse.name); + } + try testing.expectEqual(expected.len, rec.events.items.len); + for (expected, rec.events.items) |want, got| { + try testing.expectEqualStrings(want, got); + } +} + +test "non-contiguous tool_call deltas: re-emission of a closed index is dropped" { + // Degenerate backend behavior: a delta for an already-closed wire + // index. We must not reopen the block; instead drop the fragment and + // log. The successfully-closed prior blocks remain intact. + 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"}}]} + , + \\{"choices":[{"delta":{"tool_calls":[{"index":0,"id":"c0","type":"function","function":{"name":"ping","arguments":"{\"x\":1}"}}]}}]} + , + \\{"choices":[{"delta":{"tool_calls":[{"index":1,"id":"c1","type":"function","function":{"name":"ping","arguments":"{\"y\":2}"}}]}}]} + , + // Delta for already-closed index 0: must be dropped. + \\{"choices":[{"delta":{"tool_calls":[{"index":0,"function":{"arguments":",extra"}}]}}]} + , + \\{"choices":[{"delta":{},"finish_reason":"tool_calls"}]} + , + "[DONE]", + }; + + try runStreamedTurn(allocator, &conv, &recv, &events); + + // Two well-formed tool_use blocks in the final message, args unaffected + // by the dropped fragment. + const asst = conv.messages.items[1]; + try testing.expectEqual(@as(usize, 2), asst.content.items.len); + try testing.expectEqualStrings("{\"x\":1}", asst.content.items[0].ToolUse.input.items); + try testing.expectEqualStrings("{\"y\":2}", asst.content.items[1].ToolUse.input.items); + + // Callback sequence: index 0 closed cleanly before any stray delta. + // There must be exactly one block_complete[0] in the event log + // (i.e. the stray delta did not produce a second open/close cycle). + var n_complete_0: usize = 0; + for (rec.events.items) |e| { + if (std.mem.eql(u8, e, "block_complete[0]")) n_complete_0 += 1; + } + try testing.expectEqual(@as(usize, 1), n_complete_0); +} + 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 |