//! OpenAI Responses API streaming provider (ChatGPT-subscription Codex).
//!
//! Mirrors `provider_openai_chat.zig` in shape — a transient request object
//! that opens the HTTP stream and a heap-pinned `ResumableResponse` that pumps
//! SSE bytes into `Event`s — but speaks the Responses streaming protocol
//! (typed `response.*` events) instead of Chat Completions `choices[].delta`.
//!
//! Event → block mapping:
//!   - `response.output_text.delta`            → Text block deltas
//!   - `response.reasoning_summary_text.delta` → Thinking block deltas
//!   - `response.output_item.added` (function_call)  → opens a ToolUse block
//!   - `response.function_call_arguments.delta`      → ToolUse input deltas
//!   - `response.output_item.done`  (function_call)  → closes the ToolUse
//!   - `response.completed`                          → usage + finalize
//!   - `error` / `response.failed`                   → malformed-stream error
//!
//! NOTE: the Responses-backed Codex path could not be verified against live
//! ChatGPT-subscription credentials; the request/stream shapes follow the
//! OpenAI Responses API docs and the open-source Codex client. Fixture tests
//! exercise the state machine; live verification is still required.

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_responses_json.zig");
const config_mod = @import("config.zig");

const Event = stream_mod.Event;
const EventQueue = stream_mod.EventQueue;

pub const OpenAIResponsesDialect = enum {
    public,
    codex,
};

const decodeNameInPlace = provider_mod.decodeNameInPlace;

pub const OpenAIResponsesRequest = struct {
    allocator: Allocator,
    io: Io,
    config: *const config_mod.OpenAIResponsesConfig,
    dialect: OpenAIResponsesDialect = .public,
    http_client: *http.Client,
    diag: ?*provider_mod.ProviderDiagnostic = null,

    pub fn open(
        self: *OpenAIResponsesRequest,
        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),
        };
        rr.state.signature_origin = try conversation.SignatureOrigin.init(
            self.allocator,
            if (self.dialect == .codex) .openai_codex_responses else .openai_responses,
            self.config.base_url,
            self.config.model,
        );
        errdefer {
            rr.parser.deinit();
            rr.state.deinit();
        }

        const url = try responsesURL(self.allocator, self.config.base_url);
        defer self.allocator.free(url);
        const uri = try Uri.parse(url);

        const body = try json_mod.serializeRequest(self.allocator, self.config, conv, tools, switch (self.dialect) {
            .public => .public,
            .codex => .codex,
        });
        defer self.allocator.free(body);
        std.log.debug("openai_responses => {s}", .{body});

        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 },
        };
        const extra_headers = try provider_mod.mergeHeaders(
            self.allocator,
            &base_headers,
            self.config.extra_headers,
        );
        defer self.allocator.free(extra_headers);

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

        if (@intFromEnum(rr.response.head.status) >= 400) {
            return provider_mod.classifyErrorResponse(self.allocator, &rr.response, &rr.transfer_buf, self.diag, "openai_responses");
        }

        rr.body_reader = rr.response.reader(&rr.transfer_buf);
        return rr;
    }
};

// Appends `/responses` to `base_url` (with trailing slashes trimmed). The
// caller is responsible for any path segment preceding `/responses` — for
// the Codex ChatGPT-subscription endpoint, that means putting `/codex`
// (or any other prefix) in `base_url` directly. No `endsWith` guessing: the
// contract is the same regardless of `dialect`.
fn responsesURL(allocator: Allocator, base_url: []const u8) ![]u8 {
    const trimmed = std.mem.trim(u8, base_url, "/");
    return std.fmt.allocPrint(allocator, "{s}/responses", .{trimmed});
}

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_buf: [4096]u8 = undefined,
    body_reader: *std.Io.Reader = undefined,
    chunk: [4096]u8 = undefined,

    req_open: bool = false,
    done: bool = false,

    pub const ProduceStatus = provider_mod.ProviderStream.ProduceStatus;

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

    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) {
            error.EndOfStream => {
                try self.finishStream(out);
                return .response_complete;
            },
            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_responses <= {s}", .{ev_payload});
            // The Responses stream has no `[DONE]` sentinel; `response.completed`
            // is the terminal event.
            if (std.mem.eql(u8, ev_payload, "[DONE]")) {
                try self.finishStream(out);
                return .response_complete;
            }
            const terminal = try handleEvent(self.allocator, ev_payload, &self.state, out);
            if (terminal) {
                try self.finishStream(out);
                return .response_complete;
            }
        }
        return .more;
    }

    fn finishStream(self: *ResumableResponse, out: *EventQueue) !void {
        if (self.done) return;
        self.done = true;
        try self.state.finalize(out, self.conv);
    }
};

const ActiveBlock = enum { none, text, thinking };

