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, 66 insertions, 87 deletions

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;