diff options
Diffstat (limited to 'libpanto/src/anthropic_messages_json.zig')
| -rw-r--r-- | libpanto/src/anthropic_messages_json.zig | 638 |
1 files changed, 638 insertions, 0 deletions
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)); +} |