refactor: Agent.run() -> Stream -> Stream.next() -> Event

converted the main agent loop from push-based (into callbacks on a
Receiver vtable) to pull-based, where `next()` re-enters a state-machine
Stream until the next event can be returned.

1 files changed, 247 insertions, 201 deletions

diff --git a/libpanto/src/provider_anthropic_messages.zig b/libpanto/src/provider_anthropic_messages.zig
index 55efb5f..4bd146a 100644
--- a/libpanto/src/provider_anthropic_messages.zig
+++ b/libpanto/src/provider_anthropic_messages.zig
@@ -20,11 +20,15 @@ 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.
@@ -36,26 +40,29 @@ pub const AnthropicMessagesRequest = struct {
     /// Optional diagnostic side-channel; see `OpenAIChatRequest.diag`.
     diag: ?*provider_mod.ProviderDiagnostic = null,
 
-    pub fn streamStep(
+    /// 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,
-        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;
+    ) !*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),
         };
-    }
+        errdefer {
+            rr.parser.deinit();
+            rr.state.deinit();
+        }
 
-    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",
@@ -75,7 +82,7 @@ pub const AnthropicMessagesRequest = struct {
             .{ .name = "anthropic-version", .value = self.config.api_version },
         };
 
-        var req = try self.http_client.request(.POST, uri, .{
+        rr.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.
@@ -83,22 +90,25 @@ pub const AnthropicMessagesRequest = struct {
             .keep_alive = false,
             .redirect_behavior = .not_allowed,
         });
-        defer req.deinit();
+        rr.req_open = true;
+        errdefer {
+            rr.req.deinit();
+            rr.req_open = false;
+        }
 
-        req.transfer_encoding = .{ .content_length = body.len };
+        rr.req.transfer_encoding = .{ .content_length = body.len };
 
         var send_buf: [4096]u8 = undefined;
-        var bw = try req.sendBodyUnflushed(&send_buf);
+        var bw = try rr.req.sendBodyUnflushed(&send_buf);
         try bw.writer.writeAll(body);
         try bw.end();
-        try req.connection.?.flush();
+        try rr.req.connection.?.flush();
 
         var redirect_buf: [1024]u8 = undefined;
-        var response = try req.receiveHead(&redirect_buf);
+        rr.response = try rr.req.receiveHead(&redirect_buf);
 
-        if (@intFromEnum(response.head.status) >= 400) {
-            var transfer_buf: [4096]u8 = undefined;
-            const body_reader = response.reader(&transfer_buf);
+        if (@intFromEnum(rr.response.head.status) >= 400) {
+            const body_reader = rr.response.reader(&rr.transfer_buf);
             var err_buf: std.ArrayList(u8) = .empty;
             defer err_buf.deinit(self.allocator);
             var tmp: [1024]u8 = undefined;
@@ -108,7 +118,7 @@ pub const AnthropicMessagesRequest = struct {
                 try err_buf.appendSlice(self.allocator, tmp[0..n]);
                 if (err_buf.items.len > 16 * 1024) break;
             }
-            const status: u16 = @intFromEnum(response.head.status);
+            const status: u16 = @intFromEnum(rr.response.head.status);
             std.log.err("anthropic_messages HTTP {d}: {s}", .{ status, err_buf.items });
             // Anthropic rejects oversized requests with HTTP 400 and a
             // "prompt is too long" message; `classifyHttpStatus` maps that to
@@ -117,52 +127,104 @@ pub const AnthropicMessagesRequest = struct {
             const classified = provider_mod.classifyHttpStatus(status, err_buf.items);
             if (self.diag) |d| {
                 d.status_code = status;
-                d.retry_after_ms = provider_mod.retryAfterFromHead(response.head);
+                d.retry_after_ms = provider_mod.retryAfterFromHead(rr.response.head);
             }
             return classified;
         }
 
-        var transfer_buf: [4096]u8 = undefined;
-        const body_reader = response.reader(&transfer_buf);
+        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,
 
-        var parser = sse_mod.SSEParser.init(self.allocator);
-        defer parser.deinit();
+    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,
 
-        var state: StreamState = .init(self.allocator);
-        defer state.deinit();
+    req_open: bool = false,
+    done: bool = false,
 
-        // 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,
-                // Transport failure before the message completed: retryable.
-                else => return error.ProviderStreamMalformed,
-            };
-            if (n == 0) continue;
+    pub const ProduceStatus = provider_mod.ProviderStream.ProduceStatus;
 
-            const events = try parser.feed(chunk[0..n]);
-            defer parser.freeEvents(events);
+    /// 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 };
+    }
 
-            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;
-                }
+    const vtable: provider_mod.ProviderStream.VTable = .{
+        .produce = produceVT,
+        .deinit = deinitVT,
+    };
+
+    fn produceVT(ptr: *anyopaque, out: *EventQueue) anyerror!ProduceStatus {
+        const self: *ResumableResponse = @ptrCast(@alignCast(ptr));
+        return self.produce(out);
+    }
+
+    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;
+    }
 
