phase 3 part 1

1 files changed, 275 insertions, 20 deletions

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);
+}