initial commit, moved libpanto over from the pantograph repo

1 files changed, 1267 insertions, 0 deletions

diff --git a/src/provider_anthropic_messages.zig b/src/provider_anthropic_messages.zig
new file mode 100644
index 0000000..955991a
--- /dev/null
+++ b/src/provider_anthropic_messages.zig
@@ -0,0 +1,1267 @@
+//! 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));
+}