initial commit, moved libpanto over from the pantograph repo

1 files changed, 1306 insertions, 0 deletions

diff --git a/src/provider_openai_chat.zig b/src/provider_openai_chat.zig
new file mode 100644
index 0000000..4ac1e11
--- /dev/null
+++ b/src/provider_openai_chat.zig
@@ -0,0 +1,1306 @@
+//! OpenAI Chat Completions streaming provider.
+//!
+//! Wire format reference: https://platform.openai.com/docs/api-reference/chat/streaming
+//!
+//! Responsibilities:
+//!   - Convert `Conversation` → request JSON (delegated to openai_chat_json.zig)
+//!   - POST to `{base_url}/chat/completions` with `stream: true`
+//!   - Read the chunked body, feed bytes through SSEParser
+//!   - Parse each event payload, drive the block boundary state machine,
+//!     and emit Receiver callbacks
+//!   - Assemble the final Message and emit onMessageComplete
+
+const std = @import("std");
+const Allocator = std.mem.Allocator;
+const Io = std.Io;
+const http = std.http;
+const Uri = std.Uri;
+
+const conversation = @import("conversation.zig");
+const provider_mod = @import("provider.zig");
+const stream_mod = @import("stream.zig");
+const sse_mod = @import("sse.zig");
+const json_mod = @import("openai_chat_json.zig");
+const config_mod = @import("config.zig");
+
+const Event = stream_mod.Event;
+const EventQueue = stream_mod.EventQueue;
+
+const decodeNameInPlace = provider_mod.decodeNameInPlace;
+
+/// 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, tool_use };
+
+/// A single OpenAI Chat streaming request. Transient: constructed per
+/// `streamStep`, holds only borrowed state (allocator, io, the global HTTP
+/// client, and the active config). Carries nothing across requests, so it
+/// is created inline by the free `streamStep` entry point below.
+pub const OpenAIChatRequest = struct {
+    allocator: Allocator,
+    io: Io,
+    config: *const config_mod.OpenAIChatConfig,
+    http_client: *http.Client,
+    /// Optional diagnostic side-channel. When non-null, classified failures
+    /// stash the HTTP status code and any `Retry-After` here for the agent's
+    /// retry policy. Strings written here are not owned by the diagnostic;
+    /// they live only as long as this request object.
+    diag: ?*provider_mod.ProviderDiagnostic = null,
+
+    /// Open the streaming HTTP request and return a heap-allocated
+    /// resumable response. Performs the POST and reads response headers
+    /// (classifying any >=400 status into a provider error), but does NOT
+    /// pump the body — that happens lazily in `ResumableResponse.produce`.
+    ///
+    /// On success the caller owns the returned `*ResumableResponse` and must
+    /// `deinit` it. On failure nothing is allocated.
+    pub fn open(
+        self: *OpenAIChatRequest,
+        conv: *conversation.Conversation,
+        tools: *const provider_mod.ToolRegistry,
+    ) !*ResumableResponse {
+        const rr = try self.allocator.create(ResumableResponse);
+        errdefer self.allocator.destroy(rr);
+        rr.* = .{
+            .allocator = self.allocator,
+            .conv = conv,
+            .parser = sse_mod.SSEParser.init(self.allocator),
+            .state = .init(self.allocator),
+        };
+        errdefer {
+            rr.parser.deinit();
+            rr.state.deinit();
+        }
+
+        // Build URL: "{base_url}/chat/completions"
+        const url = try std.fmt.allocPrint(
+            self.allocator,
+            "{s}/chat/completions",
+            .{self.config.base_url},
+        );
+        defer self.allocator.free(url);
+
+        const uri = try Uri.parse(url);
+
+        // Build the request body.
+        const body = try json_mod.serializeRequest(self.allocator, self.config, conv, tools);
+        defer self.allocator.free(body);
+        std.log.debug("openai_chat => {s}", .{body});
+
+        // Auth header
+        const auth_value = try std.fmt.allocPrint(
+            self.allocator,
+            "Bearer {s}",
+            .{self.config.api_key},
+        );
+        defer self.allocator.free(auth_value);
+
+        const base_headers = [_]http.Header{
+            .{ .name = "content-type", .value = "application/json" },
+            .{ .name = "accept", .value = "text/event-stream" },
+            .{ .name = "authorization", .value = auth_value },
+        };
+        // Merge provider `extra_headers` (e.g. Copilot editor identity, or
+        // auth-exchange-derived headers) onto the base set. Freed at the end
+        // of `open` — after the request body has been flushed.
+        const extra_headers = try provider_mod.mergeHeaders(
+            self.allocator,
+            &base_headers,
+            self.config.extra_headers,
+        );
+        defer self.allocator.free(extra_headers);
+
+        // Open the request. We can't use `fetch()` because it buffers the
+        // response; we want to stream the body as it arrives. The request
+        // is moved into the heap struct so the body reader (which borrows
+        // `&rr.response`) stays valid across `produce` calls.
+        rr.response = try provider_mod.sendRequest(self.http_client, uri, extra_headers, body, &rr.req);
+        rr.req_open = true;
+        errdefer {
+            rr.req.deinit();
+            rr.req_open = false;
+        }
+
+        // A >=400 status classifies into a retryable/terminal provider error.
+        // HTTP 400 with a context marker becomes `ContextOverflow` so the
+        // caller can compact and retry rather than hard-fail.
+        if (@intFromEnum(rr.response.head.status) >= 400) {
+            return provider_mod.classifyErrorResponse(self.allocator, &rr.response, &rr.transfer_buf, self.diag, "openai_chat");
+        }
+
+        // Bind the streaming body reader. Valid for the lifetime of `rr`
+        // (it borrows `&rr.response` and `&rr.transfer_buf`, both pinned).
+        rr.body_reader = rr.response.reader(&rr.transfer_buf);
+        return rr;
+    }
+};
+
+/// A resumable OpenAI Chat streaming response. Owns the pinned HTTP
+/// request/response, the body reader's transfer buffer, the `SSEParser`,
+/// and the block-assembly `StreamState`. `produce` pumps just enough bytes
+/// to emit one or more events into the queue, or reports the response is
+/// complete (its assistant message committed to the conversation).
+///
+/// Must be heap-allocated and never moved: `body_reader` borrows
+/// `&self.response`.
+pub const ResumableResponse = struct {
+    allocator: Allocator,
+    conv: *conversation.Conversation,
+    parser: sse_mod.SSEParser,
+    state: StreamState,
+
+    req: http.Client.Request = undefined,
+    response: http.Client.Response = undefined,
+    /// Transfer buffer backing `body_reader`. Pinned in the heap struct.
+    transfer_buf: [4096]u8 = undefined,
+    /// The streaming body reader, bound in `open` after a 2xx response.
+    body_reader: *std.Io.Reader = undefined,
+    /// Chunk scratch for `readVec`.
+    chunk: [4096]u8 = undefined,
+
+    /// True once `req` has been initialized (so `deinit` knows to free it).
+    req_open: bool = false,
+    /// Set once the response is fully decoded and finalized.
+    done: bool = false,
+
+    pub const ProduceStatus = provider_mod.ProviderStream.ProduceStatus;
+
+    /// Wrap this response in the provider-agnostic `ProviderStream` the agent
+    /// loop drives.
+    pub fn providerStream(self: *ResumableResponse) provider_mod.ProviderStream {
+        return .{ .ptr = self, .vtable = &vtable };
+    }
+
+    const vtable: provider_mod.ProviderStream.VTable = .{
+        .produce = produceVT,
+        .deinit = deinitVT,
+        .last_error = lastErrorVT,
+    };
+
+    fn lastErrorVT(ptr: *anyopaque) ?[]const u8 {
+        const self: *ResumableResponse = @ptrCast(@alignCast(ptr));
+        return self.state.stream_error_message;
+    }
+
+    fn produceVT(ptr: *anyopaque, out: *EventQueue) anyerror!ProduceStatus {
+        const self: *ResumableResponse = @ptrCast(@alignCast(ptr));
+        return self.produce(out);
+    }
+
+    fn deinitVT(ptr: *anyopaque) void {
+        const self: *ResumableResponse = @ptrCast(@alignCast(ptr));
+        self.deinit();
+    }
+
+    pub fn deinit(self: *ResumableResponse) void {
+        if (self.req_open) self.req.deinit();
+        self.parser.deinit();
+        self.state.deinit();
+        self.allocator.destroy(self);
+    }
+
+    /// Pump the response: read one chunk, feed it through the SSE parser,
+    /// and decode each SSE event into zero or more `Event`s appended to
+    /// `out`. Returns `.more` if the caller should pump again, or
+    /// `.response_complete` once the terminal (`[DONE]` or EOF) has been
+    /// reached and the assistant message has been committed + a final
+    /// `message_complete` pushed.
+    ///
+    /// Reading and finalizing here means a single `produce` call may push
+    /// several events; the `Stream` drains the queue before pumping again.
+    pub fn produce(self: *ResumableResponse, out: *EventQueue) !ProduceStatus {
+        if (self.done) return .response_complete;
+
+        var vecs: [1][]u8 = .{&self.chunk};
+        const n = self.body_reader.readVec(&vecs) catch |err| switch (err) {
+            // Stream ended without [DONE]. Some servers and proxies omit it
+            // (or drop the trailing usage chunk). Finalize with whatever
+            // we've got — usage will be null in that case, which is fine.
+            error.EndOfStream => {
+                try self.finishStream(out);
+                return .response_complete;
+            },
+            // A transport read failure mid-stream (reset, TLS, timeout)
+            // before `[DONE]` means no assistant message was committed.
+            // Surface it as a retryable malformed-stream error.
+            else => return error.ProviderStreamMalformed,
+        };
+        if (n == 0) return .more;
+
+        const events = try self.parser.feed(self.chunk[0..n]);
+        defer self.parser.freeEvents(events);
+
+        for (events) |ev_payload| {
+            std.log.debug("openai_chat <= {s}", .{ev_payload});
+            if (std.mem.eql(u8, ev_payload, "[DONE]")) {
+                try self.finishStream(out);
+                return .response_complete;
+            }
+            try handleEvent(self.allocator, ev_payload, &self.state, out);
+            // Note: we do NOT bail when state.end_of_stream is set.
+            // OpenAI emits the terminating `usage` chunk *after* the
+            // chunk carrying finish_reason, then sends `[DONE]`. If
+            // we returned on finish_reason we'd never capture usage.
+            // `[DONE]` is the authoritative end-of-stream marker.
+        }
+        return .more;
+    }
+
+    fn finishStream(self: *ResumableResponse, out: *EventQueue) !void {
+        if (self.done) return;
+        self.done = true;
+        try self.state.finalize(out, self.conv);
+    }
+};
+
+/// 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.
+///
+/// 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,
+    /// Set when the wire stream signals end-of-turn (finish_reason or [DONE]).
+    /// Tells the outer read loop to stop pulling more events.
+    end_of_stream: bool = false,
+    /// 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.
+    block_index: usize = 0,
+
+    /// 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, in stream order.
+    blocks: std.ArrayList(conversation.ContentBlock) = .empty,
+
+    /// 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),
+
+    /// Token counts from the terminating chunk's `usage` block. Only
+    /// populated when the server sent `usage` (i.e. the request used
+    /// `stream_options.include_usage: true` AND the server honored it).
+    usage: ?provider_mod.Usage = null,
+
+    /// Owned, human-readable description of a mid-stream error embedded in an
+    /// HTTP-200 SSE body, surfaced to the agent via `ProviderStream.lastError`
+    /// so the retry notice can explain *why*.
+    stream_error_message: ?[]u8 = null,
+
+    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,
+        /// 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,
+        /// 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);
+            self.name_buf.deinit(allocator);
+            self.arguments.deinit(allocator);
+        }
+    };
+
+    fn init(allocator: Allocator) StreamState {
+        return .{
+            .allocator = allocator,
+            .closed_tool_indices = std.AutoHashMap(usize, void).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);
+        if (self.active_tool) |*tu| tu.deinit(self.allocator);
+        self.closed_tool_indices.deinit();
+        if (self.stream_error_message) |s| self.allocator.free(s);
+    }
+
+    /// Record a readable description of an embedded stream error, combining
+    /// the error `type` and `message` into one owned string (either may be
+    /// absent). Replaces any previous value.
+    fn setStreamErrorMessage(self: *StreamState, kind: ?[]const u8, message: ?[]const u8) !void {
+        if (self.stream_error_message) |old| self.allocator.free(old);
+        self.stream_error_message = try provider_mod.formatStreamError(self.allocator, kind, message);
+    }
+
+    /// Close the active text/thinking block (if any) and emit
+    /// onBlockComplete. Ownership of `current_buf` transfers into the
+    /// appended block.
+    fn closeActive(self: *StreamState, out: *EventQueue) !void {
+        if (self.active == .none) return;
+
+        const block: conversation.ContentBlock = switch (self.active) {
+            .text => .{ .Text = self.current_buf },
+            .thinking => .{ .Thinking = .{ .text = self.current_buf } },
+            .tool_use, .none => unreachable,
+        };
+        self.current_buf = .empty;
+
+        try self.blocks.append(self.allocator, block);
+        try out.push(.{ .block_complete = .{
+            .index = self.block_index,
+            .block = self.blocks.items[self.blocks.items.len - 1],
+        } });
+
+        self.active = .none;
+    }
+
+    /// Open a new text/thinking block, possibly closing a prior one.
+    fn openBlock(
+        self: *StreamState,
+        new_active: ActiveBlock,
+        out: *EventQueue,
+    ) !void {
+        std.debug.assert(new_active == .text or new_active == .thinking);
+        if (self.active == new_active) return;
+        if (self.active != .none) {
+            try self.closeActive(out);
+            self.block_index += 1;
+        }
+        self.active = new_active;
+        const block_type: provider_mod.ContentBlockType = switch (new_active) {
+            .text => .Text,
+            .thinking => .Thinking,
+            .tool_use, .none => unreachable,
+        };
+        try out.push(.{ .block_start = .{ .block_type = block_type, .index = self.block_index } });
+    }
+
+    fn appendDelta(
+        self: *StreamState,
+        out: *EventQueue,
+        delta: []const u8,
+    ) !void {
+        try self.current_buf.appendSlice(self.allocator, delta);
+        // Dupe into the queue arena: the raw `delta` borrows the transient
+        // SSE payload that `produce` frees before `next()` reads the queue.
+        try out.push(.{ .content_delta = .{
+            .index = self.block_index,
+            .delta = try out.dupeBytes(delta),
+        } });
+    }
+
+    /// 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,
+        out: *EventQueue,
+        d: json_mod.ToolCallDelta,
+    ) !void {
+        // 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(out);
+        }
+
+        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(out);
+                self.block_index += 1;
+            }
+            self.active_tool = .{ .block_index = self.block_index };
+            self.current_tool_index = d.index;
+            self.block_index += 1;
+        }
+        const tu = &self.active_tool.?;
+
+        // Some OpenAI-compatible backends send identity as fragments
+        // (`call_` then `xyz`), while others resend the cumulative/full value
+        // on later argument chunks. Accept both without turning repeats into
+        // runaway ids like `call_xyzcall_xyz...`.
+        if (d.id) |s| try mergeStreamingField(self.allocator, &tu.id_buf, s);
+        if (d.name) |s| try mergeStreamingField(self.allocator, &tu.name_buf, s);
+
+        // 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(out, tu);
+            // Fire `tool_details` 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: consumers that need the canonical
+            // value can read it from the assembled ContentBlock at
+            // block_complete.
+            try self.emitDetailsIfReady(out, tu);
+            try tu.arguments.appendSlice(self.allocator, a);
+            // Dupe into the queue arena (the SSE payload is freed before
+            // `next()` reads the queue).
+            try out.push(.{ .content_delta = .{
+                .index = tu.block_index,
+                .delta = try out.dupeBytes(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(out, tu);
+        }
+    }
+
+    /// Fire `tool_details` once both id and name are non-empty. No-op if
+    /// already fired or if either field is still empty. Requires that
+    /// `block_start` has already been emitted. Slices are duped into the
+    /// queue arena because `id_buf`/`name_buf` may still grow (and realloc)
+    /// on later fragments.
+    fn emitDetailsIfReady(
+        self: *StreamState,
+        out: *EventQueue,
+        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 out.push(.{ .tool_details = .{
+            .index = tu.block_index,
+            .id = try out.dupeBytes(tu.id_buf.items),
+            .name = try out.dupeBytes(tu.name_buf.items),
+        } });
+    }
+
+    /// Close the currently-active tool_use (if any), emitting block_start
+    /// (if it wasn't already), block_complete, and recording the wire
+    /// index as closed. No-op if there's no active tool_use.
+    fn closeActiveTool(self: *StreamState, out: *EventQueue) !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;
+        }
+
+        // The model echoes the wire-encoded tool name (`__` for `.`).
+        // Decode in place now that the full name is assembled, so the
+        // conversation, receiver callbacks, and dispatch all see the
+        // internal (dotted) name. Decoding never grows the buffer.
+        decodeNameInPlace(&tu.name_buf);
+
+        // If no arguments ever arrived, we haven't emitted block_start
+        // yet — do it now so the consumer sees a balanced start/complete.
+        try self.emitStartIfNeeded(out, &tu);
+        // Last chance to fire details if a fragmented-identity provider
+        // only finished id/name accumulation at the very end.
+        try self.emitDetailsIfReady(out, &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 out.push(.{ .block_complete = .{
+            .index = tu.block_index,
+            .block = self.blocks.items[self.blocks.items.len - 1],
+        } });
+    }
+
+    /// Emit `block_start(.ToolUse, ...)` once per in-progress tool use.
+    /// Callers must invoke this before the first `content_delta` or
+    /// `block_complete` for the block. Identity (id/name) is *not* passed at
+    /// start — consumers get identity from `tool_details` or the assembled
+    /// ContentBlock at block_complete time.
+    fn emitStartIfNeeded(
+        self: *StreamState,
+        out: *EventQueue,
+        tu: *ToolUseInProgress,
+    ) !void {
+        _ = self;
+        if (tu.started) return;
+        tu.started = true;
+        try out.push(.{ .block_start = .{ .block_type = .ToolUse, .index = tu.block_index } });
+    }
+
+    fn mergeStreamingField(
+        allocator: Allocator,
+        buf: *conversation.TextualBlock,
+        piece: []const u8,
+    ) !void {
+        if (piece.len == 0) return;
+        if (buf.items.len == 0) {
+            try buf.appendSlice(allocator, piece);
+        } else if (std.mem.eql(u8, buf.items, piece)) {
+            return;
+        } else if (std.mem.startsWith(u8, piece, buf.items)) {
+            buf.clearRetainingCapacity();
+            try buf.appendSlice(allocator, piece);
+        } else {
+            try buf.appendSlice(allocator, piece);
+        }
+    }
+
+    /// 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 and push the terminal `message_complete`.
+    fn finalize(
+        self: *StreamState,
+        out: *EventQueue,
+        conv: *conversation.Conversation,
+    ) !void {
+        if (self.finalized) return;
+        self.finalized = true;
+
+        try self.closeActive(out);
+        try self.closeActiveTool(out);
+
+        // 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, self.usage);
+
+        const msg = conv.messages.items[conv.messages.items.len - 1];
+        try out.push(.{ .message_complete = .{ .message = msg, .usage = self.usage } });
+    }
+};
+
+fn handleEvent(
+    allocator: Allocator,
+    payload: []const u8,
+    state: *StreamState,
+    out: *EventQueue,
+) !void {
+    var parsed = try json_mod.parseStreamEvent(allocator, payload);
+    defer parsed.deinit();
+    const d = parsed.delta;
+
+    // Usage block arrives in the terminating chunk (after finish_reason,
+    // with an empty `choices` array). Capture it; `finalize` delivers it
+    // as part of `onMessageComplete`. OpenAI bills `prompt_tokens` as
+    // the *total* input including cached tokens; we split them so
+    // callers don't have to.
+    if (d.usage) |u| {
+        const prompt: u64 = u.prompt_tokens orelse 0;
+        const cached: u64 = u.cached_prompt_tokens orelse 0;
+        const fresh: u64 = if (cached > prompt) 0 else prompt - cached;
+        state.usage = .{
+            .input = fresh,
+            .output = u.completion_tokens orelse 0,
+            .cache_read = cached,
+            .cache_write = 0, // OpenAI doesn't bill a cache-write premium.
+            .reasoning = u.reasoning_tokens orelse 0,
+        };
+    }
+
+    // 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,
+            });
+        }
+        // Stash a readable description so the agent's retry notice can
+        // explain *why* the stream failed. Owned by `state`.
+        state.setStreamErrorMessage(d.error_type, d.error_message) catch {};
+        return error.ProviderStreamMalformed;
+    }
+
+    if (!state.started and d.role != null) {
+        state.started = true;
+        try out.push(.{ .message_start = .assistant });
+    }
+
+    if (d.reasoning_content) |rc| if (rc.len > 0) {
+        if (!state.started) {
+            state.started = true;
+            try out.push(.{ .message_start = .assistant });
+        }
+        try state.openBlock(.thinking, out);
+        try state.appendDelta(out, rc);
+    };
+
+    if (d.content) |c| if (c.len > 0) {
+        if (!state.started) {
+            state.started = true;
+            try out.push(.{ .message_start = .assistant });
+        }
+        try state.openBlock(.text, out);
+        try state.appendDelta(out, c);
+    };
+
+    if (d.tool_calls.len > 0) {
+        if (!state.started) {
+            state.started = true;
+            try out.push(.{ .message_start = .assistant });
+        }
+        for (d.tool_calls) |tc| try state.applyToolCallDelta(out, tc);
+    }
+
+    if (d.finish_reason) |_| {
+        state.end_of_stream = true;
+    }
+}
+
+// -----------------------------------------------------------------------------
+// Tests
+// -----------------------------------------------------------------------------
+
+const testing = std.testing;
+
+/// Feed a sequence of SSE event payloads through the state machine as if
+/// they had been delivered by the wire, finalizing into `conv`. The decoded
+/// `Event`s are recorded as compact strings (the same schema the old
+/// RecordingReceiver used) so callback-ordering assertions are preserved.
+fn runStreamedTurn(
+    allocator: Allocator,
+    conv: *conversation.Conversation,
+    rec: ?*EventRecorder,
+    events: []const []const u8,
+) !void {
+    var state: StreamState = .init(allocator);
+    defer state.deinit();
+
+    var queue = EventQueue.init(allocator);
+    defer queue.deinit();
+
+    for (events) |payload| {
+        if (std.mem.eql(u8, payload, "[DONE]")) break;
+        // Process every chunk through to [DONE], including the
+        // post-finish_reason usage chunk. Mirrors the production pump in
+        // ResumableResponse.produce.
+        try handleEvent(allocator, payload, &state, &queue);
+    }
+    try state.finalize(&queue, conv);
+
+    // Drain into the recorder before the arena resets. The queue holds all
+    // events from this turn; popping records each, and the final null-pop
+    // resets the arena.
+    while (queue.pop()) |ev| {
+        if (rec) |r| try r.record(ev);
+    }
+}
+
+/// Records decoded `Event`s as compact strings for ordering assertions.
+const EventRecorder = struct {
+    allocator: Allocator,
+    events: std.ArrayList([]const u8) = .empty,
+
+    fn deinit(self: *EventRecorder) void {
+        for (self.events.items) |e| self.allocator.free(e);
+        self.events.deinit(self.allocator);
+    }
+
+    fn push(self: *EventRecorder, 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 record(self: *EventRecorder, ev: Event) !void {
+        switch (ev) {
+            .message_start => try self.push("msg_start", .{}),
+            .block_start => |b| try self.push("block_start[{d}]:{s}", .{ b.index, @tagName(b.block_type) }),
+            .tool_details => |t| try self.push("tool_details[{d}]:{s}:{s}", .{ t.index, t.id, t.name }),
+            .content_delta => |d| try self.push("delta[{d}]:{s}", .{ d.index, d.delta }),
+            .block_complete => |b| try self.push("block_complete[{d}]", .{b.index}),
+            .message_complete => |m| {
+                if (m.usage) |u| {
+                    try self.push(
+                        "msg_complete[usage:in={d},out={d},cr={d},cw={d},rsn={d}]",
+                        .{ u.input, u.output, u.cache_read, u.cache_write, u.reasoning },
+                    );
+                } else {
+                    try self.push("msg_complete[usage:null]", .{});
+                }
+            },
+            else => {},
+        }
+    }
+};
+
+/// Test helper: append a single-text user message. `addUserMessage` now
+/// takes a block slice (symmetric with `addAssistantMessage`); this wraps
+/// the common plain-text case the tests below use.
+fn addUserText(conv: *conversation.Conversation, text: []const u8) !void {
+    const tb = try conversation.textualBlockFromSlice(conv.allocator, text);
+    var block: conversation.ContentBlock = .{ .Text = tb };
+    errdefer block.deinit(conv.allocator);
+    try conv.addUserMessage(&.{block});
+}
+
+test "two streamed turns persist assistant replies in the conversation" {
+    // Regression test for the bug where `finish_reason` arrived before
+    // `[DONE]` and `finalize` early-returned without appending the assistant
+    // message, so follow-up turns were sent without prior responses.
+
+    const allocator = testing.allocator;
+
+    var conv = conversation.Conversation.init(allocator);
+    defer conv.deinit();
+
+    try conv.addSystemMessage("You are a helpful assistant.");
+    try addUserText(&conv, "hello!");
+
+    const turn1 = [_][]const u8{
+        \\{"choices":[{"delta":{"role":"assistant"}}]}
+        ,
+        \\{"choices":[{"delta":{"content":"Hello! "}}]}
+        ,
+        \\{"choices":[{"delta":{"content":"How can I help you today?"}}]}
+        ,
+        \\{"choices":[{"delta":{},"finish_reason":"stop"}]}
+        ,
+        "[DONE]",
+    };
+    try runStreamedTurn(allocator, &conv, null, &turn1);
+
+    try testing.expectEqual(@as(usize, 3), conv.messages.items.len);
+    try testing.expectEqual(conversation.MessageRole.assistant, conv.messages.items[2].role);
+    try testing.expectEqualStrings(
+        "Hello! How can I help you today?",
+        conv.messages.items[2].content.items[0].Text.items,
+    );
+
+    // Second user turn: the assistant must still see its prior response.
+    try addUserText(&conv, "how did you respond to my greeting just now?");
+
+    const turn2 = [_][]const u8{
+        \\{"choices":[{"delta":{"role":"assistant"}}]}
+        ,
+        \\{"choices":[{"delta":{"content":"I replied: \"Hello! How can I help you today?\""}}]}
+        ,
+        \\{"choices":[{"delta":{},"finish_reason":"stop"}]}
+        ,
+        "[DONE]",
+    };
+    try runStreamedTurn(allocator, &conv, null, &turn2);
+
+    // System + user + assistant + user + assistant = 5 messages.
+    try testing.expectEqual(@as(usize, 5), conv.messages.items.len);
+    try testing.expectEqual(conversation.MessageRole.assistant, conv.messages.items[4].role);
+    try testing.expectEqualStrings(
+        "I replied: \"Hello! How can I help you today?\"",
+        conv.messages.items[4].content.items[0].Text.items,
+    );
+}
+
+test "openai_chat: terminating usage chunk lands on message_complete with split cache_read" {
+    const allocator = testing.allocator;
+
+    var conv = conversation.Conversation.init(allocator);
+    defer conv.deinit();
+    try addUserText(&conv, "hi");
+
+    var rec = EventRecorder{ .allocator = allocator };
+    defer rec.deinit();
+
+    const events = [_][]const u8{
+        \\{"choices":[{"delta":{"role":"assistant"}}]}
+        ,
+        \\{"choices":[{"delta":{"content":"hi"}}]}
+        ,
+        \\{"choices":[{"delta":{},"finish_reason":"stop"}]}
+        ,
+        // OpenAI's terminating chunk: empty choices, top-level usage.
+        \\{"choices":[],"usage":{"prompt_tokens":150,"completion_tokens":42,"prompt_tokens_details":{"cached_tokens":120},"completion_tokens_details":{"reasoning_tokens":18}}}
+        ,
+        "[DONE]",
+    };
+    try runStreamedTurn(allocator, &conv, &rec, &events);
+
+    var found: ?[]const u8 = null;
+    for (rec.events.items) |s| {
+        if (std.mem.startsWith(u8, s, "msg_complete[")) found = s;
+    }
+    try testing.expect(found != null);
+    // 150 prompt - 120 cached = 30 fresh input.
+    try testing.expectEqualStrings("msg_complete[usage:in=30,out=42,cr=120,cw=0,rsn=18]", found.?);
+}
+
+test "openai_chat: omitted stream usage yields null on message_complete" {
+    const allocator = testing.allocator;
+
+    var conv = conversation.Conversation.init(allocator);
+    defer conv.deinit();
+    try addUserText(&conv, "hi");
+
+    var rec = EventRecorder{ .allocator = allocator };
+    defer rec.deinit();
+
+    const events = [_][]const u8{
+        \\{"choices":[{"delta":{"role":"assistant"}}]}
+        ,
+        \\{"choices":[{"delta":{"content":"hi"}}]}
+        ,
+        \\{"choices":[{"delta":{},"finish_reason":"stop"}]}
+        ,
+        "[DONE]",
+    };
+    try runStreamedTurn(allocator, &conv, &rec, &events);
+
+    var found: ?[]const u8 = null;
+    for (rec.events.items) |s| {
+        if (std.mem.startsWith(u8, s, "msg_complete")) found = s;
+    }
+    try testing.expect(found != null);
+    try testing.expectEqualStrings("msg_complete[usage:null]", found.?);
+}
+
+test "openai_chat: empty content alongside reasoning does not split thinking block" {
+    const allocator = testing.allocator;
+
+    var conv = conversation.Conversation.init(allocator);
+    defer conv.deinit();
+    try addUserText(&conv, "think");
+
+    var rec = EventRecorder{ .allocator = allocator };
+    defer rec.deinit();
+
+    const events = [_][]const u8{
+        \\{"choices":[{"delta":{"role":"assistant"}}]}
+        ,
+        \\{"choices":[{"delta":{"reasoning_content":"rea","content":""}}]}
+        ,
+        \\{"choices":[{"delta":{"reasoning_content":"son","content":""}}]}
+        ,
+        \\{"choices":[{"delta":{"reasoning_content":"ing","content":""}}]}
+        ,
+        \\{"choices":[{"delta":{},"finish_reason":"stop"}]}
+        ,
+        "[DONE]",
+    };
+    try runStreamedTurn(allocator, &conv, &rec, &events);
+
+    const expected = [_][]const u8{
+        "msg_start",
+        "block_start[0]:Thinking",
+        "delta[0]:rea",
+        "delta[0]:son",
+        "delta[0]:ing",
+        "block_complete[0]",
+        "msg_complete[usage:null]",
+    };
+    try testing.expectEqual(expected.len, rec.events.items.len);
+    for (expected, rec.events.items) |want, got| {
+        try testing.expectEqualStrings(want, got);
+    }
+
+    const asst = conv.messages.items[1];
+    try testing.expectEqual(@as(usize, 1), asst.content.items.len);
+    try testing.expectEqualStrings("reasoning", asst.content.items[0].Thinking.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 addUserText(&conv, "call something");
+
+    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, null, &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 "repeated full tool_call id and name do not accumulate" {
+    const allocator = testing.allocator;
+
+    var conv = conversation.Conversation.init(allocator);
+    defer conv.deinit();
+    try addUserText(&conv, "read it");
+
+    const events = [_][]const u8{
+        \\{"choices":[{"delta":{"role":"assistant"}}]}
+        ,
+        \\{"choices":[{"delta":{"tool_calls":[{"index":0,"id":"chatcmpl-tool-abc","type":"function","function":{"name":"std__read","arguments":"{\"path\""}}]}}]}
+        ,
+        \\{"choices":[{"delta":{"tool_calls":[{"index":0,"id":"chatcmpl-tool-abc","function":{"name":"std__read","arguments":":\"a\"}"}}]}}]}
+        ,
+        \\{"choices":[{"delta":{},"finish_reason":"tool_calls"}]}
+        ,
+        "[DONE]",
+    };
+
+    try runStreamedTurn(allocator, &conv, null, &events);
+
+    const tu = conv.messages.items[1].content.items[0].ToolUse;
+    try testing.expectEqualStrings("chatcmpl-tool-abc", tu.id);
+    try testing.expectEqualStrings("std.read", tu.name);
+    try testing.expectEqualStrings("{\"path\":\"a\"}", tu.input.items);
+}
+
+test "cumulative tool_call id and name replace their prefix" {
+    const allocator = testing.allocator;
+
+    var conv = conversation.Conversation.init(allocator);
+    defer conv.deinit();
+    try addUserText(&conv, "read it");
+
+    const events = [_][]const u8{
+        \\{"choices":[{"delta":{"role":"assistant"}}]}
+        ,
+        \\{"choices":[{"delta":{"tool_calls":[{"index":0,"id":"chatcmpl-tool-","type":"function","function":{"name":"std_","arguments":"{\"path\""}}]}}]}
+        ,
+        \\{"choices":[{"delta":{"tool_calls":[{"index":0,"id":"chatcmpl-tool-abc","function":{"name":"std__read","arguments":":\"a\"}"}}]}}]}
+        ,
+        \\{"choices":[{"delta":{},"finish_reason":"tool_calls"}]}
+        ,
+        "[DONE]",
+    };
+
+    try runStreamedTurn(allocator, &conv, null, &events);
+
+    const tu = conv.messages.items[1].content.items[0].ToolUse;
+    try testing.expectEqualStrings("chatcmpl-tool-abc", tu.id);
+    try testing.expectEqualStrings("std.read", tu.name);
+    try testing.expectEqualStrings("{\"path\":\"a\"}", tu.input.items);
+}
+
+test "inbound wire tool name is decoded to dotted form (even split across __)" {
+    // The model echoes the wire name it was given (`std__read`). It is
+    // decoded to the internal `std.read` for the conversation/session/
+    // dispatch. The decode happens after full assembly, so a `__` split
+    // across two deltas (`std_` + `_read`) decodes correctly.
+    const allocator = testing.allocator;
+
+    var conv = conversation.Conversation.init(allocator);
+    defer conv.deinit();
+    try addUserText(&conv, "read a file");
+
+    const events = [_][]const u8{
+        \\{"choices":[{"delta":{"role":"assistant"}}]}
+        ,
+        \\{"choices":[{"delta":{"tool_calls":[{"index":0,"id":"c1","type":"function","function":{"name":"std_"}}]}}]}
+        ,
+        \\{"choices":[{"delta":{"tool_calls":[{"index":0,"function":{"name":"_read"}}]}}]}
+        ,
+        \\{"choices":[{"delta":{"tool_calls":[{"index":0,"function":{"arguments":"{}"}}]}}]}
+        ,
+        \\{"choices":[{"delta":{},"finish_reason":"tool_calls"}]}
+        ,
+        "[DONE]",
+    };
+
+    try runStreamedTurn(allocator, &conv, null, &events);
+
+    const tu = conv.messages.items[1].content.items[0].ToolUse;
+    try testing.expectEqualStrings("std.read", tu.name);
+}
+
+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 addUserText(&conv, "ping four hosts");
+
+    var rec: EventRecorder = .{ .allocator = allocator };
+    defer rec.deinit();
+
+    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, &rec, &events);
+
+    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]",
+        // No usage chunk in this fixture (older test data) — record
+        // shows null.
+        "msg_complete[usage:null]",
+    };
+
+    // 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 addUserText(&conv, "go");
+
+    var rec: EventRecorder = .{ .allocator = allocator };
+    defer rec.deinit();
+
+    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, &rec, &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 "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 addUserText(&conv, "go");
+
+    var rec: EventRecorder = .{ .allocator = allocator };
+    defer rec.deinit();
+
+    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, &rec, &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
+    // 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 addUserText(&conv, "ring it");
+
+    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, null, &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);
+}