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.

8 files changed, 1336 insertions, 1028 deletions

diff --git a/docs/libpanto-bindings.md b/docs/libpanto-bindings.md
index 9b9ef2c..3c1a355 100644
--- a/docs/libpanto-bindings.md
+++ b/docs/libpanto-bindings.md
@@ -183,7 +183,7 @@ for v1.
 
 ---
 
-## Phase 0 — internal refactor of `libpanto` to a pull API
+## Phase 0 — internal refactor of `libpanto` to a pull API **(COMPLETE)**
 
 The real work. Everything after this is wrapping.
 
diff --git a/libpanto/src/agent.zig b/libpanto/src/agent.zig
index 122da1a..6749ffd 100644
--- a/libpanto/src/agent.zig
+++ b/libpanto/src/agent.zig
@@ -29,6 +29,7 @@ const Allocator = std.mem.Allocator;
 const Io = std.Io;
 
 const provider_mod = @import("provider.zig");
+const stream_mod = @import("stream.zig");
 const config_mod = @import("config.zig");
 const conversation = @import("conversation.zig");
 const compaction_mod = @import("compaction.zig");
@@ -44,6 +45,8 @@ pub const Tool = tool_mod.Tool;
 pub const ToolSource = tool_source_mod.ToolSource;
 pub const ToolRegistry = tool_registry_mod.ToolRegistry;
 
+const Event = stream_mod.Event;
+
 const Entry = tool_registry_mod.Entry;
 
 pub const Config = config_mod.Config;
@@ -122,38 +125,6 @@ fn cloneBlock(alloc: Allocator, block: conversation.ContentBlock) !conversation.
     };
 }
 
