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`
//!   - Read the chunked body, feed bytes through SSEParser
//!   - Parse each event payload, drive a thin assembly loop, and emit Receiver
//!     callbacks. Anthropic gives us explicit block boundaries, so no
//!     state-machine inference is needed.
//!   - Assemble the final Message and emit onMessageComplete.

const std = @import("std");
const Allocator = std.mem.Allocator;
const Io = std.Io;
const http = std.http;
const Uri = std.Uri;

const conversation = @import("conversation.zig");
const provider_mod = @import("provider.zig");
const sse_mod = @import("sse.zig");
const json_mod = @import("anthropic_messages_json.zig");
const config_mod = @import("config.zig");
const tool_registry_mod = @import("tool_registry.zig");

/// 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,

    pub fn streamStep(
        self: *AnthropicMessagesRequest,
        conv: *conversation.Conversation,
        tools: *const provider_mod.ToolRegistry,
        receiver: *provider_mod.Receiver,
    ) !void {
        // Outer wrapper guarantees `onError` is called exactly once if
        // anything fails.
        self.streamStepInner(conv, tools, receiver) catch |err| {
            receiver.onError(err);
            return err;
        };
    }

    fn streamStepInner(
        self: *AnthropicMessagesRequest,
        conv: *conversation.Conversation,
        tools: *const provider_mod.ToolRegistry,
        receiver: *provider_mod.Receiver,
    ) !void {
        const url = try std.fmt.allocPrint(
            self.allocator,
            "{s}/v1/messages",
            .{self.config.base_url},
        );
        defer self.allocator.free(url);

        const uri = try Uri.parse(url);

        const body = try json_mod.serializeRequest(self.allocator, self.config, conv, tools);
        defer self.allocator.free(body);

        const extra_headers = [_]http.Header{
            .{ .name = "content-type", .value = "application/json" },
            .{ .name = "accept", .value = "text/event-stream" },
            .{ .name = "x-api-key", .value = self.config.api_key },
            .{ .name = "anthropic-version", .value = self.config.api_version },
        };

        var req = try self.http_client.request(.POST, uri, .{
            .extra_headers = &extra_headers,
            // Disable compression: gzip buffers small SSE frames, defeating
            // the streaming property we paid for `stream: true` to get.
            .headers = .{ .accept_encoding = .{ .override = "identity" } },
            .keep_alive = false,
            .redirect_behavior = .not_allowed,
        });
        defer req.deinit();

        req.transfer_encoding = .{ .content_length = body.len };

        var send_buf: [4096]u8 = undefined;
        var bw = try req.sendBodyUnflushed(&send_buf);
        try bw.writer.writeAll(body);
        try bw.end();
        try req.connection.?.flush();

        var redirect_buf: [1024]u8 = undefined;
        var response = try req.receiveHead(&redirect_buf);

        if (@intFromEnum(response.head.status) >= 400) {
            var transfer_buf: [4096]u8 = undefined;
            const body_reader = response.reader(&transfer_buf);
            var err_buf: std.ArrayList(u8) = .empty;
            defer err_buf.deinit(self.allocator);
            var tmp: [1024]u8 = undefined;
            while (true) {
                const n = body_reader.readSliceShort(&tmp) catch break;
                if (n == 0) break;
                try err_buf.appendSlice(self.allocator, tmp[0..n]);
                if (err_buf.items.len > 16 * 1024) break;
            }
            std.log.err("anthropic_messages HTTP {d}: {s}", .{
                @intFromEnum(response.head.status),
                err_buf.items,
            });
            // Anthropic rejects oversized requests with HTTP 400 and a
            // "prompt is too long" message; surface as ContextOverflow so
            // the caller can compact and retry.
            if (@intFromEnum(response.head.status) == 400 and provider_mod.isContextOverflowBody(err_buf.items)) {
                return error.ContextOverflow;
            }
            return error.HttpError;
        }

        var transfer_buf: [4096]u8 = undefined;
        const body_reader = response.reader(&transfer_buf);

        var parser = sse_mod.SSEParser.init(self.allocator);
        defer parser.deinit();

        var state: StreamState = .init(self.allocator);
        defer state.deinit();

        // Use `readVec` so we return to the event loop as soon as *any*
        // bytes arrive, rather than waiting for the buffer to fill.
        // `readSliceShort` blocks until EOF or full, which defeats streaming.
        //
        // Per `std.Io.Reader.readVec` docs: `n == 0` does NOT mean EOF;
        // EOF is signalled only via `error.EndOfStream`. Breaking on
        // `n == 0` truncates the response mid-stream.
        var chunk: [4096]u8 = undefined;
        var vecs: [1][]u8 = .{&chunk};
        while (true) {
            const n = body_reader.readVec(&vecs) catch |err| switch (err) {
                error.EndOfStream => break,
                else => return err,
            };
            if (n == 0) continue;

            const events = try parser.feed(chunk[0..n]);
            defer parser.freeEvents(events);

            for (events) |ev_payload| {
                std.log.debug("anthropic_messages <= {s}", .{ev_payload});
                try handleEvent(self.allocator, ev_payload, &state, receiver);
                if (state.end_of_stream) {
                    try state.finalize(receiver, conv);
                    return;
                }
            }
        }

        // Stream ended without an explicit message_stop. Finalize anyway.
        try state.finalize(receiver, conv);
    }
};

/// State maintained across the streaming response.
///
/// Anthropic gives us explicit block boundaries (`content_block_start` /
/// `content_block_stop`), so we don't need to infer transitions like
/// `provider_openai_chat` does. We just track the currently-open block.
const StreamState = struct {
    allocator: Allocator,
    started: bool = false,
    end_of_stream: bool = false,
    finalized: bool = false,

    /// The block currently being assembled (if any).
    active: ?ActiveBlock = null,

    /// Assembled blocks for the final message, in stream order.
    blocks: std.ArrayList(conversation.ContentBlock) = .empty,

    /// 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,
    stop_reason: ?[]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.stop_reason) |s| self.allocator.free(s);
    }

    fn ensureStarted(self: *StreamState, receiver: *provider_mod.Receiver) !void {
        if (self.started) return;
        self.started = true;
        try receiver.onMessageStart(.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,
        receiver: *provider_mod.Receiver,
        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 receiver.onBlockStart(bt, wire_index);
            // Anthropic delivers tool id+name whole on content_block_start,
            // so we can fire onToolDetails immediately — before any arg
            // deltas. If the wire was malformed and either field is
            // missing, skip: closeBlock will drop the block defensively.
            if (kind == .tool_use) {
                if (ab.tool_id != null and ab.tool_name != null) {
                    try receiver.onToolDetails(wire_index, ab.tool_id.?, ab.tool_name.?);
                }
            }
        }
    }

    fn appendTextDelta(
        self: *StreamState,
        receiver: *provider_mod.Receiver,
        delta: []const u8,
    ) !void {
        const a = &(self.active orelse return);
        if (a.kind == .unsupported) return;
        try a.text_buf.appendSlice(self.allocator, delta);
        try receiver.onContentDelta(a.wire_index, delta);
    }

    /// 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,
        receiver: *provider_mod.Receiver,
        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 receiver.onContentDelta(a.wire_index, delta);
    }

    fn setSignature(self: *StreamState, sig: []const u8) !void {
        const a = &(self.active orelse return);
        if (a.signature) |old| self.allocator.free(old);
        a.signature = try self.allocator.dupe(u8, sig);
    }

    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;
    }

    /// Close the active block: append it to `blocks` and emit onBlockComplete.
    fn closeBlock(
        self: *StreamState,
        receiver: *provider_mod.Receiver,
    ) !void {
        var a = self.active orelse return;
        self.active = null;

        if (a.kind == .unsupported) {
            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 receiver.onBlockComplete(a.wire_index, self.blocks.items[self.blocks.items.len - 1]);
    }

    /// Drop the active block without emitting a completion callback.
    /// Used when an unexpected `content_block_start` arrives before the
    /// previous block closed.
    fn discardActive(self: *StreamState) void {
        if (self.active) |*a| {
            a.text_buf.deinit(self.allocator);
            if (a.signature) |sig| self.allocator.free(sig);
            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,
        receiver: *provider_mod.Receiver,
        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(receiver);
        }

        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;
        try conv.addAssistantMessageWithUsage(moved_blocks, usage);

        const msg = conv.messages.items[conv.messages.items.len - 1];
        try receiver.onMessageComplete(msg, usage);
    }
};

fn handleEvent(
    allocator: Allocator,
    payload: []const u8,
    state: *StreamState,
    receiver: *provider_mod.Receiver,
) !void {
    var parsed = try json_mod.parseStreamEvent(allocator, payload);
    defer parsed.deinit();

    switch (parsed.event) {
        .message_start => |s| {
            try state.ensureStarted(receiver);
            state.mergeUsage(s.usage);
        },
        .content_block_start => |s| {
            try state.ensureStarted(receiver);
            const kind: StreamState.BlockKind = switch (s.kind) {
                .text => .text,
                .thinking => .thinking,
                .tool_use => .tool_use,
                .unknown => .unsupported,
            };
            try state.openBlock(receiver, s.index, kind, s.tool_id, s.tool_name);
        },
        .content_block_delta => |d| {
            if (d.text_delta) |t| try state.appendTextDelta(receiver, t);
            if (d.thinking_delta) |t| try state.appendTextDelta(receiver, t);
            if (d.signature_delta) |sig| try state.setSignature(sig);
            if (d.input_json_delta) |j| try state.appendInputJsonDelta(receiver, j);
        },
        .content_block_stop => |s| {
            if (state.active) |a| {
                if (a.wire_index == s.index) try state.closeBlock(receiver);
            }
        },
        .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 });
            }
            return error.StreamError;
        },
        .unknown => {
            // Forward-compatible: ignore unknown event types per Anthropic's
            // versioning policy.
        },
    }
}

// -----------------------------------------------------------------------------
// Tests
// -----------------------------------------------------------------------------

const testing = std.testing;

/// Recording Receiver that captures the callback sequence for assertions.
const RecordingReceiver = struct {
    allocator: Allocator,
    events: std.ArrayList(Event) = .empty,

    const Event = union(enum) {
        message_start: conversation.MessageRole,
        block_start: struct {
            kind: provider_mod.ContentBlockType,
            index: usize,
        },
        delta: struct {
            index: usize,
            bytes: []const u8, // owned copy
        },
        block_complete: struct {
            index: usize,
            kind: provider_mod.ContentBlockType,
            text: []const u8, // owned copy
            signature: ?[]const u8 = null, // owned copy when present
        },
        message_complete: ?provider_mod.Usage,
        err: anyerror,
    };

    fn init(allocator: Allocator) RecordingReceiver {
        return .{ .allocator = allocator };
    }

    fn deinit(self: *RecordingReceiver) void {
        for (self.events.items) |ev| {
            switch (ev) {
                .delta => |d| self.allocator.free(d.bytes),
                .block_complete => |b| {
                    self.allocator.free(b.text);
                    if (b.signature) |s| self.allocator.free(s);
                },
                else => {},
            }
        }
        self.events.deinit(self.allocator);
    }

    fn receiver(self: *RecordingReceiver) provider_mod.Receiver {
        return .{ .ptr = self, .vtable = &vt };
    }

    const vt: provider_mod.ReceiverVTable = .{
        .onMessageStart = onMessageStart,
        .onBlockStart = onBlockStart,
        .onToolDetails = onToolDetails,
        .onContentDelta = onContentDelta,
        .onBlockComplete = onBlockComplete,
        .onMessageComplete = onMessageComplete,
        .onError = onError,
    };

    fn onMessageStart(ptr: *anyopaque, role: conversation.MessageRole) anyerror!void {
        const self: *RecordingReceiver = @ptrCast(@alignCast(ptr));
        try self.events.append(self.allocator, .{ .message_start = role });
    }
    fn onBlockStart(
        ptr: *anyopaque,
        bt: provider_mod.ContentBlockType,
        idx: usize,
    ) anyerror!void {
        const self: *RecordingReceiver = @ptrCast(@alignCast(ptr));
        try self.events.append(self.allocator, .{ .block_start = .{ .kind = bt, .index = idx } });
    }
    fn onToolDetails(
        _: *anyopaque,
        _: usize,
        _: []const u8,
        _: []const u8,
    ) anyerror!void {
        // Anthropic delivers identity at block_start time; the existing
        // tests assert tool-use blocks via the ContentBlock in conv after
        // finalize, not via the event stream. We accept and drop these
        // here to keep the test recorder schema stable.
    }
    fn onContentDelta(ptr: *anyopaque, idx: usize, delta: []const u8) anyerror!void {
        const self: *RecordingReceiver = @ptrCast(@alignCast(ptr));
        const copy = try self.allocator.dupe(u8, delta);
        try self.events.append(self.allocator, .{ .delta = .{ .index = idx, .bytes = copy } });
    }
    fn onBlockComplete(
        ptr: *anyopaque,
        idx: usize,
        block: conversation.ContentBlock,
    ) anyerror!void {
        const self: *RecordingReceiver = @ptrCast(@alignCast(ptr));
        switch (block) {
            .Text => |tb| {
                const txt = try self.allocator.dupe(u8, tb.items);
                try self.events.append(self.allocator, .{ .block_complete = .{
                    .kind = .Text,
                    .index = idx,
                    .text = txt,
                } });
            },
            .Thinking => |tb| {
                const txt = try self.allocator.dupe(u8, tb.text.items);
                const sig = if (tb.signature) |s| try self.allocator.dupe(u8, s) else null;
                try self.events.append(self.allocator, .{ .block_complete = .{
                    .kind = .Thinking,
                    .index = idx,
                    .text = txt,
                    .signature = sig,
                } });
            },
            else => {},
        }
    }
    fn onMessageComplete(ptr: *anyopaque, _: conversation.Message, usage: ?provider_mod.Usage) anyerror!void {
        const self: *RecordingReceiver = @ptrCast(@alignCast(ptr));
        try self.events.append(self.allocator, .{ .message_complete = usage });
    }
    fn onError(ptr: *anyopaque, err: anyerror) void {
        const self: *RecordingReceiver = @ptrCast(@alignCast(ptr));
        self.events.append(self.allocator, .{ .err = err }) catch {};
    }
};

fn runStreamedTurn(
    allocator: Allocator,
    conv: *conversation.Conversation,
    receiver: *provider_mod.Receiver,
    events: []const []const u8,
) !void {
    var state: StreamState = .init(allocator);
    defer state.deinit();

    for (events) |payload| {
        try handleEvent(allocator, payload, &state, receiver);
        if (state.end_of_stream) break;
    }
    try state.finalize(receiver, conv);
}

test "streams a text-only turn end-to-end" {
    const allocator = testing.allocator;

    var conv = conversation.Conversation.init(allocator);
    defer conv.deinit();
    try conv.addUserMessage("hello");

    var rec = RecordingReceiver.init(allocator);
    defer rec.deinit();
    var recv = rec.receiver();

    const events = [_][]const u8{
        \\{"type":"message_start","message":{"id":"msg_1","type":"message","role":"assistant","content":[],"model":"claude"}}
        ,
        \\{"type":"content_block_start","index":0,"content_block":{"type":"text","text":""}}
        ,
        \\{"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"Hello"}}
        ,
        \\{"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"!"}}
        ,
        \\{"type":"content_block_stop","index":0}
        ,
        \\{"type":"message_delta","delta":{"stop_reason":"end_turn"}}
        ,
        \\{"type":"message_stop"}
        ,
    };

    try runStreamedTurn(allocator, &conv, &recv, &events);

    // Conversation now holds the assistant reply.
    try testing.expectEqual(@as(usize, 2), conv.messages.items.len);
    try testing.expectEqual(conversation.MessageRole.assistant, conv.messages.items[1].role);
    try testing.expectEqual(@as(usize, 1), conv.messages.items[1].content.items.len);
    try testing.expectEqualStrings(
        "Hello!",
        conv.messages.items[1].content.items[0].Text.items,
    );

    // Callback sequence: msg_start, block_start, delta, delta, block_complete, msg_complete.
    try testing.expectEqual(@as(usize, 6), rec.events.items.len);
    try testing.expectEqual(conversation.MessageRole.assistant, rec.events.items[0].message_start);
    try testing.expectEqual(provider_mod.ContentBlockType.Text, rec.events.items[1].block_start.kind);
    try testing.expectEqualStrings("Hello", rec.events.items[2].delta.bytes);
    try testing.expectEqualStrings("!", rec.events.items[3].delta.bytes);
    try testing.expectEqualStrings("Hello!", rec.events.items[4].block_complete.text);
    try testing.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 conv.addUserMessage("hi");

    var rec = RecordingReceiver.init(allocator);
    defer rec.deinit();
    var recv = rec.receiver();

    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, &recv, &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 conv.addUserMessage("hi");

    var rec = RecordingReceiver.init(allocator);
    defer rec.deinit();
    var recv = rec.receiver();

    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, &recv, &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 conv.addUserMessage("solve");

    var rec = RecordingReceiver.init(allocator);
    defer rec.deinit();
    var recv = rec.receiver();

    const events = [_][]const u8{
        \\{"type":"message_start","message":{"role":"assistant"}}
        ,
        \\{"type":"content_block_start","index":0,"content_block":{"type":"thinking","thinking":"","signature":""}}
        ,
        \\{"type":"content_block_delta","index":0,"delta":{"type":"thinking_delta","thinking":"step one"}}
        ,
        \\{"type":"content_block_delta","index":0,"delta":{"type":"thinking_delta","thinking":" step two"}}
        ,
        \\{"type":"content_block_delta","index":0,"delta":{"type":"signature_delta","signature":"EqQBabc"}}
        ,
        \\{"type":"content_block_stop","index":0}
        ,
        \\{"type":"content_block_start","index":1,"content_block":{"type":"text","text":""}}
        ,
        \\{"type":"content_block_delta","index":1,"delta":{"type":"text_delta","text":"answer"}}
        ,
        \\{"type":"content_block_stop","index":1}
        ,
        \\{"type":"message_stop"}
        ,
    };

    try runStreamedTurn(allocator, &conv, &recv, &events);

    // The assistant message has Thinking + Text, with signature on the Thinking.
    const asst = conv.messages.items[1];
    try testing.expectEqual(@as(usize, 2), asst.content.items.len);
    try testing.expectEqualStrings("step one step two", asst.content.items[0].Thinking.text.items);
    try testing.expectEqualStrings("EqQBabc", asst.content.items[0].Thinking.signature.?);
    try testing.expectEqualStrings("answer", asst.content.items[1].Text.items);
}

test "signature-only thinking block (display omitted)" {
    // Anthropic emits a thinking block with only a signature_delta when
    // `display: "omitted"` is configured. Verify we still capture the
    // signature with empty thinking text.
    const allocator = testing.allocator;

    var conv = conversation.Conversation.init(allocator);
    defer conv.deinit();
    try conv.addUserMessage("hi");

    var rec = RecordingReceiver.init(allocator);
    defer rec.deinit();
    var recv = rec.receiver();

    const events = [_][]const u8{
        \\{"type":"message_start","message":{"role":"assistant"}}
        ,
        \\{"type":"content_block_start","index":0,"content_block":{"type":"thinking","thinking":"","signature":""}}
        ,
        \\{"type":"content_block_delta","index":0,"delta":{"type":"signature_delta","signature":"sig123"}}
        ,
        \\{"type":"content_block_stop","index":0}
        ,
        \\{"type":"content_block_start","index":1,"content_block":{"type":"text","text":""}}
        ,
        \\{"type":"content_block_delta","index":1,"delta":{"type":"text_delta","text":"hi back"}}
        ,
        \\{"type":"content_block_stop","index":1}
        ,
        \\{"type":"message_stop"}
        ,
    };

    try runStreamedTurn(allocator, &conv, &recv, &events);

    const asst = conv.messages.items[1];
    try testing.expectEqual(@as(usize, 2), asst.content.items.len);
    try testing.expectEqualStrings("", asst.content.items[0].Thinking.text.items);
    try testing.expectEqualStrings("sig123", asst.content.items[0].Thinking.signature.?);
}

test "ping and unknown events are ignored" {
    const allocator = testing.allocator;

    var conv = conversation.Conversation.init(allocator);
    defer conv.deinit();
    try conv.addUserMessage("hi");

    var rec = RecordingReceiver.init(allocator);
    defer rec.deinit();
    var recv = rec.receiver();

    const events = [_][]const u8{
        \\{"type":"message_start","message":{"role":"assistant"}}
        ,
        \\{"type":"ping"}
        ,
        \\{"type":"content_block_start","index":0,"content_block":{"type":"text","text":""}}
        ,
        \\{"type":"ping"}
        ,
        \\{"type":"future_event_type","whatever":true}
        ,
        \\{"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"ok"}}
        ,
        \\{"type":"content_block_stop","index":0}
        ,
        \\{"type":"message_stop"}
        ,
    };

    try runStreamedTurn(allocator, &conv, &recv, &events);

    try testing.expectEqualStrings(
        "ok",
        conv.messages.items[1].content.items[0].Text.items,
    );
}

test "tool_use blocks are captured with id, name, and assembled input" {
    const allocator = testing.allocator;

    var conv = conversation.Conversation.init(allocator);
    defer conv.deinit();
    try conv.addUserMessage("use a tool");

    var rec = RecordingReceiver.init(allocator);
    defer rec.deinit();
    var recv = rec.receiver();

    const events = [_][]const u8{
        \\{"type":"message_start","message":{"role":"assistant"}}
        ,
        \\{"type":"content_block_start","index":0,"content_block":{"type":"tool_use","id":"tu_1","name":"calc","input":{}}}
        ,
        \\{"type":"content_block_delta","index":0,"delta":{"type":"input_json_delta","partial_json":"{\"x\":"}}
        ,
        \\{"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, &recv, &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 conv.addUserMessage("use a tool");

    var rec = RecordingReceiver.init(allocator);
    defer rec.deinit();
    var recv = rec.receiver();

    const events = [_][]const u8{
        \\{"type":"message_start","message":{"role":"assistant"}}
        ,
        \\{"type":"content_block_start","index":0,"content_block":{"type":"tool_use","id":"t1","name":"calc__sum","input":{}}}
        ,
        \\{"type":"content_block_stop","index":0}
        ,
        \\{"type":"message_stop"}
        ,
    };

    try runStreamedTurn(allocator, &conv, &recv, &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 conv.addUserMessage("hi");

    var rec = RecordingReceiver.init(allocator);
    defer rec.deinit();
    var recv = rec.receiver();

    var state: StreamState = .init(allocator);
    defer state.deinit();

    try handleEvent(
        allocator,
        \\{"type":"message_start","message":{"role":"assistant"}}
    ,
        &state,
        &recv,
    );

    const result = handleEvent(
        allocator,
        \\{"type":"error","error":{"type":"overloaded_error","message":"too busy"}}
    ,
        &state,
        &recv,
    );
    try testing.expectError(error.StreamError, result);
}

test "two streamed turns persist assistant replies in the conversation" {
    // Same regression scenario as the openai_chat test, adapted to Anthropic.
    const allocator = testing.allocator;

    var conv = conversation.Conversation.init(allocator);
    defer conv.deinit();
    try conv.addSystemMessage("Be brief.");
    try conv.addUserMessage("hi");

    var rec = RecordingReceiver.init(allocator);
    defer rec.deinit();
    var recv = rec.receiver();

    const turn1 = [_][]const u8{
        \\{"type":"message_start","message":{"role":"assistant"}}
        ,
        \\{"type":"content_block_start","index":0,"content_block":{"type":"text","text":""}}
        ,
        \\{"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"Hi!"}}
        ,
        \\{"type":"content_block_stop","index":0}
        ,
        \\{"type":"message_stop"}
        ,
    };
    try runStreamedTurn(allocator, &conv, &recv, &turn1);

    try conv.addUserMessage("what did you say?");

    const turn2 = [_][]const u8{
        \\{"type":"message_start","message":{"role":"assistant"}}
        ,
        \\{"type":"content_block_start","index":0,"content_block":{"type":"text","text":""}}
        ,
        \\{"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"I said: Hi!"}}
        ,
        \\{"type":"content_block_stop","index":0}
        ,
        \\{"type":"message_stop"}
        ,
    };
    try runStreamedTurn(allocator, &conv, &recv, &turn2);

    // system + user + assistant + user + assistant = 5
    try testing.expectEqual(@as(usize, 5), conv.messages.items.len);
    try testing.expectEqualStrings(
        "I said: Hi!",
        conv.messages.items[4].content.items[0].Text.items,
    );
}