From 02b4c7a35ac0b714bc045d54fb4bb45d1ce4e490 Mon Sep 17 00:00:00 2001 From: t Date: Sat, 13 Jun 2026 23:45:59 -0600 Subject: Add Codex Responses support and session debugging Teach provider config and auth resolution about the Codex Responses dialect, including user-facing `style = "openai_responses"` with `dialect = "codex"`. Serialize and parse Responses traffic with provider-specific reasoning replay, assistant phase metadata, and robust function-call assembly keyed by `output_index` so streamed tool inputs survive proxy quirks and empty terminal payloads. Also persist thinking origins and message metadata across sessions, add the Anthropic interleaved-thinking header switch, write per-session debug logs, and improve the TUI and scripts for inspecting tool output and session costs. --- libpanto/src/provider_openai_responses.zig | 626 ++++++++++++++++++++++++++--- 1 file changed, 578 insertions(+), 48 deletions(-) (limited to 'libpanto/src/provider_openai_responses.zig') diff --git a/libpanto/src/provider_openai_responses.zig b/libpanto/src/provider_openai_responses.zig index 92a309b..c172f6d 100644 --- a/libpanto/src/provider_openai_responses.zig +++ b/libpanto/src/provider_openai_responses.zig @@ -36,6 +36,11 @@ const tool_registry_mod = @import("tool_registry.zig"); const Event = stream_mod.Event; const EventQueue = stream_mod.EventQueue; +pub const OpenAIResponsesDialect = enum { + public, + codex, +}; + fn decodeNameInPlace(name_buf: *conversation.TextualBlock) void { const decoded = tool_registry_mod.decodeName(name_buf.items, name_buf.items); name_buf.items.len = decoded.len; @@ -45,6 +50,7 @@ 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, @@ -61,17 +67,27 @@ pub const OpenAIResponsesRequest = struct { .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 std.fmt.allocPrint(self.allocator, "{s}/responses", .{self.config.base_url}); + 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); + 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); @@ -123,8 +139,18 @@ pub const OpenAIResponsesRequest = struct { if (err_buf.items.len > 16 * 1024) break; } const status: u16 = @intFromEnum(rr.response.head.status); - std.log.err("openai_responses HTTP {d}: {s}", .{ status, err_buf.items }); const classified = provider_mod.classifyHttpStatus(status, err_buf.items); + // 401/403 is routinely recovered by the turn-runner's forced token + // refresh + reopen (see `driveTurn` in tui_app.zig). Logging it at + // `.err` surfaces a scary "token expired" line for what is a + // transparent, recoverable refresh. Demote it to `.debug` (still + // captured in the debug log); the retry layer raises a hard error + // only if recovery ultimately fails. + if (classified == error.ProviderAuthFailed) { + std.log.debug("openai_responses HTTP {d} (recoverable auth): {s}", .{ status, err_buf.items }); + } else { + std.log.err("openai_responses HTTP {d}: {s}", .{ status, err_buf.items }); + } if (self.diag) |d| { d.status_code = status; d.retry_after_ms = retry_after_ms; @@ -137,6 +163,16 @@ pub const OpenAIResponsesRequest = struct { } }; +// 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, @@ -226,6 +262,29 @@ pub const ResumableResponse = struct { 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, @@ -233,14 +292,32 @@ const StreamState = struct { 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 tool calls keyed by their streaming `item_id`. - tools: std.StringArrayHashMapUnmanaged(ToolUseInProgress) = .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, @@ -258,11 +335,12 @@ const StreamState = struct { 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| { - self.allocator.free(e.key_ptr.*); e.value_ptr.deinit(self.allocator); } self.tools.deinit(self.allocator); @@ -284,10 +362,11 @@ const StreamState = struct { if (self.active == .none) return; const block: conversation.ContentBlock = switch (self.active) { .text => .{ .Text = self.current_buf }, - .thinking => .{ .Thinking = .{ .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, @@ -296,6 +375,19 @@ const StreamState = struct { 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; @@ -320,24 +412,46 @@ const StreamState = struct { } }); } - /// Open a new ToolUse block for a `function_call` output item. + /// 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, - item_id: []const u8, + 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); - const key = try self.allocator.dupe(u8, item_id); - try self.tools.put(self.allocator, key, tu); + 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) { @@ -350,8 +464,11 @@ const StreamState = struct { self.block_index += 1; } - fn appendToolArgs(self: *StreamState, out: *EventQueue, item_id: []const u8, delta: []const u8) !void { - const tu = self.tools.getPtr(item_id) orelse return; + /// 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, @@ -359,36 +476,51 @@ const StreamState = struct { } }); } - /// Close a ToolUse block on `output_item.done`. `final_args` (when the - /// done event carries the full arguments) overrides the accumulated ones. + /// 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, - item_id: []const u8, + output_index: ?usize, final_args: ?[]const u8, ) !void { - const entry = self.tools.fetchSwapRemove(item_id) orelse return; - self.allocator.free(entry.key); - var tu = entry.value; + 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) { - tu.deinit(self.allocator); + // 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; } - if (final_args) |fa| { - tu.arguments.clearRetainingCapacity(); - try tu.arguments.appendSlice(self.allocator, fa); - } - decodeNameInPlace(&tu.name_buf); - const id_owned = try tu.id_buf.toOwnedSlice(self.allocator); - const name_owned = try tu.name_buf.toOwnedSlice(self.allocator); + const input = try conversation.textualBlockFromSlice(self.allocator, normalizedArguments(tu.arguments.items)); + decodeNameInPlace(&tu.name_buf); const block: conversation.ContentBlock = .{ .ToolUse = .{ - .id = id_owned, - .name = name_owned, - .input = tu.arguments, + .id = try self.allocator.dupe(u8, tu.id_buf.items), + .name = try self.allocator.dupe(u8, tu.name_buf.items), + .input = input, } }; - tu.arguments = .empty; try self.blocks.append(self.allocator, block); try out.push(.{ .block_complete = .{ .index = tu.block_index, @@ -402,19 +534,38 @@ const StreamState = struct { try self.closeActive(out); // Close any tool calls that never received an explicit done event. var it = self.tools.iterator(); - var leftover: std.ArrayList([]const u8) = .empty; - defer leftover.deinit(self.allocator); - while (it.next()) |e| leftover.append(self.allocator, e.key_ptr.*) catch {}; - for (leftover.items) |k| try self.closeToolUse(out, k, null); + 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( @@ -445,23 +596,45 @@ fn handleEvent( if (ev.item_type) |it| { if (std.mem.eql(u8, it, "function_call")) { try state.ensureStarted(out); - if (ev.item_id) |id| try state.openToolUse(out, id, ev.call_id, ev.name); + 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.item_id) |id| { - if (ev.delta) |d| try state.appendToolArgs(out, id, d); - } + 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")) { - if (ev.item_id) |id| try state.closeToolUse(out, id, ev.arguments); + 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; @@ -593,13 +766,13 @@ test "responses stream: function call assembles a ToolUse" { defer rec.deinit(); const events = [_][]const u8{ - \\{"type":"response.output_item.added","item":{"type":"function_call","id":"fc_1","call_id":"call_9","name":"std__read"}} + \\{"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","item_id":"fc_1","delta":"{\"path\":"} + \\{"type":"response.function_call_arguments.delta","output_index":0,"item_id":"fc_1","delta":"{\"path\":"} , - \\{"type":"response.function_call_arguments.delta","item_id":"fc_1","delta":"\"a\"}"} + \\{"type":"response.function_call_arguments.delta","output_index":0,"item_id":"fc_1","delta":"\"a\"}"} , - \\{"type":"response.output_item.done","item":{"type":"function_call","id":"fc_1","call_id":"call_9","name":"std__read","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":"{\"path\":\"a\"}"}} , \\{"type":"response.completed","response":{"usage":{"input_tokens":3,"output_tokens":1}}} , @@ -628,6 +801,195 @@ test "responses stream: function call assembles a ToolUse" { 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); @@ -635,13 +997,13 @@ test "responses stream: text then tool call keeps block order" { try addUserText(&conv, "go"); const events = [_][]const u8{ - \\{"type":"response.output_text.delta","item_id":"msg_1","delta":"working"} + \\{"type":"response.output_text.delta","item_id":"msg_1","output_index":0,"delta":"working"} , - \\{"type":"response.output_item.added","item":{"type":"function_call","id":"fc_1","call_id":"c1","name":"ping"}} + \\{"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","item_id":"fc_1","delta":"{}"} + \\{"type":"response.function_call_arguments.delta","output_index":1,"item_id":"fc_1","delta":"{}"} , - \\{"type":"response.output_item.done","item":{"type":"function_call","id":"fc_1","call_id":"c1","name":"ping","arguments":"{}"}} + \\{"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}}} , @@ -654,3 +1016,171 @@ test "responses stream: text then tool call keeps block order" { 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