/// Streaming-response assembly state.
///
/// Modeled on `provider_openai_chat.zig`'s `StreamState`: one active
/// text/thinking block at a time, plus a set of in-progress tool calls. The
/// Responses protocol is friendlier than Chat Completions here — every
/// function-call event carries an explicit `item_id` (and `output_index`),
/// and the lifecycle is spelled out by `output_item.added` →
/// `function_call_arguments.delta`* → `function_call_arguments.done`/
/// `output_item.done` → `response.completed`. So tool calls are keyed by
/// `item_id` (the stable identity) and need none of Chat Completions'
/// contiguity inference.
///
/// Argument-accumulation rule (the crux of the tool-input correctness): the
/// concatenation of `function_call_arguments.delta` fragments is the reliable
/// source of the tool input. A terminal `function_call_arguments.done` /
/// `output_item.done` / `response.completed` event also restates the full
/// `arguments`, but we apply it only as a *non-empty* override: a restated
/// value can improve the accumulation (e.g. if it is more complete) but never
/// wipe it. This matters because these events are observed to restate
/// `arguments` as `""` once the value has already streamed via deltas, and an
/// unconditional overwrite there destroys the real input — the original
/// empty-tool-input bug. Empty arguments normalize to `"{}"` so a tool never
/// receives an unparseable empty string.
const StreamState = struct {
    allocator: Allocator,
    started: bool = false,
    finalized: bool = false,
    active: ActiveBlock = .none,
    block_index: usize = 0,
    current_buf: conversation.TextualBlock = .empty,
    current_thinking_signature: ?[]const u8 = null,
    signature_origin: ?conversation.SignatureOrigin = null,
    assistant_phase: ?AssistantPhase = null,
    blocks: std.ArrayList(conversation.ContentBlock) = .empty,
    /// In-progress and completed tool calls, keyed by `output_index`, in
    /// first-seen order. Completed calls are retained (flagged `closed`) so
    /// the terminal `response.completed`, which restates every output item,
    /// does not re-emit a call that `output_item.done` already closed.
    ///
    /// We key by `output_index`, NOT the per-item `item_id`. The OpenAI spec
    /// promises a stable `item.id` per output item, but the GitHub Copilot
    /// Responses proxy emits a *fresh, opaque `item.id` on every event* for
    /// the same call (`output_item.added`, each `…arguments.delta`,
    /// `output_item.done`, and the `response.completed` restatement all
    /// differ). Keying by `item_id` there made dedup miss, so one call
    /// fanned out into three identical ToolUse blocks sharing one `call_id`.
    /// `output_index` is stable across all of a call's events on both
    /// backends — and is what `provider_openai_chat` keys on, too.
    tools: std.AutoArrayHashMapUnmanaged(usize, ToolUseInProgress) = .empty,
    usage: ?provider_mod.Usage = null,
    stream_error_message: ?[]u8 = null,

    const ToolUseInProgress = struct {
        block_index: usize,
        /// Set once `block_complete` has been emitted for this call.
        closed: bool = false,
        id_buf: conversation.TextualBlock = .empty, // call_id
        name_buf: conversation.TextualBlock = .empty,
        arguments: conversation.TextualBlock = .empty,

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

    fn deinit(self: *StreamState) void {
        self.current_buf.deinit(self.allocator);
        if (self.current_thinking_signature) |s| self.allocator.free(s);
        if (self.signature_origin) |*o| o.deinit(self.allocator);
        for (self.blocks.items) |*b| b.deinit(self.allocator);
        self.blocks.deinit(self.allocator);
        var it = self.tools.iterator();
        while (it.next()) |e| {
            e.value_ptr.deinit(self.allocator);
        }
        self.tools.deinit(self.allocator);
        if (self.stream_error_message) |s| self.allocator.free(s);
    }

    fn setStreamErrorMessage(self: *StreamState, message: []const u8) void {
        if (self.stream_error_message) |old| self.allocator.free(old);
        self.stream_error_message = self.allocator.dupe(u8, message) catch null;
    }

    fn ensureStarted(self: *StreamState, out: *EventQueue) !void {
        if (self.started) return;
        self.started = true;
        try out.push(.{ .message_start = .assistant });
    }

    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, .signature = self.current_thinking_signature } },
            .none => unreachable,
        };
        self.current_buf = .empty;
        self.current_thinking_signature = null;
        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;
    }

    fn setThinkingSignature(self: *StreamState, signature: []const u8) !void {
        if (self.current_thinking_signature) |old| self.allocator.free(old);
        self.current_thinking_signature = try self.allocator.dupe(u8, signature);
    }

    fn setAssistantPhase(self: *StreamState, phase: []const u8) void {
        if (std.mem.eql(u8, phase, "commentary")) {
            self.assistant_phase = .commentary;
        } else if (std.mem.eql(u8, phase, "final_answer")) {
            self.assistant_phase = .final_answer;
        }
    }

    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,
            .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);
        try out.push(.{ .content_delta = .{
            .index = self.block_index,
            .delta = try out.dupeBytes(delta),
        } });
    }

    /// Normalize a final/seed arguments string: an empty value becomes `"{}"`
    /// so a tool never sees an unparseable empty input.
    fn normalizedArguments(arguments: []const u8) []const u8 {
        return if (arguments.len == 0) "{}" else arguments;
    }

    /// Resolve an in-progress (not-yet-closed) tool call by `output_index`.
    /// Returns null for an unknown index or an already-closed call.
    fn lookupTool(self: *StreamState, output_index: ?usize) ?*ToolUseInProgress {
        const idx = output_index orelse return null;
        const tu = self.tools.getPtr(idx) orelse return null;
        if (tu.closed) return null;
        return tu;
    }

    /// Open a ToolUse block for a `function_call` output item. No-op if a
    /// call at this `output_index` already exists (open or closed): the call
    /// is announced once by `output_item.added`, then restated by
    /// `output_item.done` and again by `response.completed` — and on the
    /// Copilot proxy each restatement carries a different `item_id`, so
    /// `output_index` is the only reliable dedup key.
    fn openToolUse(
        self: *StreamState,
        out: *EventQueue,
        output_index: ?usize,
        call_id: ?[]const u8,
        name: ?[]const u8,
    ) !void {
        const idx = output_index orelse return;
        if (self.tools.contains(idx)) return;
        // Close any active text/thinking block so the tool gets its own index.
        if (self.active != .none) {
            try self.closeActive(out);
            self.block_index += 1;
        }
        var tu: ToolUseInProgress = .{ .block_index = self.block_index };
        errdefer tu.deinit(self.allocator);
        if (call_id) |c| try tu.id_buf.appendSlice(self.allocator, c);
        if (name) |nm| try tu.name_buf.appendSlice(self.allocator, nm);
        try self.tools.put(self.allocator, idx, tu);

        try out.push(.{ .block_start = .{ .block_type = .ToolUse, .index = self.block_index } });
        if (call_id != null and name != null) {
            try out.push(.{ .tool_details = .{
                .index = self.block_index,
                .id = try out.dupeBytes(call_id.?),
                .name = try out.dupeBytes(name.?),
            } });
        }
        self.block_index += 1;
    }

    /// Append a streaming argument fragment to the matching call. Fragments
    /// for an unknown or already-closed call are dropped (the protocol always
    /// opens a call before streaming its arguments).
    fn appendToolArgs(self: *StreamState, out: *EventQueue, output_index: ?usize, delta: []const u8) !void {
        const tu = self.lookupTool(output_index) orelse return;
        try tu.arguments.appendSlice(self.allocator, delta);
        try out.push(.{ .content_delta = .{
            .index = tu.block_index,
            .delta = try out.dupeBytes(delta),
        } });
    }

    /// Override the matching call's arguments with a restated full value, but
    /// only when non-empty — terminal events frequently restate already-
    /// streamed arguments as `""`, which must not wipe the accumulated input.
    fn setToolArgs(self: *StreamState, output_index: ?usize, arguments: []const u8) !void {
        if (arguments.len == 0) return;
        const tu = self.lookupTool(output_index) orelse return;
        tu.arguments.clearRetainingCapacity();
        try tu.arguments.appendSlice(self.allocator, arguments);
    }

    /// Close a ToolUse block. `final_args` (when a `done`/`completed` event
    /// restates the full arguments) overrides the accumulated value only when
    /// non-empty. A call whose identity never resolved (missing id or name)
    /// is dropped. Idempotent: closing an already-closed call is a no-op.
    fn closeToolUse(
        self: *StreamState,
        out: *EventQueue,
        output_index: ?usize,
        final_args: ?[]const u8,
    ) !void {
        const tu = self.lookupTool(output_index) orelse return;
        if (final_args) |fa| {
            if (fa.len > 0) {
                tu.arguments.clearRetainingCapacity();
                try tu.arguments.appendSlice(self.allocator, fa);
            }
        }
        tu.closed = true;

        if (tu.id_buf.items.len == 0 or tu.name_buf.items.len == 0) {
            // Identity never resolved — nothing dispatchable. Free the buffers
            // now; the (empty) entry stays in the map for dedup + final free.
            tu.id_buf.clearAndFree(self.allocator);
            tu.name_buf.clearAndFree(self.allocator);
            tu.arguments.clearAndFree(self.allocator);
            return;
        }

        const input = try conversation.textualBlockFromSlice(self.allocator, normalizedArguments(tu.arguments.items));
        decodeNameInPlace(&tu.name_buf);
        const block: conversation.ContentBlock = .{ .ToolUse = .{
            .id = try self.allocator.dupe(u8, tu.id_buf.items),
            .name = try self.allocator.dupe(u8, tu.name_buf.items),
            .input = input,
        } };
        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],
        } });
    }

    fn finalize(self: *StreamState, out: *EventQueue, conv: *conversation.Conversation) !void {
        if (self.finalized) return;
        self.finalized = true;
        try self.closeActive(out);
        // Close any tool calls that never received an explicit done event.
        var it = self.tools.iterator();
        while (it.next()) |e| {
            if (!e.value_ptr.closed) try self.closeToolUse(out, e.key_ptr.*, null);
        }

        const moved_blocks = try self.blocks.toOwnedSlice(self.allocator);
        defer self.allocator.free(moved_blocks);
        if (self.signature_origin) |origin| {
            try conversation.setThinkingOrigins(
                self.allocator,
                moved_blocks,
                origin.api_style,
                origin.base_url,
                origin.model,
            );
        }
        try conv.addAssistantMessage(moved_blocks, self.usage);
        if (self.assistant_phase) |phase| {
            const md = switch (phase) {
                .commentary => openai_phase_commentary_metadata,
                .final_answer => openai_phase_final_answer_metadata,
            };
            conv.messages.items[conv.messages.items.len - 1].metadata = try conv.allocator.dupe(u8, md);
        }
        const msg = conv.messages.items[conv.messages.items.len - 1];
        try out.push(.{ .message_complete = .{ .message = msg, .usage = self.usage } });
    }
};

