From 73645129a9de90f867908d35e77c9252bae4e534 Mon Sep 17 00:00:00 2001 From: t Date: Fri, 5 Jun 2026 13:55:20 -0600 Subject: 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. --- libpanto/src/provider.zig | 153 ++++++++++++++++++++-------------------------- 1 file changed, 66 insertions(+), 87 deletions(-) (limited to 'libpanto/src/provider.zig') 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; -- cgit v1.3