From df2edee86eec2a8deb0ad57b5d20552199c12b65 Mon Sep 17 00:00:00 2001 From: T Date: Mon, 25 May 2026 13:24:02 -0700 Subject: phase 1 done --- libpanto/src/config.zig | 46 +++- libpanto/src/json.zig | 4 - libpanto/src/openai_chat_json.zig | 324 +++++++++++++++++++++++++ libpanto/src/provider.zig | 47 ++-- libpanto/src/provider_openai.zig | 6 - libpanto/src/provider_openai_chat.zig | 443 ++++++++++++++++++++++++++++++++++ libpanto/src/root.zig | 11 +- libpanto/src/sse.zig | 137 +++++------ 8 files changed, 898 insertions(+), 120 deletions(-) delete mode 100644 libpanto/src/json.zig create mode 100644 libpanto/src/openai_chat_json.zig delete mode 100644 libpanto/src/provider_openai.zig create mode 100644 libpanto/src/provider_openai_chat.zig (limited to 'libpanto') diff --git a/libpanto/src/config.zig b/libpanto/src/config.zig index 295f2c6..74bdcc7 100644 --- a/libpanto/src/config.zig +++ b/libpanto/src/config.zig @@ -1,5 +1,20 @@ pub const APIStyle = enum { - anthropic, + openai_chat, +}; + +/// Reasoning intensity hint sent to providers that support it. +/// +/// `.default` omits the field entirely so the provider's own default applies. +/// `.off` sends `"none"` (supported by OpenRouter/NanoGPT; ignored or rejected +/// elsewhere). The remaining values mirror the `reasoning_effort` parameter +/// accepted by OpenAI reasoning models and OpenAI-compatible proxies. +pub const ReasoningEffort = enum { + default, + off, + minimal, + low, + medium, + high, }; pub const Config = struct { @@ -7,19 +22,32 @@ pub const Config = struct { api_key: []const u8, base_url: []const u8, model: []const u8, + reasoning: ReasoningEffort = .default, }; const t = @import("std").testing; test "Config - construct with all fields" { const cfg = Config{ - .api_style = .anthropic, - .api_key = "sk-ant-test", - .base_url = "https://api.anthropic.com", - .model = "claude-sonnet-4-20250514", + .api_style = .openai_chat, + .api_key = "sk-test", + .base_url = "https://api.openai.com/v1", + .model = "gpt-4o", + .reasoning = .high, + }; + try t.expectEqual(APIStyle.openai_chat, cfg.api_style); + try t.expectEqualStrings("sk-test", cfg.api_key); + try t.expectEqualStrings("https://api.openai.com/v1", cfg.base_url); + try t.expectEqualStrings("gpt-4o", cfg.model); + try t.expectEqual(ReasoningEffort.high, cfg.reasoning); +} + +test "Config - reasoning defaults to .default" { + const cfg = Config{ + .api_style = .openai_chat, + .api_key = "k", + .base_url = "u", + .model = "m", }; - try t.expectEqual(APIStyle.anthropic, cfg.api_style); - try t.expectEqualStrings("sk-ant-test", cfg.api_key); - try t.expectEqualStrings("https://api.anthropic.com", cfg.base_url); - try t.expectEqualStrings("claude-sonnet-4-20250514", cfg.model); + try t.expectEqual(ReasoningEffort.default, cfg.reasoning); } diff --git a/libpanto/src/json.zig b/libpanto/src/json.zig deleted file mode 100644 index 542391e..0000000 --- a/libpanto/src/json.zig +++ /dev/null @@ -1,4 +0,0 @@ -// Serialization helpers — model → wire JSON, deltas → ContentBlocks. -// Implementation pending; this module will handle: -// 1. Serialize Conversation → OpenAI request body JSON -// 2. Parse SSE chunk deltas → ContentBlock updates diff --git a/libpanto/src/openai_chat_json.zig b/libpanto/src/openai_chat_json.zig new file mode 100644 index 0000000..79cccac --- /dev/null +++ b/libpanto/src/openai_chat_json.zig @@ -0,0 +1,324 @@ +//! OpenAI Chat Completions JSON serialization and parsing. +//! +//! Two responsibilities: +//! 1. Serialize a `Conversation` into the OpenAI Chat Completions request body. +//! 2. Parse one streaming SSE event's JSON payload into a strongly-typed +//! `StreamDelta` that the provider can consume. + +const std = @import("std"); +const Allocator = std.mem.Allocator; +const Writer = std.Io.Writer; +const conversation = @import("conversation.zig"); +const config_mod = @import("config.zig"); + +/// A single parsed streaming chunk. Fields are populated only when present +/// in the wire payload; null fields signal "not in this chunk". +/// +/// `content` and `reasoning_content` slices are borrowed from the parsed +/// JSON value, which is owned by the caller's `std.json.Parsed`. +pub const StreamDelta = struct { + role: ?[]const u8 = null, + content: ?[]const u8 = null, + reasoning_content: ?[]const u8 = null, + finish_reason: ?[]const u8 = null, +}; + +/// Serialize a Conversation into a `chat/completions` request body. +/// +/// The caller owns the returned slice (allocated with `allocator`). +pub fn serializeRequest( + allocator: Allocator, + cfg: *const config_mod.Config, + conv: *const conversation.Conversation, +) ![]u8 { + var aw: Writer.Allocating = .init(allocator); + errdefer aw.deinit(); + + var s: std.json.Stringify = .{ .writer = &aw.writer }; + + try s.beginObject(); + + try s.objectField("model"); + try s.write(cfg.model); + + try s.objectField("stream"); + try s.write(true); + + switch (cfg.reasoning) { + .default => {}, + .off => { + try s.objectField("reasoning_effort"); + try s.write("none"); + }, + .minimal, .low, .medium, .high => |eff| { + try s.objectField("reasoning_effort"); + try s.write(@tagName(eff)); + }, + } + + try s.objectField("messages"); + try s.beginArray(); + for (conv.messages.items) |msg| { + try writeMessage(&s, msg, allocator); + } + try s.endArray(); + + try s.endObject(); + + return try aw.toOwnedSlice(); +} + +fn writeMessage(s: *std.json.Stringify, msg: conversation.Message, allocator: Allocator) !void { + try s.beginObject(); + + try s.objectField("role"); + try s.write(@tagName(msg.role)); + + // All roles flatten to a plain `content` string in outbound requests. + // Thinking blocks are intentionally dropped from history: the openai_chat + // dialect has no portable way to round-trip them (OpenAI/DeepSeek strip; + // OpenRouter/NanoGPT want a `reasoning` field instead of an inline block; + // none accept the `{"type":"thinking",...}` shape). Preserving reasoning + // across turns will land with tool-use in a later phase. + try s.objectField("content"); + var buf: std.ArrayList(u8) = .empty; + defer buf.deinit(allocator); + try concatTextBlocks(msg.content.items, &buf, allocator); + try s.write(buf.items); + + try s.endObject(); +} + +fn concatTextBlocks( + blocks: []const conversation.ContentBlock, + out: *std.ArrayList(u8), + allocator: Allocator, +) !void { + for (blocks) |block| { + switch (block) { + .Text => |tb| try out.appendSlice(allocator, tb.items), + // Thinking: dropped. ToolUse/ToolResult: phase 3+. + else => {}, + } + } +} + +/// Parse a single SSE event payload (the JSON object that follows "data: "). +/// +/// Returns a `StreamDelta` borrowed from `parsed`. The caller must keep +/// `parsed` alive for as long as the delta's slices are in use, then call +/// `parsed.deinit()`. +pub const ParsedDelta = struct { + parsed: std.json.Parsed(std.json.Value), + delta: StreamDelta, + + pub fn deinit(self: *ParsedDelta) void { + self.parsed.deinit(); + } +}; + +pub fn parseStreamEvent(allocator: Allocator, payload: []const u8) !ParsedDelta { + var parsed = try std.json.parseFromSlice(std.json.Value, allocator, payload, .{}); + errdefer parsed.deinit(); + + var delta: StreamDelta = .{}; + + const root = parsed.value; + if (root != .object) return .{ .parsed = parsed, .delta = delta }; + + const choices_v = root.object.get("choices") orelse return .{ .parsed = parsed, .delta = delta }; + if (choices_v != .array or choices_v.array.items.len == 0) { + return .{ .parsed = parsed, .delta = delta }; + } + const choice = choices_v.array.items[0]; + if (choice != .object) return .{ .parsed = parsed, .delta = delta }; + + if (choice.object.get("finish_reason")) |fr| { + if (fr == .string) delta.finish_reason = fr.string; + } + + if (choice.object.get("delta")) |d| { + if (d == .object) { + if (d.object.get("role")) |r| { + if (r == .string) delta.role = r.string; + } + if (d.object.get("content")) |c| { + if (c == .string) delta.content = c.string; + } + // Reasoning content lives under one of these names depending on + // the provider. We accept either. + if (d.object.get("reasoning_content")) |rc| { + if (rc == .string) delta.reasoning_content = rc.string; + } else if (d.object.get("reasoning")) |rc| { + if (rc == .string) delta.reasoning_content = rc.string; + } + } + } + + return .{ .parsed = parsed, .delta = delta }; +} + +// ----------------------------------------------------------------------------- +// Tests +// ----------------------------------------------------------------------------- + +const testing = std.testing; + +fn testConfig(model: []const u8) config_mod.Config { + return .{ + .api_style = .openai_chat, + .api_key = "k", + .base_url = "u", + .model = model, + }; +} + +test "serializeRequest - system + user" { + const allocator = testing.allocator; + + var conv = conversation.Conversation.init(allocator); + defer conv.deinit(); + + try conv.addSystemMessage("You are helpful."); + try conv.addUserMessage("Hello!"); + + const cfg = testConfig("gpt-4o"); + const body = try serializeRequest(allocator, &cfg, &conv); + defer allocator.free(body); + + var parsed = try std.json.parseFromSlice(std.json.Value, allocator, body, .{}); + defer parsed.deinit(); + + const root = parsed.value.object; + try testing.expectEqualStrings("gpt-4o", root.get("model").?.string); + try testing.expect(root.get("stream").?.bool); + // reasoning_effort is omitted when set to .default. + try testing.expect(root.get("reasoning_effort") == null); + + const msgs = root.get("messages").?.array.items; + try testing.expectEqual(@as(usize, 2), msgs.len); + try testing.expectEqualStrings("system", msgs[0].object.get("role").?.string); + try testing.expectEqualStrings("You are helpful.", msgs[0].object.get("content").?.string); + try testing.expectEqualStrings("user", msgs[1].object.get("role").?.string); + try testing.expectEqualStrings("Hello!", msgs[1].object.get("content").?.string); +} + +test "serializeRequest - assistant Thinking blocks are stripped from outbound history" { + const allocator = testing.allocator; + + var conv = conversation.Conversation.init(allocator); + defer conv.deinit(); + + try conv.addAssistantMessage(&.{ + .{ .Thinking = try conversation.textualBlockFromSlice(allocator, "thinking step") }, + .{ .Text = try conversation.textualBlockFromSlice(allocator, "answer here") }, + }); + + const cfg = testConfig("gpt-4o"); + const body = try serializeRequest(allocator, &cfg, &conv); + defer allocator.free(body); + + var parsed = try std.json.parseFromSlice(std.json.Value, allocator, body, .{}); + defer parsed.deinit(); + + const msg = parsed.value.object.get("messages").?.array.items[0]; + try testing.expectEqualStrings("assistant", msg.object.get("role").?.string); + // Content is a flat string, only the Text block survives. + const content = msg.object.get("content").?.string; + try testing.expectEqualStrings("answer here", content); +} + +test "serializeRequest - reasoning effort level included when set" { + const allocator = testing.allocator; + + var conv = conversation.Conversation.init(allocator); + defer conv.deinit(); + try conv.addUserMessage("Hi"); + + var cfg = testConfig("gpt-4o"); + cfg.reasoning = .high; + + const body = try serializeRequest(allocator, &cfg, &conv); + defer allocator.free(body); + + var parsed = try std.json.parseFromSlice(std.json.Value, allocator, body, .{}); + defer parsed.deinit(); + + try testing.expectEqualStrings( + "high", + parsed.value.object.get("reasoning_effort").?.string, + ); +} + +test "serializeRequest - reasoning .off sends \"none\"" { + const allocator = testing.allocator; + + var conv = conversation.Conversation.init(allocator); + defer conv.deinit(); + try conv.addUserMessage("Hi"); + + var cfg = testConfig("gpt-4o"); + cfg.reasoning = .off; + + const body = try serializeRequest(allocator, &cfg, &conv); + defer allocator.free(body); + + var parsed = try std.json.parseFromSlice(std.json.Value, allocator, body, .{}); + defer parsed.deinit(); + + try testing.expectEqualStrings( + "none", + parsed.value.object.get("reasoning_effort").?.string, + ); +} + +test "parseStreamEvent - role only" { + const allocator = testing.allocator; + const payload = + \\{"id":"chatcmpl-1","choices":[{"index":0,"delta":{"role":"assistant"},"finish_reason":null}]} + ; + + var pd = try parseStreamEvent(allocator, payload); + defer pd.deinit(); + + try testing.expectEqualStrings("assistant", pd.delta.role.?); + try testing.expect(pd.delta.content == null); + try testing.expect(pd.delta.finish_reason == null); +} + +test "parseStreamEvent - content delta" { + const allocator = testing.allocator; + const payload = + \\{"choices":[{"delta":{"content":"Hello"},"finish_reason":null}]} + ; + + var pd = try parseStreamEvent(allocator, payload); + defer pd.deinit(); + + try testing.expectEqualStrings("Hello", pd.delta.content.?); +} + +test "parseStreamEvent - finish_reason stop" { + const allocator = testing.allocator; + const payload = + \\{"choices":[{"delta":{},"finish_reason":"stop"}]} + ; + + var pd = try parseStreamEvent(allocator, payload); + defer pd.deinit(); + + try testing.expectEqualStrings("stop", pd.delta.finish_reason.?); + try testing.expect(pd.delta.content == null); +} + +test "parseStreamEvent - reasoning_content" { + const allocator = testing.allocator; + const payload = + \\{"choices":[{"delta":{"reasoning_content":"hmm"}}]} + ; + + var pd = try parseStreamEvent(allocator, payload); + defer pd.deinit(); + + try testing.expectEqualStrings("hmm", pd.delta.reasoning_content.?); +} diff --git a/libpanto/src/provider.zig b/libpanto/src/provider.zig index 8d6eea5..16c3dd6 100644 --- a/libpanto/src/provider.zig +++ b/libpanto/src/provider.zig @@ -14,36 +14,53 @@ pub const BlockMeta = struct { tool_name: ?[]const u8 = null, }; +/// Vtable for receiving streaming events from a Provider. +/// +/// The lifecycle callbacks (`onMessageStart` ... `onMessageComplete`) return +/// `anyerror!void`. Returning an error aborts the in-flight turn: the Provider +/// stops streaming, calls `onError(err)`, and propagates `err` out of +/// `streamStep`. No partial assistant message is appended to the conversation. +/// +/// `onError` is the receiver's cleanup hook. It fires exactly once per failed +/// turn, whether the error originated in the receiver itself (a write failure) +/// or in the Provider (HTTP/parse/stream failure). It is the last callback the +/// receiver will see for that turn. `onError` itself cannot fail; receivers +/// must swallow secondary failures during cleanup. pub const ReceiverVTable = struct { - onMessageStart: *const fn (*anyopaque, conversation.MessageRole) void, - onBlockStart: *const fn (*anyopaque, ContentBlockType, usize, ?BlockMeta) void, - onContentDelta: *const fn (*anyopaque, usize, []const u8) void, - onBlockComplete: *const fn (*anyopaque, usize, conversation.ContentBlock) void, - onMessageComplete: *const fn (*anyopaque, conversation.Message) void, + onMessageStart: *const fn (*anyopaque, conversation.MessageRole) anyerror!void, + onBlockStart: *const fn (*anyopaque, ContentBlockType, usize, ?BlockMeta) anyerror!void, + onContentDelta: *const fn (*anyopaque, usize, []const u8) anyerror!void, + onBlockComplete: *const fn (*anyopaque, usize, conversation.ContentBlock) anyerror!void, + onMessageComplete: *const fn (*anyopaque, conversation.Message) anyerror!void, + onError: *const fn (*anyopaque, anyerror) void, }; pub const Receiver = struct { ptr: *anyopaque, vtable: *const ReceiverVTable, - pub fn onMessageStart(self: Receiver, role: conversation.MessageRole) void { - self.vtable.onMessageStart(self.ptr, role); + pub fn onMessageStart(self: Receiver, role: conversation.MessageRole) !void { + try self.vtable.onMessageStart(self.ptr, role); } - pub fn onBlockStart(self: Receiver, block_type: ContentBlockType, index: usize, meta: ?BlockMeta) void { - self.vtable.onBlockStart(self.ptr, block_type, index, meta); + pub fn onBlockStart(self: Receiver, block_type: ContentBlockType, index: usize, meta: ?BlockMeta) !void { + try self.vtable.onBlockStart(self.ptr, block_type, index, meta); } - pub fn onContentDelta(self: Receiver, block_index: usize, delta: []const u8) void { - self.vtable.onContentDelta(self.ptr, block_index, delta); + pub fn onContentDelta(self: Receiver, block_index: usize, delta: []const u8) !void { + try self.vtable.onContentDelta(self.ptr, block_index, delta); } - pub fn onBlockComplete(self: Receiver, block_index: usize, block: conversation.ContentBlock) void { - self.vtable.onBlockComplete(self.ptr, block_index, block); + pub fn onBlockComplete(self: Receiver, block_index: usize, block: conversation.ContentBlock) !void { + try self.vtable.onBlockComplete(self.ptr, block_index, block); } - pub fn onMessageComplete(self: Receiver, message: conversation.Message) void { - self.vtable.onMessageComplete(self.ptr, message); + pub fn onMessageComplete(self: Receiver, message: conversation.Message) !void { + try self.vtable.onMessageComplete(self.ptr, message); + } + + pub fn onError(self: Receiver, err: anyerror) void { + self.vtable.onError(self.ptr, err); } }; diff --git a/libpanto/src/provider_openai.zig b/libpanto/src/provider_openai.zig deleted file mode 100644 index c459a9e..0000000 --- a/libpanto/src/provider_openai.zig +++ /dev/null @@ -1,6 +0,0 @@ -// OpenAI-compatible provider implementation. -// Implementation pending; this module will: -// - Convert Conversation → OpenAI wire JSON -// - Make HTTP POST with stream: true -// - Parse SSE events, drive block boundary state machine -// - Call the full Receiver callback sequence diff --git a/libpanto/src/provider_openai_chat.zig b/libpanto/src/provider_openai_chat.zig new file mode 100644 index 0000000..e1b9f81 --- /dev/null +++ b/libpanto/src/provider_openai_chat.zig @@ -0,0 +1,443 @@ +//! OpenAI Chat Completions streaming provider. +//! +//! Wire format reference: https://platform.openai.com/docs/api-reference/chat/streaming +//! +//! Responsibilities: +//! - Convert `Conversation` → request JSON (delegated to openai_chat_json.zig) +//! - POST to `{base_url}/chat/completions` with `stream: true` +//! - Read the chunked body, feed bytes through SSEParser +//! - Parse each event payload, drive the block boundary state machine, +//! and emit Receiver callbacks +//! - Assemble the final Message and emit onMessageComplete + +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 sse_mod = @import("sse.zig"); +const json_mod = @import("openai_chat_json.zig"); +const config_mod = @import("config.zig"); + +/// Active streaming block type tracked by the state machine. Mirrors the +/// `ContentBlock` union variants but adds `.none` for "no block open yet". +const ActiveBlock = enum { none, text, thinking }; + +pub const OpenAIChatProvider = struct { + allocator: Allocator, + io: Io, + config: config_mod.Config, + http_client: http.Client, + + pub fn init(allocator: Allocator, io: Io, cfg: config_mod.Config) OpenAIChatProvider { + return .{ + .allocator = allocator, + .io = io, + .config = cfg, + .http_client = .{ .allocator = allocator, .io = io }, + }; + } + + pub fn deinit(self: *OpenAIChatProvider) void { + self.http_client.deinit(); + } + + /// Return a `Provider` interface bound to this concrete provider. + pub fn provider(self: *OpenAIChatProvider) provider_mod.Provider { + return .{ .ptr = self, .vtable = &vtable }; + } + + const vtable: provider_mod.ProviderVTable = .{ + .streamStep = vtableStreamStep, + .deinit = vtableDeinit, + }; + + fn vtableStreamStep( + ptr: *anyopaque, + conv: *conversation.Conversation, + receiver: *provider_mod.Receiver, + ) anyerror!void { + const self: *OpenAIChatProvider = @ptrCast(@alignCast(ptr)); + return self.streamStep(conv, receiver); + } + + fn vtableDeinit(ptr: *anyopaque) void { + const self: *OpenAIChatProvider = @ptrCast(@alignCast(ptr)); + self.deinit(); + } + + pub fn streamStep( + self: *OpenAIChatProvider, + conv: *conversation.Conversation, + receiver: *provider_mod.Receiver, + ) !void { + // Outer wrapper guarantees `onError` is called exactly once if + // anything fails — whether the receiver, the HTTP transport, or the + // SSE/JSON parsers. The inner `streamStepInner` does the real work. + self.streamStepInner(conv, receiver) catch |err| { + receiver.onError(err); + return err; + }; + } + + fn streamStepInner( + self: *OpenAIChatProvider, + conv: *conversation.Conversation, + receiver: *provider_mod.Receiver, + ) !void { + // Build URL: "{base_url}/chat/completions" + const url = try std.fmt.allocPrint( + self.allocator, + "{s}/chat/completions", + .{self.config.base_url}, + ); + defer self.allocator.free(url); + + const uri = try Uri.parse(url); + + // Build the request body. + const body = try json_mod.serializeRequest(self.allocator, &self.config, conv); + defer self.allocator.free(body); + + // Auth header + const auth_value = try std.fmt.allocPrint( + self.allocator, + "Bearer {s}", + .{self.config.api_key}, + ); + defer self.allocator.free(auth_value); + + const extra_headers = [_]http.Header{ + .{ .name = "content-type", .value = "application/json" }, + .{ .name = "accept", .value = "text/event-stream" }, + .{ .name = "authorization", .value = auth_value }, + }; + + // Open the request. We can't use `fetch()` because it buffers the + // response; we want to stream the body as it arrives. + var req = try self.http_client.request(.POST, uri, .{ + .extra_headers = &extra_headers, + .keep_alive = false, + .redirect_behavior = .not_allowed, + }); + defer req.deinit(); + + req.transfer_encoding = .{ .content_length = body.len }; + + var send_buf: [4096]u8 = undefined; + var bw = try req.sendBodyUnflushed(&send_buf); + try bw.writer.writeAll(body); + try bw.end(); + try req.connection.?.flush(); + + // Receive response headers. + var redirect_buf: [1024]u8 = undefined; + var response = try req.receiveHead(&redirect_buf); + + if (@intFromEnum(response.head.status) >= 400) { + // Drain body for diagnostics. + var transfer_buf: [4096]u8 = undefined; + const body_reader = response.reader(&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; + } + std.log.err("openai_chat HTTP {d}: {s}", .{ + @intFromEnum(response.head.status), + err_buf.items, + }); + return error.HttpError; + } + + // Stream the body through the SSE parser and event handler. + var transfer_buf: [4096]u8 = undefined; + const body_reader = response.reader(&transfer_buf); + + var parser = sse_mod.SSEParser.init(self.allocator); + defer parser.deinit(); + + var state: StreamState = .init(self.allocator); + defer state.deinit(); + + var chunk: [4096]u8 = undefined; + while (true) { + const n = try body_reader.readSliceShort(&chunk); + if (n == 0) break; + + const events = try parser.feed(chunk[0..n]); + defer parser.freeEvents(events); + + for (events) |ev_payload| { + if (std.mem.eql(u8, ev_payload, "[DONE]")) { + try state.finalize(receiver, conv); + return; + } + try handleEvent(self.allocator, ev_payload, &state, receiver); + if (state.end_of_stream) { + try state.finalize(receiver, conv); + return; + } + } + } + + // Stream ended without [DONE] or finish_reason. Finalize anyway. + try state.finalize(receiver, conv); + } +}; + +/// State maintained across the streaming response: which block is currently +/// being assembled, accumulated content, and the assistant message being +/// built up for the final `onMessageComplete` callback. +const StreamState = struct { + allocator: Allocator, + started: bool = false, + /// Set when the wire stream signals end-of-turn (finish_reason or [DONE]). + /// Tells the outer read loop to stop pulling more events. + end_of_stream: bool = false, + /// Set once `finalize` has run, to make it idempotent. + finalized: bool = false, + active: ActiveBlock = .none, + block_index: usize = 0, + + /// Buffer for the currently-streaming block's content. + /// Owned by this state until the block is completed, at which point + /// ownership transfers to the assembled Message. + current_buf: conversation.TextualBlock = .empty, + + /// Assembled blocks for the final message. + blocks: std.ArrayList(conversation.ContentBlock) = .empty, + + 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); + } + + /// Close the active block (if any) and emit onBlockComplete. + /// Ownership of `current_buf` transfers into the appended block. + fn closeActive(self: *StreamState, receiver: *provider_mod.Receiver) !void { + if (self.active == .none) return; + + const block: conversation.ContentBlock = switch (self.active) { + .text => .{ .Text = self.current_buf }, + .thinking => .{ .Thinking = self.current_buf }, + .none => unreachable, + }; + // The buffer ownership has moved into `block`; replace with empty. + self.current_buf = .empty; + + try self.blocks.append(self.allocator, block); + try receiver.onBlockComplete(self.block_index, self.blocks.items[self.blocks.items.len - 1]); + + self.active = .none; + } + + /// Open a new block of the given type, possibly closing a prior block. + fn openBlock( + self: *StreamState, + new_active: ActiveBlock, + receiver: *provider_mod.Receiver, + ) !void { + if (self.active == new_active) return; + if (self.active != .none) { + try self.closeActive(receiver); + self.block_index += 1; + } + self.active = new_active; + const block_type: provider_mod.ContentBlockType = switch (new_active) { + .text => .Text, + .thinking => .Thinking, + .none => unreachable, + }; + try receiver.onBlockStart(block_type, self.block_index, null); + } + + fn appendDelta( + self: *StreamState, + receiver: *provider_mod.Receiver, + delta: []const u8, + ) !void { + try self.current_buf.appendSlice(self.allocator, delta); + try receiver.onContentDelta(self.block_index, delta); + } + + /// End the stream: close any open block and emit onMessageComplete. + /// Ownership of the assembled blocks transfers into the conversation. + fn finalize( + self: *StreamState, + receiver: *provider_mod.Receiver, + conv: *conversation.Conversation, + ) !void { + if (self.finalized) return; + self.finalized = true; + + try self.closeActive(receiver); + + // Move blocks into a fresh conversation message. + const moved_blocks = try self.blocks.toOwnedSlice(self.allocator); + defer self.allocator.free(moved_blocks); + + try conv.addAssistantMessage(moved_blocks); + + // The conversation now owns the block buffers. Build a Message view + // for the callback (borrowed from the conversation). + const msg = conv.messages.items[conv.messages.items.len - 1]; + try receiver.onMessageComplete(msg); + } +}; + +fn handleEvent( + allocator: Allocator, + payload: []const u8, + state: *StreamState, + receiver: *provider_mod.Receiver, +) !void { + var parsed = try json_mod.parseStreamEvent(allocator, payload); + defer parsed.deinit(); + const d = parsed.delta; + + if (!state.started and d.role != null) { + state.started = true; + try receiver.onMessageStart(.assistant); + } + + if (d.reasoning_content) |rc| { + if (!state.started) { + state.started = true; + try receiver.onMessageStart(.assistant); + } + try state.openBlock(.thinking, receiver); + try state.appendDelta(receiver, rc); + } + + if (d.content) |c| { + if (!state.started) { + state.started = true; + try receiver.onMessageStart(.assistant); + } + try state.openBlock(.text, receiver); + try state.appendDelta(receiver, c); + } + + if (d.finish_reason) |_| { + state.end_of_stream = true; + } +} + +// ----------------------------------------------------------------------------- +// Tests +// ----------------------------------------------------------------------------- + +const testing = std.testing; + +/// A no-op Receiver that drops every callback. Useful when the test cares +/// about post-stream conversation state rather than callback observability. +const NoopReceiver = struct { + fn make() provider_mod.Receiver { + return .{ .ptr = @constCast(@ptrCast(&dummy)), .vtable = &vt }; + } + var dummy: u8 = 0; + const vt: provider_mod.ReceiverVTable = .{ + .onMessageStart = noopMsgStart, + .onBlockStart = noopBlockStart, + .onContentDelta = noopDelta, + .onBlockComplete = noopBlockComplete, + .onMessageComplete = noopMsgComplete, + .onError = noopErr, + }; + fn noopMsgStart(_: *anyopaque, _: conversation.MessageRole) anyerror!void {} + fn noopBlockStart(_: *anyopaque, _: provider_mod.ContentBlockType, _: usize, _: ?provider_mod.BlockMeta) anyerror!void {} + fn noopDelta(_: *anyopaque, _: usize, _: []const u8) anyerror!void {} + fn noopBlockComplete(_: *anyopaque, _: usize, _: conversation.ContentBlock) anyerror!void {} + fn noopMsgComplete(_: *anyopaque, _: conversation.Message) anyerror!void {} + fn noopErr(_: *anyopaque, _: anyerror) void {} +}; + +/// Feed a sequence of SSE event payloads through the state machine as if +/// they had been delivered by the wire, finalizing into `conv`. +fn runStreamedTurn( + allocator: Allocator, + conv: *conversation.Conversation, + receiver: *provider_mod.Receiver, + events: []const []const u8, +) !void { + var state: StreamState = .init(allocator); + defer state.deinit(); + + for (events) |payload| { + if (std.mem.eql(u8, payload, "[DONE]")) break; + try handleEvent(allocator, payload, &state, receiver); + if (state.end_of_stream) break; + } + try state.finalize(receiver, conv); +} + +test "two streamed turns persist assistant replies in the conversation" { + // Regression test for the bug where `finish_reason` arrived before + // `[DONE]` and `finalize` early-returned without appending the assistant + // message, so follow-up turns were sent without prior responses. + + const allocator = testing.allocator; + + var conv = conversation.Conversation.init(allocator); + defer conv.deinit(); + + try conv.addSystemMessage("You are a helpful assistant."); + try conv.addUserMessage("hello!"); + + var recv = NoopReceiver.make(); + + const turn1 = [_][]const u8{ + \\{"choices":[{"delta":{"role":"assistant"}}]} + , + \\{"choices":[{"delta":{"content":"Hello! "}}]} + , + \\{"choices":[{"delta":{"content":"How can I help you today?"}}]} + , + \\{"choices":[{"delta":{},"finish_reason":"stop"}]} + , + "[DONE]", + }; + try runStreamedTurn(allocator, &conv, &recv, &turn1); + + try testing.expectEqual(@as(usize, 3), conv.messages.items.len); + try testing.expectEqual(conversation.MessageRole.assistant, conv.messages.items[2].role); + try testing.expectEqualStrings( + "Hello! How can I help you today?", + conv.messages.items[2].content.items[0].Text.items, + ); + + // Second user turn: the assistant must still see its prior response. + try conv.addUserMessage("how did you respond to my greeting just now?"); + + const turn2 = [_][]const u8{ + \\{"choices":[{"delta":{"role":"assistant"}}]} + , + \\{"choices":[{"delta":{"content":"I replied: \"Hello! How can I help you today?\""}}]} + , + \\{"choices":[{"delta":{},"finish_reason":"stop"}]} + , + "[DONE]", + }; + try runStreamedTurn(allocator, &conv, &recv, &turn2); + + // System + user + assistant + user + assistant = 5 messages. + try testing.expectEqual(@as(usize, 5), conv.messages.items.len); + try testing.expectEqual(conversation.MessageRole.assistant, conv.messages.items[4].role); + try testing.expectEqualStrings( + "I replied: \"Hello! How can I help you today?\"", + conv.messages.items[4].content.items[0].Text.items, + ); +} diff --git a/libpanto/src/root.zig b/libpanto/src/root.zig index 72ceb97..9eb6d0e 100644 --- a/libpanto/src/root.zig +++ b/libpanto/src/root.zig @@ -4,6 +4,11 @@ pub const agent = @import("agent.zig"); pub const config = @import("config.zig"); pub const sse = @import("sse.zig"); -// Internal modules — not re-exported as public API -// pub const json = @import("json.zig"); -// pub const provider_openai = @import("provider_openai.zig"); +// Internal modules — exposed so their tests run with the library test suite, +// but not part of the stable public API. +pub const openai_chat_json = @import("openai_chat_json.zig"); +pub const provider_openai_chat = @import("provider_openai_chat.zig"); + +test { + @import("std").testing.refAllDecls(@This()); +} diff --git a/libpanto/src/sse.zig b/libpanto/src/sse.zig index 150c3ea..c54d532 100644 --- a/libpanto/src/sse.zig +++ b/libpanto/src/sse.zig @@ -1,99 +1,81 @@ const std = @import("std"); -/// Parsed SSE event data — the content after "data: " (prefix removed). -/// For multi-line data events, lines are joined with newlines per the SSE spec. -/// -/// Borrowed from the parser's internal buffer. Call `SSEParser.freeEvents()` -/// to free the events list when done (event data itself is borrowed and -/// does not need individual freeing). -const Event = []const u8; +/// A parsed SSE event payload: the concatenated `data:` content. +/// Owned by the caller; free with `freeEvents`. +const Event = []u8; /// Incremental SSE line parser. /// /// The HTTP client delivers arbitrary-sized read buffers; this module -/// reassembles them into complete `data: ...\n\n` events. -/// -/// Event data is borrowed from the internal buffer. Call `freeEvents()` to -/// free the events list returned by `feed`; the event data itself is borrowed -/// from the parser's buffer and does not need individual freeing. +/// reassembles them into complete `data: ...\n\n` events and returns +/// owned event payload slices. pub const SSEParser = struct { buf: std.ArrayList(u8) = .empty, allocator: std.mem.Allocator, pub fn init(allocator: std.mem.Allocator) SSEParser { - return .{ - .allocator = allocator, - }; + return .{ .allocator = allocator }; } /// Feed a chunk of raw bytes. Returns a list of complete SSE events - /// found in the buffer. Returns an empty list if no complete events - /// are available yet. - /// - /// Event data is borrowed from the internal buffer. Call `freeEvents()` - /// on the returned list when done. + /// found in the buffer (may be empty). The caller owns the returned + /// slice and every event within; free with `freeEvents`. pub fn feed(self: *SSEParser, chunk: []const u8) ![]Event { try self.buf.appendSlice(self.allocator, chunk); - var events = std.ArrayList(Event).init(self.allocator); - errdefer events.deinit(); + var events: std.ArrayList(Event) = .empty; + errdefer { + for (events.items) |ev| self.allocator.free(ev); + events.deinit(self.allocator); + } - var write_pos: usize = 0; var scan_pos: usize = 0; - while (std.mem.indexOf(u8, self.buf.items[scan_pos..], "\n\n")) |rel_delim| { const delim = scan_pos + rel_delim; const event_bytes = self.buf.items[scan_pos..delim]; - // Compact data lines in-place. Compacted data is always shorter - // than the original event bytes (we strip at least "data: " per - // line), so write_pos <= scan_pos and we never overwrite data - // we haven't yet read. - const data_start = write_pos; + // Collect data: lines into an owned payload string. + var payload: std.ArrayList(u8) = .empty; + errdefer payload.deinit(self.allocator); + var line_start: usize = 0; var has_data = false; - while (line_start < event_bytes.len) { const newline = std.mem.indexOfScalarPos(u8, event_bytes, line_start, '\n') orelse event_bytes.len; const line = event_bytes[line_start..newline]; if (std.mem.startsWith(u8, line, "data: ")) { - if (has_data) { - self.buf.items[write_pos] = '\n'; - write_pos += 1; - } - const content = line["data: ".len..]; - std.mem.copyForwards(u8, self.buf.items[write_pos .. write_pos + content.len], content); - write_pos += content.len; + if (has_data) try payload.append(self.allocator, '\n'); + try payload.appendSlice(self.allocator, line["data: ".len..]); has_data = true; } - // Ignore other SSE fields (event:, id:, retry:) and comments (starting with :) + // Ignore other SSE fields (event:, id:, retry:) and comments. line_start = newline + 1; } if (has_data) { - try events.append(self.buf.items[data_start..write_pos]); + try events.append(self.allocator, try payload.toOwnedSlice(self.allocator)); + } else { + payload.deinit(self.allocator); } scan_pos = delim + 2; } - // Shift remaining (incomplete) bytes to the front, retaining - // the backing allocation for reuse. + // Drop processed bytes; retain the buffer allocation. const remaining_len = self.buf.items.len - scan_pos; if (remaining_len > 0 and scan_pos > 0) { std.mem.copyForwards(u8, self.buf.items[0..remaining_len], self.buf.items[scan_pos..]); } self.buf.shrinkRetainingCapacity(remaining_len); - return events.toOwnedSlice(); + return events.toOwnedSlice(self.allocator); } /// Free a list of events returned by `feed`. - /// Event data is borrowed from the parser's buffer and does not need - /// to be freed individually; only the events array itself is freed. pub fn freeEvents(self: *SSEParser, events: []Event) void { + for (events) |ev| self.allocator.free(ev); self.allocator.free(events); } @@ -102,98 +84,87 @@ pub const SSEParser = struct { } }; -test "SSEParser - single complete event" { - const allocator = std.testing.allocator; +// ----------------------------------------------------------------------------- +// Tests +// ----------------------------------------------------------------------------- + +const testing = std.testing; - var parser = SSEParser.init(allocator); +test "SSEParser - single complete event" { + var parser = SSEParser.init(testing.allocator); defer parser.deinit(); const events = try parser.feed("data: hello\n\n"); defer parser.freeEvents(events); - try std.testing.expectEqual(@as(usize, 1), events.len); - try std.testing.expectEqualStrings("hello", events[0]); + try testing.expectEqual(@as(usize, 1), events.len); + try testing.expectEqualStrings("hello", events[0]); } test "SSEParser - partial then complete" { - const allocator = std.testing.allocator; - - var parser = SSEParser.init(allocator); + var parser = SSEParser.init(testing.allocator); defer parser.deinit(); - // Feed partial — no complete event yet const events1 = try parser.feed("data: hel"); defer parser.freeEvents(events1); - try std.testing.expectEqual(@as(usize, 0), events1.len); + try testing.expectEqual(@as(usize, 0), events1.len); - // Feed remainder — completes the event const events2 = try parser.feed("lo\n\n"); defer parser.freeEvents(events2); - try std.testing.expectEqual(@as(usize, 1), events2.len); - try std.testing.expectEqualStrings("hello", events2[0]); + try testing.expectEqual(@as(usize, 1), events2.len); + try testing.expectEqualStrings("hello", events2[0]); } test "SSEParser - multiple events in single chunk" { - const allocator = std.testing.allocator; - - var parser = SSEParser.init(allocator); + var parser = SSEParser.init(testing.allocator); defer parser.deinit(); const events = try parser.feed("data: one\n\ndata: two\n\n"); defer parser.freeEvents(events); - try std.testing.expectEqual(@as(usize, 2), events.len); - try std.testing.expectEqualStrings("one", events[0]); - try std.testing.expectEqualStrings("two", events[1]); + try testing.expectEqual(@as(usize, 2), events.len); + try testing.expectEqualStrings("one", events[0]); + try testing.expectEqualStrings("two", events[1]); } test "SSEParser - data DONE signal" { - const allocator = std.testing.allocator; - - var parser = SSEParser.init(allocator); + var parser = SSEParser.init(testing.allocator); defer parser.deinit(); const events = try parser.feed("data: [DONE]\n\n"); defer parser.freeEvents(events); - try std.testing.expectEqual(@as(usize, 1), events.len); - try std.testing.expectEqualStrings("[DONE]", events[0]); + try testing.expectEqual(@as(usize, 1), events.len); + try testing.expectEqualStrings("[DONE]", events[0]); } test "SSEParser - empty event (keep-alive)" { - const allocator = std.testing.allocator; - - var parser = SSEParser.init(allocator); + var parser = SSEParser.init(testing.allocator); defer parser.deinit(); - // A blank line (just \n\n) is a keep-alive comment with no data const events = try parser.feed("\n\n"); defer parser.freeEvents(events); - try std.testing.expectEqual(@as(usize, 0), events.len); + try testing.expectEqual(@as(usize, 0), events.len); } test "SSEParser - ignores non-data fields" { - const allocator = std.testing.allocator; - - var parser = SSEParser.init(allocator); + var parser = SSEParser.init(testing.allocator); defer parser.deinit(); const events = try parser.feed("event: message\ndata: payload\nid: 42\n\n"); defer parser.freeEvents(events); - try std.testing.expectEqual(@as(usize, 1), events.len); - try std.testing.expectEqualStrings("payload", events[0]); + try testing.expectEqual(@as(usize, 1), events.len); + try testing.expectEqualStrings("payload", events[0]); } test "SSEParser - multi-line data joined with newline" { - const allocator = std.testing.allocator; - - var parser = SSEParser.init(allocator); + var parser = SSEParser.init(testing.allocator); defer parser.deinit(); const events = try parser.feed("data: line1\ndata: line2\n\n"); defer parser.freeEvents(events); - try std.testing.expectEqual(@as(usize, 1), events.len); - try std.testing.expectEqualStrings("line1\nline2", events[0]); + try testing.expectEqual(@as(usize, 1), events.len); + try testing.expectEqualStrings("line1\nline2", events[0]); } -- cgit v1.3