path: root/libpanto/src/provider_openai_chat.zig

//! 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 sse_mod = @import("sse.zig");
const json_mod = @import("openai_chat_json.zig");
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, tool_use };

pub const OpenAIChatProvider = struct {
    allocator: Allocator,
    io: Io,
    config: config_mod.OpenAIChatConfig,
    http_client: http.Client,

    pub fn init(allocator: Allocator, io: Io, cfg: config_mod.OpenAIChatConfig) OpenAIChatProvider {
        return .{
            .allocator = allocator,
            .io = io,
            .config = cfg,
            .http_client = .{ .allocator = allocator, .io = io },
        };
    }

    pub fn deinit(self: *OpenAIChatProvider) void {
        self.http_client.deinit();
    }

    /// Return a `Provider` interface bound to this concrete provider.
    pub fn provider(self: *OpenAIChatProvider) provider_mod.Provider {
        return .{ .ptr = self, .vtable = &vtable };
    }

    const vtable: provider_mod.ProviderVTable = .{
        .streamStep = vtableStreamStep,
        .deinit = vtableDeinit,
    };

    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, tools, receiver);
    }

    /// Called via the `Provider` interface. Tears down the impl AND frees
    /// its heap allocation, since `Provider.init` is the one that allocated
    /// it. Direct stack-allocated users (tests, embedders) call `deinit`
    /// themselves and never hit this path.
    fn vtableDeinit(ptr: *anyopaque) void {
        const self: *OpenAIChatProvider = @ptrCast(@alignCast(ptr));
        const allocator = self.allocator;
        self.deinit();
        allocator.destroy(self);
    }

    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, tools, receiver) catch |err| {
            receiver.onError(err);
            return err;
        };
    }

    fn streamStepInner(
        self: *OpenAIChatProvider,
        conv: *conversation.Conversation,
        tools: *const provider_mod.ToolRegistry,
        receiver: *provider_mod.Receiver,
    ) !void {
        // 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);

        // Auth header
        const auth_value = try std.fmt.allocPrint(
            self.allocator,
            "Bearer {s}",
            .{self.config.api_key},
        );
        defer self.allocator.free(auth_value);

        const extra_headers = [_]http.Header{
            .{ .name = "content-type", .value = "application/json" },
            .{ .name = "accept", .value = "text/event-stream" },
            .{ .name = "authorization", .value = auth_value },
        };

        // Open the request. We can't use `fetch()` because it buffers the
        // response; we want to stream the body as it arrives.
        var req = try self.http_client.request(.POST, uri, .{
            .extra_headers = &extra_headers,
            // Disable compression: gzip buffers small SSE frames, defeating
            // the streaming property we paid for `stream: true` to get.
            .headers = .{ .accept_encoding = .{ .override = "identity" } },
            .keep_alive = false,
            .redirect_behavior = .not_allowed,
        });
        defer req.deinit();

        req.transfer_encoding = .{ .content_length = body.len };

        var send_buf: [4096]u8 = undefined;
        var bw = try req.sendBodyUnflushed(&send_buf);
        try bw.writer.writeAll(body);
        try bw.end();
        try req.connection.?.flush();

        // Receive response headers.
        var redirect_buf: [1024]u8 = undefined;
        var response = try req.receiveHead(&redirect_buf);

        if (@intFromEnum(response.head.status) >= 400) {
            // Drain body for diagnostics.
            var transfer_buf: [4096]u8 = undefined;
            const body_reader = response.reader(&transfer_buf);
            var err_buf: std.ArrayList(u8) = .empty;
            defer err_buf.deinit(self.allocator);
            var tmp: [1024]u8 = undefined;
            while (true) {
                const n = body_reader.readSliceShort(&tmp) catch break;
                if (n == 0) break;
                try err_buf.appendSlice(self.allocator, tmp[0..n]);
                if (err_buf.items.len > 16 * 1024) break;
            }
            std.log.err("openai_chat HTTP {d}: {s}", .{
                @intFromEnum(response.head.status),
                err_buf.items,
            });
            return error.HttpError;
        }

        // Stream the body through the SSE parser and event handler.
        var transfer_buf: [4096]u8 = undefined;
        const body_reader = response.reader(&transfer_buf);

        var parser = sse_mod.SSEParser.init(self.allocator);
        defer parser.deinit();

        var state: StreamState = .init(self.allocator);
        defer state.deinit();

        // 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) {
            const n = body_reader.readVec(&vecs) catch |err| switch (err) {
                error.EndOfStream => break,
                else => return err,
            };
            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;
                }
                try handleEvent(self.allocator, ev_payload, &state, receiver);
                if (state.end_of_stream) {
                    try state.finalize(receiver, conv);
                    return;
                }
            }
        }

        // Stream ended without [DONE] or finish_reason. Finalize anyway.
        try state.finalize(receiver, 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),

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

    /// 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 } },
            .tool_use, .none => unreachable,
        };
        self.current_buf = .empty;

        try self.blocks.append(self.allocator, block);
        try receiver.onBlockComplete(self.block_index, 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,
        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);
            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 receiver.onBlockStart(block_type, self.block_index);
    }

    fn appendDelta(
        self: *StreamState,
        receiver: *provider_mod.Receiver,
        delta: []const u8,
    ) !void {
        try self.current_buf.appendSlice(self.allocator, delta);
        try receiver.onContentDelta(self.block_index, 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,
        receiver: *provider_mod.Receiver,
        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(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;
            }
            self.active_tool = .{ .block_index = self.block_index };
            self.current_tool_index = d.index;
            self.block_index += 1;
        }
        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
        // 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 args begin. The first argument
        // fragment is our signal that identity is likely settled enough
        // to render. If the block closes before any args arrive (zero-arg
        // tool), `closeActiveTool` emits the start there.
        if (d.arguments) |a| {
            try self.emitStartIfNeeded(receiver, tu);
            // Fire `onToolDetails` as soon as both id and name are
            // known. We can't know identity is *final* until the block
            // closes (a later delta could append more bytes), but in
            // practice OpenAI sends each whole on the first delta. A
            // pathological backend that streams id/name across many
            // chunks would have us emit a truncated value here. We
            // accept that trade-off: receivers that need the canonical
            // value can read it from the assembled ContentBlock at
            // onBlockComplete.
            try self.emitDetailsIfReady(receiver, tu);
            try tu.arguments.appendSlice(self.allocator, a);
            try receiver.onContentDelta(tu.block_index, a);
        } else {
            // Identity-only chunk (no args yet). Still try to emit
            // details, in case both fields are now populated.
            if (tu.started) try self.emitDetailsIfReady(receiver, tu);
        }
    }

    /// Fire `onToolDetails` once both id and name are non-empty. No-op if
    /// already fired or if either field is still empty. Requires that
    /// `onBlockStart` has already been emitted.
    fn emitDetailsIfReady(
        self: *StreamState,
        receiver: *provider_mod.Receiver,
        tu: *ToolUseInProgress,
    ) !void {
        _ = self;
        if (tu.details_emitted) return;
        if (!tu.started) return;
        if (tu.id_buf.items.len == 0 or tu.name_buf.items.len == 0) return;
        tu.details_emitted = true;
        try receiver.onToolDetails(tu.block_index, tu.id_buf.items, tu.name_buf.items);
    }

    /// Close the currently-active tool_use (if any), emitting onBlockStart
    /// (if it wasn't already), onBlockComplete, and recording the wire
    /// index as closed. No-op if there's no active tool_use.
    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);
        // Last chance to fire details if a fragmented-identity provider
        // only finished id/name accumulation at the very end.
        try self.emitDetailsIfReady(receiver, &tu);

        const id_owned = try tu.id_buf.toOwnedSlice(self.allocator);
        const name_owned = try tu.name_buf.toOwnedSlice(self.allocator);
        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,
        tu: *ToolUseInProgress,
    ) !void {
        _ = self;
        if (tu.started) return;
        tu.started = true;
        try receiver.onBlockStart(.ToolUse, tu.block_index);
    }

    /// 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,
        conv: *conversation.Conversation,
    ) !void {
        if (self.finalized) return;
        self.finalized = true;

        try self.closeActive(receiver);
        try self.closeActiveTool(receiver);

        // 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);

        const msg = conv.messages.items[conv.messages.items.len - 1];
        try receiver.onMessageComplete(msg);
    }
};

