1 files changed, 179 insertions, 0 deletions
diff --git a/src/stream.zig b/src/stream.zig
new file mode 100644
index 0000000..64748b2
--- /dev/null
+++ b/src/stream.zig
@@ -0,0 +1,179 @@
+//! 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,
+
+    /// One tool result is available. The payload is a user-role carrier
+    /// containing exactly one `ToolResult` block, keyed by `tool_use_id`.
+    /// This may arrive before the aggregate `tool_dispatch_complete` event.
+    tool_dispatch_result: ToolDispatchComplete,
+
+    /// 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;
+    }
+};