path: root/libpanto/src/provider_anthropic_messages.zig

//! 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));
}