diff options
Diffstat (limited to 'libpanto/src/provider_openai_responses.zig')
| -rw-r--r-- | libpanto/src/provider_openai_responses.zig | 656 |
1 files changed, 656 insertions, 0 deletions
diff --git a/libpanto/src/provider_openai_responses.zig b/libpanto/src/provider_openai_responses.zig new file mode 100644 index 0000000..92a309b --- /dev/null +++ b/libpanto/src/provider_openai_responses.zig @@ -0,0 +1,656 @@ +//! 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 tool_registry_mod = @import("tool_registry.zig"); + +const Event = stream_mod.Event; +const EventQueue = stream_mod.EventQueue; + +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; +} + +pub const OpenAIResponsesRequest = struct { + allocator: Allocator, + io: Io, + config: *const config_mod.OpenAIResponsesConfig, + 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), + }; + errdefer { + rr.parser.deinit(); + rr.state.deinit(); + } + + const url = try std.fmt.allocPrint(self.allocator, "{s}/responses", .{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); + defer self.allocator.free(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.req = try self.http_client.request(.POST, uri, .{ + .extra_headers = extra_headers, + .headers = .{ .accept_encoding = .{ .override = "identity" } }, + .keep_alive = false, + .redirect_behavior = .not_allowed, + }); + rr.req_open = true; + errdefer { + rr.req.deinit(); + rr.req_open = false; + } + + rr.req.transfer_encoding = .{ .content_length = body.len }; + var send_buf: [4096]u8 = undefined; + var bw = try rr.req.sendBodyUnflushed(&send_buf); + try bw.writer.writeAll(body); + try bw.end(); + try rr.req.connection.?.flush(); + + var redirect_buf: [1024]u8 = undefined; + rr.response = try rr.req.receiveHead(&redirect_buf); + + if (@intFromEnum(rr.response.head.status) >= 400) { + const retry_after_ms = provider_mod.retryAfterFromHead(rr.response.head); + const body_reader = rr.response.reader(&rr.transfer_buf); + var err_buf: std.ArrayList(u8) = .empty; + defer err_buf.deinit(self.allocator); + var tmp: [1024]u8 = undefined; + while (true) { + const n = body_reader.readSliceShort(&tmp) catch break; + if (n == 0) break; + try err_buf.appendSlice(self.allocator, tmp[0..n]); + 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); + if (self.diag) |d| { + d.status_code = status; + d.retry_after_ms = retry_after_ms; + } + return classified; + } + + rr.body_reader = rr.response.reader(&rr.transfer_buf); + return rr; + } +}; + +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 }; + +const StreamState = struct { + allocator: Allocator, + started: bool = false, + finalized: bool = false, + active: ActiveBlock = .none, + block_index: usize = 0, + current_buf: conversation.TextualBlock = .empty, + blocks: std.ArrayList(conversation.ContentBlock) = .empty, + /// In-progress tool calls keyed by their streaming `item_id`. + tools: std.StringArrayHashMapUnmanaged(ToolUseInProgress) = .empty, + usage: ?provider_mod.Usage = null, + stream_error_message: ?[]u8 = null, + + const ToolUseInProgress = struct { + block_index: usize, + 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); + 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); + 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 } }, + .none => unreachable, + }; + self.current_buf = .empty; + 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 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), + } }); + } + + /// Open a new ToolUse block for a `function_call` output item. + fn openToolUse( + self: *StreamState, + out: *EventQueue, + item_id: []const u8, + call_id: ?[]const u8, + name: ?[]const u8, + ) !void { + // 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 }; + 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 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; + } + + fn appendToolArgs(self: *StreamState, out: *EventQueue, item_id: []const u8, delta: []const u8) !void { + const tu = self.tools.getPtr(item_id) orelse return; + try tu.arguments.appendSlice(self.allocator, delta); + try out.push(.{ .content_delta = .{ + .index = tu.block_index, + .delta = try out.dupeBytes(delta), + } }); + } + + /// Close a ToolUse block on `output_item.done`. `final_args` (when the + /// done event carries the full arguments) overrides the accumulated ones. + fn closeToolUse( + self: *StreamState, + out: *EventQueue, + item_id: []const u8, + final_args: ?[]const u8, + ) !void { + const entry = self.tools.fetchSwapRemove(item_id) orelse return; + self.allocator.free(entry.key); + var tu = entry.value; + + if (tu.id_buf.items.len == 0 or tu.name_buf.items.len == 0) { + tu.deinit(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 block: conversation.ContentBlock = .{ .ToolUse = .{ + .id = id_owned, + .name = name_owned, + .input = tu.arguments, + } }; + tu.arguments = .empty; + 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(); + 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); + + const moved_blocks = try self.blocks.toOwnedSlice(self.allocator); + defer self.allocator.free(moved_blocks); + try conv.addAssistantMessage(moved_blocks, self.usage); + const msg = conv.messages.items[conv.messages.items.len - 1]; + try out.push(.{ .message_complete = .{ .message = msg, .usage = self.usage } }); + } +}; + +/// 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); + if (ev.item_id) |id| try state.openToolUse(out, id, ev.call_id, ev.name); + } + } + }, + .function_call_arguments_delta => { + if (ev.item_id) |id| { + if (ev.delta) |d| try state.appendToolArgs(out, id, d); + } + }, + .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); + } + } + }, + .completed => { + 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","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","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.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: 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","delta":"working"} + , + \\{"type":"response.output_item.added","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.output_item.done","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); +} |