From 621d7fee0ace729f8d44126032d2c6e13f72ee7f Mon Sep 17 00:00:00 2001 From: t Date: Tue, 7 Jul 2026 11:29:23 -0600 Subject: Move libpanto projects to libpantograph dependency Remove the in-repo libpanto sources and binding projects from pantograph. Consume libpantograph through the Zig package URL at code.tjp.lol/libpantograph.git, including the Lua module artifact used by CLI extensions. --- libpanto/src/provider_openai_responses.zig | 1139 ---------------------------- 1 file changed, 1139 deletions(-) delete mode 100644 libpanto/src/provider_openai_responses.zig (limited to 'libpanto/src/provider_openai_responses.zig') diff --git a/libpanto/src/provider_openai_responses.zig b/libpanto/src/provider_openai_responses.zig deleted file mode 100644 index d764e6f..0000000 --- a/libpanto/src/provider_openai_responses.zig +++ /dev/null @@ -1,1139 +0,0 @@ -//! 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