const AssistantPhase = enum { commentary, final_answer };
const openai_phase_commentary_metadata = "{\"openai_responses_phase\":\"commentary\"}";
const openai_phase_final_answer_metadata = "{\"openai_responses_phase\":\"final_answer\"}";

/// Handle one parsed event. Returns true when the stream is terminal
/// (`response.completed`) so the caller can finalize.
fn handleEvent(
    allocator: Allocator,
    payload: []const u8,
    state: *StreamState,
    out: *EventQueue,
) !bool {
    var ev = try json_mod.parseStreamEvent(allocator, payload);
    defer ev.deinit();

    switch (ev.kind) {
        .output_text_delta => {
            if (ev.delta) |d| {
                try state.ensureStarted(out);
                try state.openBlock(.text, out);
                try state.appendDelta(out, d);
            }
        },
        .reasoning_summary_delta => {
            if (ev.delta) |d| {
                try state.ensureStarted(out);
                try state.openBlock(.thinking, out);
                try state.appendDelta(out, d);
            }
        },
        .output_item_added => {
            if (ev.item_type) |it| {
                if (std.mem.eql(u8, it, "function_call")) {
                    try state.ensureStarted(out);
                    try state.openToolUse(out, ev.output_index, ev.call_id, ev.name);
                    // `output_item.added` sometimes seeds the full args.
                    if (ev.arguments) |args| try state.setToolArgs(ev.output_index, args);
                }
            }
        },
        .function_call_arguments_delta => {
            if (ev.delta) |d| try state.appendToolArgs(out, ev.output_index, d);
        },
        .function_call_arguments_done => {
            if (ev.arguments) |args| try state.setToolArgs(ev.output_index, args);
        },
        .output_item_done => {
            if (ev.item_type) |it| {
                if (std.mem.eql(u8, it, "function_call")) {
                    try state.ensureStarted(out);
                    try state.openToolUse(out, ev.output_index, ev.call_id, ev.name);
                    try state.closeToolUse(out, ev.output_index, ev.arguments);
                } else if (std.mem.eql(u8, it, "reasoning")) {
                    if (ev.reasoning_item_json) |sig| {
                        try state.ensureStarted(out);
                        try state.openBlock(.thinking, out);
                        try state.setThinkingSignature(sig);
                        try state.closeActive(out);
                        state.block_index += 1;
                    }
                } else if (std.mem.eql(u8, it, "message")) {
                    if (ev.item_phase) |phase| state.setAssistantPhase(phase);
                }
            }
        },
        .completed => {
            // `response.completed` restates every output item; open+close any
            // function call not already emitted via `output_item.done`.
            for (ev.completed_items) |item| {
                try state.ensureStarted(out);
                try state.openToolUse(out, item.output_index, item.call_id, item.name);
                try state.closeToolUse(out, item.output_index, item.arguments);
            }
            if (ev.usage) |u| {
                const cached = u.cached_tokens;
                const total_in = u.input_tokens;
                const fresh = if (cached > total_in) 0 else total_in - cached;
                state.usage = .{
                    .input = fresh,
                    .output = u.output_tokens,
                    .cache_read = cached,
                    .cache_write = 0,
                    .reasoning = u.reasoning_tokens,
                };
            }
            return true;
        },
        .failed, .err => {
            if (ev.error_message) |m| {
                std.log.err("openai_responses stream error: {s}", .{m});
                state.setStreamErrorMessage(m);
            }
            return error.ProviderStreamMalformed;
        },
        .other => {},
    }
    return false;
}

