summaryrefslogtreecommitdiff
path: root/libpanto/src/provider_anthropic_messages.zig
diff options
context:
space:
mode:
Diffstat (limited to 'libpanto/src/provider_anthropic_messages.zig')
-rw-r--r--libpanto/src/provider_anthropic_messages.zig1267
1 files changed, 0 insertions, 1267 deletions
diff --git a/libpanto/src/provider_anthropic_messages.zig b/libpanto/src/provider_anthropic_messages.zig
deleted file mode 100644
index 955991a..0000000
--- a/libpanto/src/provider_anthropic_messages.zig
+++ /dev/null
@@ -1,1267 +0,0 @@
-//! 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`
-//! (the provider owns the current `/v1` suffix; a future wire revision
-//! would add a new `anthropic_messages_v2` API style rather than guessing
-//! from the configured base URL)
-//! - 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 stream_mod = @import("stream.zig");
-const sse_mod = @import("sse.zig");
-const json_mod = @import("anthropic_messages_json.zig");
-const config_mod = @import("config.zig");
-const tool_registry_mod = @import("tool_registry.zig");
-
-const Event = stream_mod.Event;
-const EventQueue = stream_mod.EventQueue;
-
-/// A single Anthropic Messages streaming request. Transient: constructed
-/// per `streamStep`, holds only borrowed state (allocator, io, the global
-/// HTTP client, and the active config). Carries nothing across requests.
-pub const AnthropicMessagesRequest = struct {
- allocator: Allocator,
- io: Io,
- config: *const config_mod.AnthropicMessagesConfig,
- http_client: *http.Client,
- /// Optional diagnostic side-channel; see `OpenAIChatRequest.diag`.
- diag: ?*provider_mod.ProviderDiagnostic = null,
-
- /// Open the streaming HTTP request and return a heap-allocated resumable
- /// response. Reads response headers (classifying any >=400 status) but
- /// does not pump the body — that happens lazily in
- /// `ResumableResponse.produce`. On success the caller owns the returned
- /// `*ResumableResponse` and must `deinit` it.
- pub fn open(
- self: *AnthropicMessagesRequest,
- conv: *conversation.Conversation,
- tools: *const provider_mod.ToolRegistry,
- ) !*ResumableResponse {
- const rr = try self.allocator.create(ResumableResponse);
- errdefer self.allocator.destroy(rr);
- rr.* = .{
- .allocator = self.allocator,
- .conv = conv,
- .parser = sse_mod.SSEParser.init(self.allocator),
- .state = .init(self.allocator),
- };
- rr.state.signature_origin = try conversation.SignatureOrigin.init(
- self.allocator,
- .anthropic_messages,
- self.config.base_url,
- self.config.model,
- );
- errdefer {
- rr.parser.deinit();
- rr.state.deinit();
- }
-
- const trimmed_base = std.mem.trim(u8, self.config.base_url, "/");
- const url = try std.fmt.allocPrint(
- self.allocator,
- "{s}/v1/messages",
- .{trimmed_base},
- );
- defer self.allocator.free(url);
-
- const uri = try Uri.parse(url);
-
- const body = try json_mod.serializeRequest(self.allocator, self.config, conv, tools);
- defer self.allocator.free(body);
- std.log.debug("anthropic_messages => {s}", .{body});
-
- // Build headers. Standard Anthropic-compatible backends use
- // `x-api-key`. OAuth-backed Anthropic-compatible providers (e.g.
- // Copilot) opt into `Authorization: Bearer ...` via
- // `config.use_bearer_auth`, which is derived from the configured auth
- // family rather than guessed from the URL/headers. The four base
- // headers are always present; the interleaved-thinking beta header is
- // added only when the config explicitly requests manual extended
- // thinking with interleaving. It is intentionally NOT sent for
- // `.adaptive` (interleaving is automatic there and the header causes
- // 400s on some backends) or `.disabled`.
- const use_bearer_auth = self.config.use_bearer_auth;
- const auth_value = if (use_bearer_auth)
- try std.fmt.allocPrint(
- self.allocator,
- "Bearer {s}",
- .{self.config.api_key},
- )
- else
- "";
- defer if (use_bearer_auth) self.allocator.free(auth_value);
- const auth_header: http.Header = if (use_bearer_auth)
- .{ .name = "authorization", .value = auth_value }
- else
- .{ .name = "x-api-key", .value = self.config.api_key };
- var headers_buf: [5]http.Header = .{
- .{ .name = "content-type", .value = "application/json" },
- .{ .name = "accept", .value = "text/event-stream" },
- auth_header,
- .{ .name = "anthropic-version", .value = self.config.api_version },
- undefined, // slot reserved for the optional beta header
- };
- const send_interleaved = self.config.thinking == .enabled and
- self.config.thinking_interleaved;
- if (send_interleaved) {
- headers_buf[4] = .{
- .name = "anthropic-beta",
- .value = "interleaved-thinking-2025-05-14",
- };
- }
- const base_headers = headers_buf[0..if (send_interleaved) @as(usize, 5) else @as(usize, 4)];
- // Merge any provider `extra_headers` onto the base set. Freed at the
- // end of `open` — after the request body has been flushed.
- const extra_headers = try provider_mod.mergeHeaders(
- self.allocator,
- base_headers,
- self.config.extra_headers,
- );
- defer self.allocator.free(extra_headers);
-
- rr.response = try provider_mod.sendRequest(self.http_client, uri, extra_headers, body, &rr.req);
- rr.req_open = true;
- errdefer {
- rr.req.deinit();
- rr.req_open = false;
- }
-
- // A >=400 status maps to a retryable/terminal provider error. Anthropic
- // rejects oversized requests with HTTP 400 + "prompt is too long",
- // which `classifyErrorResponse` maps to ContextOverflow (compact+retry).
- if (@intFromEnum(rr.response.head.status) >= 400) {
- return provider_mod.classifyErrorResponse(self.allocator, &rr.response, &rr.transfer_buf, self.diag, "anthropic_messages");
- }
-
- rr.body_reader = rr.response.reader(&rr.transfer_buf);
- return rr;
- }
-};
-
-/// A resumable Anthropic Messages streaming response. Owns the pinned HTTP
-/// request/response, the body reader's transfer buffer, the `SSEParser`, and
-/// the block-assembly `StreamState`. Must be heap-allocated and never moved:
-/// `body_reader` borrows `&self.response`.
-pub const ResumableResponse = struct {
- allocator: Allocator,
- conv: *conversation.Conversation,
- parser: sse_mod.SSEParser,
- state: StreamState,
-
- req: http.Client.Request = undefined,
- response: http.Client.Response = undefined,
- transfer_buf: [4096]u8 = undefined,
- body_reader: *std.Io.Reader = undefined,
- chunk: [4096]u8 = undefined,
-
- req_open: bool = false,
- done: bool = false,
-
- pub const ProduceStatus = provider_mod.ProviderStream.ProduceStatus;
-
- /// Wrap this response in the provider-agnostic `ProviderStream` the agent
- /// loop drives.
- pub fn providerStream(self: *ResumableResponse) provider_mod.ProviderStream {
- return .{ .ptr = self, .vtable = &vtable };
- }
-
- const vtable: provider_mod.ProviderStream.VTable = .{
- .produce = produceVT,
- .deinit = deinitVT,
- .last_error = lastErrorVT,
- };
-
- fn produceVT(ptr: *anyopaque, out: *EventQueue) anyerror!ProduceStatus {
- const self: *ResumableResponse = @ptrCast(@alignCast(ptr));
- return self.produce(out);
- }
-
- fn lastErrorVT(ptr: *anyopaque) ?[]const u8 {
- const self: *ResumableResponse = @ptrCast(@alignCast(ptr));
- return self.state.stream_error_message;
- }
-
- fn deinitVT(ptr: *anyopaque) void {
- const self: *ResumableResponse = @ptrCast(@alignCast(ptr));
- self.deinit();
- }
-
- pub fn deinit(self: *ResumableResponse) void {
- if (self.req_open) self.req.deinit();
- self.parser.deinit();
- self.state.deinit();
- self.allocator.destroy(self);
- }
-
- /// Pump the response: read one chunk, feed it through the SSE parser, and
- /// decode each SSE event into zero or more `Event`s appended to `out`.
- /// Returns `.more` if the caller should pump again, or
- /// `.response_complete` once `message_stop` (or EOF) is reached and the
- /// assistant message has been committed + a final `message_complete`
- /// pushed.
- pub fn produce(self: *ResumableResponse, out: *EventQueue) !ProduceStatus {
- if (self.done) return .response_complete;
-
- var vecs: [1][]u8 = .{&self.chunk};
- const n = self.body_reader.readVec(&vecs) catch |err| switch (err) {
- // Stream ended without an explicit message_stop. Finalize anyway.
- error.EndOfStream => {
- try self.finishStream(out);
- return .response_complete;
- },
- // Transport failure before the message completed: retryable.
- else => return error.ProviderStreamMalformed,
- };
- if (n == 0) return .more;
-
- const events = try self.parser.feed(self.chunk[0..n]);
- defer self.parser.freeEvents(events);
-
- for (events) |ev_payload| {
- std.log.debug("anthropic_messages <= {s}", .{ev_payload});
- try handleEvent(self.allocator, ev_payload, &self.state, out);
- if (self.state.end_of_stream) {
- try self.finishStream(out);
- return .response_complete;
- }
- }
- return .more;
- }
-
- fn finishStream(self: *ResumableResponse, out: *EventQueue) !void {
- if (self.done) return;
- self.done = true;
- try self.state.finalize(out, self.conv);
- }
-};
-
-/// 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,
-
- /// Accumulated token counts. Anthropic reports the input-side counts
- /// on `message_start.usage` and the final `output_tokens` (plus
- /// possibly updated input-side counts) on `message_delta.usage`.
- /// `usage_seen` distinguishes "genuinely all zero" from "never
- /// reported" — the former stamps a real `Usage` on
- /// `onMessageComplete`, the latter stamps `null`.
- usage: provider_mod.Usage = .{},
- usage_seen: bool = false,
- signature_origin: ?conversation.SignatureOrigin = null,
- stop_reason: ?[]u8 = null,
- /// Owned, human-readable description of a mid-stream `error` event
- /// (e.g. `"overloaded_error: Overloaded"`), surfaced to the agent via
- /// `ProviderStream.lastError` so the retry notice can show *why*.
- stream_error_message: ?[]u8 = null,
-
- 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,
- /// Populated for `.tool_use` blocks. Owned by this state until the
- /// block closes, at which point ownership transfers to the
- /// ToolUseBlock.
- tool_id: ?[]u8 = null,
- tool_name: ?[]u8 = null,
- };
-
- const BlockKind = enum { text, thinking, tool_use, 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);
- if (a.tool_id) |s| self.allocator.free(s);
- if (a.tool_name) |s| self.allocator.free(s);
- }
- for (self.blocks.items) |*b| b.deinit(self.allocator);
- self.blocks.deinit(self.allocator);
- if (self.signature_origin) |*o| o.deinit(self.allocator);
- if (self.stop_reason) |s| self.allocator.free(s);
- if (self.stream_error_message) |s| self.allocator.free(s);
- }
-
- fn ensureStarted(self: *StreamState, out: *EventQueue) !void {
- if (self.started) return;
- self.started = true;
- try out.push(.{ .message_start = .assistant });
- }
-
- /// Merge a wire-level usage snapshot into the accumulated counts.
- /// Missing fields mean "unchanged," not "reset to zero." Marks
- /// `usage_seen` so `finalize` delivers a non-null `Usage` to
- /// `onMessageComplete`.
- fn mergeUsage(self: *StreamState, partial: json_mod.StreamUsage) void {
- if (partial.input_tokens) |v| self.usage.input = v;
- if (partial.output_tokens) |v| self.usage.output = v;
- if (partial.cache_creation_input_tokens) |v| self.usage.cache_write = v;
- if (partial.cache_read_input_tokens) |v| self.usage.cache_read = v;
- if (partial.input_tokens != null or partial.output_tokens != null or
- partial.cache_creation_input_tokens != null or partial.cache_read_input_tokens != null)
- {
- self.usage_seen = true;
- }
- }
-
- fn openBlock(
- self: *StreamState,
- out: *EventQueue,
- wire_index: usize,
- kind: BlockKind,
- tool_id: ?[]const u8,
- tool_name: ?[]const u8,
- ) !void {
- // Defensive: if a prior block didn't get an explicit stop, drop it.
- if (self.active != null) {
- self.discardActive();
- }
- var ab: ActiveBlock = .{
- .wire_index = wire_index,
- .kind = kind,
- };
- // For tool_use blocks, capture the identity fields. Anthropic
- // delivers both whole on content_block_start. The wire name is
- // encoded (`__` for `.`); decode it here so everything downstream
- // — onToolDetails, the stored ContentBlock, session logs, and
- // dispatch — sees the internal (dotted) name. The decoded form is
- // never longer than the wire form.
- if (kind == .tool_use) {
- if (tool_id) |id| ab.tool_id = try self.allocator.dupe(u8, id);
- if (tool_name) |n| {
- // Decode `__` -> `.` into an exact-size owned buffer so the
- // stored slice is freeable as a whole allocation.
- const owned = try self.allocator.alloc(u8, n.len);
- errdefer self.allocator.free(owned);
- const decoded = tool_registry_mod.decodeName(owned, n);
- if (decoded.len == n.len) {
- ab.tool_name = owned;
- } else {
- ab.tool_name = try self.allocator.realloc(owned, decoded.len);
- }
- }
- }
- self.active = ab;
- const block_type: ?provider_mod.ContentBlockType = switch (kind) {
- .text => .Text,
- .thinking => .Thinking,
- .tool_use => .ToolUse,
- .unsupported => null,
- };
- if (block_type) |bt| {
- try out.push(.{ .block_start = .{ .block_type = bt, .index = wire_index } });
- // Anthropic delivers tool id+name whole on content_block_start,
- // so we can fire tool_details immediately — before any arg
- // deltas. If the wire was malformed and either field is
- // missing, skip: closeBlock will drop the block defensively.
- // id/name are owned by `ab` (stable) but we dupe into the queue
- // arena for a uniform borrow lifetime.
- if (kind == .tool_use) {
- if (ab.tool_id != null and ab.tool_name != null) {
- try out.push(.{ .tool_details = .{
- .index = wire_index,
- .id = try out.dupeBytes(ab.tool_id.?),
- .name = try out.dupeBytes(ab.tool_name.?),
- } });
- }
- }
- }
- }
-
- fn appendTextDelta(
- self: *StreamState,
- out: *EventQueue,
- delta: []const u8,
- ) !void {
- const a = &(self.active orelse return);
- if (a.kind == .unsupported) return;
- try a.text_buf.appendSlice(self.allocator, delta);
- // Dupe into the queue arena: `delta` borrows the transient SSE/JSON
- // payload that `produce` frees before `next()` reads the queue.
- try out.push(.{ .content_delta = .{
- .index = a.wire_index,
- .delta = try out.dupeBytes(delta),
- } });
- }
-
- /// Append a chunk of the streamed JSON arguments for the active
- /// tool_use block. No-op if the active block isn't a tool_use.
- fn appendInputJsonDelta(
- self: *StreamState,
- out: *EventQueue,
- delta: []const u8,
- ) !void {
- const a = &(self.active orelse return);
- if (a.kind != .tool_use) return;
- try a.text_buf.appendSlice(self.allocator, delta);
- try out.push(.{ .content_delta = .{
- .index = a.wire_index,
- .delta = try out.dupeBytes(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);
- }
-
- fn setStopReason(self: *StreamState, reason: ?[]const u8) !void {
- if (self.stop_reason) |old| self.allocator.free(old);
- self.stop_reason = if (reason) |r| try self.allocator.dupe(u8, r) else null;
- }
-
- /// Record a readable description of a mid-stream `error` event, combining
- /// the error `kind` and `message` into one owned string (either may be
- /// absent). Replaces any previous value.
- fn setStreamErrorMessage(self: *StreamState, kind: ?[]const u8, message: ?[]const u8) !void {
- if (self.stream_error_message) |old| self.allocator.free(old);
- self.stream_error_message = try provider_mod.formatStreamError(self.allocator, kind, message);
- }
-
- /// Close the active block: append it to `blocks` and emit block_complete.
- fn closeBlock(
- self: *StreamState,
- out: *EventQueue,
- ) !void {
- var a = self.active orelse return;
- self.active = null;
-
- if (a.kind == .unsupported) {
- a.text_buf.deinit(self.allocator);
- if (a.signature) |sig| self.allocator.free(sig);
- if (a.tool_id) |s| self.allocator.free(s);
- if (a.tool_name) |s| self.allocator.free(s);
- return;
- }
- // tool_use blocks require both id and name. If either is missing
- // (malformed stream), drop the block defensively.
- if (a.kind == .tool_use and (a.tool_id == null or a.tool_name == null)) {
- a.text_buf.deinit(self.allocator);
- if (a.tool_id) |s| self.allocator.free(s);
- if (a.tool_name) |s| self.allocator.free(s);
- 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,
- } },
- // An interrupted/malformed tool_use (incomplete or non-object
- // input JSON) is preserved as-is. The agent's dispatch path
- // detects invalid input and answers it with a synthetic error
- // ToolResult in the *following user message* — emitting a
- // ToolResult here would wrongly place it in this assistant
- // message, which Anthropic rejects.
- .tool_use => blk: {
- break :blk .{ .ToolUse = .{
- .id = a.tool_id.?,
- .name = a.tool_name.?,
- .input = a.text_buf,
- } };
- },
- .unsupported => unreachable,
- };
-
- try self.blocks.append(self.allocator, block);
- try out.push(.{ .block_complete = .{
- .index = a.wire_index,
- .block = 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);
- if (a.tool_id) |s| self.allocator.free(s);
- if (a.tool_name) |s| self.allocator.free(s);
- self.active = null;
- }
- }
-
- fn finalize(
- self: *StreamState,
- out: *EventQueue,
- conv: *conversation.Conversation,
- ) !void {
- if (self.finalized) return;
- self.finalized = true;
-
- if (self.active != null) {
- // Preserve an interrupted tool call so the agent can answer it
- // with a synthetic error ToolResult instead of invoking it.
- try self.closeBlock(out);
- }
-
- const moved_blocks = try self.blocks.toOwnedSlice(self.allocator);
- defer self.allocator.free(moved_blocks);
-
- const usage: ?provider_mod.Usage = if (self.usage_seen) self.usage else null;
- if (self.signature_origin) |origin| {
- try conversation.setThinkingOrigins(
- self.allocator,
- moved_blocks,
- origin.api_style,
- origin.base_url,
- origin.model,
- );
- }
- try conv.addAssistantMessage(moved_blocks, usage);
-
- const msg = conv.messages.items[conv.messages.items.len - 1];
- try out.push(.{ .message_complete = .{ .message = msg, .usage = usage } });
- }
-};
-
-fn handleEvent(
- allocator: Allocator,
- payload: []const u8,
- state: *StreamState,
- out: *EventQueue,
-) !void {
- var parsed = try json_mod.parseStreamEvent(allocator, payload);
- defer parsed.deinit();
-
- switch (parsed.event) {
- .message_start => |s| {
- try state.ensureStarted(out);
- state.mergeUsage(s.usage);
- },
- .content_block_start => |s| {
- try state.ensureStarted(out);
- const kind: StreamState.BlockKind = switch (s.kind) {
- .text => .text,
- .thinking => .thinking,
- .tool_use => .tool_use,
- .unknown => .unsupported,
- };
- try state.openBlock(out, s.index, kind, s.tool_id, s.tool_name);
- },
- .content_block_delta => |d| {
- if (d.text_delta) |t| try state.appendTextDelta(out, t);
- if (d.thinking_delta) |t| try state.appendTextDelta(out, t);
- if (d.signature_delta) |sig| try state.setSignature(sig);
- if (d.input_json_delta) |j| try state.appendInputJsonDelta(out, j);
- },
- .content_block_stop => |s| {
- if (state.active) |a| {
- if (a.wire_index == s.index) try state.closeBlock(out);
- }
- },
- .message_delta => |d| {
- try state.setStopReason(d.stop_reason);
- state.mergeUsage(d.usage);
- },
- .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 });
- }
- // Stash a readable description so the agent's retry notice can
- // explain *why* the stream failed instead of only showing the
- // bare error name. Owned by `state`; freed in `deinit`.
- state.setStreamErrorMessage(e.kind, e.message) catch {};
- // Mid-stream error event before the message was committed. Map
- // the common overload case to a dedicated retryable error so the
- // UI can say "overloaded" rather than "malformed stream"; other
- // kinds stay as the generic retryable malformed-stream error.
- if (e.kind) |k| {
- if (std.mem.eql(u8, k, "overloaded_error")) return error.ProviderOverloaded;
- }
- return error.ProviderStreamMalformed;
- },
- .unknown => {
- // Forward-compatible: ignore unknown event types per Anthropic's
- // versioning policy.
- },
- }
-}
-
-// -----------------------------------------------------------------------------
-// Tests
-// -----------------------------------------------------------------------------
-
-const testing = std.testing;
-
-/// Records the decoded pull `Event`s for assertions. Keeps the same typed
-/// schema the old RecordingReceiver exposed (message_start / block_start /
-/// delta / block_complete / message_complete), so the existing assertions
-/// are preserved verbatim. The recorder owns copies of all byte payloads
-/// (the queue arena is reset on full drain).
-const EventRecorder = struct {
- allocator: Allocator,
- events: std.ArrayList(Rec) = .empty,
-
- const Rec = 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: ?provider_mod.Usage,
- };
-
- fn init(allocator: Allocator) EventRecorder {
- return .{ .allocator = allocator };
- }
-
- fn deinit(self: *EventRecorder) 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);
- }
-
- /// Translate one pull `Event` into the recorder's schema. Tool identity
- /// (`tool_details`) is dropped: the anthropic tests assert tool-use
- /// blocks via the ContentBlock in conv after finalize, not via the
- /// event stream. Tool-arg `content_delta`s are also dropped here because
- /// the old recorder only recorded text/thinking deltas (it routed tool
- /// args through a separate path that didn't call onContentDelta in a way
- /// these tests observe) — we preserve that by only recording deltas for
- /// the currently text/thinking block. Since the recorder can't see block
- /// kind from a bare delta, we record every delta; the existing tests
- /// only assert delta bytes for text/thinking turns, so this is
- /// equivalent for them.
- fn record(self: *EventRecorder, ev: Event) !void {
- switch (ev) {
- .message_start => |role| try self.events.append(self.allocator, .{ .message_start = role }),
- .block_start => |b| try self.events.append(self.allocator, .{ .block_start = .{
- .kind = b.block_type,
- .index = b.index,
- } }),
- .content_delta => |d| {
- const copy = try self.allocator.dupe(u8, d.delta);
- try self.events.append(self.allocator, .{ .delta = .{ .index = d.index, .bytes = copy } });
- },
- .block_complete => |bc| switch (bc.block) {
- .Text => |tb| {
- const txt = try self.allocator.dupe(u8, tb.items);
- try self.events.append(self.allocator, .{ .block_complete = .{
- .kind = .Text,
- .index = bc.index,
- .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 = bc.index,
- .text = txt,
- .signature = sig,
- } });
- },
- else => {},
- },
- .message_complete => |m| try self.events.append(self.allocator, .{ .message_complete = m.usage }),
- else => {},
- }
- }
-};
-
-fn runStreamedTurn(
- allocator: Allocator,
- conv: *conversation.Conversation,
- rec: ?*EventRecorder,
- events: []const []const u8,
-) !void {
- var state: StreamState = .init(allocator);
- defer state.deinit();
-
- var queue = EventQueue.init(allocator);
- defer queue.deinit();
-
- for (events) |payload| {
- try handleEvent(allocator, payload, &state, &queue);
- if (state.end_of_stream) break;
- }
- try state.finalize(&queue, conv);
-
- while (queue.pop()) |ev| {
- if (rec) |r| try r.record(ev);
- }
-}
-
-/// Test helper: append a single-text user message. `addUserMessage` now
-/// takes a block slice (symmetric with `addAssistantMessage`); this wraps
-/// the common plain-text case the tests below use.
-fn addUserText(conv: *conversation.Conversation, text: []const u8) !void {
- const tb = try conversation.textualBlockFromSlice(conv.allocator, text);
- var block: conversation.ContentBlock = .{ .Text = tb };
- errdefer block.deinit(conv.allocator);
- try conv.addUserMessage(&.{block});
-}
-
-test "streams a text-only turn end-to-end" {
- const allocator = testing.allocator;
-
- var conv = conversation.Conversation.init(allocator);
- defer conv.deinit();
- try addUserText(&conv, "hello");
-
- var rec = EventRecorder.init(allocator);
- defer rec.deinit();
-
- 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, &rec, &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.expect(rec.events.items[5] == .message_complete);
- // No usage on the wire — the assertion is structural.
- try testing.expect(rec.events.items[5].message_complete == null);
-}
-
-test "anthropic: captures usage from message_start and message_delta on message_complete" {
- const allocator = testing.allocator;
-
- var conv = conversation.Conversation.init(allocator);
- defer conv.deinit();
- try addUserText(&conv, "hi");
-
- var rec = EventRecorder.init(allocator);
- defer rec.deinit();
-
- const events = [_][]const u8{
- // Initial input-side counts on message_start.
- \\{"type":"message_start","message":{"id":"m","type":"message","role":"assistant","content":[],"model":"claude","usage":{"input_tokens":100,"cache_creation_input_tokens":50,"cache_read_input_tokens":200}}}
- ,
- \\{"type":"content_block_start","index":0,"content_block":{"type":"text","text":""}}
- ,
- \\{"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"ok"}}
- ,
- \\{"type":"content_block_stop","index":0}
- ,
- // Final output count on message_delta.
- \\{"type":"message_delta","delta":{"stop_reason":"end_turn"},"usage":{"output_tokens":42}}
- ,
- \\{"type":"message_stop"}
- ,
- };
-
- try runStreamedTurn(allocator, &conv, &rec, &events);
-
- // Find the message_complete event and check its usage payload.
- var found: ?provider_mod.Usage = null;
- for (rec.events.items) |ev| {
- if (ev == .message_complete) found = ev.message_complete;
- }
- try testing.expect(found != null);
- const u = found.?;
- try testing.expectEqual(@as(u64, 100), u.input);
- try testing.expectEqual(@as(u64, 42), u.output);
- try testing.expectEqual(@as(u64, 200), u.cache_read);
- try testing.expectEqual(@as(u64, 50), u.cache_write);
- try testing.expectEqual(@as(u64, 0), u.reasoning); // Anthropic doesn't split reasoning separately.
-}
-
-test "anthropic: message_complete carries null usage when wire omits it" {
- const allocator = testing.allocator;
-
- var conv = conversation.Conversation.init(allocator);
- defer conv.deinit();
- try addUserText(&conv, "hi");
-
- var rec = EventRecorder.init(allocator);
- defer rec.deinit();
-
- const events = [_][]const u8{
- \\{"type":"message_start","message":{"id":"m","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":"hi"}}
- ,
- \\{"type":"content_block_stop","index":0}
- ,
- \\{"type":"message_delta","delta":{"stop_reason":"end_turn"}}
- ,
- \\{"type":"message_stop"}
- ,
- };
-
- try runStreamedTurn(allocator, &conv, &rec, &events);
-
- var saw_complete = false;
- for (rec.events.items) |ev| {
- if (ev == .message_complete) {
- saw_complete = true;
- try testing.expect(ev.message_complete == null);
- }
- }
- try testing.expect(saw_complete);
-}
-
-test "captures thinking signature for round-trip" {
- const allocator = testing.allocator;
-
- var conv = conversation.Conversation.init(allocator);
- defer conv.deinit();
- try addUserText(&conv, "solve");
-
- var rec = EventRecorder.init(allocator);
- defer rec.deinit();
-
- 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, &rec, &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 addUserText(&conv, "hi");
-
- var rec = EventRecorder.init(allocator);
- defer rec.deinit();
-
- 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, &rec, &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 addUserText(&conv, "hi");
-
- var rec = EventRecorder.init(allocator);
- defer rec.deinit();
-
- 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, &rec, &events);
-
- try testing.expectEqualStrings(
- "ok",
- conv.messages.items[1].content.items[0].Text.items,
- );
-}
-
-test "tool_use blocks are captured with id, name, and assembled input" {
- const allocator = testing.allocator;
-
- var conv = conversation.Conversation.init(allocator);
- defer conv.deinit();
- try addUserText(&conv, "use a tool");
-
- var rec = EventRecorder.init(allocator);
- defer rec.deinit();
-
- 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\":"}}
- ,
- \\{"type":"content_block_delta","index":0,"delta":{"type":"input_json_delta","partial_json":"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, &rec, &events);
-
- const asst = conv.messages.items[1];
- try testing.expectEqual(@as(usize, 2), asst.content.items.len);
-
- const tu = asst.content.items[0].ToolUse;
- try testing.expectEqualStrings("tu_1", tu.id);
- try testing.expectEqualStrings("calc", tu.name);
- try testing.expectEqualStrings("{\"x\":1}", tu.input.items);
-
- try testing.expectEqualStrings("done", asst.content.items[1].Text.items);
-}
-
-test "inbound wire tool name is decoded to dotted form" {
- // Anthropic delivers the (wire-encoded) name whole at
- // content_block_start; it is decoded to the internal dotted form for
- // the conversation, session logs, and dispatch.
- const allocator = testing.allocator;
-
- var conv = conversation.Conversation.init(allocator);
- defer conv.deinit();
- try addUserText(&conv, "use a tool");
-
- var rec = EventRecorder.init(allocator);
- defer rec.deinit();
-
- const events = [_][]const u8{
- \\{"type":"message_start","message":{"role":"assistant"}}
- ,
- \\{"type":"content_block_start","index":0,"content_block":{"type":"tool_use","id":"t1","name":"calc__sum","input":{}}}
- ,
- \\{"type":"content_block_stop","index":0}
- ,
- \\{"type":"message_stop"}
- ,
- };
-
- try runStreamedTurn(allocator, &conv, &rec, &events);
-
- const tu = conv.messages.items[1].content.items[0].ToolUse;
- try testing.expectEqualStrings("calc.sum", tu.name);
-}
-
-test "error event propagates as Zig error" {
- const allocator = testing.allocator;
-
- var conv = conversation.Conversation.init(allocator);
- defer conv.deinit();
- try addUserText(&conv, "hi");
-
- var queue = EventQueue.init(allocator);
- defer queue.deinit();
-
- var state: StreamState = .init(allocator);
- defer state.deinit();
-
- try handleEvent(
- allocator,
- \\{"type":"message_start","message":{"role":"assistant"}}
- ,
- &state,
- &queue,
- );
-
- const result = handleEvent(
- allocator,
- \\{"type":"error","error":{"type":"overloaded_error","message":"too busy"}}
- ,
- &state,
- &queue,
- );
- // `overloaded_error` maps to the dedicated retryable error, and the
- // provider's diagnostic is stashed for the agent's retry notice.
- try testing.expectError(error.ProviderOverloaded, result);
- try testing.expectEqualStrings("overloaded_error: too busy", state.stream_error_message.?);
-}
-
-test "non-overloaded error event stays malformed and stashes message" {
- const allocator = testing.allocator;
-
- var conv = conversation.Conversation.init(allocator);
- defer conv.deinit();
- try addUserText(&conv, "hi");
-
- var queue = EventQueue.init(allocator);
- defer queue.deinit();
-
- var state: StreamState = .init(allocator);
- defer state.deinit();
-
- const result = handleEvent(
- allocator,
- \\{"type":"error","error":{"type":"api_error","message":"boom"}}
- ,
- &state,
- &queue,
- );
- try testing.expectError(error.ProviderStreamMalformed, result);
- try testing.expectEqualStrings("api_error: boom", state.stream_error_message.?);
-}
-
-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 addUserText(&conv, "hi");
-
- var rec = EventRecorder.init(allocator);
- defer rec.deinit();
-
- 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, &rec, &turn1);
-
- try addUserText(&conv, "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, &rec, &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,
- );
-}
-
-/// Helper: whether this config sends Bearer auth instead of `x-api-key`.
-fn headerSliceUsesBearerAuth(cfg: *const config_mod.AnthropicMessagesConfig) bool {
- return cfg.use_bearer_auth;
-}
-
-/// Helper: build the header slice exactly as `open` does, given a config,
-/// and return whether the interleaved beta header is present.
-/// This lets us test the header-selection logic without a live HTTP connection.
-fn headerSliceIncludesInterleaved(cfg: *const config_mod.AnthropicMessagesConfig) bool {
- const send_interleaved = cfg.thinking == .enabled and cfg.thinking_interleaved;
- return send_interleaved;
-}
-
-test "oauth-backed anthropic auth uses bearer auth" {
- const cfg: config_mod.AnthropicMessagesConfig = .{
- .api_key = "k",
- .base_url = "https://api.individual.githubcopilot.com",
- .model = "claude-sonnet-4-5",
- .use_bearer_auth = true,
- };
- try testing.expect(headerSliceUsesBearerAuth(&cfg));
-}
-
-test "plain anthropic auth uses x-api-key" {
- const cfg: config_mod.AnthropicMessagesConfig = .{
- .api_key = "k",
- .base_url = "https://api.anthropic.com",
- .model = "claude-sonnet-4-5",
- };
- try testing.expect(!headerSliceUsesBearerAuth(&cfg));
-}
-
-test "interleaved beta header: enabled when thinking=.enabled and interleaved=true" {
- const cfg: config_mod.AnthropicMessagesConfig = .{
- .api_key = "k",
- .base_url = "u",
- .model = "m",
- .thinking = .enabled,
- .thinking_interleaved = true,
- };
- try testing.expect(headerSliceIncludesInterleaved(&cfg));
-}
-
-test "interleaved beta header: absent when thinking=.enabled and interleaved=false" {
- const cfg: config_mod.AnthropicMessagesConfig = .{
- .api_key = "k",
- .base_url = "u",
- .model = "m",
- .thinking = .enabled,
- .thinking_interleaved = false,
- };
- try testing.expect(!headerSliceIncludesInterleaved(&cfg));
-}
-
-test "interleaved beta header: absent when thinking=.adaptive even if interleaved=true" {
- const cfg: config_mod.AnthropicMessagesConfig = .{
- .api_key = "k",
- .base_url = "u",
- .model = "m",
- .thinking = .adaptive,
- .thinking_interleaved = true,
- };
- // .adaptive does not send the header; interleaving is automatic there.
- try testing.expect(!headerSliceIncludesInterleaved(&cfg));
-}
-
-test "interleaved beta header: absent when thinking=.disabled" {
- const cfg: config_mod.AnthropicMessagesConfig = .{
- .api_key = "k",
- .base_url = "u",
- .model = "m",
- .thinking = .disabled,
- .thinking_interleaved = true,
- };
- try testing.expect(!headerSliceIncludesInterleaved(&cfg));
-}
generated by cgit v1.3 (git 2.53.0) at 2026-07-09 07:15:00 +0000