path: root/libpanto/src/provider_openai_responses.zig

//! 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);
}