-        // Stream ended without an explicit message_stop. Finalize anyway.
-        try state.finalize(receiver, conv);
+    fn finishStream(self: *ResumableResponse, out: *EventQueue) !void {
+        if (self.done) return;
+        self.done = true;
+        try self.state.finalize(out, self.conv);
     }
 };
 
@@ -224,10 +286,10 @@ const StreamState = struct {
         if (self.stop_reason) |s| self.allocator.free(s);
     }
 
-    fn ensureStarted(self: *StreamState, receiver: *provider_mod.Receiver) !void {
+    fn ensureStarted(self: *StreamState, out: *EventQueue) !void {
         if (self.started) return;
         self.started = true;
-        try receiver.onMessageStart(.assistant);
+        try out.push(.{ .message_start = .assistant });
     }
 
     /// Merge a wire-level usage snapshot into the accumulated counts.
@@ -248,7 +310,7 @@ const StreamState = struct {
 
     fn openBlock(
         self: *StreamState,
-        receiver: *provider_mod.Receiver,
+        out: *EventQueue,
         wire_index: usize,
         kind: BlockKind,
         tool_id: ?[]const u8,
@@ -291,14 +353,20 @@ const StreamState = struct {
             .unsupported => null,
         };
         if (block_type) |bt| {
-            try receiver.onBlockStart(bt, wire_index);
+            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 onToolDetails immediately — before any arg
+            // 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 receiver.onToolDetails(wire_index, ab.tool_id.?, ab.tool_name.?);
+                    try out.push(.{ .tool_details = .{
+                        .index = wire_index,
+                        .id = try out.dupeBytes(ab.tool_id.?),
+                        .name = try out.dupeBytes(ab.tool_name.?),
+                    } });
                 }
             }
         }
@@ -306,26 +374,34 @@ const StreamState = struct {
 
     fn appendTextDelta(
         self: *StreamState,
-        receiver: *provider_mod.Receiver,
+        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);
-        try receiver.onContentDelta(a.wire_index, 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,
-        receiver: *provider_mod.Receiver,
+        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 receiver.onContentDelta(a.wire_index, delta);
+        try out.push(.{ .content_delta = .{
+            .index = a.wire_index,
+            .delta = try out.dupeBytes(delta),
+        } });
     }
 
     fn setSignature(self: *StreamState, sig: []const u8) !void {
@@ -339,10 +415,10 @@ const StreamState = struct {
         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.
+    /// Close the active block: append it to `blocks` and emit block_complete.
     fn closeBlock(
         self: *StreamState,
-        receiver: *provider_mod.Receiver,
+        out: *EventQueue,
     ) !void {
         var a = self.active orelse return;
         self.active = null;
@@ -389,7 +465,10 @@ const StreamState = struct {
         };
 
         try self.blocks.append(self.allocator, block);
-        try receiver.onBlockComplete(a.wire_index, self.blocks.items[self.blocks.items.len - 1]);
+        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.
@@ -407,7 +486,7 @@ const StreamState = struct {
 
     fn finalize(
         self: *StreamState,
-        receiver: *provider_mod.Receiver,
+        out: *EventQueue,
         conv: *conversation.Conversation,
     ) !void {
         if (self.finalized) return;
@@ -416,7 +495,7 @@ const StreamState = struct {
         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);
+            try self.closeBlock(out);
         }
 
         const moved_blocks = try self.blocks.toOwnedSlice(self.allocator);
@@ -426,7 +505,7 @@ const StreamState = struct {
         try conv.addAssistantMessageWithUsage(moved_blocks, usage);
 
         const msg = conv.messages.items[conv.messages.items.len - 1];
-        try receiver.onMessageComplete(msg, usage);
+        try out.push(.{ .message_complete = .{ .message = msg, .usage = usage } });
     }
 };
 
@@ -434,35 +513,35 @@ fn handleEvent(
     allocator: Allocator,
     payload: []const u8,
     state: *StreamState,
-    receiver: *provider_mod.Receiver,
+    out: *EventQueue,
 ) !void {
     var parsed = try json_mod.parseStreamEvent(allocator, payload);
     defer parsed.deinit();
 
     switch (parsed.event) {
         .message_start => |s| {
-            try state.ensureStarted(receiver);
+            try state.ensureStarted(out);
             state.mergeUsage(s.usage);
         },
         .content_block_start => |s| {
-            try state.ensureStarted(receiver);
+            try state.ensureStarted(out);
             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);
+            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(receiver, t);
-            if (d.thinking_delta) |t| try state.appendTextDelta(receiver, t);
+            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(receiver, j);
+            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(receiver);
+                if (a.wire_index == s.index) try state.closeBlock(out);
             }
         },
         .message_delta => |d| {
@@ -494,12 +573,16 @@ fn handleEvent(
 
 const testing = std.testing;
 
-/// Recording Receiver that captures the callback sequence for assertions.
-const RecordingReceiver = struct {
+/// 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(Event) = .empty,
+    events: std.ArrayList(Rec) = .empty,
 
-    const Event = union(enum) {
+    const Rec = union(enum) {
         message_start: conversation.MessageRole,
         block_start: struct {
             kind: provider_mod.ContentBlockType,
@@ -516,14 +599,13 @@ const RecordingReceiver = struct {
             signature: ?[]const u8 = null, // owned copy when present
         },
         message_complete: ?provider_mod.Usage,
-        err: anyerror,
     };
 
-    fn init(allocator: Allocator) RecordingReceiver {
+    fn init(allocator: Allocator) EventRecorder {
         return .{ .allocator = allocator };
     }
 
-    fn deinit(self: *RecordingReceiver) void {
+    fn deinit(self: *EventRecorder) void {
         for (self.events.items) |ev| {
             switch (ev) {
                 .delta => |d| self.allocator.free(d.bytes),
@@ -537,102 +619,76 @@ const RecordingReceiver = struct {
         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,
-        .onProviderRetry = onProviderRetry,
-    };
-
-    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 onProviderRetry(_: *anyopaque, _: provider_mod.ProviderRetryInfo) void {}
-    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,
-                } });
+    /// 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 } });
             },
-            .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,
-                } });
+            .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 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,
+    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, receiver);
+        try handleEvent(allocator, payload, &state, &queue);
         if (state.end_of_stream) break;
     }
-    try state.finalize(receiver, conv);
+    try state.finalize(&queue, conv);
+
+    while (queue.pop()) |ev| {
+        if (rec) |r| try r.record(ev);
+    }
 }
 
 test "streams a text-only turn end-to-end" {
@@ -642,9 +698,8 @@ test "streams a text-only turn end-to-end" {
     defer conv.deinit();
     try conv.addUserMessage("hello");
 
-    var rec = RecordingReceiver.init(allocator);
+    var rec = EventRecorder.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"}}
@@ -663,7 +718,7 @@ test "streams a text-only turn end-to-end" {
         ,
     };
 
-    try runStreamedTurn(allocator, &conv, &recv, &events);
+    try runStreamedTurn(allocator, &conv, &rec, &events);
 
     // Conversation now holds the assistant reply.
     try testing.expectEqual(@as(usize, 2), conv.messages.items.len);
@@ -693,9 +748,8 @@ test "anthropic: captures usage from message_start and message_delta on message_
     defer conv.deinit();
     try conv.addUserMessage("hi");
 
-    var rec = RecordingReceiver.init(allocator);
+    var rec = EventRecorder.init(allocator);
     defer rec.deinit();
-    var recv = rec.receiver();
 
     const events = [_][]const u8{
         // Initial input-side counts on message_start.
@@ -714,7 +768,7 @@ test "anthropic: captures usage from message_start and message_delta on message_
         ,
     };
 
-    try runStreamedTurn(allocator, &conv, &recv, &events);
+    try runStreamedTurn(allocator, &conv, &rec, &events);
 
     // Find the message_complete event and check its usage payload.
     var found: ?provider_mod.Usage = null;
@@ -737,9 +791,8 @@ test "anthropic: message_complete carries null usage when wire omits it" {
     defer conv.deinit();
     try conv.addUserMessage("hi");
 
-    var rec = RecordingReceiver.init(allocator);
+    var rec = EventRecorder.init(allocator);
     defer rec.deinit();
-    var recv = rec.receiver();
 
     const events = [_][]const u8{
         \\{"type":"message_start","message":{"id":"m","role":"assistant","content":[],"model":"claude"}}
@@ -756,7 +809,7 @@ test "anthropic: message_complete carries null usage when wire omits it" {
         ,
     };
 
-    try runStreamedTurn(allocator, &conv, &recv, &events);
+    try runStreamedTurn(allocator, &conv, &rec, &events);
 
     var saw_complete = false;
     for (rec.events.items) |ev| {
@@ -775,9 +828,8 @@ test "captures thinking signature for round-trip" {
     defer conv.deinit();
     try conv.addUserMessage("solve");
 
-    var rec = RecordingReceiver.init(allocator);
+    var rec = EventRecorder.init(allocator);
     defer rec.deinit();
-    var recv = rec.receiver();
 
     const events = [_][]const u8{
         \\{"type":"message_start","message":{"role":"assistant"}}
@@ -802,7 +854,7 @@ test "captures thinking signature for round-trip" {
         ,
     };
 
-    try runStreamedTurn(allocator, &conv, &recv, &events);
+    try runStreamedTurn(allocator, &conv, &rec, &events);
 
     // The assistant message has Thinking + Text, with signature on the Thinking.
     const asst = conv.messages.items[1];
@@ -822,9 +874,8 @@ test "signature-only thinking block (display omitted)" {
     defer conv.deinit();
     try conv.addUserMessage("hi");
 
-    var rec = RecordingReceiver.init(allocator);
+    var rec = EventRecorder.init(allocator);
     defer rec.deinit();
-    var recv = rec.receiver();
 
     const events = [_][]const u8{
         \\{"type":"message_start","message":{"role":"assistant"}}
@@ -845,7 +896,7 @@ test "signature-only thinking block (display omitted)" {
         ,
     };
 
-    try runStreamedTurn(allocator, &conv, &recv, &events);
+    try runStreamedTurn(allocator, &conv, &rec, &events);
 
     const asst = conv.messages.items[1];
     try testing.expectEqual(@as(usize, 2), asst.content.items.len);
@@ -860,9 +911,8 @@ test "ping and unknown events are ignored" {
     defer conv.deinit();
     try conv.addUserMessage("hi");
 
-    var rec = RecordingReceiver.init(allocator);
+    var rec = EventRecorder.init(allocator);
     defer rec.deinit();
-    var recv = rec.receiver();
 
     const events = [_][]const u8{
         \\{"type":"message_start","message":{"role":"assistant"}}
@@ -883,7 +933,7 @@ test "ping and unknown events are ignored" {
         ,
     };
 
-    try runStreamedTurn(allocator, &conv, &recv, &events);
+    try runStreamedTurn(allocator, &conv, &rec, &events);
 
     try testing.expectEqualStrings(
         "ok",
@@ -898,9 +948,8 @@ test "tool_use blocks are captured with id, name, and assembled input" {
     defer conv.deinit();
     try conv.addUserMessage("use a tool");
 
-    var rec = RecordingReceiver.init(allocator);
+    var rec = EventRecorder.init(allocator);
     defer rec.deinit();
-    var recv = rec.receiver();
 
     const events = [_][]const u8{
         \\{"type":"message_start","message":{"role":"assistant"}}
@@ -923,7 +972,7 @@ test "tool_use blocks are captured with id, name, and assembled input" {
         ,
     };
 
-    try runStreamedTurn(allocator, &conv, &recv, &events);
+    try runStreamedTurn(allocator, &conv, &rec, &events);
 
     const asst = conv.messages.items[1];
     try testing.expectEqual(@as(usize, 2), asst.content.items.len);
@@ -946,9 +995,8 @@ test "inbound wire tool name is decoded to dotted form" {
     defer conv.deinit();
     try conv.addUserMessage("use a tool");
 
-    var rec = RecordingReceiver.init(allocator);
+    var rec = EventRecorder.init(allocator);
     defer rec.deinit();
-    var recv = rec.receiver();
 
     const events = [_][]const u8{
         \\{"type":"message_start","message":{"role":"assistant"}}
@@ -961,7 +1009,7 @@ test "inbound wire tool name is decoded to dotted form" {
         ,
     };
 
-    try runStreamedTurn(allocator, &conv, &recv, &events);
+    try runStreamedTurn(allocator, &conv, &rec, &events);
 
     const tu = conv.messages.items[1].content.items[0].ToolUse;
     try testing.expectEqualStrings("calc.sum", tu.name);
@@ -974,9 +1022,8 @@ test "error event propagates as Zig error" {
     defer conv.deinit();
     try conv.addUserMessage("hi");
 
-    var rec = RecordingReceiver.init(allocator);
-    defer rec.deinit();
-    var recv = rec.receiver();
+    var queue = EventQueue.init(allocator);
+    defer queue.deinit();
 
     var state: StreamState = .init(allocator);
     defer state.deinit();
@@ -986,7 +1033,7 @@ test "error event propagates as Zig error" {
         \\{"type":"message_start","message":{"role":"assistant"}}
     ,
         &state,
-        &recv,
+        &queue,
     );
 
     const result = handleEvent(
@@ -994,7 +1041,7 @@ test "error event propagates as Zig error" {
         \\{"type":"error","error":{"type":"overloaded_error","message":"too busy"}}
     ,
         &state,
-        &recv,
+        &queue,
     );
     try testing.expectError(error.ProviderStreamMalformed, result);
 }
@@ -1008,9 +1055,8 @@ test "two streamed turns persist assistant replies in the conversation" {
     try conv.addSystemMessage("Be brief.");
     try conv.addUserMessage("hi");
 
-    var rec = RecordingReceiver.init(allocator);
+    var rec = EventRecorder.init(allocator);
     defer rec.deinit();
-    var recv = rec.receiver();
 
     const turn1 = [_][]const u8{
         \\{"type":"message_start","message":{"role":"assistant"}}
@@ -1024,7 +1070,7 @@ test "two streamed turns persist assistant replies in the conversation" {
         \\{"type":"message_stop"}
         ,
     };
-    try runStreamedTurn(allocator, &conv, &recv, &turn1);
+    try runStreamedTurn(allocator, &conv, &rec, &turn1);
 
     try conv.addUserMessage("what did you say?");
 
@@ -1040,7 +1086,7 @@ test "two streamed turns persist assistant replies in the conversation" {
         \\{"type":"message_stop"}
         ,
     };
-    try runStreamedTurn(allocator, &conv, &recv, &turn2);
+    try runStreamedTurn(allocator, &conv, &rec, &turn2);
 
     // system + user + assistant + user + assistant = 5
     try testing.expectEqual(@as(usize, 5), conv.messages.items.len);