summaryrefslogtreecommitdiff
path: root/libpanto/src/provider_openai_responses.zig
diff options
context:
space:
mode:
Diffstat (limited to 'libpanto/src/provider_openai_responses.zig')
-rw-r--r--libpanto/src/provider_openai_responses.zig626
1 files changed, 578 insertions, 48 deletions
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);
+}
generated by cgit v1.3 (git 2.53.0) at 2026-07-09 10:13:59 +0000