phase 2 done

1 files changed, 828 insertions, 0 deletions

diff --git a/libpanto/src/provider_anthropic_messages.zig b/libpanto/src/provider_anthropic_messages.zig
new file mode 100644
index 0000000..2914b6e
--- /dev/null
+++ b/libpanto/src/provider_anthropic_messages.zig
@@ -0,0 +1,828 @@
+//! 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");
+
+pub const AnthropicMessagesProvider = struct {
+    allocator: Allocator,
+    io: Io,
+    config: config_mod.AnthropicMessagesConfig,
+    http_client: http.Client,
+
+    pub fn init(
+        allocator: Allocator,
+        io: Io,
+        cfg: config_mod.AnthropicMessagesConfig,
+    ) AnthropicMessagesProvider {
+        return .{
+            .allocator = allocator,
+            .io = io,
+            .config = cfg,
+            .http_client = .{ .allocator = allocator, .io = io },
+        };
+    }
+
+    pub fn deinit(self: *AnthropicMessagesProvider) void {
+        self.http_client.deinit();
+    }
+
+    pub fn provider(self: *AnthropicMessagesProvider) provider_mod.Provider {
+        return .{ .ptr = self, .vtable = &vtable };
+    }
+
+    const vtable: provider_mod.ProviderVTable = .{
+        .streamStep = vtableStreamStep,
+        .deinit = vtableDeinit,
+    };
+
+    fn vtableStreamStep(
+        ptr: *anyopaque,
+        conv: *conversation.Conversation,
+        receiver: *provider_mod.Receiver,
+    ) anyerror!void {
+        const self: *AnthropicMessagesProvider = @ptrCast(@alignCast(ptr));
+        return self.streamStep(conv, receiver);
+    }
+
+    /// Called via the `Provider` interface. Tears down the impl AND frees
+    /// its heap allocation, since `Provider.init` is the one that allocated
+    /// it. Direct stack-allocated users (tests, embedders) call `deinit`
+    /// themselves and never hit this path.
+    fn vtableDeinit(ptr: *anyopaque) void {
+        const self: *AnthropicMessagesProvider = @ptrCast(@alignCast(ptr));
+        const allocator = self.allocator;
+        self.deinit();
+        allocator.destroy(self);
+    }
+
+    pub fn streamStep(
+        self: *AnthropicMessagesProvider,
+        conv: *conversation.Conversation,
+        receiver: *provider_mod.Receiver,
+    ) !void {
+        // Outer wrapper guarantees `onError` is called exactly once if
+        // anything fails.
+        self.streamStepInner(conv, receiver) catch |err| {
+            receiver.onError(err);
+            return err;
+        };
+    }
+
+    fn streamStepInner(
+        self: *AnthropicMessagesProvider,
+        conv: *conversation.Conversation,
+        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);
+        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,
+            });
+            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.
+        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) break;
+
+            const events = try parser.feed(chunk[0..n]);
+            defer parser.freeEvents(events);
+
+            for (events) |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,
+
+    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,
+    };
+
+    const BlockKind = enum { text, thinking, 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);
+        }
+        for (self.blocks.items) |*b| b.deinit(self.allocator);
+        self.blocks.deinit(self.allocator);
+    }
+
+    fn ensureStarted(self: *StreamState, receiver: *provider_mod.Receiver) !void {
+        if (self.started) return;
+        self.started = true;
+        try receiver.onMessageStart(.assistant);
+    }
+
+    fn openBlock(
+        self: *StreamState,
+        receiver: *provider_mod.Receiver,
+        wire_index: usize,
+        kind: BlockKind,
+        tool_meta: ?provider_mod.BlockMeta,
+    ) !void {
+        // Defensive: if a prior block didn't get an explicit stop, drop it.
+        if (self.active != null) {
+            self.discardActive();
+        }
+        self.active = .{
+            .wire_index = wire_index,
+            .kind = kind,
+        };
+        const block_type: ?provider_mod.ContentBlockType = switch (kind) {
+            .text => .Text,
+            .thinking => .Thinking,
+            .unsupported => null,
+        };
+        if (block_type) |bt| {
+            try receiver.onBlockStart(bt, wire_index, tool_meta);
+        }
+    }
+
+    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);
+    }
+
+    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);
+    }
+
+    /// 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) {
+            // Drop unsupported blocks (e.g. tool_use until phase 3+).
+            a.text_buf.deinit(self.allocator);
+            if (a.signature) |sig| self.allocator.free(sig);
+            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,
+            } },
+            .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);
+            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) {
+            // Wire dropped us mid-block. Close it cleanly anyway.
+            try self.closeBlock(receiver);
+        }
+
+        const moved_blocks = try self.blocks.toOwnedSlice(self.allocator);
+        defer self.allocator.free(moved_blocks);
+
+        try conv.addAssistantMessage(moved_blocks);
+
+        const msg = conv.messages.items[conv.messages.items.len - 1];
+        try receiver.onMessageComplete(msg);
+    }
+};
+
+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 => {
+            try state.ensureStarted(receiver);
+        },
+        .content_block_start => |s| {
+            try state.ensureStarted(receiver);
+            const kind: StreamState.BlockKind = switch (s.kind) {
+                .text => .text,
+                .thinking => .thinking,
+                .tool_use, .unknown => .unsupported,
+            };
+            // tool_use meta will start being honored in phase 3+. For now we
+            // mark the block unsupported and drop it.
+            try state.openBlock(receiver, s.index, kind, null);
+        },
+        .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);
+            // input_json_delta: phase 3+.
+        },
+        .content_block_stop => {
+            try state.closeBlock(receiver);
+        },
+        .message_delta => {
+            // We don't act on stop_reason directly; message_stop is the
+            // authoritative end-of-stream signal.
+        },
+        .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,
+        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,
+        .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,
+        _: ?provider_mod.BlockMeta,
+    ) anyerror!void {
+        const self: *RecordingReceiver = @ptrCast(@alignCast(ptr));
+        try self.events.append(self.allocator, .{ .block_start = .{ .kind = bt, .index = idx } });
+    }
+    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) anyerror!void {
+        const self: *RecordingReceiver = @ptrCast(@alignCast(ptr));
+        try self.events.append(self.allocator, .message_complete);
+    }
+    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.expectEqual(@as(@TypeOf(rec.events.items[5]), .message_complete), rec.events.items[5]);
+}
+
+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 dropped (deferred to phase 3)" {
+    // A tool_use block is started, receives input_json_delta events, and
+    // stops. Phase 2 should silently drop it without crashing.
+    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\":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);
+
+    // Only the text block survives.
+    const asst = conv.messages.items[1];
+    try testing.expectEqual(@as(usize, 1), asst.content.items.len);
+    try testing.expectEqualStrings("done", asst.content.items[0].Text.items);
+}
+
+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,
+    );
+}