diff options
Diffstat (limited to 'libpanto/src/stream.zig')
| -rw-r--r-- | libpanto/src/stream.zig | 179 |
1 files changed, 0 insertions, 179 deletions
diff --git a/libpanto/src/stream.zig b/libpanto/src/stream.zig deleted file mode 100644 index 64748b2..0000000 --- a/libpanto/src/stream.zig +++ /dev/null @@ -1,179 +0,0 @@ -//! 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; - } -}; |