-/// A minimal receiver that captures the assistant's streamed message for
-/// compaction. We don't need incremental events — the assembled message is
-/// read off the conversation after the turn — so all callbacks are no-ops.
-const CompactionCapture = struct {
-    allocator: Allocator,
-
-    fn receiver(self: *CompactionCapture) provider_mod.Receiver {
-        return .{ .ptr = self, .vtable = &vt };
-    }
-    fn deinit(self: *CompactionCapture) void {
-        _ = self;
-    }
-    const vt: provider_mod.ReceiverVTable = .{
-        .onMessageStart = onMessageStart,
-        .onBlockStart = onBlockStart,
-        .onToolDetails = onToolDetails,
-        .onContentDelta = onContentDelta,
-        .onBlockComplete = onBlockComplete,
-        .onMessageComplete = onMessageComplete,
-        .onError = onError,
-        .onProviderRetry = onProviderRetry,
-    };
-    fn onMessageStart(_: *anyopaque, _: conversation.MessageRole) anyerror!void {}
-    fn onBlockStart(_: *anyopaque, _: provider_mod.ContentBlockType, _: usize) anyerror!void {}
-    fn onToolDetails(_: *anyopaque, _: usize, _: []const u8, _: []const u8) anyerror!void {}
-    fn onContentDelta(_: *anyopaque, _: usize, _: []const u8) anyerror!void {}
-    fn onBlockComplete(_: *anyopaque, _: usize, _: conversation.ContentBlock) anyerror!void {}
-    fn onMessageComplete(_: *anyopaque, _: conversation.Message, _: ?provider_mod.Usage) anyerror!void {}
-    fn onError(_: *anyopaque, _: anyerror) void {}
-    fn onProviderRetry(_: *anyopaque, _: provider_mod.ProviderRetryInfo) void {}
-};
-
 fn isValidToolInput(input: []const u8) bool {
     if (input.len == 0) return true;
     if (input[0] != '{') return true; // legacy tests/tools may use opaque bytes
@@ -236,8 +207,8 @@ pub const Agent = struct {
     persist_provider: []const u8 = "",
     persist_model: []const u8 = "",
     /// Injectable streaming seam. Defaults to the real provider dispatch
-    /// (`provider_mod.streamStep`); tests override it with a stub.
-    stream_fn: provider_mod.StreamFn = provider_mod.streamStep,
+    /// (`provider_mod.openStream`); tests override it with a stub.
+    open_stream_fn: provider_mod.OpenStreamFn = provider_mod.openStream,
     /// Compaction system prompt used for automatic compaction on context
     /// overflow. Borrowed; set by the embedder (resolved from its
     /// `COMPACTION.md` layers). When null, auto-compaction is disabled and
@@ -313,24 +284,6 @@ pub const Agent = struct {
         return self.config.registry;
     }
 
-    /// Add a user message to the conversation and durably persist it
-    /// immediately (before any provider call). The user prompt is logged on
-    /// submission, so a crash before the model replies leaves a recoverable
-    /// dangling prompt in the store.
-    pub fn submitUserMessage(self: *Agent, text: []const u8) !void {
-        const start = self.conversation.messages.items.len;
-        try self.conversation.addUserMessage(text);
-        const pm = self.providerModel();
-        try turn_persist.persistTurn(
-            self.allocator,
-            self.session_store,
-            &self.conversation,
-            start,
-            pm.provider,
-            pm.model,
-        );
-    }
-
     /// Add a system message (append or replace mode) to the conversation
     /// and persist it. The persisted entry records the mode so replay
     /// reconstructs the same effective system prompt.
@@ -355,45 +308,51 @@ pub const Agent = struct {
         );
     }
 
-    /// Drive the conversation forward until the model stops calling tools.
+    /// The user's submission that opens a turn. A struct (not a bare slice)
+    /// so it can grow to carry file/image attachments alongside the chat
+    /// text without changing `run`'s signature.
+    pub const UserMessage = struct {
+        text: []const u8,
+    };
+
+    /// Submit a user message and begin a turn, returning a resumable pull
+    /// `Stream`.
     ///
-    /// Persists everything the turn appends to the conversation: assistant
-    /// messages and tool-result user messages. Persistence runs on exit
-    /// regardless of how the turn ended — including error paths that
-    /// committed messages before failing — so a partial turn is durably
-    /// logged. An automatic compaction during the turn is persisted as a
-    /// fresh compaction window instead.
-    pub fn runStep(
-        self: *Agent,
-        receiver: *provider_mod.Receiver,
-    ) !void {
+    /// The user message is appended to the conversation and durably persisted
+    /// *immediately* (before any provider call), so a crash before the model
+    /// replies leaves a recoverable dangling prompt in the store. No provider
+    /// I/O happens here: the request opens lazily on the first
+    /// `Stream.next()`.
+    ///
+    /// The returned `*Stream` is heap-allocated; the caller owns it and must
+    /// `deinit` it. Persistence of whatever the turn committed runs when the
+    /// stream reaches its terminal `turn_complete` or is `deinit`ed early
+    /// (so a partial turn is still durably logged), mirroring the previous
+    /// `runStep` exit-path guarantee.
+    ///
+    /// The agent re-reads its `config` snapshot at the top of each provider
+    /// response inside the stream, so a mid-conversation `setConfig` takes
+    /// effect at the next response boundary, never mid-stream.
+    pub fn run(self: *Agent, message: UserMessage) !*Stream {
         self.auto_compacted = false;
-        const conv = &self.conversation;
-        const start = conv.messages.items.len;
 
-        // Persist whatever the turn committed, on every exit path.
-        defer self.persistTurnTail(start) catch |e| {
-            std.log.err("session: failed to persist turn: {t}", .{e});
-        };
-
-        while (true) {
-            // Re-read the config snapshot at the top of each turn so a
-            // mid-conversation swap takes effect here, never mid-stream.
-            const cfg = self.config;
-            try self.streamWithRetries(cfg, receiver);
-
-            const last = conv.messages.items[conv.messages.items.len - 1];
-            std.debug.assert(last.role == .assistant);
-
-            // Defense-in-depth: a provider that silently committed an
-            // empty assistant message means the turn made no observable
-            // progress. Surface it instead of looping back to the prompt.
-            if (last.content.items.len == 0) return error.EmptyAssistantResponse;
-
-            if (!hasToolUseBlock(last)) return;
+        // Append + persist the user prompt up front (the dangling-prompt
+        // recovery guarantee).
+        const user_start = self.conversation.messages.items.len;
+        try self.conversation.addUserMessage(message.text);
+        const pm = self.providerModel();
+        try turn_persist.persistTurn(
+            self.allocator,
+            self.session_store,
+            &self.conversation,
+            user_start,
+            pm.provider,
+            pm.model,
+        );
 
-            try self.dispatchToolCalls(last);
-        }
+        const s = try self.allocator.create(Stream);
+        s.* = Stream.init(self);
+        return s;
     }
 
     /// Persist the messages a turn produced. When the turn auto-compacted,
@@ -429,9 +388,12 @@ pub const Agent = struct {
         return false;
     }
 
-    /// Drive one provider turn with the configured retry policy.
+    /// Open one provider response with the configured retry policy, pushing
+    /// any informational `provider_retry` events into `out`. Returns the
+    /// resumable `ProviderStream` once a request has been successfully
+    /// opened (headers received), or propagates a terminal error.
     ///
-    /// Decision path for a failed attempt:
+    /// Decision path for a failed open:
     ///   - `ContextOverflow`: compact once, then retry the same request a
     ///     single time against the compacted conversation (a one-shot path,
     ///     independent of the transient-retry budget).
@@ -442,29 +404,30 @@ pub const Agent = struct {
     ///   - anything else (auth, bad request, cancellation, local errors):
     ///     propagate immediately.
     ///
-    /// A failed attempt never mutates the conversation (providers commit the
+    /// A failed open never mutates the conversation (providers commit the
     /// assistant message only on success), so each retry runs against the
-    /// same snapshot.
-    fn streamWithRetries(
+    /// same snapshot. Mid-stream failures (surfaced from `produce`) re-enter
+    /// here via `Stream`, replaying the response from a fresh open — exactly
+    /// as the previous push loop replayed a failed `streamStep`.
+    fn openWithRetries(
         self: *Agent,
         cfg: *const Config,
-        receiver: *provider_mod.Receiver,
-    ) !void {
+        out: *stream_mod.EventQueue,
+    ) !provider_mod.ProviderStream {
         const policy = cfg.retry;
         var attempt: usize = 1;
         while (true) {
             var diag: provider_mod.ProviderDiagnostic = .{};
-            self.stream_fn(self.allocator, self.io, cfg, &self.conversation, receiver, &diag) catch |err| {
+            const ps = self.open_stream_fn(self.allocator, self.io, cfg, &self.conversation, &diag) catch |err| {
                 if (err == error.ContextOverflow) {
-                    try self.handleContextOverflow(cfg, receiver, err);
-                    return;
+                    return self.handleContextOverflow(cfg, out, err);
                 }
                 if (!provider_mod.isRetryableProviderError(err)) return err;
                 // Out of attempts: hard-fail with the last error.
                 if (attempt >= policy.max_attempts) return err;
 
                 const delay_ms = self.backoffDelayMs(policy, attempt, diag.retry_after_ms);
-                receiver.onProviderRetry(.{
+                try out.push(.{ .provider_retry = .{
                     .attempt = attempt,
                     .max_attempts = policy.max_attempts,
                     .delay_ms = delay_ms,
@@ -472,7 +435,7 @@ pub const Agent = struct {
                     .status_code = diag.status_code,
                     .retry_after_ms = diag.retry_after_ms,
                     .message = diag.message,
-                });
+                } });
                 if (delay_ms > 0) {
                     const ms: i64 = @intCast(@min(delay_ms, std.math.maxInt(i64)));
                     self.io.sleep(.fromMilliseconds(ms), .real) catch |e| return e;
@@ -480,36 +443,35 @@ pub const Agent = struct {
                 attempt += 1;
                 continue;
             };
-            return;
+            return ps;
         }
     }
 
-    /// One-shot context-overflow recovery: compact once, retry once. Mirrors
-    /// the prior inline behavior, now fired from `streamWithRetries`. The
-    /// retry is announced through `onProviderRetry` with `compaction = true`
-    /// and `delay_ms = 0`.
+    /// One-shot context-overflow recovery: compact once, retry once. Pushes
+    /// a `provider_retry` event with `compaction = true` and `delay_ms = 0`,
+    /// then re-opens the request against the compacted context. A second
+    /// overflow (or any other error) propagates.
     fn handleContextOverflow(
         self: *Agent,
         cfg: *const Config,
-        receiver: *provider_mod.Receiver,
+        out: *stream_mod.EventQueue,
         err: anyerror,
-    ) !void {
+    ) !provider_mod.ProviderStream {
         if (self.auto_compacted) return err; // already retried once this turn
         const sys = self.compaction_system_prompt orelse return err;
         const res = try self.compact(sys, null);
         if (!res.compacted) return err; // nothing to shed; give up
         self.auto_compacted = true;
-        receiver.onProviderRetry(.{
+        try out.push(.{ .provider_retry = .{
             .attempt = 1,
             .max_attempts = 2,
             .delay_ms = 0,
             .err = err,
             .compaction = true,
-        });
-        // Retry the same request against the compacted context. A second
-        // overflow (or any other error) propagates.
+        } });
+        // Retry the same request against the compacted context.
         var diag: provider_mod.ProviderDiagnostic = .{};
-        try self.stream_fn(self.allocator, self.io, cfg, &self.conversation, receiver, &diag);
+        return self.open_stream_fn(self.allocator, self.io, cfg, &self.conversation, &diag);
     }
 
     /// Compute the backoff delay (ms) for the just-failed `attempt`
@@ -829,11 +791,20 @@ pub const Agent = struct {
         try conv.addSystemMessage(system_prompt);
         try conv.addUserMessage(body);
 
-        var capture = CompactionCapture{ .allocator = alloc };
-        defer capture.deinit();
-        var recv = capture.receiver();
-
-        try self.stream_fn(alloc, self.io, cfg, &conv, &recv, null);
+        // Drive one provider response to completion, ignoring every event.
+        // Compaction doesn't need incremental output — the assembled message
+        // is read off the conversation below — so we just pump the pull
+        // stream until it commits the assistant message.
+        var queue = stream_mod.EventQueue.init(alloc);
+        defer queue.deinit();
+        var ps = try self.open_stream_fn(alloc, self.io, cfg, &conv, null);
+        defer ps.deinit();
+        while (true) {
+            const status = try ps.produce(&queue);
+            // Drain (and discard) any events to bound the queue/arena.
+            while (queue.pop()) |_| {}
+            if (status == .response_complete) break;
+        }
 
         // The provider appended an assistant message; gather its text.
         const last = conv.messages.items[conv.messages.items.len - 1];
@@ -1081,6 +1052,206 @@ pub const Agent = struct {
     }
 };
 
+/// A resumable pull handle over one agent turn.
+///
+/// `next()` pulls one `Event` at a time, driving the agent loop
+/// incrementally: open a provider response, stream its events, dispatch any
+/// tool calls between responses, and repeat until the model stops calling
+/// tools. The whole loop's state lives here (not on a stack frame), so the
+/// turn can suspend and resume between events.
+///
+/// Contract (see `stream.zig`):
+///   - an `Event` value is streaming progress, including `turn_complete`;
+///   - `null` means exhausted (already past `turn_complete`), never before;
+///   - an error is a genuine failure (network/parse/provider).
+///
+/// Event payloads borrow from the stream's decode state or the conversation
+/// and are valid only until the next `next()` call.
+pub const Stream = struct {
+    agent: *Agent,
+    queue: stream_mod.EventQueue,
+    phase: Phase,
+    /// The active provider response, when in `.streaming`.
+    response: ?provider_mod.ProviderStream = null,
+    /// First message index of this turn (for persistence).
+    start: usize,
+    /// Set once the turn's tail has been persisted (on terminal or deinit).
+    persisted: bool = false,
+    /// A terminal error to surface once any already-queued events (e.g.
+    /// `provider_retry` notices pushed before the failing attempt) have been
+    /// drained. `next()` yields the queue first, then this error.
+    pending_error: ?anyerror = null,
+
+    const Phase = enum {
+        /// Open the next provider response (with retries).
+        turn_start,
+        /// Pump the active provider response into events.
+        streaming,
+        /// A provider response completed; decide tools-vs-done.
+        after_response,
+        /// The turn reached its terminal `turn_complete`.
+        done,
+        /// A failure already propagated; `next()` is poisoned.
+        failed,
+    };
+
+    fn init(agent: *Agent) Stream {
+        return .{
+            .agent = agent,
+            .queue = stream_mod.EventQueue.init(agent.allocator),
+            .phase = .turn_start,
+            .start = agent.conversation.messages.items.len,
+        };
+    }
+
+    pub fn deinit(self: *Stream) void {
+        // Persist whatever the turn committed, on every exit path — including
+        // dropping the stream mid-turn after some messages were committed.
+        self.persistTail();
+        if (self.response) |ps| ps.deinit();
+        self.queue.deinit();
+        self.agent.allocator.destroy(self);
+    }
+
+    fn persistTail(self: *Stream) void {
+        if (self.persisted) return;
+        self.persisted = true;
+        self.agent.persistTurnTail(self.start) catch |e| {
+            std.log.err("session: failed to persist turn: {t}", .{e});
+        };
+    }
+
+    /// Pull the next event, or null past the terminal. See the contract
+    /// above.
+    pub fn next(self: *Stream) !?Event {
+        // Always drain queued events first; they borrow decode/conversation
+        // state valid until this call returns.
+        if (self.queue.pop()) |ev| return ev;
+        // Queue drained: a deferred terminal error surfaces now.
+        if (self.pending_error) |err| {
+            self.pending_error = null;
+            self.phase = .failed;
+            return err;
+        }
+
+        while (true) {
+            switch (self.phase) {
+                .done => return null,
+                .failed => return error.StreamPoisoned,
+                .turn_start => {
+                    // Re-read the config snapshot at each response boundary
+                    // so a mid-conversation swap takes effect here, never
+                    // mid-stream.
+                    const cfg = self.agent.config;
+                    const ps = self.agent.openWithRetries(cfg, &self.queue) catch |err| {
+                        // Surface the failure after any queued retry notices
+                        // (pushed before each backoff) are drained.
+                        if (self.queue.pop()) |ev| {
+                            self.pending_error = err;
+                            return ev;
+                        }
+                        self.phase = .failed;
+                        return err;
+                    };
+                    self.response = ps;
+                    self.phase = .streaming;
+                    if (self.queue.pop()) |ev| return ev; // retry notices
+                },
+                .streaming => {
+                    const ps = self.response.?;
+                    const status = ps.produce(&self.queue) catch |err| {
+                        // Mid-stream failure. The conversation was not
+                        // mutated (commit happens only at response
+                        // completion), so retry by re-opening from scratch,
+                        // exactly as the prior push loop replayed a failed
+                        // streamStep. Non-retryable errors propagate.
+                        ps.deinit();
+                        self.response = null;
+                        if (!provider_mod.isRetryableProviderError(err)) {
+                            self.phase = .failed;
+                            return err;
+                        }
+                        self.phase = .turn_start;
+                        // Emit a retry notice so consumers see the stall.
+                        const cfg = self.agent.config;
+                        const delay_ms = self.agent.backoffDelayMs(cfg.retry, 1, null);
+                        self.queue.push(.{ .provider_retry = .{
+                            .attempt = 1,
+                            .max_attempts = cfg.retry.max_attempts,
+                            .delay_ms = delay_ms,
+                            .err = err,
+                        } }) catch |e| {
+                            self.phase = .failed;
+                            return e;
+                        };
+                        if (delay_ms > 0) {
+                            const ms: i64 = @intCast(@min(delay_ms, std.math.maxInt(i64)));
+                            self.agent.io.sleep(.fromMilliseconds(ms), .real) catch |e| {
+                                self.phase = .failed;
+                                return e;
+                            };
+                        }
+                        if (self.queue.pop()) |ev| return ev;
+                        continue;
+                    };
+                    if (self.queue.pop()) |ev| return ev;
+                    if (status == .response_complete) {
+                        ps.deinit();
+                        self.response = null;
+                        self.phase = .after_response;
+                    }
+                    // else `.more`: loop and pump again.
+                },
+                .after_response => {
+                    const conv = &self.agent.conversation;
+                    const last = conv.messages.items[conv.messages.items.len - 1];
+                    std.debug.assert(last.role == .assistant);
+
+                    // Defense-in-depth: a provider that silently committed an
+                    // empty assistant message means the turn made no
+                    // observable progress. Surface it instead of looping.
+                    if (last.content.items.len == 0) {
+                        self.phase = .failed;
+                        return error.EmptyAssistantResponse;
+                    }
+
+                    if (!Agent.hasToolUseBlock(last)) {
+                        self.phase = .done;
+                        self.persistTail();
+                        return .turn_complete;
+                    }
+
+                    // Dispatch the tool calls, bracketed by boundary events.
+                    const count = toolUseCount(last);
+                    self.queue.push(.{ .tool_dispatch_start = .{ .count = count } }) catch |e| {
+                        self.phase = .failed;
+                        return e;
+                    };
+                    self.agent.dispatchToolCalls(last) catch |err| {
+                        self.phase = .failed;
+                        return err;
+                    };
+                    const result_msg = conv.messages.items[conv.messages.items.len - 1];
+                    self.queue.push(.{ .tool_dispatch_complete = .{ .message = result_msg } }) catch |e| {
+                        self.phase = .failed;
+                        return e;
+                    };
+                    self.phase = .turn_start;
+                    if (self.queue.pop()) |ev| return ev;
+                },
+            }
+        }
+    }
+};
+
+fn toolUseCount(msg: conversation.Message) usize {
+    var n: usize = 0;
+    for (msg.content.items) |block| {
+        if (block == .ToolUse) n += 1;
+    }
+    return n;
+}
+
 /// One ToolUse, as flattened into the agent's dispatch list. `result`
 /// and `err` are filled in by the worker; exactly one is non-null on
 /// successful task completion.
@@ -1249,6 +1420,17 @@ fn runSourceGroup(agent: *Agent, sg: Group.SourceGroup, calls: []FlatCall) void
 
 const testing = std.testing;
 
+/// Test helper: submit `text` and drive the whole turn to completion via the
+/// pull `Stream`, discarding every event (the agent tests assert on
+/// conversation/store state, not on the event stream). Mirrors the old
+/// `submitUserMessage` + `runStep`: it returns the same terminal error a
+/// turn would raise.
+fn drainTurn(agent: *Agent, text: []const u8) !void {
+    var s = try agent.run(.{ .text = text });
+    defer s.deinit();
+    while (try s.next()) |_| {}
+}
+
 /// Test helper: the items of a ToolResultBlock's first text part.
 fn trText(tr: conversation.ToolResultBlock) []const u8 {
     for (tr.parts.items) |p| {
@@ -1257,31 +1439,30 @@ fn trText(tr: conversation.ToolResultBlock) []const u8 {
     return "";
 }
 
-/// Test harness for the injectable `stream_fn` seam.
+/// Test harness for the injectable `open_stream_fn` seam.
 ///
-/// `provider_mod.StreamFn` carries no user context (it mirrors the real
+/// `provider_mod.OpenStreamFn` carries no user context (it mirrors the real
 /// free function exactly), so the stub parks its state in a module-level
-/// pointer that `stubStreamStep` reads. The Zig test runner executes tests
-/// serially in one process, so a single global slot is safe; each test
-/// sets it via `install` before driving the agent.
+/// pointer that `stubOpenStream` reads. The Zig test runner executes tests
+/// serially in one process, so a single global slot is safe; each test sets
+/// it via `install` before driving the agent.
 var stub_active: ?*StubProvider = null;
 
 const StubProvider = struct {
     allocator: Allocator,
     scripted: []const ScriptedTurn,
     next: usize = 0,
-    /// Number of leading stream calls that should fail with
+    /// Number of leading stream opens that should fail with
     /// `error.ContextOverflow` before any scripted turn is served. Used to
     /// drive the auto-compaction path. Decremented on each overflow.
     overflow_calls: usize = 0,
     /// A queue of provider errors to return, in order, before any scripted
-    /// turn is served. Each entry is consumed on one stream call. Used to
-    /// drive the transient-retry path. `diag_retry_after_ms`, when set on an
-    /// entry, is stashed into the caller's `ProviderDiagnostic`.
+    /// turn is served. Each entry is consumed on one open. Used to drive the
+    /// transient-retry path.
     scripted_errors: []const ScriptedError = &.{},
     error_idx: usize = 0,
-    /// Count of stream calls observed (failed + succeeded). Lets tests
-    /// assert the exact number of attempts.
+    /// Count of opens observed (failed + succeeded). Lets tests assert the
+    /// exact number of attempts.
     calls_made: usize = 0,
 
     const ScriptedError = struct {
@@ -1293,7 +1474,7 @@ const StubProvider = struct {
     const ScriptedTurn = struct {
         blocks: []const TestBlock,
         /// Optional provider usage to stamp on the committed assistant
-        /// message (mirrors a real provider's `onMessageComplete` usage).
+        /// message (mirrors a real provider's terminal usage).
         usage: ?conversation.Usage = null,
     };
 
@@ -1307,24 +1488,94 @@ const StubProvider = struct {
     };
 
     /// Point the global seam at this stub and return the function to assign
-    /// to `agent.stream_fn`. Call once per test, after constructing the
+    /// to `agent.open_stream_fn`. Call once per test, after constructing the
     /// stub on the stack.
-    fn install(self: *StubProvider) provider_mod.StreamFn {
+    fn install(self: *StubProvider) provider_mod.OpenStreamFn {
         stub_active = self;
-        return stubStreamStep;
+        return stubOpenStream;
     }
 };
 
-fn stubStreamStep(
+/// A canned resumable response: on the first `produce` it commits the
+/// scripted assistant message to the conversation and pushes the terminal
+/// `message_complete`, then reports `.response_complete`. It does not emit
+/// per-block events (the agent tests assert on conversation state, not the
+/// event stream), which is sufficient for driving the agent loop.
+const StubResponse = struct {
+    allocator: Allocator,
+    conv: *conversation.Conversation,
+    turn: StubProvider.ScriptedTurn,
+    done: bool = false,
+
+    fn create(
+        allocator: Allocator,
+        conv: *conversation.Conversation,
+        turn: StubProvider.ScriptedTurn,
+    ) !provider_mod.ProviderStream {
+        const self = try allocator.create(StubResponse);
+        self.* = .{ .allocator = allocator, .conv = conv, .turn = turn };
+        return .{ .ptr = self, .vtable = &vtable };
+    }
+
+    const vtable: provider_mod.ProviderStream.VTable = .{
+        .produce = produceVT,
+        .deinit = deinitVT,
+    };
+
+    fn produceVT(ptr: *anyopaque, out: *stream_mod.EventQueue) anyerror!provider_mod.ProviderStream.ProduceStatus {
+        const self: *StubResponse = @ptrCast(@alignCast(ptr));
+        if (self.done) return .response_complete;
+        self.done = true;
+
+        var blocks: std.ArrayList(conversation.ContentBlock) = .empty;
+        errdefer {
+            for (blocks.items) |*b| b.deinit(self.allocator);
+            blocks.deinit(self.allocator);
+        }
+        for (self.turn.blocks) |tb| {
+            switch (tb) {
+                .Text => |s| try blocks.append(self.allocator, .{
+                    .Text = try conversation.textualBlockFromSlice(self.allocator, s),
+                }),
+                .ToolUse => |tu| {
+                    const id = try self.allocator.dupe(u8, tu.id);
+                    errdefer self.allocator.free(id);
+                    const name = try self.allocator.dupe(u8, tu.name);
+                    errdefer self.allocator.free(name);
+                    var input_buf: conversation.TextualBlock = .empty;
+                    errdefer input_buf.deinit(self.allocator);
+                    try input_buf.appendSlice(self.allocator, tu.input);
+                    try blocks.append(self.allocator, .{ .ToolUse = .{
+                        .id = id,
+                        .name = name,
+                        .input = input_buf,
+                    } });
+                },
+            }
+        }
+        const moved = try blocks.toOwnedSlice(self.allocator);
+        defer self.allocator.free(moved);
+        try self.conv.addAssistantMessageWithUsage(moved, self.turn.usage);
+
+        const msg = self.conv.messages.items[self.conv.messages.items.len - 1];
+        try out.push(.{ .message_complete = .{ .message = msg, .usage = self.turn.usage } });
+        return .response_complete;
+    }
+
+    fn deinitVT(ptr: *anyopaque) void {
+        const self: *StubResponse = @ptrCast(@alignCast(ptr));
+        self.allocator.destroy(self);
+    }
+};
+
+fn stubOpenStream(
     allocator: Allocator,
     _: Io,
     _: *const config_mod.Config,
     conv: *conversation.Conversation,
-    _: *provider_mod.Receiver,
     diag: ?*provider_mod.ProviderDiagnostic,
-) anyerror!void {
+) anyerror!provider_mod.ProviderStream {
     const self = stub_active orelse return error.NoStubInstalled;
-    _ = allocator;
     self.calls_made += 1;
     if (self.error_idx < self.scripted_errors.len) {
         const e = self.scripted_errors[self.error_idx];
@@ -1342,38 +1593,7 @@ fn stubStreamStep(
     if (self.next >= self.scripted.len) return error.NoMoreScriptedTurns;
     const turn = self.scripted[self.next];
     self.next += 1;
-
-    var blocks: std.ArrayList(conversation.ContentBlock) = .empty;
-    errdefer {
-        for (blocks.items) |*b| b.deinit(self.allocator);
-        blocks.deinit(self.allocator);
-    }
-    for (turn.blocks) |tb| {
-        switch (tb) {
-            .Text => |s| {
-                try blocks.append(self.allocator, .{
-                    .Text = try conversation.textualBlockFromSlice(self.allocator, s),
-                });
-            },
-            .ToolUse => |tu| {
-                const id = try self.allocator.dupe(u8, tu.id);
-                errdefer self.allocator.free(id);
-                const name = try self.allocator.dupe(u8, tu.name);
-                errdefer self.allocator.free(name);
-                var input_buf: conversation.TextualBlock = .empty;
-                errdefer input_buf.deinit(self.allocator);
-                try input_buf.appendSlice(self.allocator, tu.input);
-                try blocks.append(self.allocator, .{ .ToolUse = .{
-                    .id = id,
-                    .name = name,
-                    .input = input_buf,
-                } });
-            },
-        }
-    }
-    const moved = try blocks.toOwnedSlice(self.allocator);
-    defer self.allocator.free(moved);
-    try conv.addAssistantMessageWithUsage(moved, turn.usage);
+    return StubResponse.create(allocator, conv, turn);
 }
 
 /// Build a stack registry + active `Config` snapshot wired together, for
@@ -1551,31 +1771,6 @@ const HardFailTool = struct {
     }
 };
 
-const NoopReceiver = struct {
-    fn make() provider_mod.Receiver {
-        return .{ .ptr = @ptrCast(@constCast(&dummy)), .vtable = &vt };
-    }
-    var dummy: u8 = 0;
-    const vt: provider_mod.ReceiverVTable = .{
-        .onMessageStart = noop1,
-        .onBlockStart = noop2,
-        .onToolDetails = noopToolDetails,
-        .onContentDelta = noop3,
-        .onBlockComplete = noop4,
-        .onMessageComplete = noop5,
-        .onError = noop6,
-        .onProviderRetry = noop7,
-    };
-    fn noop1(_: *anyopaque, _: conversation.MessageRole) anyerror!void {}
-    fn noop2(_: *anyopaque, _: provider_mod.ContentBlockType, _: usize) anyerror!void {}
-    fn noopToolDetails(_: *anyopaque, _: usize, _: []const u8, _: []const u8) anyerror!void {}
-    fn noop3(_: *anyopaque, _: usize, _: []const u8) anyerror!void {}
-    fn noop4(_: *anyopaque, _: usize, _: conversation.ContentBlock) anyerror!void {}
-    fn noop5(_: *anyopaque, _: conversation.Message, _: ?provider_mod.Usage) anyerror!void {}
-    fn noop6(_: *anyopaque, _: anyerror) void {}
-    fn noop7(_: *anyopaque, _: provider_mod.ProviderRetryInfo) void {}
-};
-
 /// An in-memory `SessionStore` test double: records every appended
 /// `DiskMessage` (role + provider/model stamp) so tests can assert the
 /// agent persisted the right turn without touching disk. Honors the store
@@ -1661,13 +1856,11 @@ test "agent persists user, assistant, and tool-result messages of a turn" {
     defer cap.deinit();
     var agent = Agent.init(allocator, io, &h.config, cap.store(), null);
     defer agent.deinit();
-    agent.stream_fn = stub.install();
+    agent.open_stream_fn = stub.install();
     agent.persist_provider = "openai";
     agent.persist_model = "gpt-4o";
 
-    var recv = NoopReceiver.make();
-    try agent.submitUserMessage("call a tool");
-    try agent.runStep(&recv);
+    try drainTurn(&agent, "call a tool");
 
     // Persisted, in order: user prompt, assistant(ToolUse), user(ToolResult),
     // assistant(text).
@@ -1701,11 +1894,9 @@ test "agent runs a turn against NullStore without persisting or erroring" {
     var ns = null_store_mod.NullStore.init(allocator);
     var agent = Agent.init(allocator, io, &h.config, ns.store(), null);
     defer agent.deinit();
-    agent.stream_fn = stub.install();
+    agent.open_stream_fn = stub.install();
 
-    var recv = NoopReceiver.make();
-    try agent.submitUserMessage("hello");
-    try agent.runStep(&recv);
+    try drainTurn(&agent, "hello");
 
     // Nothing crashed; the conversation has the user + assistant messages.
     try testing.expectEqual(@as(usize, 2), agent.conversation.messages.items.len);
@@ -1992,13 +2183,11 @@ test "runStep dispatches a tool call and loops to a final text turn" {
     var ns = null_store_mod.NullStore.init(allocator);
     var agent = Agent.init(allocator, io, &h.config, ns.store(), null);
     defer agent.deinit();
-    agent.stream_fn = stub.install();
+    agent.open_stream_fn = stub.install();
 
     const conv = &agent.conversation;
-    try conv.addUserMessage("call a tool");
 
-    var recv = NoopReceiver.make();
-    try agent.runStep(&recv);
+    try drainTurn(&agent, "call a tool");
 
     try testing.expectEqual(@as(usize, 4), conv.messages.items.len);
 
@@ -2044,13 +2233,11 @@ test "runStep dispatches multiple tool calls in parallel" {
     var ns = null_store_mod.NullStore.init(allocator);
     var agent = Agent.init(allocator, io, &h.config, ns.store(), null);
     defer agent.deinit();
-    agent.stream_fn = stub.install();
+    agent.open_stream_fn = stub.install();
 
     const conv = &agent.conversation;
-    try conv.addUserMessage("go");
 
-    var recv = NoopReceiver.make();
-    try agent.runStep(&recv);
+    try drainTurn(&agent, "go");
 
     const tr_msg = conv.messages.items[2];
     try testing.expectEqual(@as(usize, 3), tr_msg.content.items.len);
@@ -2085,13 +2272,11 @@ test "runStep: native tool handler error becomes an error result and the model g
     var ns = null_store_mod.NullStore.init(allocator);
     var agent = Agent.init(allocator, io, &h.config, ns.store(), null);
     defer agent.deinit();
-    agent.stream_fn = stub.install();
+    agent.open_stream_fn = stub.install();
 
     const conv = &agent.conversation;
-    try conv.addUserMessage("break it");
 
-    var recv = NoopReceiver.make();
-    try agent.runStep(&recv);
+    try drainTurn(&agent, "break it");
 
     // user, assistant(tool_use), user(tool_result), assistant(text)
     try testing.expectEqual(@as(usize, 4), conv.messages.items.len);
@@ -2120,13 +2305,11 @@ test "runStep: unknown tool becomes an error tool result and the loop continues"
     var ns = null_store_mod.NullStore.init(allocator);
     var agent = Agent.init(allocator, io, &h.config, ns.store(), null);
     defer agent.deinit();
-    agent.stream_fn = stub.install();
+    agent.open_stream_fn = stub.install();
 
     const conv = &agent.conversation;
-    try conv.addUserMessage("call a ghost");
 
-    var recv = NoopReceiver.make();
-    try agent.runStep(&recv);
+    try drainTurn(&agent, "call a ghost");
 
     // messages: user, assistant(tool_use), user(tool_result), assistant(text)
     try testing.expectEqual(@as(usize, 4), conv.messages.items.len);
@@ -2154,13 +2337,11 @@ test "runStep with no tool calls returns after one provider step" {
     var ns = null_store_mod.NullStore.init(allocator);
     var agent = Agent.init(allocator, io, &h.config, ns.store(), null);
     defer agent.deinit();
-    agent.stream_fn = stub.install();
+    agent.open_stream_fn = stub.install();
 
     const conv = &agent.conversation;
-    try conv.addUserMessage("hello");
 
-    var recv = NoopReceiver.make();
-    try agent.runStep(&recv);
+    try drainTurn(&agent, "hello");
 
     try testing.expectEqual(@as(usize, 2), conv.messages.items.len);
     try testing.expectEqualStrings("hi", conv.messages.items[1].content.items[0].Text.items);
@@ -2182,13 +2363,10 @@ test "runStep surfaces EmptyAssistantResponse when provider commits an empty mes
     var ns = null_store_mod.NullStore.init(allocator);
     var agent = Agent.init(allocator, io, &h.config, ns.store(), null);
     defer agent.deinit();
-    agent.stream_fn = stub.install();
+    agent.open_stream_fn = stub.install();
 
-    const conv = &agent.conversation;
-    try conv.addUserMessage("hi");
 
-    var recv = NoopReceiver.make();
-    try testing.expectError(error.EmptyAssistantResponse, agent.runStep(&recv));
+    try testing.expectError(error.EmptyAssistantResponse, drainTurn(&agent, "hi"));
 }
 
 // ------------ ToolSource tests ------------
@@ -2215,13 +2393,11 @@ test "runStep delivers all source-backed calls in one batch on one thread" {
     var ns = null_store_mod.NullStore.init(allocator);
     var agent = Agent.init(allocator, io, &h.config, ns.store(), null);
     defer agent.deinit();
-    agent.stream_fn = stub.install();
+    agent.open_stream_fn = stub.install();
 
     const conv = &agent.conversation;
-    try conv.addUserMessage("go");
 
-    var recv = NoopReceiver.make();
-    try agent.runStep(&recv);
+    try drainTurn(&agent, "go");
 
     // Locate the source and inspect its observed batches.
     const view = h.registry.lookup("lua_x") orelse return error.NotFound;
@@ -2268,13 +2444,10 @@ test "runStep: distinct sources run on distinct threads in parallel" {
     var ns = null_store_mod.NullStore.init(allocator);
     var agent = Agent.init(allocator, io, &h.config, ns.store(), null);
     defer agent.deinit();
-    agent.stream_fn = stub.install();
+    agent.open_stream_fn = stub.install();
 
-    const conv = &agent.conversation;
-    try conv.addUserMessage("go");
 
-    var recv = NoopReceiver.make();
-    try agent.runStep(&recv);
+    try drainTurn(&agent, "go");
 
     const view_a = h.registry.lookup("src_a_t") orelse return error.NotFound;
     const view_b = h.registry.lookup("src_b_t") orelse return error.NotFound;
@@ -2308,13 +2481,11 @@ test "runStep: source whole-batch error becomes per-call error results and conti
     var ns = null_store_mod.NullStore.init(allocator);
     var agent = Agent.init(allocator, io, &h.config, ns.store(), null);
     defer agent.deinit();
-    agent.stream_fn = stub.install();
+    agent.open_stream_fn = stub.install();
 
     const conv = &agent.conversation;
-    try conv.addUserMessage("kaboom");
 
-    var recv = NoopReceiver.make();
-    try agent.runStep(&recv);
+    try drainTurn(&agent, "kaboom");
 
     // user, assistant(tool_use x2), user(tool_result x2), assistant(text)
     try testing.expectEqual(@as(usize, 4), conv.messages.items.len);
@@ -2353,13 +2524,11 @@ test "runStep: mixed single Tools and source-backed tools coexist in one turn" {
     var ns = null_store_mod.NullStore.init(allocator);
     var agent = Agent.init(allocator, io, &h.config, ns.store(), null);
     defer agent.deinit();
-    agent.stream_fn = stub.install();
+    agent.open_stream_fn = stub.install();
 
     const conv = &agent.conversation;
-    try conv.addUserMessage("go");
 
-    var recv = NoopReceiver.make();
-    try agent.runStep(&recv);
+    try drainTurn(&agent, "go");
 
     const tr_msg = conv.messages.items[2];
     try testing.expectEqual(@as(usize, 3), tr_msg.content.items.len);
@@ -2406,7 +2575,7 @@ test "setConfig swaps the visible tool set between turns" {
     var ns = null_store_mod.NullStore.init(allocator);
     var agent = Agent.init(allocator, io, &cfg_a, ns.store(), null);
     defer agent.deinit();
-    agent.stream_fn = stub.install();
+    agent.open_stream_fn = stub.install();
 
     // Under A: `echo` visible, `late` not.
     try testing.expect(agent.config.registry.lookup("echo") != null);
@@ -2420,9 +2589,7 @@ test "setConfig swaps the visible tool set between turns" {
     // A real turn under B resolves `late` (which would have been
     // UnknownTool under A), then loops to the final text turn.
     const conv = &agent.conversation;
-    try conv.addUserMessage("go");
-    var recv = NoopReceiver.make();
-    try agent.runStep(&recv);
+    try drainTurn(&agent, "go");
 
     const tr = conv.messages.items[2].content.items[0].ToolResult;
     try testing.expectEqualStrings("2", tr.tool_use_id);
@@ -2450,7 +2617,7 @@ test "compact: summarizes prefix, keeps suffix, system survives" {
     var ns = null_store_mod.NullStore.init(allocator);
     var agent = Agent.init(allocator, io, &h.config, ns.store(), null);
     defer agent.deinit();
-    agent.stream_fn = stub.install();
+    agent.open_stream_fn = stub.install();
 
     const conv = &agent.conversation;
     try conv.addSystemMessage("you are helpful");
@@ -2512,7 +2679,7 @@ test "compact: restated suffix usage reconstructs a fresh cumulative chain" {
     var ns = null_store_mod.NullStore.init(allocator);
     var agent = Agent.init(allocator, io, &h.config, ns.store(), null);
     defer agent.deinit();
-    agent.stream_fn = stub.install();
+    agent.open_stream_fn = stub.install();
 
     const conv = &agent.conversation;
     // Prefix turn (will be summarized). Cumulative footprint = 500+40+10+50
@@ -2585,7 +2752,7 @@ test "compact: no-op when conversation already fits the budget" {
     var ns = null_store_mod.NullStore.init(allocator);
     var agent = Agent.init(allocator, io, &h.config, ns.store(), null);
     defer agent.deinit();
-    agent.stream_fn = stub.install();
+    agent.open_stream_fn = stub.install();
 
     const conv = &agent.conversation;
     try conv.addSystemMessage("sys");
@@ -2622,7 +2789,7 @@ test "compact: extra instructions are appended to the system prompt" {
     var ns = null_store_mod.NullStore.init(allocator);
     var agent = Agent.init(allocator, io, &h.config, ns.store(), null);
     defer agent.deinit();
-    agent.stream_fn = stub.install();
+    agent.open_stream_fn = stub.install();
 
     const conv = &agent.conversation;
     try conv.addUserMessage("question one two three");
@@ -2662,7 +2829,7 @@ test "runStep: auto-compacts on context overflow and retries once" {
     var ns = null_store_mod.NullStore.init(allocator);
     var agent = Agent.init(allocator, io, &h.config, ns.store(), null);
     defer agent.deinit();
-    agent.stream_fn = stub.install();
+    agent.open_stream_fn = stub.install();
     agent.compaction_system_prompt = "Summarize the conversation.";
 
     const conv = &agent.conversation;
@@ -2671,10 +2838,8 @@ test "runStep: auto-compacts on context overflow and retries once" {
     try conv.addAssistantMessage(&.{
         .{ .Text = try conversation.textualBlockFromSlice(allocator, "first answer with several words") },
     });
-    try conv.addUserMessage("second recent question");
 
-    var recv = NoopReceiver.make();
-    try agent.runStep(&recv);
+    try drainTurn(&agent, "second recent question");
 
     try testing.expect(agent.auto_compacted);
     // After compaction + retry: [system, summary, user q2, assistant final].
@@ -2708,55 +2873,39 @@ test "runStep: context overflow without compaction prompt propagates" {
     var ns = null_store_mod.NullStore.init(allocator);
     var agent = Agent.init(allocator, io, &h.config, ns.store(), null);
     defer agent.deinit();
-    agent.stream_fn = stub.install();
+    agent.open_stream_fn = stub.install();
     // No compaction_system_prompt set -> overflow propagates.
 
-    const conv = &agent.conversation;
-    try conv.addUserMessage("hi");
 
-    var recv = NoopReceiver.make();
-    try testing.expectError(error.ContextOverflow, agent.runStep(&recv));
+    try testing.expectError(error.ContextOverflow, drainTurn(&agent, "hi"));
 }
 
 // -----------------------------------------------------------------------------
 // Phase 6: provider retry + tool-error holistic tests
 // -----------------------------------------------------------------------------
 
-/// Receiver that records `onProviderRetry` notifications (and nothing else),
-/// so tests can assert retry scheduling without a live provider.
-const RetryRecordingReceiver = struct {
+/// Records the `provider_retry` events a turn emits, so tests can assert
+/// retry scheduling without a live provider.
+const RetryRecorder = struct {
     infos: std.ArrayList(provider_mod.ProviderRetryInfo) = .empty,
     allocator: Allocator,
 
-    fn make(self: *RetryRecordingReceiver) provider_mod.Receiver {
-        return .{ .ptr = self, .vtable = &vt };
-    }
-    fn deinit(self: *RetryRecordingReceiver) void {
+    fn deinit(self: *RetryRecorder) void {
         self.infos.deinit(self.allocator);
     }
-    const vt: provider_mod.ReceiverVTable = .{
-        .onMessageStart = onMessageStart,
-        .onBlockStart = onBlockStart,
-        .onToolDetails = onToolDetails,
-        .onContentDelta = onContentDelta,
-        .onBlockComplete = onBlockComplete,
-        .onMessageComplete = onMessageComplete,
-        .onError = onError,
-        .onProviderRetry = onProviderRetry,
-    };
-    fn onMessageStart(_: *anyopaque, _: conversation.MessageRole) anyerror!void {}
-    fn onBlockStart(_: *anyopaque, _: provider_mod.ContentBlockType, _: usize) anyerror!void {}
-    fn onToolDetails(_: *anyopaque, _: usize, _: []const u8, _: []const u8) anyerror!void {}
-    fn onContentDelta(_: *anyopaque, _: usize, _: []const u8) anyerror!void {}
-    fn onBlockComplete(_: *anyopaque, _: usize, _: conversation.ContentBlock) anyerror!void {}
-    fn onMessageComplete(_: *anyopaque, _: conversation.Message, _: ?provider_mod.Usage) anyerror!void {}
-    fn onError(_: *anyopaque, _: anyerror) void {}
-    fn onProviderRetry(ptr: *anyopaque, info: provider_mod.ProviderRetryInfo) void {
-        const self: *RetryRecordingReceiver = @ptrCast(@alignCast(ptr));
-        self.infos.append(self.allocator, info) catch {};
-    }
 };
 
+/// Drive a whole turn via the pull `Stream`, recording every
+/// `provider_retry` event into `rr` and discarding the rest. Returns the
+/// same terminal error the turn would raise.
+fn drainTurnRecording(agent: *Agent, text: []const u8, rr: *RetryRecorder) !void {
+    var s = try agent.run(.{ .text = text });
+    defer s.deinit();
+    while (try s.next()) |ev| {
+        if (ev == .provider_retry) try rr.infos.append(rr.allocator, ev.provider_retry);
+    }
+}
+
 /// Build an agent + harness with near-zero backoff so retry tests don't
 /// actually sleep. Caller owns the harness and must keep it alive.
 fn fastRetryHarness(h: *TestHarness) void {
@@ -2795,15 +2944,13 @@ test "runStep: provider 429 retries then succeeds without duplicate messages" {
     var ns = null_store_mod.NullStore.init(allocator);
     var agent = Agent.init(allocator, io, &h.config, ns.store(), null);
     defer agent.deinit();
-    agent.stream_fn = stub.install();
+    agent.open_stream_fn = stub.install();
 
     const conv = &agent.conversation;
-    try conv.addUserMessage("hi");
 
-    var rr = RetryRecordingReceiver{ .allocator = allocator };
+    var rr = RetryRecorder{ .allocator = allocator };
     defer rr.deinit();
-    var recv = rr.make();
-    try agent.runStep(&recv);
+    try drainTurnRecording(&agent, "hi", &rr);
 
     // Two failures + one success.
     try testing.expectEqual(@as(usize, 3), stub.calls_made);
@@ -2840,15 +2987,12 @@ test "runStep: provider 500 retries with backoff notification" {
     var ns = null_store_mod.NullStore.init(allocator);
     var agent = Agent.init(allocator, io, &h.config, ns.store(), null);
     defer agent.deinit();
-    agent.stream_fn = stub.install();
+    agent.open_stream_fn = stub.install();
 
-    const conv = &agent.conversation;
-    try conv.addUserMessage("hi");
 
-    var rr = RetryRecordingReceiver{ .allocator = allocator };
+    var rr = RetryRecorder{ .allocator = allocator };
     defer rr.deinit();
-    var recv = rr.make();
-    try agent.runStep(&recv);
+    try drainTurnRecording(&agent, "hi", &rr);
 
     try testing.expectEqual(@as(usize, 2), stub.calls_made);
     try testing.expectEqual(@as(usize, 1), rr.infos.items.len);
@@ -2880,15 +3024,12 @@ test "runStep: provider auth failure does not retry" {
     var ns = null_store_mod.NullStore.init(allocator);
     var agent = Agent.init(allocator, io, &h.config, ns.store(), null);
     defer agent.deinit();
-    agent.stream_fn = stub.install();
+    agent.open_stream_fn = stub.install();
 
-    const conv = &agent.conversation;
-    try conv.addUserMessage("hi");
 
-    var rr = RetryRecordingReceiver{ .allocator = allocator };
+    var rr = RetryRecorder{ .allocator = allocator };
     defer rr.deinit();
-    var recv = rr.make();
-    try testing.expectError(error.ProviderAuthFailed, agent.runStep(&recv));
+    try testing.expectError(error.ProviderAuthFailed, drainTurnRecording(&agent, "hi", &rr));
 
     // Exactly one attempt, no retry notification.
     try testing.expectEqual(@as(usize, 1), stub.calls_made);
@@ -2921,15 +3062,12 @@ test "runStep: retries exhaust and hard-fail after max_attempts" {
     var ns = null_store_mod.NullStore.init(allocator);
     var agent = Agent.init(allocator, io, &h.config, ns.store(), null);
     defer agent.deinit();
-    agent.stream_fn = stub.install();
+    agent.open_stream_fn = stub.install();
 
-    const conv = &agent.conversation;
-    try conv.addUserMessage("hi");
 
-    var rr = RetryRecordingReceiver{ .allocator = allocator };
+    var rr = RetryRecorder{ .allocator = allocator };
     defer rr.deinit();
-    var recv = rr.make();
-    try testing.expectError(error.ProviderUnavailable, agent.runStep(&recv));
+    try testing.expectError(error.ProviderUnavailable, drainTurnRecording(&agent, "hi", &rr));
 
     // 4 attempts total (max_attempts), 3 retry notifications.
     try testing.expectEqual(@as(usize, 4), stub.calls_made);
@@ -2961,15 +3099,12 @@ test "runStep: Retry-After is honored and reported" {
     var ns = null_store_mod.NullStore.init(allocator);
     var agent = Agent.init(allocator, io, &h.config, ns.store(), null);
     defer agent.deinit();
-    agent.stream_fn = stub.install();
+    agent.open_stream_fn = stub.install();
 
-    const conv = &agent.conversation;
-    try conv.addUserMessage("hi");
 
-    var rr = RetryRecordingReceiver{ .allocator = allocator };
+    var rr = RetryRecorder{ .allocator = allocator };
     defer rr.deinit();
-    var recv = rr.make();
-    try agent.runStep(&recv);
+    try drainTurnRecording(&agent, "hi", &rr);
 
     try testing.expectEqual(@as(usize, 1), rr.infos.items.len);
     // Reported Retry-After is the raw provider value...
@@ -2998,13 +3133,11 @@ test "runStep: cancellation from a tool still hard-fails" {
     var ns = null_store_mod.NullStore.init(allocator);
     var agent = Agent.init(allocator, io, &h.config, ns.store(), null);
     defer agent.deinit();
-    agent.stream_fn = stub.install();
+    agent.open_stream_fn = stub.install();
 
     const conv = &agent.conversation;
-    try conv.addUserMessage("go");
 
-    var recv = NoopReceiver.make();
-    try testing.expectError(error.Canceled, agent.runStep(&recv));
+    try testing.expectError(error.Canceled, drainTurn(&agent, "go"));
     // Turn aborts: no tool result appended (user + assistant only).
     try testing.expectEqual(@as(usize, 2), conv.messages.items.len);
 }
@@ -3030,13 +3163,11 @@ test "runStep: source per-call error produces a per-call error result and contin
     var ns = null_store_mod.NullStore.init(allocator);
     var agent = Agent.init(allocator, io, &h.config, ns.store(), null);
     defer agent.deinit();
-    agent.stream_fn = stub.install();
+    agent.open_stream_fn = stub.install();
 
     const conv = &agent.conversation;
-    try conv.addUserMessage("go");
 
-    var recv = NoopReceiver.make();
-    try agent.runStep(&recv);
+    try drainTurn(&agent, "go");
 
     try testing.expectEqual(@as(usize, 4), conv.messages.items.len);
     const tr_msg = conv.messages.items[2];
@@ -3070,7 +3201,7 @@ test "runStep: context-overflow compaction fires a compaction retry notification
     var ns = null_store_mod.NullStore.init(allocator);
     var agent = Agent.init(allocator, io, &h.config, ns.store(), null);
     defer agent.deinit();
-    agent.stream_fn = stub.install();
+    agent.open_stream_fn = stub.install();
     agent.compaction_system_prompt = "Summarize the conversation.";
 
     const conv = &agent.conversation;
@@ -3079,12 +3210,10 @@ test "runStep: context-overflow compaction fires a compaction retry notification
     try conv.addAssistantMessage(&.{
         .{ .Text = try conversation.textualBlockFromSlice(allocator, "first answer with several words") },
     });
-    try conv.addUserMessage("second recent question");
 
-    var rr = RetryRecordingReceiver{ .allocator = allocator };
+    var rr = RetryRecorder{ .allocator = allocator };
     defer rr.deinit();
-    var recv = rr.make();
-    try agent.runStep(&recv);
+    try drainTurnRecording(&agent, "second recent question", &rr);
 
     try testing.expect(agent.auto_compacted);
     // Exactly one notification, flagged as a compaction retry with no delay.
diff --git a/libpanto/src/provider.zig b/libpanto/src/provider.zig
index 0258b6b..0fe0ed4 100644
--- a/libpanto/src/provider.zig
+++ b/libpanto/src/provider.zig
@@ -4,9 +4,12 @@ const config_mod = @import("config.zig");
 const conversation = @import("conversation.zig");
 const tool_registry_mod = @import("tool_registry.zig");
 const session_mod = @import("session.zig");
+const stream_mod = @import("stream.zig");
 pub const ToolRegistry = tool_registry_mod.ToolRegistry;
 pub const Usage = session_mod.Usage;
 
+const EventQueue = stream_mod.EventQueue;
+
 pub const ContentBlockType = enum {
     Text,
     Thinking,
@@ -172,109 +175,84 @@ pub const ProviderRetryInfo = struct {
     compaction: bool = false,
 };
 
-/// Vtable for receiving streaming events from a Provider.
-///
-/// The lifecycle callbacks (`onMessageStart` ... `onMessageComplete`) return
-/// `anyerror!void`. Returning an error aborts the in-flight turn: the Provider
-/// stops streaming, calls `onError(err)`, and propagates `err` out of
-/// `streamStep`. No partial assistant message is appended to the conversation.
+/// A resumable provider streaming response: the pull-side projection of one
+/// provider HTTP turn. It wraps the per-provider `ResumableResponse` (which
+/// owns the pinned HTTP request/response, body reader, `SSEParser`, and
+/// decode state) behind a tag so the agent loop can pump any provider
+/// uniformly.
 ///
-/// Tool-use identity (`id`, `name`) is delivered via `onToolDetails`, fired
-/// once per ToolUse block at the earliest moment both fields are known. For
-/// Anthropic this is immediately after `onBlockStart`, before any deltas. For
-/// OpenAI Chat Completions this may be partway through the arg-deltas, since
-/// the wire protocol can split `id` and `name` across multiple streaming
-/// chunks. The only guarantees are: it fires strictly after the block's
-/// `onBlockStart`, strictly before its `onBlockComplete`, and at most once
-/// per ToolUse block. It never fires for non-ToolUse blocks. If a tool_use
-/// block is dropped because identity never fully arrived, `onToolDetails`
-/// (and `onBlockComplete`) never fire for it.
+/// `produce(out)` reads just enough bytes to append one or more `Event`s to
+/// `out` (or reach response-complete), so the `Stream` drains the queue
+/// before pumping again. On response completion the assistant message has
+/// been committed to the conversation and a terminal `message_complete`
+/// pushed.
 ///
-/// `onError` is the receiver's cleanup hook. It fires exactly once per failed
-/// turn, whether the error originated in the receiver itself (a write failure)
-/// or in the Provider (HTTP/parse/stream failure). It is the last callback the
-/// receiver will see for that turn. `onError` itself cannot fail; receivers
-/// must swallow secondary failures during cleanup.
+/// Tool-use identity (`id`, `name`) is delivered via a `tool_details` event,
+/// pushed once per ToolUse block at the earliest moment both fields are
+/// known. For Anthropic this is immediately after `block_start`, before any
+/// deltas. For OpenAI Chat Completions this may be partway through the
+/// arg-deltas, since the wire protocol can split `id` and `name` across
+/// multiple streaming chunks. The guarantees: it fires strictly after the
+/// block's `block_start`, strictly before its `block_complete`, and at most
+/// once per ToolUse block. It never fires for non-ToolUse blocks. If a
+/// tool_use block is dropped because identity never fully arrived, neither
+/// `tool_details` nor `block_complete` fire for it.
 ///
-/// `onMessageComplete`'s `usage` argument carries the wire-reported token
-/// counts for the just-finished assistant turn. Providers fire this exactly
-/// once per successful turn. `usage` is `null` only when the wire genuinely
-/// did not deliver any usage information — chiefly OpenAI-compatible proxies
+/// The terminal `message_complete`'s `usage` carries the wire-reported token
+/// counts for the just-finished assistant message. It is `null` only when the
+/// wire genuinely delivered no usage — chiefly OpenAI-compatible proxies
 /// (OpenRouter, vLLM, some self-hosted backends) that ignore
-/// `stream_options.include_usage`. Receivers that compute cost should record
+/// `stream_options.include_usage`. Consumers that compute cost should record
 /// the null case explicitly ("unknown") rather than treating it as zero.
-pub const ReceiverVTable = struct {
-    onMessageStart: *const fn (*anyopaque, conversation.MessageRole) anyerror!void,
-    onBlockStart: *const fn (*anyopaque, ContentBlockType, usize) anyerror!void,
-    onToolDetails: *const fn (*anyopaque, usize, []const u8, []const u8) anyerror!void,
-    onContentDelta: *const fn (*anyopaque, usize, []const u8) anyerror!void,
-    onBlockComplete: *const fn (*anyopaque, usize, conversation.ContentBlock) anyerror!void,
-    onMessageComplete: *const fn (*anyopaque, conversation.Message, ?Usage) anyerror!void,
-    onError: *const fn (*anyopaque, anyerror) void,
-    /// Fired by the agent after a provider attempt fails and before it
-    /// sleeps for the next attempt. Purely informational: receivers surface
-    /// it (e.g. a dim CLI status line) but cannot abort or alter the retry.
-    /// Never fires for tool-call failures (those become tool results) nor
-    /// for terminal provider errors. May be a no-op.
-    onProviderRetry: *const fn (*anyopaque, ProviderRetryInfo) void,
-};
-
-pub const Receiver = struct {
+pub const ProviderStream = struct {
     ptr: *anyopaque,
-    vtable: *const ReceiverVTable,
+    vtable: *const VTable,
 
-    pub fn onMessageStart(self: Receiver, role: conversation.MessageRole) !void {
-        try self.vtable.onMessageStart(self.ptr, role);
-    }
-
-    pub fn onBlockStart(self: Receiver, block_type: ContentBlockType, index: usize) !void {
-        try self.vtable.onBlockStart(self.ptr, block_type, index);
-    }
+    pub const ProduceStatus = enum { more, response_complete };
 
-    pub fn onToolDetails(self: Receiver, block_index: usize, id: []const u8, name: []const u8) !void {
-        try self.vtable.onToolDetails(self.ptr, block_index, id, name);
-    }
-
-    pub fn onContentDelta(self: Receiver, block_index: usize, delta: []const u8) !void {
-        try self.vtable.onContentDelta(self.ptr, block_index, delta);
-    }
-
-    pub fn onBlockComplete(self: Receiver, block_index: usize, block: conversation.ContentBlock) !void {
-        try self.vtable.onBlockComplete(self.ptr, block_index, block);
-    }
-
-    pub fn onMessageComplete(self: Receiver, message: conversation.Message, usage: ?Usage) !void {
-        try self.vtable.onMessageComplete(self.ptr, message, usage);
-    }
+    pub const VTable = struct {
+        /// Pump the response, appending decoded events to `out`. Returns
+        /// `.more` (pump again) or `.response_complete` (the assistant
+        /// message is committed; a terminal `message_complete` was pushed).
+        produce: *const fn (*anyopaque, *EventQueue) anyerror!ProduceStatus,
+        /// Free the response and any owned state.
+        deinit: *const fn (*anyopaque) void,
+    };
 
-    pub fn onError(self: Receiver, err: anyerror) void {
-        self.vtable.onError(self.ptr, err);
+    /// Pump the response, appending decoded events to `out`. Errors are
+    /// genuine failures (transport/parse/provider).
+    pub fn produce(self: ProviderStream, out: *EventQueue) anyerror!ProduceStatus {
+        return self.vtable.produce(self.ptr, out);
     }
 
-    pub fn onProviderRetry(self: Receiver, info: ProviderRetryInfo) void {
-        self.vtable.onProviderRetry(self.ptr, info);
+    pub fn deinit(self: ProviderStream) void {
+        self.vtable.deinit(self.ptr);
     }
 };
 
-/// Drive one streaming provider turn against the active config snapshot.
+/// Open one streaming provider turn against the active config snapshot,
+/// returning a resumable `ProviderStream`. Performs the POST and reads
+/// response headers (classifying any >=400 status into a provider error),
+/// but does not pump the body — that happens lazily via
+/// `ProviderStream.produce`.
 ///
 /// This is the single dispatch point: it switches on `cfg.provider`'s
 /// `APIStyle` tag, builds a transient per-request object bound to the
-/// process-global HTTP client, and runs it. There is no persistent
-/// provider object — every turn re-reads `cfg`, so swapping the agent's
-/// `*const Config` between turns changes provider, model, base_url, and
-/// the visible tool set with no transport teardown.
+/// process-global HTTP client, and opens it. There is no persistent provider
+/// object — every turn re-reads `cfg`, so swapping the agent's
+/// `*const Config` between turns changes provider, model, base_url, and the
+/// visible tool set with no transport teardown.
 ///
 /// The tool registry is taken from `cfg.registry`; the serializers receive
-/// it directly.
-pub fn streamStep(
+/// it directly. On success the caller owns the returned `ProviderStream` and
+/// must `deinit` it.
+pub fn openStream(
     allocator: std.mem.Allocator,
     io: std.Io,
     cfg: *const config_mod.Config,
     conv: *conversation.Conversation,
-    receiver: *Receiver,
     diag: ?*ProviderDiagnostic,
-) anyerror!void {
+) anyerror!ProviderStream {
     // Imported lazily to break the circular module graph:
     // provider.zig <- provider_openai_chat.zig <- provider.zig.
     const provider_openai_chat = @import("provider_openai_chat.zig");
@@ -289,7 +267,8 @@ pub fn streamStep(
                 .http_client = client,
                 .diag = diag,
             };
-            return req.streamStep(conv, cfg.registry, receiver);
+            const rr = try req.open(conv, cfg.registry);
+            return rr.providerStream();
         },
         .anthropic_messages => |*c| {
             var req: provider_anthropic_messages.AnthropicMessagesRequest = .{
@@ -299,22 +278,22 @@ pub fn streamStep(
                 .http_client = client,
                 .diag = diag,
             };
-            return req.streamStep(conv, cfg.registry, receiver);
+            const rr = try req.open(conv, cfg.registry);
+            return rr.providerStream();
         },
     }
 }
 
-/// The shape of `streamStep`, exposed as a function-pointer type so the
-/// agent can carry an injectable seam (real dispatch in production, a stub
-/// in tests) without resurrecting a per-provider vtable.
-pub const StreamFn = *const fn (
+/// The shape of `openStream`, exposed as a function-pointer type so the agent
+/// can carry an injectable seam (real dispatch in production, a stub in
+/// tests) without resurrecting a per-provider vtable.
+pub const OpenStreamFn = *const fn (
     allocator: std.mem.Allocator,
     io: std.Io,
     cfg: *const config_mod.Config,
     conv: *conversation.Conversation,
-    receiver: *Receiver,
     diag: ?*ProviderDiagnostic,
-) anyerror!void;
+) anyerror!ProviderStream;
 
 test "isContextOverflowBody - matches known markers, rejects others" {
     const t2 = std.testing;
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);
diff --git a/libpanto/src/provider_openai_chat.zig b/libpanto/src/provider_openai_chat.zig
index ca01d7c..30bccb6 100644
--- a/libpanto/src/provider_openai_chat.zig
+++ b/libpanto/src/provider_openai_chat.zig
@@ -18,11 +18,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("openai_chat_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;
+
 /// Decode a wire tool name (`__` -> `.`) in place within an assembled
 /// name buffer. Decoding only ever shrinks the buffer (reads stay ahead
 /// of writes), so aliasing src/dst is safe; we then truncate to the
@@ -52,27 +56,31 @@ pub const OpenAIChatRequest = struct {
     /// they live only as long as this request object.
     diag: ?*provider_mod.ProviderDiagnostic = null,
 
-    pub fn streamStep(
+    /// Open the streaming HTTP request and return a heap-allocated
+    /// resumable response. Performs the POST and reads response headers
+    /// (classifying any >=400 status into a provider error), but does NOT
+    /// pump the body — that happens lazily in `ResumableResponse.produce`.
+    ///
+    /// On success the caller owns the returned `*ResumableResponse` and must
+    /// `deinit` it. On failure nothing is allocated.
+    pub fn open(
         self: *OpenAIChatRequest,
         conv: *conversation.Conversation,
         tools: *const provider_mod.ToolRegistry,
-        receiver: *provider_mod.Receiver,
-    ) !void {
-        // Outer wrapper guarantees `onError` is called exactly once if
-        // anything fails — whether the receiver, the HTTP transport, or the
-        // SSE/JSON parsers. The inner `streamStepInner` does the real work.
-        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: *OpenAIChatRequest,
-        conv: *conversation.Conversation,
-        tools: *const provider_mod.ToolRegistry,
-        receiver: *provider_mod.Receiver,
-    ) !void {
         // Build URL: "{base_url}/chat/completions"
         const url = try std.fmt.allocPrint(
             self.allocator,
@@ -102,8 +110,10 @@ pub const OpenAIChatRequest = struct {
         };
 
         // Open the request. We can't use `fetch()` because it buffers the
-        // response; we want to stream the body as it arrives.
-        var req = try self.http_client.request(.POST, uri, .{
+        // response; we want to stream the body as it arrives. The request
+        // is moved into the heap struct so the body reader (which borrows
+        // `&rr.response`) stays valid across `produce` calls.
+        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.
@@ -111,24 +121,27 @@ pub const OpenAIChatRequest = 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();
 
         // Receive response headers.
         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) {
+        if (@intFromEnum(rr.response.head.status) >= 400) {
             // Drain body for diagnostics.
-            var transfer_buf: [4096]u8 = undefined;
-            const body_reader = response.reader(&transfer_buf);
+            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;
@@ -138,7 +151,7 @@ pub const OpenAIChatRequest = 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("openai_chat HTTP {d}: {s}", .{ status, err_buf.items });
             // Classify the status into a retryable/terminal provider error.
             // HTTP 400 with a context marker becomes `ContextOverflow` so the
@@ -146,64 +159,127 @@ pub const OpenAIChatRequest = 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;
         }
 
-        // Stream the body through the SSE parser and event handler.
-        var transfer_buf: [4096]u8 = undefined;
-        const body_reader = response.reader(&transfer_buf);
+        // Bind the streaming body reader. Valid for the lifetime of `rr`
+        // (it borrows `&rr.response` and `&rr.transfer_buf`, both pinned).
+        rr.body_reader = rr.response.reader(&rr.transfer_buf);
+        return rr;
+    }
+};
 
-        var parser = sse_mod.SSEParser.init(self.allocator);
-        defer parser.deinit();
+/// A resumable OpenAI Chat streaming response. Owns the pinned HTTP
+/// request/response, the body reader's transfer buffer, the `SSEParser`,
+/// and the block-assembly `StreamState`. `produce` pumps just enough bytes
+/// to emit one or more events into the queue, or reports the response is
+/// complete (its assistant message committed to the conversation).
+///
+/// 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 state: StreamState = .init(self.allocator);
-        defer state.deinit();
+    req: http.Client.Request = undefined,
+    response: http.Client.Response = undefined,
+    /// Transfer buffer backing `body_reader`. Pinned in the heap struct.
+    transfer_buf: [4096]u8 = undefined,
+    /// The streaming body reader, bound in `open` after a 2xx response.
+    body_reader: *std.Io.Reader = undefined,
+    /// Chunk scratch for `readVec`.
+    chunk: [4096]u8 = undefined,
 
-        // 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: a return value of 0 does NOT
-        // signal end-of-stream — it just means no new bytes were available
-        // this call, and the caller should try again. EOF is reported only
-        // via `error.EndOfStream`. Breaking on `n == 0` truncates the
-        // response and silently cuts off 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,
-                // A transport read failure mid-stream (reset, TLS, timeout)
-                // before `[DONE]` means no assistant message was committed.
-                // Surface it as a retryable malformed-stream error.
-                else => return error.ProviderStreamMalformed,
-            };
-            if (n == 0) continue;
+    /// True once `req` has been initialized (so `deinit` knows to free it).
+    req_open: bool = false,
+    /// Set once the response is fully decoded and finalized.
+    done: bool = false,
 
-            const events = try parser.feed(chunk[0..n]);
-            defer parser.freeEvents(events);
+    pub const ProduceStatus = provider_mod.ProviderStream.ProduceStatus;
 
-            for (events) |ev_payload| {
-                std.log.debug("openai_chat <= {s}", .{ev_payload});
-                if (std.mem.eql(u8, ev_payload, "[DONE]")) {
-                    try state.finalize(receiver, conv);
-                    return;
-                }
-                try handleEvent(self.allocator, ev_payload, &state, receiver);
-                // Note: we do NOT bail when state.end_of_stream is set.
-                // OpenAI emits the terminating `usage` chunk *after* the
-                // chunk carrying finish_reason, then sends `[DONE]`. If
-                // we returned on finish_reason we'd never capture usage.
-                // `[DONE]` is the authoritative end-of-stream marker.
+    /// Wrap this response in the provider-agnostic `ProviderStream` the agent
+    /// loop drives.
+    pub fn providerStream(self: *ResumableResponse) provider_mod.ProviderStream {
+        return .{ .ptr = self, .vtable = &vtable };
+    }
+
+    const vtable: provider_mod.ProviderStream.VTable = .{
+        .produce = produceVT,
+        .deinit = deinitVT,
+    };
+
+    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 the terminal (`[DONE]` or EOF) has been
+    /// reached and the assistant message has been committed + a final
+    /// `message_complete` pushed.
+    ///
+    /// Reading and finalizing here means a single `produce` call may push
+    /// several events; the `Stream` drains the queue before pumping again.
+    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 [DONE]. Some servers and proxies omit it
+            // (or drop the trailing usage chunk). Finalize with whatever
+            // we've got — usage will be null in that case, which is fine.
+            error.EndOfStream => {
+                try self.finishStream(out);
+                return .response_complete;
+            },
+            // A transport read failure mid-stream (reset, TLS, timeout)
+            // before `[DONE]` means no assistant message was committed.
+            // Surface it as a retryable malformed-stream error.
+            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("openai_chat <= {s}", .{ev_payload});
+            if (std.mem.eql(u8, ev_payload, "[DONE]")) {
+                try self.finishStream(out);
+                return .response_complete;
             }
+            try handleEvent(self.allocator, ev_payload, &self.state, out);
+            // Note: we do NOT bail when state.end_of_stream is set.
+            // OpenAI emits the terminating `usage` chunk *after* the
+            // chunk carrying finish_reason, then sends `[DONE]`. If
+            // we returned on finish_reason we'd never capture usage.
+            // `[DONE]` is the authoritative end-of-stream marker.
         }
+        return .more;
+    }
 
-        // Stream ended without [DONE]. Some servers and proxies omit it
-        // (or drop the trailing usage chunk). Finalize with whatever we've
-        // got — usage will be null in that case, which is fine.
-        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);
     }
 };
 
@@ -310,7 +386,7 @@ const StreamState = struct {
     /// Close the active text/thinking block (if any) and emit
     /// onBlockComplete. Ownership of `current_buf` transfers into the
     /// appended block.
-    fn closeActive(self: *StreamState, receiver: *provider_mod.Receiver) !void {
+    fn closeActive(self: *StreamState, out: *EventQueue) !void {
         if (self.active == .none) return;
 
         const block: conversation.ContentBlock = switch (self.active) {
@@ -321,7 +397,10 @@ const StreamState = struct {
         self.current_buf = .empty;
 
         try self.blocks.append(self.allocator, block);
-        try receiver.onBlockComplete(self.block_index, self.blocks.items[self.blocks.items.len - 1]);
+        try out.push(.{ .block_complete = .{
+            .index = self.block_index,
+            .block = self.blocks.items[self.blocks.items.len - 1],
+        } });
 
         self.active = .none;
     }
@@ -330,12 +409,12 @@ const StreamState = struct {
     fn openBlock(
         self: *StreamState,
         new_active: ActiveBlock,
-        receiver: *provider_mod.Receiver,
+        out: *EventQueue,
     ) !void {
         std.debug.assert(new_active == .text or new_active == .thinking);
         if (self.active == new_active) return;
         if (self.active != .none) {
-            try self.closeActive(receiver);
+            try self.closeActive(out);
             self.block_index += 1;
         }
         self.active = new_active;
@@ -344,16 +423,21 @@ const StreamState = struct {
             .thinking => .Thinking,
             .tool_use, .none => unreachable,
         };
-        try receiver.onBlockStart(block_type, self.block_index);
+        try out.push(.{ .block_start = .{ .block_type = block_type, .index = self.block_index } });
     }
 
     fn appendDelta(
         self: *StreamState,
-        receiver: *provider_mod.Receiver,
+        out: *EventQueue,
         delta: []const u8,
     ) !void {
         try self.current_buf.appendSlice(self.allocator, delta);
-        try receiver.onContentDelta(self.block_index, delta);
+        // Dupe into the queue arena: the raw `delta` borrows the transient
+        // SSE payload that `produce` frees before `next()` reads the queue.
+        try out.push(.{ .content_delta = .{
+            .index = self.block_index,
+            .delta = try out.dupeBytes(delta),
+        } });
     }
 
     /// Apply one streaming tool_call delta. Opens a new tool_use on the
@@ -362,7 +446,7 @@ const StreamState = struct {
     /// a malformed stream — we log and drop it.
     fn applyToolCallDelta(
         self: *StreamState,
-        receiver: *provider_mod.Receiver,
+        out: *EventQueue,
         d: json_mod.ToolCallDelta,
     ) !void {
         // Degenerate backend: a delta arrived for an index whose block we
@@ -382,14 +466,14 @@ const StreamState = struct {
         // the only signal openai_chat gives us for mid-stream tool_use
         // boundaries; see the StreamState doc-comment for the rationale.
         if (self.current_tool_index) |cur| {
-            if (cur != d.index) try self.closeActiveTool(receiver);
+            if (cur != d.index) try self.closeActiveTool(out);
         }
 
         if (self.active_tool == null) {
             // Opening a new tool_use. First close any open text/thinking
             // block so the tool_use gets its own block_index.
             if (self.active != .none) {
-                try self.closeActive(receiver);
+                try self.closeActive(out);
                 self.block_index += 1;
             }
             self.active_tool = .{ .block_index = self.block_index };
@@ -410,32 +494,39 @@ const StreamState = struct {
         // to render. If the block closes before any args arrive (zero-arg
         // tool), `closeActiveTool` emits the start there.
         if (d.arguments) |a| {
-            try self.emitStartIfNeeded(receiver, tu);
-            // Fire `onToolDetails` as soon as both id and name are
+            try self.emitStartIfNeeded(out, tu);
+            // Fire `tool_details` as soon as both id and name are
             // known. We can't know identity is *final* until the block
             // closes (a later delta could append more bytes), but in
             // practice OpenAI sends each whole on the first delta. A
             // pathological backend that streams id/name across many
             // chunks would have us emit a truncated value here. We
-            // accept that trade-off: receivers that need the canonical
+            // accept that trade-off: consumers that need the canonical
             // value can read it from the assembled ContentBlock at
-            // onBlockComplete.
-            try self.emitDetailsIfReady(receiver, tu);
+            // block_complete.
+            try self.emitDetailsIfReady(out, tu);
             try tu.arguments.appendSlice(self.allocator, a);
-            try receiver.onContentDelta(tu.block_index, a);
+            // Dupe into the queue arena (the SSE payload is freed before
+            // `next()` reads the queue).
+            try out.push(.{ .content_delta = .{
+                .index = tu.block_index,
+                .delta = try out.dupeBytes(a),
+            } });
         } else {
             // Identity-only chunk (no args yet). Still try to emit
             // details, in case both fields are now populated.
-            if (tu.started) try self.emitDetailsIfReady(receiver, tu);
+            if (tu.started) try self.emitDetailsIfReady(out, tu);
         }
     }
 
-    /// Fire `onToolDetails` once both id and name are non-empty. No-op if
+    /// Fire `tool_details` once both id and name are non-empty. No-op if
     /// already fired or if either field is still empty. Requires that
-    /// `onBlockStart` has already been emitted.
+    /// `block_start` has already been emitted. Slices are duped into the
+    /// queue arena because `id_buf`/`name_buf` may still grow (and realloc)
+    /// on later fragments.
     fn emitDetailsIfReady(
         self: *StreamState,
-        receiver: *provider_mod.Receiver,
+        out: *EventQueue,
         tu: *ToolUseInProgress,
     ) !void {
         _ = self;
@@ -443,13 +534,17 @@ const StreamState = struct {
         if (!tu.started) return;
         if (tu.id_buf.items.len == 0 or tu.name_buf.items.len == 0) return;
         tu.details_emitted = true;
-        try receiver.onToolDetails(tu.block_index, tu.id_buf.items, tu.name_buf.items);
+        try out.push(.{ .tool_details = .{
+            .index = tu.block_index,
+            .id = try out.dupeBytes(tu.id_buf.items),
+            .name = try out.dupeBytes(tu.name_buf.items),
+        } });
     }
 
-    /// Close the currently-active tool_use (if any), emitting onBlockStart
-    /// (if it wasn't already), onBlockComplete, and recording the wire
+    /// Close the currently-active tool_use (if any), emitting block_start
+    /// (if it wasn't already), block_complete, and recording the wire
     /// index as closed. No-op if there's no active tool_use.
-    fn closeActiveTool(self: *StreamState, receiver: *provider_mod.Receiver) !void {
+    fn closeActiveTool(self: *StreamState, out: *EventQueue) !void {
         var tu = self.active_tool orelse return;
         self.active_tool = null;
         const wire_index = self.current_tool_index.?;
@@ -481,12 +576,12 @@ const StreamState = struct {
         // internal (dotted) name. Decoding never grows the buffer.
         decodeNameInPlace(&tu.name_buf);
 
-        // If no arguments ever arrived, we haven't emitted onBlockStart
-        // yet — do it now so the receiver sees a balanced start/complete.
-        try self.emitStartIfNeeded(receiver, &tu);
+        // If no arguments ever arrived, we haven't emitted block_start
+        // yet — do it now so the consumer sees a balanced start/complete.
+        try self.emitStartIfNeeded(out, &tu);
         // Last chance to fire details if a fragmented-identity provider
         // only finished id/name accumulation at the very end.
-        try self.emitDetailsIfReady(receiver, &tu);
+        try self.emitDetailsIfReady(out, &tu);
 
         const id_owned = try tu.id_buf.toOwnedSlice(self.allocator);
         const name_owned = try tu.name_buf.toOwnedSlice(self.allocator);
@@ -500,39 +595,41 @@ const StreamState = struct {
         tu.arguments = .empty;
 
         try self.blocks.append(self.allocator, block);
-        try receiver.onBlockComplete(tu.block_index, self.blocks.items[self.blocks.items.len - 1]);
+        try out.push(.{ .block_complete = .{
+            .index = tu.block_index,
+            .block = self.blocks.items[self.blocks.items.len - 1],
+        } });
     }
 
-    /// Emit `onBlockStart(.ToolUse, ...)` once per in-progress tool use.
-    /// Callers must invoke this before the first `onContentDelta` or
-    /// `onBlockComplete` for the block. Identity (id/name) is *not*
-    /// passed at start — see provider.zig's ReceiverVTable docs for the
-    /// rationale. Receivers get identity from the assembled ContentBlock
-    /// at onBlockComplete time.
+    /// Emit `block_start(.ToolUse, ...)` once per in-progress tool use.
+    /// Callers must invoke this before the first `content_delta` or
+    /// `block_complete` for the block. Identity (id/name) is *not* passed at
+    /// start — consumers get identity from `tool_details` or the assembled
+    /// ContentBlock at block_complete time.
     fn emitStartIfNeeded(
         self: *StreamState,
-        receiver: *provider_mod.Receiver,
+        out: *EventQueue,
         tu: *ToolUseInProgress,
     ) !void {
         _ = self;
         if (tu.started) return;
         tu.started = true;
-        try receiver.onBlockStart(.ToolUse, tu.block_index);
+        try out.push(.{ .block_start = .{ .block_type = .ToolUse, .index = tu.block_index } });
     }
 
     /// End the stream: close any open text/thinking block, close the still-
     /// active tool_use (if any), then commit the assembled assistant
-    /// Message to the conversation.
+    /// Message to the conversation and push the terminal `message_complete`.
     fn finalize(
         self: *StreamState,
-        receiver: *provider_mod.Receiver,
+        out: *EventQueue,
         conv: *conversation.Conversation,
     ) !void {
         if (self.finalized) return;
         self.finalized = true;
 
-        try self.closeActive(receiver);
-        try self.closeActiveTool(receiver);
+        try self.closeActive(out);
+        try self.closeActiveTool(out);
 
         // Move blocks into a fresh conversation message.
         const moved_blocks = try self.blocks.toOwnedSlice(self.allocator);
@@ -541,7 +638,7 @@ const StreamState = struct {
         try conv.addAssistantMessageWithUsage(moved_blocks, self.usage);
 
         const msg = conv.messages.items[conv.messages.items.len - 1];
-        try receiver.onMessageComplete(msg, self.usage);
+        try out.push(.{ .message_complete = .{ .message = msg, .usage = self.usage } });
     }
 };
 
@@ -549,7 +646,7 @@ 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();
@@ -587,33 +684,33 @@ fn handleEvent(
 
     if (!state.started and d.role != null) {
         state.started = true;
-        try receiver.onMessageStart(.assistant);
+        try out.push(.{ .message_start = .assistant });
     }
 
     if (d.reasoning_content) |rc| {
         if (!state.started) {
             state.started = true;
-            try receiver.onMessageStart(.assistant);
+            try out.push(.{ .message_start = .assistant });
         }
-        try state.openBlock(.thinking, receiver);
-        try state.appendDelta(receiver, rc);
+        try state.openBlock(.thinking, out);
+        try state.appendDelta(out, rc);
     }
 
     if (d.content) |c| {
         if (!state.started) {
             state.started = true;
-            try receiver.onMessageStart(.assistant);
+            try out.push(.{ .message_start = .assistant });
         }
-        try state.openBlock(.text, receiver);
-        try state.appendDelta(receiver, c);
+        try state.openBlock(.text, out);
+        try state.appendDelta(out, c);
     }
 
     if (d.tool_calls.len > 0) {
         if (!state.started) {
             state.started = true;
-            try receiver.onMessageStart(.assistant);
+            try out.push(.{ .message_start = .assistant });
         }
-        for (d.tool_calls) |tc| try state.applyToolCallDelta(receiver, tc);
+        for (d.tool_calls) |tc| try state.applyToolCallDelta(out, tc);
     }
 
     if (d.finish_reason) |_| {
@@ -627,54 +724,76 @@ fn handleEvent(
 
 const testing = std.testing;
 
-/// A no-op Receiver that drops every callback. Useful when the test cares
-/// about post-stream conversation state rather than callback observability.
-const NoopReceiver = struct {
-    fn make() provider_mod.Receiver {
-        return .{ .ptr = @ptrCast(@constCast(&dummy)), .vtable = &vt };
-    }
-    var dummy: u8 = 0;
-    const vt: provider_mod.ReceiverVTable = .{
-        .onMessageStart = noopMsgStart,
-        .onBlockStart = noopBlockStart,
-        .onToolDetails = noopToolDetails,
-        .onContentDelta = noopDelta,
-        .onBlockComplete = noopBlockComplete,
-        .onMessageComplete = noopMsgComplete,
-        .onError = noopErr,
-        .onProviderRetry = noopRetry,
-    };
-    fn noopMsgStart(_: *anyopaque, _: conversation.MessageRole) anyerror!void {}
-    fn noopBlockStart(_: *anyopaque, _: provider_mod.ContentBlockType, _: usize) anyerror!void {}
-    fn noopToolDetails(_: *anyopaque, _: usize, _: []const u8, _: []const u8) anyerror!void {}
-    fn noopDelta(_: *anyopaque, _: usize, _: []const u8) anyerror!void {}
-    fn noopBlockComplete(_: *anyopaque, _: usize, _: conversation.ContentBlock) anyerror!void {}
-    fn noopMsgComplete(_: *anyopaque, _: conversation.Message, _: ?provider_mod.Usage) anyerror!void {}
-    fn noopErr(_: *anyopaque, _: anyerror) void {}
-    fn noopRetry(_: *anyopaque, _: provider_mod.ProviderRetryInfo) void {}
-};
-
 /// Feed a sequence of SSE event payloads through the state machine as if
-/// they had been delivered by the wire, finalizing into `conv`.
+/// they had been delivered by the wire, finalizing into `conv`. The decoded
+/// `Event`s are recorded as compact strings (the same schema the old
+/// RecordingReceiver used) so callback-ordering assertions are preserved.
 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| {
         if (std.mem.eql(u8, payload, "[DONE]")) break;
         // Process every chunk through to [DONE], including the
-        // post-finish_reason usage chunk. Mirrors the production loop
-        // in OpenAIChatRequest.streamStep.
-        try handleEvent(allocator, payload, &state, receiver);
+        // post-finish_reason usage chunk. Mirrors the production pump in
+        // ResumableResponse.produce.
+        try handleEvent(allocator, payload, &state, &queue);
+    }
+    try state.finalize(&queue, conv);
+
+    // Drain into the recorder before the arena resets. The queue holds all
+    // events from this turn; popping records each, and the final null-pop
+    // resets the arena.
+    while (queue.pop()) |ev| {
+        if (rec) |r| try r.record(ev);
     }
-    try state.finalize(receiver, conv);
 }
 
+/// Records decoded `Event`s as compact strings for ordering assertions.
+const EventRecorder = struct {
+    allocator: Allocator,
+    events: std.ArrayList([]const u8) = .empty,
+
+    fn deinit(self: *EventRecorder) void {
+        for (self.events.items) |e| self.allocator.free(e);
+        self.events.deinit(self.allocator);
+    }
+
+    fn push(self: *EventRecorder, comptime fmt: []const u8, args: anytype) !void {
+        const owned = try std.fmt.allocPrint(self.allocator, fmt, args);
+        try self.events.append(self.allocator, owned);
+    }
+
+    fn record(self: *EventRecorder, ev: Event) !void {
+        switch (ev) {
+            .message_start => try self.push("msg_start", .{}),
+            .block_start => |b| try self.push("block_start[{d}]:{s}", .{ b.index, @tagName(b.block_type) }),
+            .tool_details => |t| try self.push("tool_details[{d}]:{s}:{s}", .{ t.index, t.id, t.name }),
+            .content_delta => |d| try self.push("delta[{d}]:{s}", .{ d.index, d.delta }),
+            .block_complete => |b| try self.push("block_complete[{d}]", .{b.index}),
+            .message_complete => |m| {
+                if (m.usage) |u| {
+                    try self.push(
+                        "msg_complete[usage:in={d},out={d},cr={d},cw={d},rsn={d}]",
+                        .{ u.input, u.output, u.cache_read, u.cache_write, u.reasoning },
+                    );
+                } else {
+                    try self.push("msg_complete[usage:null]", .{});
+                }
+            },
+            else => {},
+        }
+    }
+};
+
 test "two streamed turns persist assistant replies in the conversation" {
     // Regression test for the bug where `finish_reason` arrived before
     // `[DONE]` and `finalize` early-returned without appending the assistant
@@ -688,8 +807,6 @@ test "two streamed turns persist assistant replies in the conversation" {
     try conv.addSystemMessage("You are a helpful assistant.");
     try conv.addUserMessage("hello!");
 
-    var recv = NoopReceiver.make();
-
     const turn1 = [_][]const u8{
         \\{"choices":[{"delta":{"role":"assistant"}}]}
         ,
@@ -701,7 +818,7 @@ test "two streamed turns persist assistant replies in the conversation" {
         ,
         "[DONE]",
     };
-    try runStreamedTurn(allocator, &conv, &recv, &turn1);
+    try runStreamedTurn(allocator, &conv, null, &turn1);
 
     try testing.expectEqual(@as(usize, 3), conv.messages.items.len);
     try testing.expectEqual(conversation.MessageRole.assistant, conv.messages.items[2].role);
@@ -722,7 +839,7 @@ test "two streamed turns persist assistant replies in the conversation" {
         ,
         "[DONE]",
     };
-    try runStreamedTurn(allocator, &conv, &recv, &turn2);
+    try runStreamedTurn(allocator, &conv, null, &turn2);
 
     // System + user + assistant + user + assistant = 5 messages.
     try testing.expectEqual(@as(usize, 5), conv.messages.items.len);
@@ -740,12 +857,8 @@ test "openai_chat: terminating usage chunk lands on message_complete with split
     defer conv.deinit();
     try conv.addUserMessage("hi");
 
-    var rec = RecordingReceiver{ .allocator = allocator };
-    defer {
-        for (rec.events.items) |s| allocator.free(s);
-        rec.events.deinit(allocator);
-    }
-    var recv = rec.receiver();
+    var rec = EventRecorder{ .allocator = allocator };
+    defer rec.deinit();
 
     const events = [_][]const u8{
         \\{"choices":[{"delta":{"role":"assistant"}}]}
@@ -759,7 +872,7 @@ test "openai_chat: terminating usage chunk lands on message_complete with split
         ,
         "[DONE]",
     };
-    try runStreamedTurn(allocator, &conv, &recv, &events);
+    try runStreamedTurn(allocator, &conv, &rec, &events);
 
     var found: ?[]const u8 = null;
     for (rec.events.items) |s| {
@@ -777,12 +890,8 @@ test "openai_chat: omitted stream usage yields null on message_complete" {
     defer conv.deinit();
     try conv.addUserMessage("hi");
 
-    var rec = RecordingReceiver{ .allocator = allocator };
-    defer {
-        for (rec.events.items) |s| allocator.free(s);
-        rec.events.deinit(allocator);
-    }
-    var recv = rec.receiver();
+    var rec = EventRecorder{ .allocator = allocator };
+    defer rec.deinit();
 
     const events = [_][]const u8{
         \\{"choices":[{"delta":{"role":"assistant"}}]}
@@ -793,7 +902,7 @@ test "openai_chat: omitted stream usage yields null on message_complete" {
         ,
         "[DONE]",
     };
-    try runStreamedTurn(allocator, &conv, &recv, &events);
+    try runStreamedTurn(allocator, &conv, &rec, &events);
 
     var found: ?[]const u8 = null;
     for (rec.events.items) |s| {
@@ -814,8 +923,6 @@ test "fragmented tool_call id and name are reassembled" {
     defer conv.deinit();
     try conv.addUserMessage("call something");
 
-    var recv = NoopReceiver.make();
-
     const events = [_][]const u8{
         \\{"choices":[{"delta":{"role":"assistant"}}]}
         ,
@@ -830,7 +937,7 @@ test "fragmented tool_call id and name are reassembled" {
         "[DONE]",
     };
 
-    try runStreamedTurn(allocator, &conv, &recv, &events);
+    try runStreamedTurn(allocator, &conv, null, &events);
 
     const asst = conv.messages.items[1];
     try testing.expectEqual(@as(usize, 1), asst.content.items.len);
@@ -851,8 +958,6 @@ test "inbound wire tool name is decoded to dotted form (even split across __)" {
     defer conv.deinit();
     try conv.addUserMessage("read a file");
 
-    var recv = NoopReceiver.make();
-
     const events = [_][]const u8{
         \\{"choices":[{"delta":{"role":"assistant"}}]}
         ,
@@ -867,96 +972,12 @@ test "inbound wire tool name is decoded to dotted form (even split across __)" {
         "[DONE]",
     };
 
-    try runStreamedTurn(allocator, &conv, &recv, &events);
+    try runStreamedTurn(allocator, &conv, null, &events);
 
     const tu = conv.messages.items[1].content.items[0].ToolUse;
     try testing.expectEqualStrings("std.read", tu.name);
 }
 
-/// A Receiver that records the sequence of callback events as compact
-/// strings. Useful for asserting per-block start/complete ordering.
-const RecordingReceiver = struct {
-    allocator: Allocator,
-    events: std.ArrayList([]const u8) = .empty,
-
-    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 onProviderRetry(_: *anyopaque, _: provider_mod.ProviderRetryInfo) void {}
-
-    fn record(self: *RecordingReceiver, s: []const u8) !void {
-        const owned = try self.allocator.dupe(u8, s);
-        try self.events.append(self.allocator, owned);
-    }
-
-    fn recordFmt(self: *RecordingReceiver, comptime fmt: []const u8, args: anytype) !void {
-        const owned = try std.fmt.allocPrint(self.allocator, fmt, args);
-        try self.events.append(self.allocator, owned);
-    }
-
-    fn deinit(self: *RecordingReceiver) void {
-        for (self.events.items) |e| self.allocator.free(e);
-        self.events.deinit(self.allocator);
-    }
-
-    fn onMessageStart(ptr: *anyopaque, _: conversation.MessageRole) anyerror!void {
-        const self: *RecordingReceiver = @ptrCast(@alignCast(ptr));
-        try self.record("msg_start");
-    }
-    fn onBlockStart(
-        ptr: *anyopaque,
-        bt: provider_mod.ContentBlockType,
-        idx: usize,
-    ) anyerror!void {
-        const self: *RecordingReceiver = @ptrCast(@alignCast(ptr));
-        try self.recordFmt("block_start[{d}]:{s}", .{ idx, @tagName(bt) });
-    }
-    fn onToolDetails(
-        ptr: *anyopaque,
-        idx: usize,
-        id: []const u8,
-        name: []const u8,
-    ) anyerror!void {
-        const self: *RecordingReceiver = @ptrCast(@alignCast(ptr));
-        try self.recordFmt("tool_details[{d}]:{s}:{s}", .{ idx, id, name });
-    }
-    fn onContentDelta(ptr: *anyopaque, idx: usize, delta: []const u8) anyerror!void {
-        const self: *RecordingReceiver = @ptrCast(@alignCast(ptr));
-        try self.recordFmt("delta[{d}]:{s}", .{ idx, delta });
-    }
-    fn onBlockComplete(
-        ptr: *anyopaque,
-        idx: usize,
-        _: conversation.ContentBlock,
-    ) anyerror!void {
-        const self: *RecordingReceiver = @ptrCast(@alignCast(ptr));
-        try self.recordFmt("block_complete[{d}]", .{idx});
-    }
-    fn onMessageComplete(ptr: *anyopaque, _: conversation.Message, usage: ?provider_mod.Usage) anyerror!void {
-        const self: *RecordingReceiver = @ptrCast(@alignCast(ptr));
-        if (usage) |u| {
-            try self.recordFmt(
-                "msg_complete[usage:in={d},out={d},cr={d},cw={d},rsn={d}]",
-                .{ u.input, u.output, u.cache_read, u.cache_write, u.reasoning },
-            );
-        } else {
-            try self.record("msg_complete[usage:null]");
-        }
-    }
-    fn onError(_: *anyopaque, _: anyerror) void {}
-};
 
 test "parallel tool_calls emit one complete start/delta/complete cycle per block" {
     // Regression test: previously, the OpenAI provider deferred ALL
@@ -972,9 +993,8 @@ test "parallel tool_calls emit one complete start/delta/complete cycle per block
     defer conv.deinit();
     try conv.addUserMessage("ping four hosts");
 
-    var rec: RecordingReceiver = .{ .allocator = allocator };
+    var rec: EventRecorder = .{ .allocator = allocator };
     defer rec.deinit();
-    var recv = rec.receiver();
 
     const events = [_][]const u8{
         \\{"choices":[{"delta":{"role":"assistant"}}]}
@@ -992,7 +1012,7 @@ test "parallel tool_calls emit one complete start/delta/complete cycle per block
         "[DONE]",
     };
 
-    try runStreamedTurn(allocator, &conv, &recv, &events);
+    try runStreamedTurn(allocator, &conv, &rec, &events);
 
     const expected = [_][]const u8{
         "msg_start",
@@ -1039,9 +1059,8 @@ test "non-contiguous tool_call deltas: re-emission of a closed index is dropped"
     defer conv.deinit();
     try conv.addUserMessage("go");
 
-    var rec: RecordingReceiver = .{ .allocator = allocator };
+    var rec: EventRecorder = .{ .allocator = allocator };
     defer rec.deinit();
-    var recv = rec.receiver();
 
     const events = [_][]const u8{
         \\{"choices":[{"delta":{"role":"assistant"}}]}
@@ -1058,7 +1077,7 @@ test "non-contiguous tool_call deltas: re-emission of a closed index is dropped"
         "[DONE]",
     };
 
-    try runStreamedTurn(allocator, &conv, &recv, &events);
+    try runStreamedTurn(allocator, &conv, &rec, &events);
 
     // Two well-formed tool_use blocks in the final message, args unaffected
     // by the dropped fragment.
@@ -1088,9 +1107,8 @@ test "onToolDetails fires after id+name complete, even mid-arg-stream" {
     defer conv.deinit();
     try conv.addUserMessage("go");
 
-    var rec: RecordingReceiver = .{ .allocator = allocator };
+    var rec: EventRecorder = .{ .allocator = allocator };
     defer rec.deinit();
-    var recv = rec.receiver();
 
     const events = [_][]const u8{
         \\{"choices":[{"delta":{"role":"assistant"}}]}
@@ -1114,7 +1132,7 @@ test "onToolDetails fires after id+name complete, even mid-arg-stream" {
         "[DONE]",
     };
 
-    try runStreamedTurn(allocator, &conv, &recv, &events);
+    try runStreamedTurn(allocator, &conv, &rec, &events);
 
     // Exactly one tool_details event, fired with the id-prefix that was
     // current at first-args-arrival, and ordered between block_start and
@@ -1155,8 +1173,6 @@ test "tool_call with no arguments still finalizes a well-formed ToolUse" {
     defer conv.deinit();
     try conv.addUserMessage("ring it");
 
-    var recv = NoopReceiver.make();
-
     const events = [_][]const u8{
         \\{"choices":[{"delta":{"role":"assistant"}}]}
         ,
@@ -1167,7 +1183,7 @@ test "tool_call with no arguments still finalizes a well-formed ToolUse" {
         "[DONE]",
     };
 
-    try runStreamedTurn(allocator, &conv, &recv, &events);
+    try runStreamedTurn(allocator, &conv, null, &events);
 
     const asst = conv.messages.items[1];
     try testing.expectEqual(@as(usize, 1), asst.content.items.len);
diff --git a/libpanto/src/root.zig b/libpanto/src/root.zig
index a49a428..b5fab1a 100644
--- a/libpanto/src/root.zig
+++ b/libpanto/src/root.zig
@@ -2,6 +2,7 @@ const std = @import("std");
 
 pub const conversation = @import("conversation.zig");
 pub const provider = @import("provider.zig");
+pub const stream = @import("stream.zig");
 pub const agent = @import("agent.zig");
 pub const config = @import("config.zig");
 pub const sse = @import("sse.zig");
@@ -32,12 +33,17 @@ pub const ToolRegistry = tool_registry.ToolRegistry;
 pub const Config = config.Config;
 pub const ProviderConfig = config.ProviderConfig;
 
+// Re-export the pull-streaming surface for embedders.
+pub const Event = stream.Event;
+pub const Stream = agent.Stream;
+pub const Agent = agent.Agent;
+
 // Internal modules. Not part of the public API — callers drive turns via
-// `provider.streamStep(allocator, io, &config, conv, receiver)` (or via the
-// `Agent`, which holds a swappable `*const Config`). The process-global HTTP
-// client is initialized with `config.initHttp` / torn down with
-// `config.deinitHttp`. These impls are exposed here only so `refAllDecls`
-// picks up their tests.
+// the `Agent` (which holds a swappable `*const Config`): `agent.run()`
+// returns a `*Stream` whose `next()` pulls one `Event` at a time. The
+// process-global HTTP client is initialized with `config.initHttp` / torn
+// down with `config.deinitHttp`. These impls are exposed here only so
+// `refAllDecls` picks up their tests.
 const openai_chat_json = @import("openai_chat_json.zig");
 const provider_openai_chat = @import("provider_openai_chat.zig");
 const anthropic_messages_json = @import("anthropic_messages_json.zig");
diff --git a/libpanto/src/stream.zig b/libpanto/src/stream.zig
new file mode 100644
index 0000000..e626f31
--- /dev/null
+++ b/libpanto/src/stream.zig
@@ -0,0 +1,174 @@
+//! Pull-based streaming surface for `libpanto`.
+//!
+//! This is the spine of the language-bindings work (see
+//! `docs/libpanto-bindings.md` and `docs/phase0-pull-stream-design.md`).
+//! Instead of pushing events at a `Receiver` vtable, the agent loop is
+//! inverted into a resumable `Stream` whose `next()` *pulls* one `Event`
+//! at a time. Pull is the more primitive primitive: push composes trivially
+//! on top of it, and it maps 1:1 onto Go range-over-func iterators and
+//! Python generators.
+//!
+//! Contract (the terminal-event invariant):
+//!
+//!   - `Event` (a value)  -> streaming progress, including `turn_complete`.
+//!   - `null`             -> the stream is exhausted (already past the
+//!                           terminal `turn_complete`). Never returned
+//!                           before `turn_complete`.
+//!   - `error.X`          -> a genuine failure (network, parse, provider).
+//!
+//! Event payloads borrow from state owned by the stream or the
+//! conversation. **An `Event` is valid only until the next `next()` call.**
+//! Consumers that need to retain data copy it out before advancing.
+
+const std = @import("std");
+const Allocator = std.mem.Allocator;
+
+const conversation = @import("conversation.zig");
+const provider_mod = @import("provider.zig");
+const session_mod = @import("session.zig");
+
+pub const ContentBlockType = provider_mod.ContentBlockType;
+pub const Usage = session_mod.Usage;
+pub const ProviderRetryInfo = provider_mod.ProviderRetryInfo;
+
+/// The single success-only event type every binding marshals. Mirrors the
+/// former `ReceiverVTable` callbacks plus the agent's tool-dispatch
+/// boundaries. Provider failures are NOT a variant here — they surface as
+/// the `!` in `Stream.next() !?Event`.
+pub const Event = union(enum) {
+    /// An assistant message began streaming.
+    message_start: conversation.MessageRole,
+
+    /// A content block opened. `index` is the receiver-facing block index.
+    block_start: BlockStart,
+
+    /// Tool identity resolved for a ToolUse block (id + name both known).
+    /// Fires once per ToolUse block, after its `block_start` and before its
+    /// `block_complete`. Borrowed slices; valid until the next `next()`.
+    tool_details: ToolDetails,
+
+    /// Streaming content for the open block (text, thinking, or tool args).
+    /// `delta` is borrowed; valid until the next `next()`.
+    content_delta: ContentDelta,
+
+    /// A content block closed. `block` is borrowed from the message under
+    /// construction; valid until the next `next()`.
+    block_complete: BlockComplete,
+
+    /// One assistant message finished streaming (one provider response).
+    /// In a tool-using turn this fires once per assistant message, not once
+    /// per turn — `turn_complete` is the turn-level terminal. `message` is
+    /// borrowed from the conversation.
+    message_complete: MessageComplete,
+
+    /// Provider retry scheduled before the agent sleeps for the next
+    /// attempt. Purely informational; simple consumers ignore it.
+    provider_retry: ProviderRetryInfo,
+
+    /// The agent began dispatching the tool calls in the just-completed
+    /// assistant message. Marks the boundary between a provider stream and
+    /// concurrent tool execution.
+    tool_dispatch_start: ToolDispatchStart,
+
+    /// The agent finished dispatching tools and appended a user(ToolResult)
+    /// message to the conversation. `message` is borrowed.
+    tool_dispatch_complete: ToolDispatchComplete,
+
+    /// The turn terminal: the model stopped calling tools and the turn is
+    /// done. Emitted exactly once, after the final `message_complete` and
+    /// any tool dispatch. Every `next()` after this returns `null`.
+    turn_complete,
+
+    pub const BlockStart = struct {
+        block_type: ContentBlockType,
+        index: usize,
+    };
+    pub const ToolDetails = struct {
+        index: usize,
+        id: []const u8,
+        name: []const u8,
+    };
+    pub const ContentDelta = struct {
+        index: usize,
+        delta: []const u8,
+    };
+    pub const BlockComplete = struct {
+        index: usize,
+        block: conversation.ContentBlock,
+    };
+    pub const MessageComplete = struct {
+        message: conversation.Message,
+        usage: ?Usage,
+    };
+    pub const ToolDispatchStart = struct {
+        count: usize,
+    };
+    pub const ToolDispatchComplete = struct {
+        message: conversation.Message,
+    };
+};
+
+/// A small FIFO of decoded-but-not-yet-yielded events. One `parser.feed()`
+/// can yield several SSE events, each of which can produce several `Event`s;
+/// the provider decode step appends them here and `Stream.next()` drains
+/// the queue before pulling more bytes.
+///
+/// Transient byte payloads (delta text, tool id/name) are duped into a
+/// queue-owned arena via `dupeBytes`, so they survive the provider freeing
+/// its SSE/JSON scratch and any reallocation of the provider's accumulation
+/// buffers. The arena (and the event list) are reset when the queue fully
+/// drains, so memory is bounded by the events produced from a single byte
+/// chunk. Events whose payloads live in the conversation (`block_complete`,
+/// `message_complete`, `tool_dispatch_complete`) borrow directly and are
+/// not duped — the conversation outlives the `next()` step.
+pub const EventQueue = struct {
+    items: std.ArrayList(Event) = .empty,
+    head: usize = 0,
+    arena: std.heap.ArenaAllocator,
+    allocator: Allocator,
+
+    pub fn init(allocator: Allocator) EventQueue {
+        return .{
+            .allocator = allocator,
+            .arena = std.heap.ArenaAllocator.init(allocator),
+        };
+    }
+
+    pub fn deinit(self: *EventQueue) void {
+        self.items.deinit(self.allocator);
+        self.arena.deinit();
+    }
+
+    pub fn push(self: *EventQueue, ev: Event) !void {
+        try self.items.append(self.allocator, ev);
+    }
+
+    /// Copy transient bytes into the queue's arena. The returned slice is
+    /// valid until the queue next fully drains (i.e. until the consumer has
+    /// pulled every queued event). Providers MUST route any byte payload
+    /// borrowed from SSE/JSON scratch or a reallocating buffer through here
+    /// before queueing it on an event.
+    pub fn dupeBytes(self: *EventQueue, bytes: []const u8) ![]const u8 {
+        return self.arena.allocator().dupe(u8, bytes);
+    }
+
+    pub fn isEmpty(self: *const EventQueue) bool {
+        return self.head >= self.items.items.len;
+    }
+
+    /// Pop the next event, or null if empty. Resets the backing list and
+    /// arena when drained so they can be refilled for the next chunk without
+    /// unbounded growth.
+    pub fn pop(self: *EventQueue) ?Event {
+        if (self.head >= self.items.items.len) {
+            // Drained: reset list + arena to reuse for the next chunk.
+            self.items.clearRetainingCapacity();
+            self.head = 0;
+            _ = self.arena.reset(.retain_capacity);
+            return null;
+        }
+        const ev = self.items.items[self.head];
+        self.head += 1;
+        return ev;
+    }
+};
diff --git a/src/main.zig b/src/main.zig
index 9059a6f..ff00689 100644
--- a/src/main.zig
+++ b/src/main.zig
@@ -45,144 +45,94 @@ test {
     _ = command_compaction;
 }
 
-const Receiver = panto.provider.Receiver;
-const ReceiverVTable = panto.provider.ReceiverVTable;
 const ContentBlockType = panto.provider.ContentBlockType;
 const MessageRole = panto.conversation.MessageRole;
+const Event = panto.stream.Event;
 
-/// Receiver that prints streaming deltas to stdout. Thinking blocks are
-/// dimmed with ANSI escape codes; text blocks render plain.
-///
-/// Persistence is owned by the agent now; the receiver is display-only.
-const CLIReceiver = struct {
+/// Display state for the pull-stream CLI renderer. Prints streaming deltas
+/// to stdout; thinking blocks are dimmed with ANSI escape codes, text blocks
+/// render plain. Persistence is owned by the agent; the renderer is
+/// display-only.
+const CLIRenderer = struct {
     stdout: *std.Io.Writer,
     file: *std.Io.File.Writer,
     allocator: std.mem.Allocator,
 
-    pub fn receiver(self: *CLIReceiver) Receiver {
-        return .{ .ptr = self, .vtable = &vtable };
-    }
-
-    /// Per-turn reset hook. Currently a no-op (no per-turn receiver state),
+    /// Per-turn reset hook. Currently a no-op (no per-turn render state),
     /// retained as the REPL's turn-boundary signal.
-    pub fn beginTurn(self: *CLIReceiver) void {
+    pub fn beginTurn(self: *CLIRenderer) void {
         _ = self;
     }
 
-    pub fn deinit(self: *CLIReceiver) void {
+    pub fn deinit(self: *CLIRenderer) void {
         _ = self;
     }
 
-    const vtable: ReceiverVTable = .{
-        .onMessageStart = onMessageStart,
-        .onBlockStart = onBlockStart,
-        .onToolDetails = onToolDetails,
-        .onContentDelta = onContentDelta,
-        .onBlockComplete = onBlockComplete,
-        .onMessageComplete = onMessageComplete,
-        .onError = onError,
-        .onProviderRetry = onProviderRetry,
-    };
-
-    /// Surface provider retry scheduling as a dim status line so the user
-    /// can see the agent is waiting on the provider rather than hung.
-    fn onProviderRetry(ptr: *anyopaque, info: panto.provider.ProviderRetryInfo) void {
-        const self: *CLIReceiver = @ptrCast(@alignCast(ptr));
-        if (info.compaction) {
-            self.stdout.print(
-                "\x1b[2mcontext overflow: compacting and retrying\x1b[0m\n",
-                .{},
-            ) catch {};
-        } else {
-            const secs = @as(f64, @floatFromInt(info.delay_ms)) / 1000.0;
-            self.stdout.print(
-                "\x1b[2mprovider unavailable ({s}): retrying in {d:.1}s (attempt {d}/{d})\x1b[0m\n",
-                .{ @errorName(info.err), secs, info.attempt + 1, info.max_attempts },
-            ) catch {};
-        }
-        self.file.flush() catch {};
-    }
-
-    /// The print-based CLI defers tool-name rendering to onBlockComplete
-    /// to avoid cursor gymnastics (we can't go back and edit the prefix).
-    /// Receivers backed by a TUI capable of in-place updates would render
-    /// the name as soon as it's known here.
-    fn onToolDetails(
-        ptr: *anyopaque,
-        index: usize,
-        id: []const u8,
-        name: []const u8,
-    ) anyerror!void {
-        _ = ptr;
-        _ = index;
-        _ = id;
-        _ = name;
-    }
-
-    fn onMessageStart(ptr: *anyopaque, role: MessageRole) anyerror!void {
-        _ = ptr;
-        _ = role;
-    }
-
-    fn onBlockStart(
-        ptr: *anyopaque,
-        block_type: ContentBlockType,
-        index: usize,
-    ) anyerror!void {
-        _ = index;
-        const self: *CLIReceiver = @ptrCast(@alignCast(ptr));
-        switch (block_type) {
-            .Thinking => try self.stdout.writeAll("\x1b[2m[thinking] "),
-            // Tool name is not known reliably at start time (OpenAI may
-            // stream id/name across fragments). Open with a bare prefix;
-            // the tool name lands at onBlockComplete from the assembled
-            // ContentBlock.
-            .ToolUse => try self.stdout.writeAll("\n\x1b[36mtool: \x1b[0m"),
-            else => {},
-        }
-        try self.file.flush();
-    }
-
-    fn onContentDelta(ptr: *anyopaque, index: usize, delta: []const u8) anyerror!void {
-        _ = index;
-        const self: *CLIReceiver = @ptrCast(@alignCast(ptr));
-        try self.stdout.writeAll(delta);
-        try self.file.flush();
-    }
-
-    fn onBlockComplete(
-        ptr: *anyopaque,
-        index: usize,
-        block: panto.conversation.ContentBlock,
-    ) anyerror!void {
-        _ = index;
-        const self: *CLIReceiver = @ptrCast(@alignCast(ptr));
-        switch (block) {
-            .Thinking => try self.stdout.writeAll("\x1b[0m\n"),
-            // Append the tool name now that we know it for certain.
-            .ToolUse => |tu| try self.stdout.print("\x1b[36m : ({s})\x1b[0m\n", .{tu.name}),
-            else => {},
+    /// Render one pull `Event`. Mirrors the prior push receiver's output
+    /// exactly: dim `[thinking]`, a `tool:` prefix opened at block start
+    /// with the name appended at block_complete, a trailing newline at each
+    /// `message_complete`, and a dim retry status line.
+    fn render(self: *CLIRenderer, ev: Event) !void {
+        switch (ev) {
+            .message_start => {},
+            .block_start => |b| {
+                switch (b.block_type) {
+                    .Thinking => try self.stdout.writeAll("\x1b[2m[thinking] "),
+                    // Tool name is not known reliably at start time (OpenAI
+                    // may stream id/name across fragments). Open with a bare
+                    // prefix; the name lands at block_complete from the
+                    // assembled ContentBlock.
+                    .ToolUse => try self.stdout.writeAll("\n\x1b[36mtool: \x1b[0m"),
+                    else => {},
+                }
+                try self.file.flush();
+            },
+            // The print-based CLI defers tool-name rendering to
+            // block_complete to avoid cursor gymnastics (we can't go back and
+            // edit the prefix), so tool_details is a no-op here.
+            .tool_details => {},
+            .content_delta => |d| {
+                try self.stdout.writeAll(d.delta);
+                try self.file.flush();
+            },
+            .block_complete => |b| {
+                switch (b.block) {
+                    .Thinking => try self.stdout.writeAll("\x1b[0m\n"),
+                    // Append the tool name now that we know it for certain.
+                    .ToolUse => |tu| try self.stdout.print("\x1b[36m : ({s})\x1b[0m\n", .{tu.name}),
+                    else => {},
+                }
+                try self.file.flush();
+            },
+            .message_complete => {
+                try self.stdout.writeAll("\n");
+                try self.file.flush();
+            },
+            // Surface provider retry scheduling as a dim status line so the
+            // user can see the agent is waiting on the provider, not hung.
+            .provider_retry => |info| {
+                if (info.compaction) {
+                    self.stdout.print(
+                        "\x1b[2mcontext overflow: compacting and retrying\x1b[0m\n",
+                        .{},
+                    ) catch {};
+                } else {
+                    const secs = @as(f64, @floatFromInt(info.delay_ms)) / 1000.0;
+                    self.stdout.print(
+                        "\x1b[2mprovider unavailable ({s}): retrying in {d:.1}s (attempt {d}/{d})\x1b[0m\n",
+                        .{ @errorName(info.err), secs, info.attempt + 1, info.max_attempts },
+                    ) catch {};
+                }
+                self.file.flush() catch {};
+            },
+            .tool_dispatch_start, .tool_dispatch_complete, .turn_complete => {},
         }
-        try self.file.flush();
-    }
-
-    fn onMessageComplete(
-        ptr: *anyopaque,
-        message: panto.conversation.Message,
-        usage: ?panto.session_manager.Usage,
-    ) anyerror!void {
-        const self: *CLIReceiver = @ptrCast(@alignCast(ptr));
-        _ = message;
-        _ = usage;
-        try self.stdout.writeAll("\n");
-        try self.file.flush();
     }
 
     /// Reset any in-progress display state after a failed turn. Errors here
     /// must be swallowed — we're already in the error path.
-    fn onError(ptr: *anyopaque, err: anyerror) void {
+    fn renderError(self: *CLIRenderer, err: anyerror) void {
         _ = &err;
-        const self: *CLIReceiver = @ptrCast(@alignCast(ptr));
         // If we were rendering a Thinking block, clear the dim style so
         // subsequent output (including the "[error: ...]" line) is readable.
         self.stdout.writeAll("\x1b[0m") catch {};
@@ -190,6 +140,16 @@ const CLIReceiver = struct {
     }
 };
 
+/// Drive a whole turn: open the pull `Stream` and render every event until
+/// it terminates (`turn_complete` → `next()` returns null). A failure
+/// surfaces as the error from `next()`; the caller renders it. The stream is
+/// always `deinit`ed (persisting the turn tail) on every exit path.
+fn driveTurn(agent: *panto.agent.Agent, message: panto.agent.Agent.UserMessage, renderer: *CLIRenderer) !void {
+    var stream = try agent.run(message);
+    defer stream.deinit();
+    while (try stream.next()) |ev| try renderer.render(ev);
+}
+
 /// Spin up a Lua interpreter, run a no-op, tear it down. Catches
 /// link/compile errors on the Lua dependency at the earliest moment.
 /// Logs only in debug builds; release builds run silently.
@@ -532,13 +492,12 @@ pub fn main(init: std.process.Init) !void {
     try stdout.print("> ", .{});
     try stdout_file.flush();
 
-    var cli_recv = CLIReceiver{
+    var cli_renderer = CLIRenderer{
         .stdout = stdout,
         .file = &stdout_file,
         .allocator = alloc,
     };
-    defer cli_recv.deinit();
-    var recv = cli_recv.receiver();
+    defer cli_renderer.deinit();
 
     // Build the slash-command registry and register builtins, then append
     // any commands declared by Lua extensions (harvested at load time).
@@ -609,15 +568,14 @@ pub fn main(init: std.process.Init) !void {
             continue;
         }
 
-        // Submit + persist the user prompt, then drive the turn. The agent
-        // owns the conversation and persists everything it generates
+        // Drive the turn. `run` submits + persists the user prompt and the
+        // agent owns the conversation, persisting everything it generates
         // (assistant/tool-result messages, and the compaction window on
         // auto-compaction) through its session store — the CLI no longer
         // touches persistence.
-        try agent.submitUserMessage(line);
-
-        cli_recv.beginTurn();
-        agent.runStep(&recv) catch |err| {
+        cli_renderer.beginTurn();
+        driveTurn(&agent, .{ .text = line }, &cli_renderer) catch |err| {
+            cli_renderer.renderError(err);
             try stdout.print("\n[error: {s}]\n", .{@errorName(err)});
         };