From f026bb81ae68f516910d0eb4f23c9344dd36b62b Mon Sep 17 00:00:00 2001 From: T Date: Mon, 25 May 2026 21:50:43 -0600 Subject: phase 2 done --- libpanto/src/anthropic_messages_json.zig | 638 +++++++++++++++++++++ libpanto/src/config.zig | 66 ++- libpanto/src/conversation.zig | 27 +- libpanto/src/openai_chat_json.zig | 7 +- libpanto/src/provider.zig | 30 + libpanto/src/provider_anthropic_messages.zig | 828 +++++++++++++++++++++++++++ libpanto/src/provider_openai_chat.zig | 24 +- libpanto/src/root.zig | 19 +- 8 files changed, 1606 insertions(+), 33 deletions(-) create mode 100644 libpanto/src/anthropic_messages_json.zig create mode 100644 libpanto/src/provider_anthropic_messages.zig (limited to 'libpanto/src') diff --git a/libpanto/src/anthropic_messages_json.zig b/libpanto/src/anthropic_messages_json.zig new file mode 100644 index 0000000..8f44edc --- /dev/null +++ b/libpanto/src/anthropic_messages_json.zig @@ -0,0 +1,638 @@ +//! Anthropic Messages API JSON serialization and event parsing. +//! +//! Two responsibilities: +//! 1. Serialize a `Conversation` into a `/v1/messages` request body. +//! Anthropic differs from OpenAI in several ways — see `serializeRequest`. +//! 2. Parse one streaming SSE event payload (the JSON object after `data: `) +//! into a strongly-typed `StreamEvent` that the provider can consume. +//! +//! Wire format reference: +//! https://platform.claude.com/docs/en/build-with-claude/streaming +//! https://platform.claude.com/docs/en/build-with-claude/extended-thinking + +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"); + +// ----------------------------------------------------------------------------- +// Request serialization +// ----------------------------------------------------------------------------- + +/// Serialize a Conversation into a `/v1/messages` request body. +/// +/// Differences from OpenAI Chat Completions: +/// - System messages are extracted and concatenated into a top-level +/// `system` string. They do not appear in the `messages` array. +/// - `content` is always an array of typed blocks, never a bare string. +/// - `max_tokens` is required. +/// +/// Caller owns the returned slice. +pub fn serializeRequest( + allocator: Allocator, + cfg: *const config_mod.AnthropicMessagesConfig, + 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("max_tokens"); + try s.write(cfg.max_tokens); + + try s.objectField("stream"); + try s.write(true); + + // Build and emit the concatenated system prompt, if any. + var system_buf: std.ArrayList(u8) = .empty; + defer system_buf.deinit(allocator); + try collectSystemPrompt(conv, &system_buf, allocator); + if (system_buf.items.len > 0) { + try s.objectField("system"); + try s.write(system_buf.items); + } + + // Emit messages (everything that isn't .system). + try s.objectField("messages"); + try s.beginArray(); + for (conv.messages.items) |msg| { + if (msg.role == .system) continue; + try writeMessage(&s, msg, allocator); + } + try s.endArray(); + + try s.endObject(); + + return try aw.toOwnedSlice(); +} + +fn collectSystemPrompt( + conv: *const conversation.Conversation, + out: *std.ArrayList(u8), + allocator: Allocator, +) !void { + var first = true; + for (conv.messages.items) |msg| { + if (msg.role != .system) continue; + for (msg.content.items) |block| { + if (block != .Text) continue; + if (!first) try out.append(allocator, '\n'); + try out.appendSlice(allocator, block.Text.items); + first = false; + } + } +} + +fn writeMessage( + s: *std.json.Stringify, + msg: conversation.Message, + allocator: Allocator, +) !void { + _ = allocator; + try s.beginObject(); + + try s.objectField("role"); + try s.write(@tagName(msg.role)); + + try s.objectField("content"); + try s.beginArray(); + for (msg.content.items) |block| { + try writeBlock(s, block); + } + try s.endArray(); + + try s.endObject(); +} + +fn writeBlock(s: *std.json.Stringify, block: conversation.ContentBlock) !void { + switch (block) { + .Text => |tb| { + try s.beginObject(); + try s.objectField("type"); + try s.write("text"); + try s.objectField("text"); + try s.write(tb.items); + try s.endObject(); + }, + .Thinking => |tb| { + // Anthropic requires the signature field to round-trip a thinking + // block. If we don't have one (e.g. block was synthesized from a + // non-Anthropic provider), skip the block: Anthropic will reject + // unsigned thinking on incoming messages. + const sig = tb.signature orelse return; + try s.beginObject(); + try s.objectField("type"); + try s.write("thinking"); + try s.objectField("thinking"); + try s.write(tb.text.items); + try s.objectField("signature"); + try s.write(sig); + try s.endObject(); + }, + // ToolUse / ToolResult: phase 3+. + .ToolUse, .ToolResult => {}, + } +} + +// ----------------------------------------------------------------------------- +// Streaming event parsing +// ----------------------------------------------------------------------------- + +/// Parsed shape of one SSE event payload. The `tag` field selects which of +/// the payload variants is populated. +/// +/// String slices are borrowed from the underlying `std.json.Parsed`. The +/// caller must keep `parsed` alive (or copy the slices) until the event is +/// fully consumed. +pub const StreamEventTag = enum { + message_start, + content_block_start, + content_block_delta, + content_block_stop, + message_delta, + message_stop, + ping, + @"error", + unknown, +}; + +pub const ContentBlockKind = enum { text, thinking, tool_use, unknown }; + +pub const StreamEvent = union(StreamEventTag) { + message_start: struct { + role: ?[]const u8 = null, + }, + content_block_start: struct { + index: usize, + kind: ContentBlockKind, + // For tool_use blocks (phase 3+): + tool_id: ?[]const u8 = null, + tool_name: ?[]const u8 = null, + }, + content_block_delta: struct { + index: usize, + text_delta: ?[]const u8 = null, + thinking_delta: ?[]const u8 = null, + signature_delta: ?[]const u8 = null, + input_json_delta: ?[]const u8 = null, + }, + content_block_stop: struct { + index: usize, + }, + message_delta: struct { + stop_reason: ?[]const u8 = null, + }, + message_stop: void, + ping: void, + @"error": struct { + kind: ?[]const u8 = null, + message: ?[]const u8 = null, + }, + unknown: void, +}; + +pub const ParsedStreamEvent = struct { + parsed: std.json.Parsed(std.json.Value), + event: StreamEvent, + + pub fn deinit(self: *ParsedStreamEvent) void { + self.parsed.deinit(); + } +}; + +pub fn parseStreamEvent(allocator: Allocator, payload: []const u8) !ParsedStreamEvent { + var parsed = try std.json.parseFromSlice(std.json.Value, allocator, payload, .{}); + errdefer parsed.deinit(); + + const root = parsed.value; + if (root != .object) return .{ .parsed = parsed, .event = .unknown }; + + const type_v = root.object.get("type") orelse return .{ .parsed = parsed, .event = .unknown }; + if (type_v != .string) return .{ .parsed = parsed, .event = .unknown }; + + const ty = type_v.string; + if (std.mem.eql(u8, ty, "message_start")) { + var role: ?[]const u8 = null; + if (root.object.get("message")) |m| { + if (m == .object) { + if (m.object.get("role")) |r| { + if (r == .string) role = r.string; + } + } + } + return .{ .parsed = parsed, .event = .{ .message_start = .{ .role = role } } }; + } + + if (std.mem.eql(u8, ty, "content_block_start")) { + const idx = readIndex(root) orelse 0; + var kind: ContentBlockKind = .unknown; + var tool_id: ?[]const u8 = null; + var tool_name: ?[]const u8 = null; + if (root.object.get("content_block")) |cb| { + if (cb == .object) { + if (cb.object.get("type")) |t| { + if (t == .string) { + if (std.mem.eql(u8, t.string, "text")) { + kind = .text; + } else if (std.mem.eql(u8, t.string, "thinking")) { + kind = .thinking; + } else if (std.mem.eql(u8, t.string, "tool_use")) { + kind = .tool_use; + } + } + } + if (cb.object.get("id")) |i| { + if (i == .string) tool_id = i.string; + } + if (cb.object.get("name")) |n| { + if (n == .string) tool_name = n.string; + } + } + } + return .{ .parsed = parsed, .event = .{ .content_block_start = .{ + .index = idx, + .kind = kind, + .tool_id = tool_id, + .tool_name = tool_name, + } } }; + } + + if (std.mem.eql(u8, ty, "content_block_delta")) { + const idx = readIndex(root) orelse 0; + var text_delta: ?[]const u8 = null; + var thinking_delta: ?[]const u8 = null; + var signature_delta: ?[]const u8 = null; + var input_json_delta: ?[]const u8 = null; + if (root.object.get("delta")) |d| { + if (d == .object) { + const dt = blk: { + if (d.object.get("type")) |t| { + if (t == .string) break :blk t.string; + } + break :blk ""; + }; + if (std.mem.eql(u8, dt, "text_delta")) { + if (d.object.get("text")) |v| if (v == .string) { + text_delta = v.string; + }; + } else if (std.mem.eql(u8, dt, "thinking_delta")) { + if (d.object.get("thinking")) |v| if (v == .string) { + thinking_delta = v.string; + }; + } else if (std.mem.eql(u8, dt, "signature_delta")) { + if (d.object.get("signature")) |v| if (v == .string) { + signature_delta = v.string; + }; + } else if (std.mem.eql(u8, dt, "input_json_delta")) { + if (d.object.get("partial_json")) |v| if (v == .string) { + input_json_delta = v.string; + }; + } + } + } + return .{ .parsed = parsed, .event = .{ .content_block_delta = .{ + .index = idx, + .text_delta = text_delta, + .thinking_delta = thinking_delta, + .signature_delta = signature_delta, + .input_json_delta = input_json_delta, + } } }; + } + + if (std.mem.eql(u8, ty, "content_block_stop")) { + const idx = readIndex(root) orelse 0; + return .{ .parsed = parsed, .event = .{ .content_block_stop = .{ .index = idx } } }; + } + + if (std.mem.eql(u8, ty, "message_delta")) { + var stop_reason: ?[]const u8 = null; + if (root.object.get("delta")) |d| { + if (d == .object) { + if (d.object.get("stop_reason")) |sr| { + if (sr == .string) stop_reason = sr.string; + } + } + } + return .{ .parsed = parsed, .event = .{ .message_delta = .{ .stop_reason = stop_reason } } }; + } + + if (std.mem.eql(u8, ty, "message_stop")) { + return .{ .parsed = parsed, .event = .message_stop }; + } + + if (std.mem.eql(u8, ty, "ping")) { + return .{ .parsed = parsed, .event = .ping }; + } + + if (std.mem.eql(u8, ty, "error")) { + var kind: ?[]const u8 = null; + var message: ?[]const u8 = null; + if (root.object.get("error")) |e| { + if (e == .object) { + if (e.object.get("type")) |t| if (t == .string) { + kind = t.string; + }; + if (e.object.get("message")) |m| if (m == .string) { + message = m.string; + }; + } + } + return .{ .parsed = parsed, .event = .{ .@"error" = .{ .kind = kind, .message = message } } }; + } + + return .{ .parsed = parsed, .event = .unknown }; +} + +fn readIndex(root: std.json.Value) ?usize { + const v = root.object.get("index") orelse return null; + if (v != .integer) return null; + if (v.integer < 0) return null; + return @intCast(v.integer); +} + +// ----------------------------------------------------------------------------- +// Tests +// ----------------------------------------------------------------------------- + +const testing = std.testing; + +fn testConfig(model: []const u8) config_mod.AnthropicMessagesConfig { + return .{ + .api_key = "k", + .base_url = "u", + .model = model, + .max_tokens = 1024, + }; +} + +test "serializeRequest - system extracted into top-level field" { + 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("claude-sonnet-4-20250514"); + 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("claude-sonnet-4-20250514", root.get("model").?.string); + try testing.expect(root.get("stream").?.bool); + try testing.expectEqual(@as(i64, 1024), root.get("max_tokens").?.integer); + try testing.expectEqualStrings("You are helpful.", root.get("system").?.string); + + // Only the user message appears in `messages`. + const msgs = root.get("messages").?.array.items; + try testing.expectEqual(@as(usize, 1), msgs.len); + try testing.expectEqualStrings("user", msgs[0].object.get("role").?.string); + + // Content is an array of typed blocks, never a bare string. + const content = msgs[0].object.get("content").?.array.items; + try testing.expectEqual(@as(usize, 1), content.len); + try testing.expectEqualStrings("text", content[0].object.get("type").?.string); + try testing.expectEqualStrings("Hello!", content[0].object.get("text").?.string); +} + +test "serializeRequest - multiple system messages concatenated with newlines" { + const allocator = testing.allocator; + + var conv = conversation.Conversation.init(allocator); + defer conv.deinit(); + + try conv.addSystemMessage("Be terse."); + try conv.addSystemMessage("Be accurate."); + try conv.addUserMessage("Hi"); + + const cfg = testConfig("claude-x"); + 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( + "Be terse.\nBe accurate.", + parsed.value.object.get("system").?.string, + ); +} + +test "serializeRequest - no system messages omits the system field" { + const allocator = testing.allocator; + + var conv = conversation.Conversation.init(allocator); + defer conv.deinit(); + try conv.addUserMessage("Hi"); + + const cfg = testConfig("claude-x"); + 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.expect(parsed.value.object.get("system") == null); +} + +test "serializeRequest - signed assistant Thinking blocks round-trip" { + const allocator = testing.allocator; + + var conv = conversation.Conversation.init(allocator); + defer conv.deinit(); + + const sig = try allocator.dupe(u8, "EqQBCgIYAhIM1gbcDa9GJwZA"); + try conv.addAssistantMessage(&.{ + .{ .Thinking = .{ + .text = try conversation.textualBlockFromSlice(allocator, "let me think"), + .signature = sig, + } }, + .{ .Text = try conversation.textualBlockFromSlice(allocator, "the answer is 42") }, + }); + + const cfg = testConfig("claude-x"); + 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); + + const content = msg.object.get("content").?.array.items; + try testing.expectEqual(@as(usize, 2), content.len); + + try testing.expectEqualStrings("thinking", content[0].object.get("type").?.string); + try testing.expectEqualStrings("let me think", content[0].object.get("thinking").?.string); + try testing.expectEqualStrings( + "EqQBCgIYAhIM1gbcDa9GJwZA", + content[0].object.get("signature").?.string, + ); + + try testing.expectEqualStrings("text", content[1].object.get("type").?.string); + try testing.expectEqualStrings("the answer is 42", content[1].object.get("text").?.string); +} + +test "serializeRequest - unsigned Thinking blocks are dropped" { + // Anthropic rejects thinking blocks without a valid signature on inbound + // messages. If a block lacks one (e.g. from a different provider or an + // interrupted stream), omit it rather than send a guaranteed-400 request. + const allocator = testing.allocator; + + var conv = conversation.Conversation.init(allocator); + defer conv.deinit(); + + try conv.addAssistantMessage(&.{ + .{ .Thinking = .{ + .text = try conversation.textualBlockFromSlice(allocator, "unsigned thinking"), + .signature = null, + } }, + .{ .Text = try conversation.textualBlockFromSlice(allocator, "answer") }, + }); + + const cfg = testConfig("claude-x"); + 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 content = parsed.value.object.get("messages").?.array.items[0] + .object.get("content").?.array.items; + try testing.expectEqual(@as(usize, 1), content.len); + try testing.expectEqualStrings("text", content[0].object.get("type").?.string); +} + +test "parseStreamEvent - message_start" { + const allocator = testing.allocator; + const payload = + \\{"type":"message_start","message":{"id":"msg_1","type":"message","role":"assistant","content":[],"model":"claude","stop_reason":null}} + ; + var pe = try parseStreamEvent(allocator, payload); + defer pe.deinit(); + try testing.expectEqual(StreamEventTag.message_start, @as(StreamEventTag, pe.event)); + try testing.expectEqualStrings("assistant", pe.event.message_start.role.?); +} + +test "parseStreamEvent - content_block_start text" { + const allocator = testing.allocator; + const payload = + \\{"type":"content_block_start","index":0,"content_block":{"type":"text","text":""}} + ; + var pe = try parseStreamEvent(allocator, payload); + defer pe.deinit(); + try testing.expectEqual(StreamEventTag.content_block_start, @as(StreamEventTag, pe.event)); + try testing.expectEqual(@as(usize, 0), pe.event.content_block_start.index); + try testing.expectEqual(ContentBlockKind.text, pe.event.content_block_start.kind); +} + +test "parseStreamEvent - content_block_start thinking" { + const allocator = testing.allocator; + const payload = + \\{"type":"content_block_start","index":0,"content_block":{"type":"thinking","thinking":"","signature":""}} + ; + var pe = try parseStreamEvent(allocator, payload); + defer pe.deinit(); + try testing.expectEqual(ContentBlockKind.thinking, pe.event.content_block_start.kind); +} + +test "parseStreamEvent - text_delta" { + const allocator = testing.allocator; + const payload = + \\{"type":"content_block_delta","index":1,"delta":{"type":"text_delta","text":"Hello"}} + ; + var pe = try parseStreamEvent(allocator, payload); + defer pe.deinit(); + try testing.expectEqual(@as(usize, 1), pe.event.content_block_delta.index); + try testing.expectEqualStrings("Hello", pe.event.content_block_delta.text_delta.?); + try testing.expect(pe.event.content_block_delta.thinking_delta == null); +} + +test "parseStreamEvent - thinking_delta" { + const allocator = testing.allocator; + const payload = + \\{"type":"content_block_delta","index":0,"delta":{"type":"thinking_delta","thinking":"step 1"}} + ; + var pe = try parseStreamEvent(allocator, payload); + defer pe.deinit(); + try testing.expectEqualStrings("step 1", pe.event.content_block_delta.thinking_delta.?); +} + +test "parseStreamEvent - signature_delta" { + const allocator = testing.allocator; + const payload = + \\{"type":"content_block_delta","index":0,"delta":{"type":"signature_delta","signature":"abc123"}} + ; + var pe = try parseStreamEvent(allocator, payload); + defer pe.deinit(); + try testing.expectEqualStrings("abc123", pe.event.content_block_delta.signature_delta.?); +} + +test "parseStreamEvent - content_block_stop" { + const allocator = testing.allocator; + const payload = + \\{"type":"content_block_stop","index":2} + ; + var pe = try parseStreamEvent(allocator, payload); + defer pe.deinit(); + try testing.expectEqual(@as(usize, 2), pe.event.content_block_stop.index); +} + +test "parseStreamEvent - message_delta stop_reason" { + const allocator = testing.allocator; + const payload = + \\{"type":"message_delta","delta":{"stop_reason":"end_turn","stop_sequence":null}} + ; + var pe = try parseStreamEvent(allocator, payload); + defer pe.deinit(); + try testing.expectEqualStrings("end_turn", pe.event.message_delta.stop_reason.?); +} + +test "parseStreamEvent - message_stop" { + const allocator = testing.allocator; + const payload = + \\{"type":"message_stop"} + ; + var pe = try parseStreamEvent(allocator, payload); + defer pe.deinit(); + try testing.expectEqual(StreamEventTag.message_stop, @as(StreamEventTag, pe.event)); +} + +test "parseStreamEvent - ping" { + const allocator = testing.allocator; + const payload = "{\"type\":\"ping\"}"; + var pe = try parseStreamEvent(allocator, payload); + defer pe.deinit(); + try testing.expectEqual(StreamEventTag.ping, @as(StreamEventTag, pe.event)); +} + +test "parseStreamEvent - error" { + const allocator = testing.allocator; + const payload = + \\{"type":"error","error":{"type":"overloaded_error","message":"too busy"}} + ; + var pe = try parseStreamEvent(allocator, payload); + defer pe.deinit(); + try testing.expectEqualStrings("overloaded_error", pe.event.@"error".kind.?); + try testing.expectEqualStrings("too busy", pe.event.@"error".message.?); +} + +test "parseStreamEvent - unknown type" { + const allocator = testing.allocator; + const payload = + \\{"type":"some_future_event","extra":"data"} + ; + var pe = try parseStreamEvent(allocator, payload); + defer pe.deinit(); + try testing.expectEqual(StreamEventTag.unknown, @as(StreamEventTag, pe.event)); +} diff --git a/libpanto/src/config.zig b/libpanto/src/config.zig index 74bdcc7..e395230 100644 --- a/libpanto/src/config.zig +++ b/libpanto/src/config.zig @@ -1,5 +1,14 @@ +//! Per-provider configuration. +//! +//! `Config` is a tagged union keyed by `APIStyle`. Each variant carries the +//! settings specific to one wire dialect. New providers add a new tag and a +//! new payload struct here; nothing else in libpanto needs a central enum +//! refresh. + +/// The wire dialect a provider speaks. pub const APIStyle = enum { openai_chat, + anthropic_messages, }; /// Reasoning intensity hint sent to providers that support it. @@ -17,37 +26,62 @@ pub const ReasoningEffort = enum { high, }; -pub const Config = struct { - api_style: APIStyle, +pub const OpenAIChatConfig = struct { api_key: []const u8, base_url: []const u8, model: []const u8, reasoning: ReasoningEffort = .default, }; +pub const AnthropicMessagesConfig = struct { + api_key: []const u8, + base_url: []const u8, + model: []const u8, + /// Value sent in the `anthropic-version` header. + api_version: []const u8 = "2023-06-01", + /// Required by Anthropic's Messages API. + max_tokens: u32 = 4096, +}; + +pub const Config = union(APIStyle) { + openai_chat: OpenAIChatConfig, + anthropic_messages: AnthropicMessagesConfig, + + pub fn style(self: Config) APIStyle { + return @as(APIStyle, self); + } +}; + const t = @import("std").testing; -test "Config - construct with all fields" { - const cfg = Config{ - .api_style = .openai_chat, +test "Config - openai_chat variant" { + const cfg: Config = .{ .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); + } }; + try t.expectEqual(APIStyle.openai_chat, cfg.style()); + try t.expectEqualStrings("sk-test", cfg.openai_chat.api_key); + try t.expectEqual(ReasoningEffort.high, cfg.openai_chat.reasoning); +} + +test "Config - anthropic_messages variant" { + const cfg: Config = .{ .anthropic_messages = .{ + .api_key = "sk-ant-test", + .base_url = "https://api.anthropic.com", + .model = "claude-sonnet-4-20250514", + } }; + try t.expectEqual(APIStyle.anthropic_messages, cfg.style()); + try t.expectEqualStrings("2023-06-01", cfg.anthropic_messages.api_version); + try t.expectEqual(@as(u32, 4096), cfg.anthropic_messages.max_tokens); } -test "Config - reasoning defaults to .default" { - const cfg = Config{ - .api_style = .openai_chat, +test "Config - openai_chat reasoning defaults to .default" { + const cfg: Config = .{ .openai_chat = .{ .api_key = "k", .base_url = "u", .model = "m", - }; - try t.expectEqual(ReasoningEffort.default, cfg.reasoning); + } }; + try t.expectEqual(ReasoningEffort.default, cfg.openai_chat.reasoning); } diff --git a/libpanto/src/conversation.zig b/libpanto/src/conversation.zig index 0e9a4e9..6776412 100644 --- a/libpanto/src/conversation.zig +++ b/libpanto/src/conversation.zig @@ -13,6 +13,22 @@ pub fn textualBlockFromSlice(alloc: Allocator, slice: []const u8) !TextualBlock return buf; } +/// A reasoning/thinking block from the assistant. +/// +/// `text` is the streamed reasoning content. `signature` is the opaque +/// integrity token Anthropic emits with extended-thinking responses and +/// requires echoed back verbatim on follow-up turns. It is optional because +/// other providers (OpenAI-compatible APIs) do not produce it. +pub const ThinkingBlock = struct { + text: TextualBlock = .empty, + signature: ?[]const u8 = null, + + pub fn deinit(self: *ThinkingBlock, alloc: Allocator) void { + self.text.deinit(alloc); + if (self.signature) |sig| alloc.free(sig); + } +}; + pub const ToolUseBlock = struct { id: []const u8, name: []const u8, @@ -37,13 +53,16 @@ pub const ToolResultBlock = struct { pub const ContentBlock = union(enum) { Text: TextualBlock, - Thinking: TextualBlock, + Thinking: ThinkingBlock, ToolUse: ToolUseBlock, ToolResult: ToolResultBlock, pub fn deinit(self: *ContentBlock, alloc: Allocator) void { switch (self.*) { - inline else => |*b| b.deinit(alloc), + .Text => |*b| b.deinit(alloc), + .Thinking => |*b| b.deinit(alloc), + .ToolUse => |*b| b.deinit(alloc), + .ToolResult => |*b| b.deinit(alloc), } } }; @@ -175,11 +194,11 @@ test "ContentBlock - Thinking variant" { defer conv.deinit(); try conv.addAssistantMessage(&.{ - .{ .Thinking = try textualBlockFromSlice(allocator, "hmm...") }, + .{ .Thinking = .{ .text = try textualBlockFromSlice(allocator, "hmm...") } }, .{ .Text = try textualBlockFromSlice(allocator, "answer") }, }); try std.testing.expectEqual(@as(usize, 2), conv.messages.items[0].content.items.len); - try std.testing.expectEqualStrings("hmm...", conv.messages.items[0].content.items[0].Thinking.items); + try std.testing.expectEqualStrings("hmm...", conv.messages.items[0].content.items[0].Thinking.text.items); try std.testing.expectEqualStrings("answer", conv.messages.items[0].content.items[1].Text.items); } diff --git a/libpanto/src/openai_chat_json.zig b/libpanto/src/openai_chat_json.zig index 79cccac..465d951 100644 --- a/libpanto/src/openai_chat_json.zig +++ b/libpanto/src/openai_chat_json.zig @@ -28,7 +28,7 @@ pub const StreamDelta = struct { /// The caller owns the returned slice (allocated with `allocator`). pub fn serializeRequest( allocator: Allocator, - cfg: *const config_mod.Config, + cfg: *const config_mod.OpenAIChatConfig, conv: *const conversation.Conversation, ) ![]u8 { var aw: Writer.Allocating = .init(allocator); @@ -164,9 +164,8 @@ pub fn parseStreamEvent(allocator: Allocator, payload: []const u8) !ParsedDelta const testing = std.testing; -fn testConfig(model: []const u8) config_mod.Config { +fn testConfig(model: []const u8) config_mod.OpenAIChatConfig { return .{ - .api_style = .openai_chat, .api_key = "k", .base_url = "u", .model = model, @@ -210,7 +209,7 @@ test "serializeRequest - assistant Thinking blocks are stripped from outbound hi defer conv.deinit(); try conv.addAssistantMessage(&.{ - .{ .Thinking = try conversation.textualBlockFromSlice(allocator, "thinking step") }, + .{ .Thinking = .{ .text = try conversation.textualBlockFromSlice(allocator, "thinking step") } }, .{ .Text = try conversation.textualBlockFromSlice(allocator, "answer here") }, }); diff --git a/libpanto/src/provider.zig b/libpanto/src/provider.zig index 16c3dd6..6d97bb7 100644 --- a/libpanto/src/provider.zig +++ b/libpanto/src/provider.zig @@ -1,5 +1,6 @@ const std = @import("std"); const conversation = @import("conversation.zig"); +const config_mod = @import("config.zig"); pub const ContentBlockType = enum { Text, @@ -73,6 +74,35 @@ pub const Provider = struct { ptr: *anyopaque, vtable: *const ProviderVTable, + /// Construct the concrete provider implementation that matches `cfg`'s + /// API style, heap-allocate it, and return a `Provider` interface bound + /// to it. `deinit` tears down the impl and frees the allocation. + /// + /// The concrete provider types (`OpenAIChatProvider`, etc.) are + /// implementation details; callers interact only with this interface. + pub fn init( + allocator: std.mem.Allocator, + io: std.Io, + cfg: config_mod.Config, + ) !Provider { + // Imported lazily to break the circular module graph: + // provider.zig <- provider_openai_chat.zig <- provider.zig. + const provider_openai_chat = @import("provider_openai_chat.zig"); + const provider_anthropic_messages = @import("provider_anthropic_messages.zig"); + switch (cfg) { + .openai_chat => |c| { + const impl = try allocator.create(provider_openai_chat.OpenAIChatProvider); + impl.* = provider_openai_chat.OpenAIChatProvider.init(allocator, io, c); + return impl.provider(); + }, + .anthropic_messages => |c| { + const impl = try allocator.create(provider_anthropic_messages.AnthropicMessagesProvider); + impl.* = provider_anthropic_messages.AnthropicMessagesProvider.init(allocator, io, c); + return impl.provider(); + }, + } + } + pub fn streamStep(self: Provider, conv: *conversation.Conversation, receiver: *Receiver) anyerror!void { return self.vtable.streamStep(self.ptr, conv, receiver); } diff --git a/libpanto/src/provider_anthropic_messages.zig b/libpanto/src/provider_anthropic_messages.zig new file mode 100644 index 0000000..2914b6e --- /dev/null +++ b/libpanto/src/provider_anthropic_messages.zig @@ -0,0 +1,828 @@ +//! Anthropic Messages API streaming provider. +//! +//! Wire format reference: +//! https://platform.claude.com/docs/en/build-with-claude/streaming +//! +//! Responsibilities: +//! - Convert `Conversation` → request JSON (delegated to anthropic_messages_json.zig) +//! - POST to `{base_url}/v1/messages` with `stream: true` +//! - Read the chunked body, feed bytes through SSEParser +//! - Parse each event payload, drive a thin assembly loop, and emit Receiver +//! callbacks. Anthropic gives us explicit block boundaries, so no +//! state-machine inference is needed. +//! - 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("anthropic_messages_json.zig"); +const config_mod = @import("config.zig"); + +pub const AnthropicMessagesProvider = struct { + allocator: Allocator, + io: Io, + config: config_mod.AnthropicMessagesConfig, + http_client: http.Client, + + pub fn init( + allocator: Allocator, + io: Io, + cfg: config_mod.AnthropicMessagesConfig, + ) AnthropicMessagesProvider { + return .{ + .allocator = allocator, + .io = io, + .config = cfg, + .http_client = .{ .allocator = allocator, .io = io }, + }; + } + + pub fn deinit(self: *AnthropicMessagesProvider) void { + self.http_client.deinit(); + } + + pub fn provider(self: *AnthropicMessagesProvider) 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: *AnthropicMessagesProvider = @ptrCast(@alignCast(ptr)); + return self.streamStep(conv, receiver); + } + + /// Called via the `Provider` interface. Tears down the impl AND frees + /// its heap allocation, since `Provider.init` is the one that allocated + /// it. Direct stack-allocated users (tests, embedders) call `deinit` + /// themselves and never hit this path. + fn vtableDeinit(ptr: *anyopaque) void { + const self: *AnthropicMessagesProvider = @ptrCast(@alignCast(ptr)); + const allocator = self.allocator; + self.deinit(); + allocator.destroy(self); + } + + pub fn streamStep( + self: *AnthropicMessagesProvider, + conv: *conversation.Conversation, + receiver: *provider_mod.Receiver, + ) !void { + // Outer wrapper guarantees `onError` is called exactly once if + // anything fails. + self.streamStepInner(conv, receiver) catch |err| { + receiver.onError(err); + return err; + }; + } + + fn streamStepInner( + self: *AnthropicMessagesProvider, + conv: *conversation.Conversation, + receiver: *provider_mod.Receiver, + ) !void { + const url = try std.fmt.allocPrint( + self.allocator, + "{s}/v1/messages", + .{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); + defer self.allocator.free(body); + + const extra_headers = [_]http.Header{ + .{ .name = "content-type", .value = "application/json" }, + .{ .name = "accept", .value = "text/event-stream" }, + .{ .name = "x-api-key", .value = self.config.api_key }, + .{ .name = "anthropic-version", .value = self.config.api_version }, + }; + + var req = try self.http_client.request(.POST, uri, .{ + .extra_headers = &extra_headers, + // Disable compression: gzip buffers small SSE frames, defeating + // the streaming property we paid for `stream: true` to get. + .headers = .{ .accept_encoding = .{ .override = "identity" } }, + .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(); + + var redirect_buf: [1024]u8 = undefined; + var response = try req.receiveHead(&redirect_buf); + + if (@intFromEnum(response.head.status) >= 400) { + 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("anthropic_messages HTTP {d}: {s}", .{ + @intFromEnum(response.head.status), + err_buf.items, + }); + return error.HttpError; + } + + 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(); + + // Use `readVec` so we return to the event loop as soon as *any* + // bytes arrive, rather than waiting for the buffer to fill. + // `readSliceShort` blocks until EOF or full, which defeats streaming. + var chunk: [4096]u8 = undefined; + var vecs: [1][]u8 = .{&chunk}; + while (true) { + const n = body_reader.readVec(&vecs) catch |err| switch (err) { + error.EndOfStream => break, + else => return err, + }; + if (n == 0) break; + + const events = try parser.feed(chunk[0..n]); + defer parser.freeEvents(events); + + for (events) |ev_payload| { + try handleEvent(self.allocator, ev_payload, &state, receiver); + if (state.end_of_stream) { + try state.finalize(receiver, conv); + return; + } + } + } + + // Stream ended without an explicit message_stop. Finalize anyway. + try state.finalize(receiver, conv); + } +}; + +/// State maintained across the streaming response. +/// +/// Anthropic gives us explicit block boundaries (`content_block_start` / +/// `content_block_stop`), so we don't need to infer transitions like +/// `provider_openai_chat` does. We just track the currently-open block. +const StreamState = struct { + allocator: Allocator, + started: bool = false, + end_of_stream: bool = false, + finalized: bool = false, + + /// The block currently being assembled (if any). + active: ?ActiveBlock = null, + + /// Assembled blocks for the final message, in stream order. + blocks: std.ArrayList(conversation.ContentBlock) = .empty, + + const ActiveBlock = struct { + /// Index reported on the wire (Anthropic's content-array index). + wire_index: usize, + kind: BlockKind, + text_buf: conversation.TextualBlock = .empty, + signature: ?[]const u8 = null, + }; + + const BlockKind = enum { text, thinking, unsupported }; + + fn init(allocator: Allocator) StreamState { + return .{ .allocator = allocator }; + } + + fn deinit(self: *StreamState) void { + if (self.active) |*a| { + a.text_buf.deinit(self.allocator); + if (a.signature) |sig| self.allocator.free(sig); + } + for (self.blocks.items) |*b| b.deinit(self.allocator); + self.blocks.deinit(self.allocator); + } + + fn ensureStarted(self: *StreamState, receiver: *provider_mod.Receiver) !void { + if (self.started) return; + self.started = true; + try receiver.onMessageStart(.assistant); + } + + fn openBlock( + self: *StreamState, + receiver: *provider_mod.Receiver, + wire_index: usize, + kind: BlockKind, + tool_meta: ?provider_mod.BlockMeta, + ) !void { + // Defensive: if a prior block didn't get an explicit stop, drop it. + if (self.active != null) { + self.discardActive(); + } + self.active = .{ + .wire_index = wire_index, + .kind = kind, + }; + const block_type: ?provider_mod.ContentBlockType = switch (kind) { + .text => .Text, + .thinking => .Thinking, + .unsupported => null, + }; + if (block_type) |bt| { + try receiver.onBlockStart(bt, wire_index, tool_meta); + } + } + + fn appendTextDelta( + self: *StreamState, + receiver: *provider_mod.Receiver, + delta: []const u8, + ) !void { + const a = &(self.active orelse return); + if (a.kind == .unsupported) return; + try a.text_buf.appendSlice(self.allocator, delta); + try receiver.onContentDelta(a.wire_index, delta); + } + + fn setSignature(self: *StreamState, sig: []const u8) !void { + const a = &(self.active orelse return); + if (a.signature) |old| self.allocator.free(old); + a.signature = try self.allocator.dupe(u8, sig); + } + + /// Close the active block: append it to `blocks` and emit onBlockComplete. + fn closeBlock( + self: *StreamState, + receiver: *provider_mod.Receiver, + ) !void { + var a = self.active orelse return; + self.active = null; + + if (a.kind == .unsupported) { + // Drop unsupported blocks (e.g. tool_use until phase 3+). + a.text_buf.deinit(self.allocator); + if (a.signature) |sig| self.allocator.free(sig); + return; + } + + const block: conversation.ContentBlock = switch (a.kind) { + .text => blk: { + if (a.signature) |sig| self.allocator.free(sig); + break :blk .{ .Text = a.text_buf }; + }, + .thinking => .{ .Thinking = .{ + .text = a.text_buf, + .signature = a.signature, + } }, + .unsupported => unreachable, + }; + + try self.blocks.append(self.allocator, block); + try receiver.onBlockComplete(a.wire_index, self.blocks.items[self.blocks.items.len - 1]); + } + + /// Drop the active block without emitting a completion callback. + /// Used when an unexpected `content_block_start` arrives before the + /// previous block closed. + fn discardActive(self: *StreamState) void { + if (self.active) |*a| { + a.text_buf.deinit(self.allocator); + if (a.signature) |sig| self.allocator.free(sig); + self.active = null; + } + } + + fn finalize( + self: *StreamState, + receiver: *provider_mod.Receiver, + conv: *conversation.Conversation, + ) !void { + if (self.finalized) return; + self.finalized = true; + + if (self.active != null) { + // Wire dropped us mid-block. Close it cleanly anyway. + try self.closeBlock(receiver); + } + + const moved_blocks = try self.blocks.toOwnedSlice(self.allocator); + defer self.allocator.free(moved_blocks); + + try conv.addAssistantMessage(moved_blocks); + + 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(); + + switch (parsed.event) { + .message_start => { + try state.ensureStarted(receiver); + }, + .content_block_start => |s| { + try state.ensureStarted(receiver); + const kind: StreamState.BlockKind = switch (s.kind) { + .text => .text, + .thinking => .thinking, + .tool_use, .unknown => .unsupported, + }; + // tool_use meta will start being honored in phase 3+. For now we + // mark the block unsupported and drop it. + try state.openBlock(receiver, s.index, kind, null); + }, + .content_block_delta => |d| { + if (d.text_delta) |t| try state.appendTextDelta(receiver, t); + if (d.thinking_delta) |t| try state.appendTextDelta(receiver, t); + if (d.signature_delta) |sig| try state.setSignature(sig); + // input_json_delta: phase 3+. + }, + .content_block_stop => { + try state.closeBlock(receiver); + }, + .message_delta => { + // We don't act on stop_reason directly; message_stop is the + // authoritative end-of-stream signal. + }, + .message_stop => { + state.end_of_stream = true; + }, + .ping => {}, + .@"error" => |e| { + if (!@import("builtin").is_test) { + std.log.err("anthropic stream error: {?s}: {?s}", .{ e.kind, e.message }); + } + return error.StreamError; + }, + .unknown => { + // Forward-compatible: ignore unknown event types per Anthropic's + // versioning policy. + }, + } +} + +// ----------------------------------------------------------------------------- +// Tests +// ----------------------------------------------------------------------------- + +const testing = std.testing; + +/// Recording Receiver that captures the callback sequence for assertions. +const RecordingReceiver = struct { + allocator: Allocator, + events: std.ArrayList(Event) = .empty, + + const Event = union(enum) { + message_start: conversation.MessageRole, + block_start: struct { + kind: provider_mod.ContentBlockType, + index: usize, + }, + delta: struct { + index: usize, + bytes: []const u8, // owned copy + }, + block_complete: struct { + index: usize, + kind: provider_mod.ContentBlockType, + text: []const u8, // owned copy + signature: ?[]const u8 = null, // owned copy when present + }, + message_complete, + err: anyerror, + }; + + fn init(allocator: Allocator) RecordingReceiver { + return .{ .allocator = allocator }; + } + + fn deinit(self: *RecordingReceiver) void { + for (self.events.items) |ev| { + switch (ev) { + .delta => |d| self.allocator.free(d.bytes), + .block_complete => |b| { + self.allocator.free(b.text); + if (b.signature) |s| self.allocator.free(s); + }, + else => {}, + } + } + self.events.deinit(self.allocator); + } + + fn receiver(self: *RecordingReceiver) provider_mod.Receiver { + return .{ .ptr = self, .vtable = &vt }; + } + + const vt: provider_mod.ReceiverVTable = .{ + .onMessageStart = onMessageStart, + .onBlockStart = onBlockStart, + .onContentDelta = onContentDelta, + .onBlockComplete = onBlockComplete, + .onMessageComplete = onMessageComplete, + .onError = onError, + }; + + fn onMessageStart(ptr: *anyopaque, role: conversation.MessageRole) anyerror!void { + const self: *RecordingReceiver = @ptrCast(@alignCast(ptr)); + try self.events.append(self.allocator, .{ .message_start = role }); + } + fn onBlockStart( + ptr: *anyopaque, + bt: provider_mod.ContentBlockType, + idx: usize, + _: ?provider_mod.BlockMeta, + ) anyerror!void { + const self: *RecordingReceiver = @ptrCast(@alignCast(ptr)); + try self.events.append(self.allocator, .{ .block_start = .{ .kind = bt, .index = idx } }); + } + fn onContentDelta(ptr: *anyopaque, idx: usize, delta: []const u8) anyerror!void { + const self: *RecordingReceiver = @ptrCast(@alignCast(ptr)); + const copy = try self.allocator.dupe(u8, delta); + try self.events.append(self.allocator, .{ .delta = .{ .index = idx, .bytes = copy } }); + } + fn onBlockComplete( + ptr: *anyopaque, + idx: usize, + block: conversation.ContentBlock, + ) anyerror!void { + const self: *RecordingReceiver = @ptrCast(@alignCast(ptr)); + switch (block) { + .Text => |tb| { + const txt = try self.allocator.dupe(u8, tb.items); + try self.events.append(self.allocator, .{ .block_complete = .{ + .kind = .Text, + .index = idx, + .text = txt, + } }); + }, + .Thinking => |tb| { + const txt = try self.allocator.dupe(u8, tb.text.items); + const sig = if (tb.signature) |s| try self.allocator.dupe(u8, s) else null; + try self.events.append(self.allocator, .{ .block_complete = .{ + .kind = .Thinking, + .index = idx, + .text = txt, + .signature = sig, + } }); + }, + else => {}, + } + } + fn onMessageComplete(ptr: *anyopaque, _: conversation.Message) anyerror!void { + const self: *RecordingReceiver = @ptrCast(@alignCast(ptr)); + try self.events.append(self.allocator, .message_complete); + } + fn onError(ptr: *anyopaque, err: anyerror) void { + const self: *RecordingReceiver = @ptrCast(@alignCast(ptr)); + self.events.append(self.allocator, .{ .err = err }) catch {}; + } +}; + +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| { + try handleEvent(allocator, payload, &state, receiver); + if (state.end_of_stream) break; + } + try state.finalize(receiver, conv); +} + +test "streams a text-only turn end-to-end" { + const allocator = testing.allocator; + + var conv = conversation.Conversation.init(allocator); + defer conv.deinit(); + try conv.addUserMessage("hello"); + + var rec = RecordingReceiver.init(allocator); + defer rec.deinit(); + var recv = rec.receiver(); + + const events = [_][]const u8{ + \\{"type":"message_start","message":{"id":"msg_1","type":"message","role":"assistant","content":[],"model":"claude"}} + , + \\{"type":"content_block_start","index":0,"content_block":{"type":"text","text":""}} + , + \\{"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"Hello"}} + , + \\{"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"!"}} + , + \\{"type":"content_block_stop","index":0} + , + \\{"type":"message_delta","delta":{"stop_reason":"end_turn"}} + , + \\{"type":"message_stop"} + , + }; + + try runStreamedTurn(allocator, &conv, &recv, &events); + + // Conversation now holds the assistant reply. + try testing.expectEqual(@as(usize, 2), conv.messages.items.len); + try testing.expectEqual(conversation.MessageRole.assistant, conv.messages.items[1].role); + try testing.expectEqual(@as(usize, 1), conv.messages.items[1].content.items.len); + try testing.expectEqualStrings( + "Hello!", + conv.messages.items[1].content.items[0].Text.items, + ); + + // Callback sequence: msg_start, block_start, delta, delta, block_complete, msg_complete. + try testing.expectEqual(@as(usize, 6), rec.events.items.len); + try testing.expectEqual(conversation.MessageRole.assistant, rec.events.items[0].message_start); + try testing.expectEqual(provider_mod.ContentBlockType.Text, rec.events.items[1].block_start.kind); + try testing.expectEqualStrings("Hello", rec.events.items[2].delta.bytes); + try testing.expectEqualStrings("!", rec.events.items[3].delta.bytes); + try testing.expectEqualStrings("Hello!", rec.events.items[4].block_complete.text); + try testing.expectEqual(@as(@TypeOf(rec.events.items[5]), .message_complete), rec.events.items[5]); +} + +test "captures thinking signature for round-trip" { + const allocator = testing.allocator; + + var conv = conversation.Conversation.init(allocator); + defer conv.deinit(); + try conv.addUserMessage("solve"); + + var rec = RecordingReceiver.init(allocator); + defer rec.deinit(); + var recv = rec.receiver(); + + const events = [_][]const u8{ + \\{"type":"message_start","message":{"role":"assistant"}} + , + \\{"type":"content_block_start","index":0,"content_block":{"type":"thinking","thinking":"","signature":""}} + , + \\{"type":"content_block_delta","index":0,"delta":{"type":"thinking_delta","thinking":"step one"}} + , + \\{"type":"content_block_delta","index":0,"delta":{"type":"thinking_delta","thinking":" step two"}} + , + \\{"type":"content_block_delta","index":0,"delta":{"type":"signature_delta","signature":"EqQBabc"}} + , + \\{"type":"content_block_stop","index":0} + , + \\{"type":"content_block_start","index":1,"content_block":{"type":"text","text":""}} + , + \\{"type":"content_block_delta","index":1,"delta":{"type":"text_delta","text":"answer"}} + , + \\{"type":"content_block_stop","index":1} + , + \\{"type":"message_stop"} + , + }; + + try runStreamedTurn(allocator, &conv, &recv, &events); + + // The assistant message has Thinking + Text, with signature on the Thinking. + const asst = conv.messages.items[1]; + try testing.expectEqual(@as(usize, 2), asst.content.items.len); + try testing.expectEqualStrings("step one step two", asst.content.items[0].Thinking.text.items); + try testing.expectEqualStrings("EqQBabc", asst.content.items[0].Thinking.signature.?); + try testing.expectEqualStrings("answer", asst.content.items[1].Text.items); +} + +test "signature-only thinking block (display omitted)" { + // Anthropic emits a thinking block with only a signature_delta when + // `display: "omitted"` is configured. Verify we still capture the + // signature with empty thinking text. + const allocator = testing.allocator; + + var conv = conversation.Conversation.init(allocator); + defer conv.deinit(); + try conv.addUserMessage("hi"); + + var rec = RecordingReceiver.init(allocator); + defer rec.deinit(); + var recv = rec.receiver(); + + const events = [_][]const u8{ + \\{"type":"message_start","message":{"role":"assistant"}} + , + \\{"type":"content_block_start","index":0,"content_block":{"type":"thinking","thinking":"","signature":""}} + , + \\{"type":"content_block_delta","index":0,"delta":{"type":"signature_delta","signature":"sig123"}} + , + \\{"type":"content_block_stop","index":0} + , + \\{"type":"content_block_start","index":1,"content_block":{"type":"text","text":""}} + , + \\{"type":"content_block_delta","index":1,"delta":{"type":"text_delta","text":"hi back"}} + , + \\{"type":"content_block_stop","index":1} + , + \\{"type":"message_stop"} + , + }; + + try runStreamedTurn(allocator, &conv, &recv, &events); + + const asst = conv.messages.items[1]; + try testing.expectEqual(@as(usize, 2), asst.content.items.len); + try testing.expectEqualStrings("", asst.content.items[0].Thinking.text.items); + try testing.expectEqualStrings("sig123", asst.content.items[0].Thinking.signature.?); +} + +test "ping and unknown events are ignored" { + const allocator = testing.allocator; + + var conv = conversation.Conversation.init(allocator); + defer conv.deinit(); + try conv.addUserMessage("hi"); + + var rec = RecordingReceiver.init(allocator); + defer rec.deinit(); + var recv = rec.receiver(); + + const events = [_][]const u8{ + \\{"type":"message_start","message":{"role":"assistant"}} + , + \\{"type":"ping"} + , + \\{"type":"content_block_start","index":0,"content_block":{"type":"text","text":""}} + , + \\{"type":"ping"} + , + \\{"type":"future_event_type","whatever":true} + , + \\{"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"ok"}} + , + \\{"type":"content_block_stop","index":0} + , + \\{"type":"message_stop"} + , + }; + + try runStreamedTurn(allocator, &conv, &recv, &events); + + try testing.expectEqualStrings( + "ok", + conv.messages.items[1].content.items[0].Text.items, + ); +} + +test "tool_use blocks are dropped (deferred to phase 3)" { + // A tool_use block is started, receives input_json_delta events, and + // stops. Phase 2 should silently drop it without crashing. + const allocator = testing.allocator; + + var conv = conversation.Conversation.init(allocator); + defer conv.deinit(); + try conv.addUserMessage("use a tool"); + + var rec = RecordingReceiver.init(allocator); + defer rec.deinit(); + var recv = rec.receiver(); + + const events = [_][]const u8{ + \\{"type":"message_start","message":{"role":"assistant"}} + , + \\{"type":"content_block_start","index":0,"content_block":{"type":"tool_use","id":"tu_1","name":"calc","input":{}}} + , + \\{"type":"content_block_delta","index":0,"delta":{"type":"input_json_delta","partial_json":"{\"x\":1}"}} + , + \\{"type":"content_block_stop","index":0} + , + \\{"type":"content_block_start","index":1,"content_block":{"type":"text","text":""}} + , + \\{"type":"content_block_delta","index":1,"delta":{"type":"text_delta","text":"done"}} + , + \\{"type":"content_block_stop","index":1} + , + \\{"type":"message_stop"} + , + }; + + try runStreamedTurn(allocator, &conv, &recv, &events); + + // Only the text block survives. + const asst = conv.messages.items[1]; + try testing.expectEqual(@as(usize, 1), asst.content.items.len); + try testing.expectEqualStrings("done", asst.content.items[0].Text.items); +} + +test "error event propagates as Zig error" { + const allocator = testing.allocator; + + var conv = conversation.Conversation.init(allocator); + defer conv.deinit(); + try conv.addUserMessage("hi"); + + var rec = RecordingReceiver.init(allocator); + defer rec.deinit(); + var recv = rec.receiver(); + + var state: StreamState = .init(allocator); + defer state.deinit(); + + try handleEvent( + allocator, + \\{"type":"message_start","message":{"role":"assistant"}} + , + &state, + &recv, + ); + + const result = handleEvent( + allocator, + \\{"type":"error","error":{"type":"overloaded_error","message":"too busy"}} + , + &state, + &recv, + ); + try testing.expectError(error.StreamError, result); +} + +test "two streamed turns persist assistant replies in the conversation" { + // Same regression scenario as the openai_chat test, adapted to Anthropic. + const allocator = testing.allocator; + + var conv = conversation.Conversation.init(allocator); + defer conv.deinit(); + try conv.addSystemMessage("Be brief."); + try conv.addUserMessage("hi"); + + var rec = RecordingReceiver.init(allocator); + defer rec.deinit(); + var recv = rec.receiver(); + + const turn1 = [_][]const u8{ + \\{"type":"message_start","message":{"role":"assistant"}} + , + \\{"type":"content_block_start","index":0,"content_block":{"type":"text","text":""}} + , + \\{"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"Hi!"}} + , + \\{"type":"content_block_stop","index":0} + , + \\{"type":"message_stop"} + , + }; + try runStreamedTurn(allocator, &conv, &recv, &turn1); + + try conv.addUserMessage("what did you say?"); + + const turn2 = [_][]const u8{ + \\{"type":"message_start","message":{"role":"assistant"}} + , + \\{"type":"content_block_start","index":0,"content_block":{"type":"text","text":""}} + , + \\{"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"I said: Hi!"}} + , + \\{"type":"content_block_stop","index":0} + , + \\{"type":"message_stop"} + , + }; + try runStreamedTurn(allocator, &conv, &recv, &turn2); + + // system + user + assistant + user + assistant = 5 + try testing.expectEqual(@as(usize, 5), conv.messages.items.len); + try testing.expectEqualStrings( + "I said: Hi!", + conv.messages.items[4].content.items[0].Text.items, + ); +} diff --git a/libpanto/src/provider_openai_chat.zig b/libpanto/src/provider_openai_chat.zig index e1b9f81..c03d797 100644 --- a/libpanto/src/provider_openai_chat.zig +++ b/libpanto/src/provider_openai_chat.zig @@ -29,10 +29,10 @@ const ActiveBlock = enum { none, text, thinking }; pub const OpenAIChatProvider = struct { allocator: Allocator, io: Io, - config: config_mod.Config, + config: config_mod.OpenAIChatConfig, http_client: http.Client, - pub fn init(allocator: Allocator, io: Io, cfg: config_mod.Config) OpenAIChatProvider { + pub fn init(allocator: Allocator, io: Io, cfg: config_mod.OpenAIChatConfig) OpenAIChatProvider { return .{ .allocator = allocator, .io = io, @@ -64,9 +64,15 @@ pub const OpenAIChatProvider = struct { return self.streamStep(conv, receiver); } + /// Called via the `Provider` interface. Tears down the impl AND frees + /// its heap allocation, since `Provider.init` is the one that allocated + /// it. Direct stack-allocated users (tests, embedders) call `deinit` + /// themselves and never hit this path. fn vtableDeinit(ptr: *anyopaque) void { const self: *OpenAIChatProvider = @ptrCast(@alignCast(ptr)); + const allocator = self.allocator; self.deinit(); + allocator.destroy(self); } pub fn streamStep( @@ -120,6 +126,9 @@ pub const OpenAIChatProvider = struct { // response; we want to stream the body as it arrives. var req = try self.http_client.request(.POST, uri, .{ .extra_headers = &extra_headers, + // Disable compression: gzip buffers small SSE frames, defeating + // the streaming property we paid for `stream: true` to get. + .headers = .{ .accept_encoding = .{ .override = "identity" } }, .keep_alive = false, .redirect_behavior = .not_allowed, }); @@ -167,9 +176,16 @@ pub const OpenAIChatProvider = struct { var state: StreamState = .init(self.allocator); defer state.deinit(); + // Use `readVec` so we return to the event loop as soon as *any* + // bytes arrive, rather than waiting for the buffer to fill. + // `readSliceShort` blocks until EOF or full, which defeats streaming. var chunk: [4096]u8 = undefined; + var vecs: [1][]u8 = .{&chunk}; while (true) { - const n = try body_reader.readSliceShort(&chunk); + const n = body_reader.readVec(&vecs) catch |err| switch (err) { + error.EndOfStream => break, + else => return err, + }; if (n == 0) break; const events = try parser.feed(chunk[0..n]); @@ -232,7 +248,7 @@ const StreamState = struct { const block: conversation.ContentBlock = switch (self.active) { .text => .{ .Text = self.current_buf }, - .thinking => .{ .Thinking = self.current_buf }, + .thinking => .{ .Thinking = .{ .text = self.current_buf } }, .none => unreachable, }; // The buffer ownership has moved into `block`; replace with empty. diff --git a/libpanto/src/root.zig b/libpanto/src/root.zig index 9eb6d0e..a5cf8b5 100644 --- a/libpanto/src/root.zig +++ b/libpanto/src/root.zig @@ -4,11 +4,20 @@ pub const agent = @import("agent.zig"); pub const config = @import("config.zig"); pub const sse = @import("sse.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"); +// Internal modules. Not part of the public API — callers construct providers +// via `provider.Provider.init(allocator, io, config)`, which picks the right +// concrete implementation based on the config's API style. These are exposed +// here only so `refAllDecls` picks up their tests. +const openai_chat_json = @import("openai_chat_json.zig"); +const provider_openai_chat = @import("provider_openai_chat.zig"); +const anthropic_messages_json = @import("anthropic_messages_json.zig"); +const provider_anthropic_messages = @import("provider_anthropic_messages.zig"); test { - @import("std").testing.refAllDecls(@This()); + const std = @import("std"); + std.testing.refAllDecls(@This()); + std.testing.refAllDecls(openai_chat_json); + std.testing.refAllDecls(provider_openai_chat); + std.testing.refAllDecls(anthropic_messages_json); + std.testing.refAllDecls(provider_anthropic_messages); } -- cgit v1.3