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, 85 insertions, 127 deletions

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