// ===========================================================================
// Tests
// ===========================================================================

const testing = std.testing;

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 {
        try self.events.append(self.allocator, try std.fmt.allocPrint(self.allocator, fmt, args));
    }
    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[in={d},out={d},cr={d},rsn={d}]", .{ u.input, u.output, u.cache_read, u.reasoning });
                } else try self.push("msg_complete[null]", .{});
            },
            else => {},
        }
    }
};

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

    var terminal = false;
    for (events) |payload| {
        terminal = try handleEvent(allocator, payload, &state, &queue);
        if (terminal) break;
    }
    try state.finalize(&queue, conv);
    while (queue.pop()) |ev| {
        if (rec) |r| try r.record(ev);
    }
}

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 "responses stream: reasoning then text then completed" {
    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{
        \\{"type":"response.reasoning_summary_text.delta","item_id":"rs_1","delta":"thinking…"}
        ,
        \\{"type":"response.output_text.delta","item_id":"msg_1","delta":"Hello"}
        ,
        \\{"type":"response.output_text.delta","item_id":"msg_1","delta":" there"}
        ,
        \\{"type":"response.completed","response":{"usage":{"input_tokens":10,"output_tokens":5,"output_tokens_details":{"reasoning_tokens":2}}}}
        ,
    };
    try runStreamedTurn(allocator, &conv, &rec, &events);

    const asst = conv.messages.items[1];
    try testing.expectEqual(@as(usize, 2), asst.content.items.len);
    try testing.expectEqualStrings("thinking…", asst.content.items[0].Thinking.text.items);
    try testing.expectEqualStrings("Hello there", asst.content.items[1].Text.items);

    // Usage stamped.
    try testing.expect(asst.usage != null);
    try testing.expectEqual(@as(u64, 5), asst.usage.?.output);
    try testing.expectEqual(@as(u64, 2), asst.usage.?.reasoning);
}

