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.zig1139
1 files changed, 0 insertions, 1139 deletions
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);
-}
generated by cgit v1.3 (git 2.53.0) at 2026-07-09 06:30:32 +0000