initial commit, moved libpanto over from the pantograph repo

1 files changed, 1139 insertions, 0 deletions

diff --git a/src/provider_openai_responses.zig b/src/provider_openai_responses.zig
new file mode 100644
index 0000000..d764e6f
--- /dev/null
+++ b/src/provider_openai_responses.zig
@@ -0,0 +1,1139 @@
+//! 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);
+}