fn handleEvent(
    allocator: Allocator,
    payload: []const u8,
    state: *StreamState,
    receiver: *provider_mod.Receiver,
) !void {
    var parsed = try json_mod.parseStreamEvent(allocator, payload);
    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);
    }

    if (d.reasoning_content) |rc| {
        if (!state.started) {
            state.started = true;
            try receiver.onMessageStart(.assistant);
        }
        try state.openBlock(.thinking, receiver);
        try state.appendDelta(receiver, rc);
    }

    if (d.content) |c| {
        if (!state.started) {
            state.started = true;
            try receiver.onMessageStart(.assistant);
        }
        try state.openBlock(.text, receiver);
        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;
    }
}

// -----------------------------------------------------------------------------
// Tests
// -----------------------------------------------------------------------------

const testing = std.testing;

/// A no-op Receiver that drops every callback. Useful when the test cares
/// about post-stream conversation state rather than callback observability.
const NoopReceiver = struct {
    fn make() provider_mod.Receiver {
        return .{ .ptr = @constCast(@ptrCast(&dummy)), .vtable = &vt };
    }
    var dummy: u8 = 0;
    const vt: provider_mod.ReceiverVTable = .{
        .onMessageStart = noopMsgStart,
        .onBlockStart = noopBlockStart,
        .onToolDetails = noopToolDetails,
        .onContentDelta = noopDelta,
        .onBlockComplete = noopBlockComplete,
        .onMessageComplete = noopMsgComplete,
        .onError = noopErr,
    };
    fn noopMsgStart(_: *anyopaque, _: conversation.MessageRole) anyerror!void {}
    fn noopBlockStart(_: *anyopaque, _: provider_mod.ContentBlockType, _: usize) anyerror!void {}
    fn noopToolDetails(_: *anyopaque, _: usize, _: []const u8, _: []const u8) anyerror!void {}
    fn noopDelta(_: *anyopaque, _: usize, _: []const u8) anyerror!void {}
    fn noopBlockComplete(_: *anyopaque, _: usize, _: conversation.ContentBlock) anyerror!void {}
    fn noopMsgComplete(_: *anyopaque, _: conversation.Message) anyerror!void {}
    fn noopErr(_: *anyopaque, _: anyerror) void {}
};

