From f83578fdc9264019a1a1cef8c5484a161167d3dd Mon Sep 17 00:00:00 2001 From: t Date: Tue, 7 Jul 2026 11:26:32 -0600 Subject: initial commit, moved libpanto over from the pantograph repo --- src/provider_openai_responses.zig | 1139 +++++++++++++++++++++++++++++++++++++ 1 file changed, 1139 insertions(+) create mode 100644 src/provider_openai_responses.zig (limited to 'src/provider_openai_responses.zig') 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); +} -- cgit v1.3