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

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

1 files changed, 458 insertions, 329 deletions

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.