test "responses stream: function call assembles a ToolUse" {
    const allocator = testing.allocator;
    var conv = conversation.Conversation.init(allocator);
    defer conv.deinit();
    try addUserText(&conv, "call it");

    var rec = EventRecorder{ .allocator = allocator };
    defer rec.deinit();

    const events = [_][]const u8{
        \\{"type":"response.output_item.added","output_index":0,"item":{"type":"function_call","id":"fc_1","call_id":"call_9","name":"std__read"}}
        ,
        \\{"type":"response.function_call_arguments.delta","output_index":0,"item_id":"fc_1","delta":"{\"path\":"}
        ,
        \\{"type":"response.function_call_arguments.delta","output_index":0,"item_id":"fc_1","delta":"\"a\"}"}
        ,
        \\{"type":"response.output_item.done","output_index":0,"item":{"type":"function_call","id":"fc_1","call_id":"call_9","name":"std__read","arguments":"{\"path\":\"a\"}"}}
        ,
        \\{"type":"response.completed","response":{"usage":{"input_tokens":3,"output_tokens":1}}}
        ,
    };
    try runStreamedTurn(allocator, &conv, &rec, &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_9", tu.id);
    // Wire name `std__read` decoded to internal dotted form.
    try testing.expectEqualStrings("std.read", tu.name);
    try testing.expectEqualStrings("{\"path\":\"a\"}", tu.input.items);

    // Callback order: start → details → deltas → complete.
    const expect = [_][]const u8{
        "msg_start",
        "block_start[0]:ToolUse",
        "tool_details[0]:call_9:std__read",
        "delta[0]:{\"path\":",
        "delta[0]:\"a\"}",
        "block_complete[0]",
        "msg_complete[in=3,out=1,cr=0,rsn=0]",
    };
    try testing.expectEqual(expect.len, rec.events.items.len);
    for (expect, rec.events.items) |w, g| try testing.expectEqualStrings(w, g);
}

test "responses stream: function call arguments can be keyed by output_index" {
    const allocator = testing.allocator;
    var conv = conversation.Conversation.init(allocator);
    defer conv.deinit();
    try addUserText(&conv, "call it");

    const events = [_][]const u8{
        \\{"type":"response.output_item.added","output_index":0,"item":{"type":"function_call","id":"fc_1","call_id":"call_9","name":"std__read"}}
        ,
        \\{"type":"response.function_call_arguments.delta","output_index":0,"delta":"{\"path\":"}
        ,
        \\{"type":"response.function_call_arguments.delta","output_index":0,"delta":"\"a\"}"}
        ,
        \\{"type":"response.output_item.done","output_index":0,"item":{"type":"function_call","id":"fc_1","call_id":"call_9","name":"std__read","arguments":""}}
        ,
        \\{"type":"response.completed","response":{"usage":{"input_tokens":3,"output_tokens":1}}}
        ,
    };
    try runStreamedTurn(allocator, &conv, null, &events);

    const tu = conv.messages.items[1].content.items[0].ToolUse;
    try testing.expectEqualStrings("call_9", tu.id);
    try testing.expectEqualStrings("std.read", tu.name);
    try testing.expectEqualStrings("{\"path\":\"a\"}", tu.input.items);
}

test "responses stream: function call arguments on item added are retained" {
    const allocator = testing.allocator;
    var conv = conversation.Conversation.init(allocator);
    defer conv.deinit();
    try addUserText(&conv, "call it");

    const events = [_][]const u8{
        \\{"type":"response.output_item.added","output_index":0,"item":{"type":"function_call","id":"fc_1","call_id":"call_9","name":"std__bash","arguments":"{\"command\":\"ls\"}"}}
        ,
        \\{"type":"response.completed","response":{"usage":{"input_tokens":3,"output_tokens":1}}}
        ,
    };
    try runStreamedTurn(allocator, &conv, null, &events);

    const tu = conv.messages.items[1].content.items[0].ToolUse;
    try testing.expectEqualStrings("call_9", tu.id);
    try testing.expectEqualStrings("std.bash", tu.name);
    try testing.expectEqualStrings("{\"command\":\"ls\"}", tu.input.items);
}

test "responses stream: function call arguments done supplies final input" {
    const allocator = testing.allocator;
    var conv = conversation.Conversation.init(allocator);
    defer conv.deinit();
    try addUserText(&conv, "call it");

    const events = [_][]const u8{
        \\{"type":"response.output_item.added","output_index":0,"item":{"type":"function_call","id":"fc_1","call_id":"call_9","name":"std__read"}}
        ,
        \\{"type":"response.function_call_arguments.done","output_index":0,"arguments":"{\"path\":\"a\"}"}
        ,
        \\{"type":"response.output_item.done","output_index":0,"item":{"type":"function_call","id":"fc_1","call_id":"call_9","name":"std__read","arguments":""}}
        ,
        \\{"type":"response.completed","response":{"usage":{"input_tokens":3,"output_tokens":1}}}
        ,
    };
    try runStreamedTurn(allocator, &conv, null, &events);

    const tu = conv.messages.items[1].content.items[0].ToolUse;
    try testing.expectEqualStrings("call_9", tu.id);
    try testing.expectEqualStrings("std.read", tu.name);
    try testing.expectEqualStrings("{\"path\":\"a\"}", tu.input.items);
}

test "responses stream: completed output supplies final function call input" {
    const allocator = testing.allocator;
    var conv = conversation.Conversation.init(allocator);
    defer conv.deinit();
    try addUserText(&conv, "call it");

    const events = [_][]const u8{
        \\{"type":"response.output_item.added","output_index":0,"item":{"type":"function_call","id":"fc_1","call_id":"call_9","name":"std__read","arguments":""}}
        ,
        \\{"type":"response.completed","response":{"output":[{"type":"function_call","id":"fc_1","call_id":"call_9","name":"std__read","arguments":"{\"path\":\"a\"}","status":"completed"}],"usage":{"input_tokens":3,"output_tokens":1}}}
        ,
    };
    try runStreamedTurn(allocator, &conv, null, &events);

    const tu = conv.messages.items[1].content.items[0].ToolUse;
    try testing.expectEqualStrings("call_9", tu.id);
    try testing.expectEqualStrings("std.read", tu.name);
    try testing.expectEqualStrings("{\"path\":\"a\"}", tu.input.items);
}

test "responses stream: completed output does not duplicate closed function call" {
    const allocator = testing.allocator;
    var conv = conversation.Conversation.init(allocator);
    defer conv.deinit();
    try addUserText(&conv, "call it");

    const events = [_][]const u8{
        \\{"type":"response.output_item.added","output_index":0,"item":{"type":"function_call","id":"fc_1","call_id":"call_9","name":"std__read","arguments":""}}
        ,
        \\{"type":"response.function_call_arguments.delta","output_index":0,"delta":"{\"path\":\"a\"}"}
        ,
        \\{"type":"response.output_item.done","output_index":0,"item":{"type":"function_call","id":"fc_1","call_id":"call_9","name":"std__read","arguments":""}}
        ,
        \\{"type":"response.completed","response":{"output":[{"type":"function_call","id":"fc_1","call_id":"call_9","name":"std__read","arguments":"{\"path\":\"a\"}","status":"completed"}],"usage":{"input_tokens":3,"output_tokens":1}}}
        ,
    };
    try runStreamedTurn(allocator, &conv, null, &events);

    const asst = conv.messages.items[1];
    try testing.expectEqual(@as(usize, 1), asst.content.items.len);
    try testing.expectEqualStrings("{\"path\":\"a\"}", asst.content.items[0].ToolUse.input.items);
}

test "responses stream: empty function call arguments normalize to object" {
    const allocator = testing.allocator;
    var conv = conversation.Conversation.init(allocator);
    defer conv.deinit();
    try addUserText(&conv, "call it");

    const events = [_][]const u8{
        \\{"type":"response.output_item.added","output_index":0,"item":{"type":"function_call","id":"fc_1","call_id":"call_9","name":"ping","arguments":""}}
        ,
        \\{"type":"response.output_item.done","output_index":0,"item":{"type":"function_call","id":"fc_1","call_id":"call_9","name":"ping","arguments":""}}
        ,
        \\{"type":"response.completed","response":{"usage":{"input_tokens":3,"output_tokens":1}}}
        ,
    };
    try runStreamedTurn(allocator, &conv, null, &events);

    const tu = conv.messages.items[1].content.items[0].ToolUse;
    try testing.expectEqualStrings("call_9", tu.id);
    try testing.expectEqualStrings("ping", tu.name);
    try testing.expectEqualStrings("{}", tu.input.items);
}

test "responses stream: finalization closes an open function call" {
    const allocator = testing.allocator;
    var conv = conversation.Conversation.init(allocator);
    defer conv.deinit();
    try addUserText(&conv, "call it");

    const events = [_][]const u8{
        \\{"type":"response.output_item.added","output_index":0,"item":{"type":"function_call","id":"fc_1","call_id":"call_9","name":"std__bash"}}
        ,
        \\{"type":"response.function_call_arguments.delta","output_index":0,"item_id":"fc_1","delta":"{\"command\":\"ls\"}"}
        ,
        \\{"type":"response.completed","response":{"usage":{"input_tokens":3,"output_tokens":1}}}
        ,
    };
    try runStreamedTurn(allocator, &conv, null, &events);

    const tu = conv.messages.items[1].content.items[0].ToolUse;
    try testing.expectEqualStrings("call_9", tu.id);
    try testing.expectEqualStrings("std.bash", tu.name);
    try testing.expectEqualStrings("{\"command\":\"ls\"}", tu.input.items);
}

test "responses stream: streamed args survive an empty arguments.done" {
    // Codex streams the arguments as `function_call_arguments.delta`
    // fragments and then emits a terminal `function_call_arguments.done`
    // whose `arguments` field is empty (the value already arrived via the
    // deltas). The accumulated input must NOT be wiped by that empty done.
    const allocator = testing.allocator;
    var conv = conversation.Conversation.init(allocator);
    defer conv.deinit();
    try addUserText(&conv, "call it");

    const events = [_][]const u8{
        \\{"type":"response.output_item.added","output_index":0,"item":{"type":"function_call","id":"fc_1","status":"in_progress","arguments":"","call_id":"call_9","name":"std__read"}}
        ,
        \\{"type":"response.function_call_arguments.delta","item_id":"fc_1","output_index":0,"delta":"{\"path\":"}
        ,
        \\{"type":"response.function_call_arguments.delta","item_id":"fc_1","output_index":0,"delta":"\"a\"}"}
        ,
        \\{"type":"response.function_call_arguments.done","item_id":"fc_1","output_index":0,"arguments":""}
        ,
        \\{"type":"response.output_item.done","output_index":0,"item":{"type":"function_call","id":"fc_1","status":"completed","arguments":"","call_id":"call_9","name":"std__read"}}
        ,
        \\{"type":"response.completed","response":{"usage":{"input_tokens":3,"output_tokens":1}}}
        ,
    };
    try runStreamedTurn(allocator, &conv, null, &events);

    const tu = conv.messages.items[1].content.items[0].ToolUse;
    try testing.expectEqualStrings("call_9", tu.id);
    try testing.expectEqualStrings("std.read", tu.name);
    try testing.expectEqualStrings("{\"path\":\"a\"}", tu.input.items);
}

test "responses stream: text then tool call keeps block order" {
    const allocator = testing.allocator;
    var conv = conversation.Conversation.init(allocator);
    defer conv.deinit();
    try addUserText(&conv, "go");

    const events = [_][]const u8{
        \\{"type":"response.output_text.delta","item_id":"msg_1","output_index":0,"delta":"working"}
        ,
        \\{"type":"response.output_item.added","output_index":1,"item":{"type":"function_call","id":"fc_1","call_id":"c1","name":"ping"}}
        ,
        \\{"type":"response.function_call_arguments.delta","output_index":1,"item_id":"fc_1","delta":"{}"}
        ,
        \\{"type":"response.output_item.done","output_index":1,"item":{"type":"function_call","id":"fc_1","call_id":"c1","name":"ping","arguments":"{}"}}
        ,
        \\{"type":"response.completed","response":{"usage":{"input_tokens":1,"output_tokens":1}}}
        ,
    };
    try runStreamedTurn(allocator, &conv, null, &events);

    const asst = conv.messages.items[1];
    try testing.expectEqual(@as(usize, 2), asst.content.items.len);
    try testing.expectEqualStrings("working", asst.content.items[0].Text.items);
    try testing.expectEqualStrings("ping", asst.content.items[1].ToolUse.name);
    try testing.expectEqualStrings("{}", asst.content.items[1].ToolUse.input.items);
}

/// Drive raw SSE bytes through the same path `produce` uses live: feed
/// arbitrary byte chunks to the `SSEParser`, decode each event, and drain the
/// `EventQueue` after every chunk (which resets its arena). This exercises
/// payload lifetimes the all-at-once `runStreamedTurn` helper does not.
fn runRawStream(
    allocator: Allocator,
    conv: *conversation.Conversation,
    raw: []const u8,
    chunk_len: usize,
) !void {
    var parser = sse_mod.SSEParser.init(allocator);
    defer parser.deinit();
    var state: StreamState = .init(allocator);
    defer state.deinit();
    var queue = EventQueue.init(allocator);
    defer queue.deinit();

    var off: usize = 0;
    var done = false;
    while (off < raw.len and !done) {
        const end = @min(off + chunk_len, raw.len);
        const events = try parser.feed(raw[off..end]);
        defer parser.freeEvents(events);
        off = end;
        for (events) |payload| {
            if (try handleEvent(allocator, payload, &state, &queue)) {
                done = true;
                break;
            }
        }
        // Drain (and reset the arena) between chunks, as the agent loop does.
        while (queue.pop()) |_| {}
    }
    try state.finalize(&queue, conv);
    while (queue.pop()) |_| {}
}

test "responses stream: realistic codex SSE assembles tool input across chunks" {
    // A representative ChatGPT-Codex function-call stream: explicit `event:`
    // lines, args streamed as `function_call_arguments.delta` fragments, and
    // terminal `done`/`completed` events that restate `arguments` as "".
    // Sliced into small byte chunks so events straddle reads and the queue
    // arena resets mid-stream.
    const allocator = testing.allocator;

    const raw =
        "event: response.created\n" ++
        "data: {\"type\":\"response.created\",\"response\":{\"id\":\"resp_1\"}}\n\n" ++
        "event: response.output_item.added\n" ++
        "data: {\"type\":\"response.output_item.added\",\"output_index\":0,\"item\":{\"id\":\"fc_1\",\"type\":\"function_call\",\"status\":\"in_progress\",\"arguments\":\"\",\"call_id\":\"call_9\",\"name\":\"std__read\"}}\n\n" ++
        "event: response.function_call_arguments.delta\n" ++
        "data: {\"type\":\"response.function_call_arguments.delta\",\"item_id\":\"fc_1\",\"output_index\":0,\"delta\":\"{\\\"path\\\":\"}\n\n" ++
        "event: response.function_call_arguments.delta\n" ++
        "data: {\"type\":\"response.function_call_arguments.delta\",\"item_id\":\"fc_1\",\"output_index\":0,\"delta\":\"\\\"/tmp/x\\\"}\"}\n\n" ++
        "event: response.function_call_arguments.done\n" ++
        "data: {\"type\":\"response.function_call_arguments.done\",\"item_id\":\"fc_1\",\"output_index\":0,\"arguments\":\"\"}\n\n" ++
        "event: response.output_item.done\n" ++
        "data: {\"type\":\"response.output_item.done\",\"output_index\":0,\"item\":{\"id\":\"fc_1\",\"type\":\"function_call\",\"status\":\"completed\",\"arguments\":\"\",\"call_id\":\"call_9\",\"name\":\"std__read\"}}\n\n" ++
        "event: response.completed\n" ++
        "data: {\"type\":\"response.completed\",\"response\":{\"output\":[{\"id\":\"fc_1\",\"type\":\"function_call\",\"status\":\"completed\",\"arguments\":\"{\\\"path\\\":\\\"/tmp/x\\\"}\",\"call_id\":\"call_9\",\"name\":\"std__read\"}],\"usage\":{\"input_tokens\":10,\"output_tokens\":5}}}\n\n";

    // Try several chunk sizes so events land on different read boundaries.
    for ([_]usize{ 1, 7, 64, raw.len }) |chunk_len| {
        var c = conversation.Conversation.init(allocator);
        defer c.deinit();
        try addUserText(&c, "read the file");
        try runRawStream(allocator, &c, raw, chunk_len);

        const asst = c.messages.items[1];
        try testing.expectEqual(@as(usize, 1), asst.content.items.len);
        const tu = asst.content.items[0].ToolUse;
        try testing.expectEqualStrings("call_9", tu.id);
        try testing.expectEqualStrings("std.read", tu.name);
        try testing.expectEqualStrings("{\"path\":\"/tmp/x\"}", tu.input.items);
    }
}

test "responses stream: parallel function calls assemble distinct ToolUse blocks" {
    // The model emits several function calls in one turn, each with its own
    // item_id / output_index / call_id, with their argument deltas
    // interleaved. Each must become its own ToolUse block with the right
    // input — no cross-talk between calls.
    const allocator = testing.allocator;
    var conv = conversation.Conversation.init(allocator);
    defer conv.deinit();
    try addUserText(&conv, "ls three dirs");

    const events = [_][]const u8{
        \\{"type":"response.output_item.added","output_index":0,"item":{"type":"function_call","id":"fc_a","call_id":"call_a","name":"std__shell","arguments":""}}
        ,
        \\{"type":"response.output_item.added","output_index":1,"item":{"type":"function_call","id":"fc_b","call_id":"call_b","name":"std__shell","arguments":""}}
        ,
        \\{"type":"response.function_call_arguments.delta","item_id":"fc_a","output_index":0,"delta":"{\"command\":\"ls a\"}"}
        ,
        \\{"type":"response.function_call_arguments.delta","item_id":"fc_b","output_index":1,"delta":"{\"command\":"}
        ,
        \\{"type":"response.output_item.added","output_index":2,"item":{"type":"function_call","id":"fc_c","call_id":"call_c","name":"std__shell","arguments":""}}
        ,
        \\{"type":"response.function_call_arguments.delta","item_id":"fc_b","output_index":1,"delta":"\"ls b\"}"}
        ,
        \\{"type":"response.function_call_arguments.delta","item_id":"fc_c","output_index":2,"delta":"{\"command\":\"ls c\"}"}
        ,
        \\{"type":"response.output_item.done","output_index":0,"item":{"type":"function_call","id":"fc_a","call_id":"call_a","name":"std__shell","arguments":""}}
        ,
        \\{"type":"response.output_item.done","output_index":1,"item":{"type":"function_call","id":"fc_b","call_id":"call_b","name":"std__shell","arguments":""}}
        ,
        \\{"type":"response.output_item.done","output_index":2,"item":{"type":"function_call","id":"fc_c","call_id":"call_c","name":"std__shell","arguments":""}}
        ,
        \\{"type":"response.completed","response":{"output":[{"type":"function_call","id":"fc_a","call_id":"call_a","name":"std__shell","arguments":"{\"command\":\"ls a\"}"},{"type":"function_call","id":"fc_b","call_id":"call_b","name":"std__shell","arguments":"{\"command\":\"ls b\"}"},{"type":"function_call","id":"fc_c","call_id":"call_c","name":"std__shell","arguments":"{\"command\":\"ls c\"}"}],"usage":{"input_tokens":5,"output_tokens":9}}}
        ,
    };
    try runStreamedTurn(allocator, &conv, null, &events);

    const asst = conv.messages.items[1];
    try testing.expectEqual(@as(usize, 3), asst.content.items.len);
    try testing.expectEqualStrings("call_a", asst.content.items[0].ToolUse.id);
    try testing.expectEqualStrings("{\"command\":\"ls a\"}", asst.content.items[0].ToolUse.input.items);
    try testing.expectEqualStrings("call_b", asst.content.items[1].ToolUse.id);
    try testing.expectEqualStrings("{\"command\":\"ls b\"}", asst.content.items[1].ToolUse.input.items);
    try testing.expectEqualStrings("call_c", asst.content.items[2].ToolUse.id);
    try testing.expectEqualStrings("{\"command\":\"ls c\"}", asst.content.items[2].ToolUse.input.items);
    for (asst.content.items) |b| try testing.expectEqualStrings("std.shell", b.ToolUse.name);
}

test "responses stream: one call with a mutating item_id stays a single block" {
    // Regression for the GitHub Copilot Responses proxy: it emits a *fresh*,
    // opaque `item_id` on every event for the SAME call — the
    // `output_item.added`, each `…arguments.delta`, the `…arguments.done`, the
    // `output_item.done`, and the `response.completed` restatement all carry
    // different `item_id`s. Only `output_index` (and `call_id`) are stable.
    // Keying tool calls by `item_id` made dedup miss, fanning this single call
    // out into three identical ToolUse blocks sharing one `call_id` (which in
    // turn stranded two UI result boxes at the `(…)` placeholder). Keying by
    // `output_index` must collapse it back to exactly one block. A leading
    // reasoning item (also with mutating ids) must not spawn a phantom tool.
    const allocator = testing.allocator;
    var conv = conversation.Conversation.init(allocator);
    defer conv.deinit();
    try addUserText(&conv, "what's in the current directory?");

    const events = [_][]const u8{
        \\{"type":"response.output_item.added","output_index":0,"item":{"type":"reasoning","id":"rsn_AAAA"}}
        ,
        \\{"type":"response.output_item.done","output_index":0,"item":{"type":"reasoning","id":"rsn_BBBB"}}
        ,
        \\{"type":"response.output_item.added","output_index":1,"item":{"type":"function_call","id":"itm_AAAA","call_id":"call_KOEB","name":"std__shell","arguments":""}}
        ,
        \\{"type":"response.function_call_arguments.delta","output_index":1,"item_id":"itm_BBBB","delta":"{\"command\":"}
        ,
        \\{"type":"response.function_call_arguments.delta","output_index":1,"item_id":"itm_CCCC","delta":"\"pwd && ls -la\"}"}
        ,
        \\{"type":"response.function_call_arguments.done","output_index":1,"item_id":"itm_DDDD","arguments":""}
        ,
        \\{"type":"response.output_item.done","output_index":1,"item":{"type":"function_call","id":"itm_EEEE","call_id":"call_KOEB","name":"std__shell","arguments":"{\"command\":\"pwd && ls -la\"}"}}
        ,
        \\{"type":"response.completed","response":{"output":[{"type":"reasoning","id":"rsn_CCCC"},{"type":"function_call","id":"itm_FFFF","call_id":"call_KOEB","name":"std__shell","arguments":"{\"command\":\"pwd && ls -la\"}"}],"usage":{"input_tokens":5,"output_tokens":9}}}
        ,
    };
    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_KOEB", tu.id);
    try testing.expectEqualStrings("std.shell", tu.name);
    try testing.expectEqualStrings("{\"command\":\"pwd && ls -la\"}", tu.input.items);
}