/// Feed a sequence of SSE event payloads through the state machine as if
/// they had been delivered by the wire, finalizing into `conv`.
fn runStreamedTurn(
    allocator: Allocator,
    conv: *conversation.Conversation,
    receiver: *provider_mod.Receiver,
    events: []const []const u8,
) !void {
    var state: StreamState = .init(allocator);
    defer state.deinit();

    for (events) |payload| {
        if (std.mem.eql(u8, payload, "[DONE]")) break;
        try handleEvent(allocator, payload, &state, receiver);
        if (state.end_of_stream) break;
    }
    try state.finalize(receiver, conv);
}

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 conv.addUserMessage("hello!");

    var recv = NoopReceiver.make();

    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, &recv, &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 conv.addUserMessage("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, &recv, &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 "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);
}

/// 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,
        .onToolDetails = onToolDetails,
        .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 onToolDetails(
        ptr: *anyopaque,
        idx: usize,
        id: []const u8,
        name: []const u8,
    ) anyerror!void {
        const self: *RecordingReceiver = @ptrCast(@alignCast(ptr));
        try self.recordFmt("tool_details[{d}]:{s}:{s}", .{ idx, id, name });
    }
    fn onContentDelta(ptr: *anyopaque, idx: usize, delta: []const u8) anyerror!void {
        const self: *RecordingReceiver = @ptrCast(@alignCast(ptr));
        try self.recordFmt("delta[{d}]:{s}", .{ idx, delta });
    }
    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",
        "tool_details[0]:c0:ping",
        "delta[0]:{\"host\":\"a\"}",
        "block_complete[0]",
        "block_start[1]:ToolUse",
        "tool_details[1]:c1:ping",
        "delta[1]:{\"host\":\"b\"}",
        "block_complete[1]",
        "block_start[2]:ToolUse",
        "tool_details[2]:c2:ping",
        "delta[2]:{\"host\":\"c\"}",
        "block_complete[2]",
        "block_start[3]:ToolUse",
        "tool_details[3]:c3:ping",
        "delta[3]:{\"host\":\"d\"}",
        "block_complete[3]",
        "msg_complete",
    };

    // 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 "onToolDetails fires after id+name complete, even mid-arg-stream" {
    // Fragmented-identity provider: id arrives split across two chunks,
    // and an arg fragment appears between them. `onToolDetails` must
    // wait until both id and name are non-empty (i.e. on the chunk that
    // completes id), and must fire exactly once, before block_complete.
    const allocator = testing.allocator;

    var conv = conversation.Conversation.init(allocator);
    defer conv.deinit();
    try conv.addUserMessage("go");

    var rec: RecordingReceiver = .{ .allocator = allocator };
    defer rec.deinit();
    var recv = rec.receiver();

    const events = [_][]const u8{
        \\{"choices":[{"delta":{"role":"assistant"}}]}
        ,
        // Identity-only chunk: name arrives whole, id starts. No args yet,
        // so onBlockStart hasn't fired and details can't either.
        \\{"choices":[{"delta":{"tool_calls":[{"index":0,"id":"call_","type":"function","function":{"name":"ping"}}]}}]}
        ,
        // First arg chunk: onBlockStart fires. id is still "call_" — not
        // empty — and name is non-empty, so onToolDetails fires here
        // with whatever id we have so far (`call_`).
        \\{"choices":[{"delta":{"tool_calls":[{"index":0,"function":{"arguments":"{\"x\":"}}]}}]}
        ,
        // Mid-stream id completion + second arg chunk. onToolDetails has
        // already fired so it does NOT fire again, even though id grew.
        // The final ContentBlock will carry the full "call_xyz" id.
        \\{"choices":[{"delta":{"tool_calls":[{"index":0,"id":"xyz","function":{"arguments":"1}"}}]}}]}
        ,
        \\{"choices":[{"delta":{},"finish_reason":"tool_calls"}]}
        ,
        "[DONE]",
    };

    try runStreamedTurn(allocator, &conv, &recv, &events);

    // Exactly one tool_details event, fired with the id-prefix that was
    // current at first-args-arrival, and ordered between block_start and
    // block_complete.
    var n_details: usize = 0;
    var details_pos: ?usize = null;
    var block_start_pos: ?usize = null;
    var block_complete_pos: ?usize = null;
    for (rec.events.items, 0..) |e, i| {
        if (std.mem.startsWith(u8, e, "tool_details[")) {
            n_details += 1;
            details_pos = i;
            try testing.expectEqualStrings("tool_details[0]:call_:ping", e);
        } else if (std.mem.eql(u8, e, "block_start[0]:ToolUse")) {
            block_start_pos = i;
        } else if (std.mem.eql(u8, e, "block_complete[0]")) {
            block_complete_pos = i;
        }
    }
    try testing.expectEqual(@as(usize, 1), n_details);
    try testing.expect(block_start_pos.? < details_pos.?);
    try testing.expect(details_pos.? < block_complete_pos.?);

    // Final ContentBlock has the full id assembled from both fragments.
    const asst = conv.messages.items[1];
    try testing.expectEqualStrings("call_xyz", asst.content.items[0].ToolUse.id);
    try testing.expectEqualStrings("ping", asst.content.items[0].ToolUse.name);
    try testing.expectEqualStrings("{\"x\":1}", asst.content.items[0].ToolUse.input.items);
}

test "tool_call with no arguments still finalizes a well-formed ToolUse" {
    // Some providers may emit a tool call with no arguments at all (e.g. a
    // zero-arg tool). The state machine should still emit onBlockStart
    // 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);
}