diff options
Diffstat (limited to 'libpanto')
| -rw-r--r-- | libpanto/src/agent.zig | 787 | ||||
| -rw-r--r-- | libpanto/src/provider.zig | 153 | ||||
| -rw-r--r-- | libpanto/src/provider_anthropic_messages.zig | 448 | ||||
| -rw-r--r-- | libpanto/src/provider_openai_chat.zig | 572 | ||||
| -rw-r--r-- | libpanto/src/root.zig | 16 | ||||
| -rw-r--r-- | libpanto/src/stream.zig | 174 |
6 files changed, 1250 insertions, 900 deletions
diff --git a/libpanto/src/agent.zig b/libpanto/src/agent.zig index 122da1a..6749ffd 100644 --- a/libpanto/src/agent.zig +++ b/libpanto/src/agent.zig @@ -29,6 +29,7 @@ const Allocator = std.mem.Allocator; const Io = std.Io; const provider_mod = @import("provider.zig"); +const stream_mod = @import("stream.zig"); const config_mod = @import("config.zig"); const conversation = @import("conversation.zig"); const compaction_mod = @import("compaction.zig"); @@ -44,6 +45,8 @@ pub const Tool = tool_mod.Tool; pub const ToolSource = tool_source_mod.ToolSource; pub const ToolRegistry = tool_registry_mod.ToolRegistry; +const Event = stream_mod.Event; + const Entry = tool_registry_mod.Entry; pub const Config = config_mod.Config; @@ -122,38 +125,6 @@ fn cloneBlock(alloc: Allocator, block: conversation.ContentBlock) !conversation. }; } -/// A minimal receiver that captures the assistant's streamed message for -/// compaction. We don't need incremental events — the assembled message is -/// read off the conversation after the turn — so all callbacks are no-ops. -const CompactionCapture = struct { - allocator: Allocator, - - fn receiver(self: *CompactionCapture) provider_mod.Receiver { - return .{ .ptr = self, .vtable = &vt }; - } - fn deinit(self: *CompactionCapture) void { - _ = self; - } - const vt: provider_mod.ReceiverVTable = .{ - .onMessageStart = onMessageStart, - .onBlockStart = onBlockStart, - .onToolDetails = onToolDetails, - .onContentDelta = onContentDelta, - .onBlockComplete = onBlockComplete, - .onMessageComplete = onMessageComplete, - .onError = onError, - .onProviderRetry = onProviderRetry, - }; - fn onMessageStart(_: *anyopaque, _: conversation.MessageRole) anyerror!void {} - fn onBlockStart(_: *anyopaque, _: provider_mod.ContentBlockType, _: usize) anyerror!void {} - fn onToolDetails(_: *anyopaque, _: usize, _: []const u8, _: []const u8) anyerror!void {} - fn onContentDelta(_: *anyopaque, _: usize, _: []const u8) anyerror!void {} - fn onBlockComplete(_: *anyopaque, _: usize, _: conversation.ContentBlock) anyerror!void {} - fn onMessageComplete(_: *anyopaque, _: conversation.Message, _: ?provider_mod.Usage) anyerror!void {} - fn onError(_: *anyopaque, _: anyerror) void {} - fn onProviderRetry(_: *anyopaque, _: provider_mod.ProviderRetryInfo) void {} -}; - fn isValidToolInput(input: []const u8) bool { if (input.len == 0) return true; if (input[0] != '{') return true; // legacy tests/tools may use opaque bytes @@ -236,8 +207,8 @@ pub const Agent = struct { persist_provider: []const u8 = "", persist_model: []const u8 = "", /// Injectable streaming seam. Defaults to the real provider dispatch - /// (`provider_mod.streamStep`); tests override it with a stub. - stream_fn: provider_mod.StreamFn = provider_mod.streamStep, + /// (`provider_mod.openStream`); tests override it with a stub. + open_stream_fn: provider_mod.OpenStreamFn = provider_mod.openStream, /// Compaction system prompt used for automatic compaction on context /// overflow. Borrowed; set by the embedder (resolved from its /// `COMPACTION.md` layers). When null, auto-compaction is disabled and @@ -313,24 +284,6 @@ pub const Agent = struct { return self.config.registry; } - /// Add a user message to the conversation and durably persist it - /// immediately (before any provider call). The user prompt is logged on - /// submission, so a crash before the model replies leaves a recoverable - /// dangling prompt in the store. - pub fn submitUserMessage(self: *Agent, text: []const u8) !void { - const start = self.conversation.messages.items.len; - try self.conversation.addUserMessage(text); - const pm = self.providerModel(); - try turn_persist.persistTurn( - self.allocator, - self.session_store, - &self.conversation, - start, - pm.provider, - pm.model, - ); - } - /// Add a system message (append or replace mode) to the conversation /// and persist it. The persisted entry records the mode so replay /// reconstructs the same effective system prompt. @@ -355,45 +308,51 @@ pub const Agent = struct { ); } - /// Drive the conversation forward until the model stops calling tools. + /// The user's submission that opens a turn. A struct (not a bare slice) + /// so it can grow to carry file/image attachments alongside the chat + /// text without changing `run`'s signature. + pub const UserMessage = struct { + text: []const u8, + }; + + /// Submit a user message and begin a turn, returning a resumable pull + /// `Stream`. /// - /// Persists everything the turn appends to the conversation: assistant - /// messages and tool-result user messages. Persistence runs on exit - /// regardless of how the turn ended — including error paths that - /// committed messages before failing — so a partial turn is durably - /// logged. An automatic compaction during the turn is persisted as a - /// fresh compaction window instead. - pub fn runStep( - self: *Agent, - receiver: *provider_mod.Receiver, - ) !void { + /// The user message is appended to the conversation and durably persisted + /// *immediately* (before any provider call), so a crash before the model + /// replies leaves a recoverable dangling prompt in the store. No provider + /// I/O happens here: the request opens lazily on the first + /// `Stream.next()`. + /// + /// The returned `*Stream` is heap-allocated; the caller owns it and must + /// `deinit` it. Persistence of whatever the turn committed runs when the + /// stream reaches its terminal `turn_complete` or is `deinit`ed early + /// (so a partial turn is still durably logged), mirroring the previous + /// `runStep` exit-path guarantee. + /// + /// The agent re-reads its `config` snapshot at the top of each provider + /// response inside the stream, so a mid-conversation `setConfig` takes + /// effect at the next response boundary, never mid-stream. + pub fn run(self: *Agent, message: UserMessage) !*Stream { self.auto_compacted = false; - const conv = &self.conversation; - const start = conv.messages.items.len; - // Persist whatever the turn committed, on every exit path. - defer self.persistTurnTail(start) catch |e| { - std.log.err("session: failed to persist turn: {t}", .{e}); - }; - - while (true) { - // Re-read the config snapshot at the top of each turn so a - // mid-conversation swap takes effect here, never mid-stream. - const cfg = self.config; - try self.streamWithRetries(cfg, receiver); - - const last = conv.messages.items[conv.messages.items.len - 1]; - std.debug.assert(last.role == .assistant); - - // Defense-in-depth: a provider that silently committed an - // empty assistant message means the turn made no observable - // progress. Surface it instead of looping back to the prompt. - if (last.content.items.len == 0) return error.EmptyAssistantResponse; - - if (!hasToolUseBlock(last)) return; + // Append + persist the user prompt up front (the dangling-prompt + // recovery guarantee). + const user_start = self.conversation.messages.items.len; + try self.conversation.addUserMessage(message.text); + const pm = self.providerModel(); + try turn_persist.persistTurn( + self.allocator, + self.session_store, + &self.conversation, + user_start, + pm.provider, + pm.model, + ); - try self.dispatchToolCalls(last); - } + const s = try self.allocator.create(Stream); + s.* = Stream.init(self); + return s; } /// Persist the messages a turn produced. When the turn auto-compacted, @@ -429,9 +388,12 @@ pub const Agent = struct { return false; } - /// Drive one provider turn with the configured retry policy. + /// Open one provider response with the configured retry policy, pushing + /// any informational `provider_retry` events into `out`. Returns the + /// resumable `ProviderStream` once a request has been successfully + /// opened (headers received), or propagates a terminal error. /// - /// Decision path for a failed attempt: + /// Decision path for a failed open: /// - `ContextOverflow`: compact once, then retry the same request a /// single time against the compacted conversation (a one-shot path, /// independent of the transient-retry budget). @@ -442,29 +404,30 @@ pub const Agent = struct { /// - anything else (auth, bad request, cancellation, local errors): /// propagate immediately. /// - /// A failed attempt never mutates the conversation (providers commit the + /// A failed open never mutates the conversation (providers commit the /// assistant message only on success), so each retry runs against the - /// same snapshot. - fn streamWithRetries( + /// same snapshot. Mid-stream failures (surfaced from `produce`) re-enter + /// here via `Stream`, replaying the response from a fresh open — exactly + /// as the previous push loop replayed a failed `streamStep`. + fn openWithRetries( self: *Agent, cfg: *const Config, - receiver: *provider_mod.Receiver, - ) !void { + out: *stream_mod.EventQueue, + ) !provider_mod.ProviderStream { const policy = cfg.retry; var attempt: usize = 1; while (true) { var diag: provider_mod.ProviderDiagnostic = .{}; - self.stream_fn(self.allocator, self.io, cfg, &self.conversation, receiver, &diag) catch |err| { + const ps = self.open_stream_fn(self.allocator, self.io, cfg, &self.conversation, &diag) catch |err| { if (err == error.ContextOverflow) { - try self.handleContextOverflow(cfg, receiver, err); - return; + return self.handleContextOverflow(cfg, out, err); } if (!provider_mod.isRetryableProviderError(err)) return err; // Out of attempts: hard-fail with the last error. if (attempt >= policy.max_attempts) return err; const delay_ms = self.backoffDelayMs(policy, attempt, diag.retry_after_ms); - receiver.onProviderRetry(.{ + try out.push(.{ .provider_retry = .{ .attempt = attempt, .max_attempts = policy.max_attempts, .delay_ms = delay_ms, @@ -472,7 +435,7 @@ pub const Agent = struct { .status_code = diag.status_code, .retry_after_ms = diag.retry_after_ms, .message = diag.message, - }); + } }); if (delay_ms > 0) { const ms: i64 = @intCast(@min(delay_ms, std.math.maxInt(i64))); self.io.sleep(.fromMilliseconds(ms), .real) catch |e| return e; @@ -480,36 +443,35 @@ pub const Agent = struct { attempt += 1; continue; }; - return; + return ps; } } - /// One-shot context-overflow recovery: compact once, retry once. Mirrors - /// the prior inline behavior, now fired from `streamWithRetries`. The - /// retry is announced through `onProviderRetry` with `compaction = true` - /// and `delay_ms = 0`. + /// One-shot context-overflow recovery: compact once, retry once. Pushes + /// a `provider_retry` event with `compaction = true` and `delay_ms = 0`, + /// then re-opens the request against the compacted context. A second + /// overflow (or any other error) propagates. fn handleContextOverflow( self: *Agent, cfg: *const Config, - receiver: *provider_mod.Receiver, + out: *stream_mod.EventQueue, err: anyerror, - ) !void { + ) !provider_mod.ProviderStream { if (self.auto_compacted) return err; // already retried once this turn const sys = self.compaction_system_prompt orelse return err; const res = try self.compact(sys, null); if (!res.compacted) return err; // nothing to shed; give up self.auto_compacted = true; - receiver.onProviderRetry(.{ + try out.push(.{ .provider_retry = .{ .attempt = 1, .max_attempts = 2, .delay_ms = 0, .err = err, .compaction = true, - }); - // Retry the same request against the compacted context. A second - // overflow (or any other error) propagates. + } }); + // Retry the same request against the compacted context. var diag: provider_mod.ProviderDiagnostic = .{}; - try self.stream_fn(self.allocator, self.io, cfg, &self.conversation, receiver, &diag); + return self.open_stream_fn(self.allocator, self.io, cfg, &self.conversation, &diag); } /// Compute the backoff delay (ms) for the just-failed `attempt` @@ -829,11 +791,20 @@ pub const Agent = struct { try conv.addSystemMessage(system_prompt); try conv.addUserMessage(body); - var capture = CompactionCapture{ .allocator = alloc }; - defer capture.deinit(); - var recv = capture.receiver(); - - try self.stream_fn(alloc, self.io, cfg, &conv, &recv, null); + // Drive one provider response to completion, ignoring every event. + // Compaction doesn't need incremental output — the assembled message + // is read off the conversation below — so we just pump the pull + // stream until it commits the assistant message. + var queue = stream_mod.EventQueue.init(alloc); + defer queue.deinit(); + var ps = try self.open_stream_fn(alloc, self.io, cfg, &conv, null); + defer ps.deinit(); + while (true) { + const status = try ps.produce(&queue); + // Drain (and discard) any events to bound the queue/arena. + while (queue.pop()) |_| {} + if (status == .response_complete) break; + } // The provider appended an assistant message; gather its text. const last = conv.messages.items[conv.messages.items.len - 1]; @@ -1081,6 +1052,206 @@ pub const Agent = struct { } }; +/// A resumable pull handle over one agent turn. +/// +/// `next()` pulls one `Event` at a time, driving the agent loop +/// incrementally: open a provider response, stream its events, dispatch any +/// tool calls between responses, and repeat until the model stops calling +/// tools. The whole loop's state lives here (not on a stack frame), so the +/// turn can suspend and resume between events. +/// +/// Contract (see `stream.zig`): +/// - an `Event` value is streaming progress, including `turn_complete`; +/// - `null` means exhausted (already past `turn_complete`), never before; +/// - an error is a genuine failure (network/parse/provider). +/// +/// Event payloads borrow from the stream's decode state or the conversation +/// and are valid only until the next `next()` call. +pub const Stream = struct { + agent: *Agent, + queue: stream_mod.EventQueue, + phase: Phase, + /// The active provider response, when in `.streaming`. + response: ?provider_mod.ProviderStream = null, + /// First message index of this turn (for persistence). + start: usize, + /// Set once the turn's tail has been persisted (on terminal or deinit). + persisted: bool = false, + /// A terminal error to surface once any already-queued events (e.g. + /// `provider_retry` notices pushed before the failing attempt) have been + /// drained. `next()` yields the queue first, then this error. + pending_error: ?anyerror = null, + + const Phase = enum { + /// Open the next provider response (with retries). + turn_start, + /// Pump the active provider response into events. + streaming, + /// A provider response completed; decide tools-vs-done. + after_response, + /// The turn reached its terminal `turn_complete`. + done, + /// A failure already propagated; `next()` is poisoned. + failed, + }; + + fn init(agent: *Agent) Stream { + return .{ + .agent = agent, + .queue = stream_mod.EventQueue.init(agent.allocator), + .phase = .turn_start, + .start = agent.conversation.messages.items.len, + }; + } + + pub fn deinit(self: *Stream) void { + // Persist whatever the turn committed, on every exit path — including + // dropping the stream mid-turn after some messages were committed. + self.persistTail(); + if (self.response) |ps| ps.deinit(); + self.queue.deinit(); + self.agent.allocator.destroy(self); + } + + fn persistTail(self: *Stream) void { + if (self.persisted) return; + self.persisted = true; + self.agent.persistTurnTail(self.start) catch |e| { + std.log.err("session: failed to persist turn: {t}", .{e}); + }; + } + + /// Pull the next event, or null past the terminal. See the contract + /// above. + pub fn next(self: *Stream) !?Event { + // Always drain queued events first; they borrow decode/conversation + // state valid until this call returns. + if (self.queue.pop()) |ev| return ev; + // Queue drained: a deferred terminal error surfaces now. + if (self.pending_error) |err| { + self.pending_error = null; + self.phase = .failed; + return err; + } + + while (true) { + switch (self.phase) { + .done => return null, + .failed => return error.StreamPoisoned, + .turn_start => { + // Re-read the config snapshot at each response boundary + // so a mid-conversation swap takes effect here, never + // mid-stream. + const cfg = self.agent.config; + const ps = self.agent.openWithRetries(cfg, &self.queue) catch |err| { + // Surface the failure after any queued retry notices + // (pushed before each backoff) are drained. + if (self.queue.pop()) |ev| { + self.pending_error = err; + return ev; + } + self.phase = .failed; + return err; + }; + self.response = ps; + self.phase = .streaming; + if (self.queue.pop()) |ev| return ev; // retry notices + }, + .streaming => { + const ps = self.response.?; + const status = ps.produce(&self.queue) catch |err| { + // Mid-stream failure. The conversation was not + // mutated (commit happens only at response + // completion), so retry by re-opening from scratch, + // exactly as the prior push loop replayed a failed + // streamStep. Non-retryable errors propagate. + ps.deinit(); + self.response = null; + if (!provider_mod.isRetryableProviderError(err)) { + self.phase = .failed; + return err; + } + self.phase = .turn_start; + // Emit a retry notice so consumers see the stall. + const cfg = self.agent.config; + const delay_ms = self.agent.backoffDelayMs(cfg.retry, 1, null); + self.queue.push(.{ .provider_retry = .{ + .attempt = 1, + .max_attempts = cfg.retry.max_attempts, + .delay_ms = delay_ms, + .err = err, + } }) catch |e| { + self.phase = .failed; + return e; + }; + if (delay_ms > 0) { + const ms: i64 = @intCast(@min(delay_ms, std.math.maxInt(i64))); + self.agent.io.sleep(.fromMilliseconds(ms), .real) catch |e| { + self.phase = .failed; + return e; + }; + } + if (self.queue.pop()) |ev| return ev; + continue; + }; + if (self.queue.pop()) |ev| return ev; + if (status == .response_complete) { + ps.deinit(); + self.response = null; + self.phase = .after_response; + } + // else `.more`: loop and pump again. + }, + .after_response => { + const conv = &self.agent.conversation; + const last = conv.messages.items[conv.messages.items.len - 1]; + std.debug.assert(last.role == .assistant); + + // Defense-in-depth: a provider that silently committed an + // empty assistant message means the turn made no + // observable progress. Surface it instead of looping. + if (last.content.items.len == 0) { + self.phase = .failed; + return error.EmptyAssistantResponse; + } + + if (!Agent.hasToolUseBlock(last)) { + self.phase = .done; + self.persistTail(); + return .turn_complete; + } + + // Dispatch the tool calls, bracketed by boundary events. + const count = toolUseCount(last); + self.queue.push(.{ .tool_dispatch_start = .{ .count = count } }) catch |e| { + self.phase = .failed; + return e; + }; + self.agent.dispatchToolCalls(last) catch |err| { + self.phase = .failed; + return err; + }; + const result_msg = conv.messages.items[conv.messages.items.len - 1]; + self.queue.push(.{ .tool_dispatch_complete = .{ .message = result_msg } }) catch |e| { + self.phase = .failed; + return e; + }; + self.phase = .turn_start; + if (self.queue.pop()) |ev| return ev; + }, + } + } + } +}; + +fn toolUseCount(msg: conversation.Message) usize { + var n: usize = 0; + for (msg.content.items) |block| { + if (block == .ToolUse) n += 1; + } + return n; +} + /// One ToolUse, as flattened into the agent's dispatch list. `result` /// and `err` are filled in by the worker; exactly one is non-null on /// successful task completion. @@ -1249,6 +1420,17 @@ fn runSourceGroup(agent: *Agent, sg: Group.SourceGroup, calls: []FlatCall) void const testing = std.testing; +/// Test helper: submit `text` and drive the whole turn to completion via the +/// pull `Stream`, discarding every event (the agent tests assert on +/// conversation/store state, not on the event stream). Mirrors the old +/// `submitUserMessage` + `runStep`: it returns the same terminal error a +/// turn would raise. +fn drainTurn(agent: *Agent, text: []const u8) !void { + var s = try agent.run(.{ .text = text }); + defer s.deinit(); + while (try s.next()) |_| {} +} + /// Test helper: the items of a ToolResultBlock's first text part. fn trText(tr: conversation.ToolResultBlock) []const u8 { for (tr.parts.items) |p| { @@ -1257,31 +1439,30 @@ fn trText(tr: conversation.ToolResultBlock) []const u8 { return ""; } -/// Test harness for the injectable `stream_fn` seam. +/// Test harness for the injectable `open_stream_fn` seam. /// -/// `provider_mod.StreamFn` carries no user context (it mirrors the real +/// `provider_mod.OpenStreamFn` carries no user context (it mirrors the real /// free function exactly), so the stub parks its state in a module-level -/// pointer that `stubStreamStep` reads. The Zig test runner executes tests -/// serially in one process, so a single global slot is safe; each test -/// sets it via `install` before driving the agent. +/// pointer that `stubOpenStream` reads. The Zig test runner executes tests +/// serially in one process, so a single global slot is safe; each test sets +/// it via `install` before driving the agent. var stub_active: ?*StubProvider = null; const StubProvider = struct { allocator: Allocator, scripted: []const ScriptedTurn, next: usize = 0, - /// Number of leading stream calls that should fail with + /// Number of leading stream opens that should fail with /// `error.ContextOverflow` before any scripted turn is served. Used to /// drive the auto-compaction path. Decremented on each overflow. overflow_calls: usize = 0, /// A queue of provider errors to return, in order, before any scripted - /// turn is served. Each entry is consumed on one stream call. Used to - /// drive the transient-retry path. `diag_retry_after_ms`, when set on an - /// entry, is stashed into the caller's `ProviderDiagnostic`. + /// turn is served. Each entry is consumed on one open. Used to drive the + /// transient-retry path. scripted_errors: []const ScriptedError = &.{}, error_idx: usize = 0, - /// Count of stream calls observed (failed + succeeded). Lets tests - /// assert the exact number of attempts. + /// Count of opens observed (failed + succeeded). Lets tests assert the + /// exact number of attempts. calls_made: usize = 0, const ScriptedError = struct { @@ -1293,7 +1474,7 @@ const StubProvider = struct { const ScriptedTurn = struct { blocks: []const TestBlock, /// Optional provider usage to stamp on the committed assistant - /// message (mirrors a real provider's `onMessageComplete` usage). + /// message (mirrors a real provider's terminal usage). usage: ?conversation.Usage = null, }; @@ -1307,24 +1488,94 @@ const StubProvider = struct { }; /// Point the global seam at this stub and return the function to assign - /// to `agent.stream_fn`. Call once per test, after constructing the + /// to `agent.open_stream_fn`. Call once per test, after constructing the /// stub on the stack. - fn install(self: *StubProvider) provider_mod.StreamFn { + fn install(self: *StubProvider) provider_mod.OpenStreamFn { stub_active = self; - return stubStreamStep; + return stubOpenStream; } }; -fn stubStreamStep( +/// A canned resumable response: on the first `produce` it commits the +/// scripted assistant message to the conversation and pushes the terminal +/// `message_complete`, then reports `.response_complete`. It does not emit +/// per-block events (the agent tests assert on conversation state, not the +/// event stream), which is sufficient for driving the agent loop. +const StubResponse = struct { + allocator: Allocator, + conv: *conversation.Conversation, + turn: StubProvider.ScriptedTurn, + done: bool = false, + + fn create( + allocator: Allocator, + conv: *conversation.Conversation, + turn: StubProvider.ScriptedTurn, + ) !provider_mod.ProviderStream { + const self = try allocator.create(StubResponse); + self.* = .{ .allocator = allocator, .conv = conv, .turn = turn }; + return .{ .ptr = self, .vtable = &vtable }; + } + + const vtable: provider_mod.ProviderStream.VTable = .{ + .produce = produceVT, + .deinit = deinitVT, + }; + + fn produceVT(ptr: *anyopaque, out: *stream_mod.EventQueue) anyerror!provider_mod.ProviderStream.ProduceStatus { + const self: *StubResponse = @ptrCast(@alignCast(ptr)); + if (self.done) return .response_complete; + self.done = true; + + var blocks: std.ArrayList(conversation.ContentBlock) = .empty; + errdefer { + for (blocks.items) |*b| b.deinit(self.allocator); + blocks.deinit(self.allocator); + } + for (self.turn.blocks) |tb| { + switch (tb) { + .Text => |s| try blocks.append(self.allocator, .{ + .Text = try conversation.textualBlockFromSlice(self.allocator, s), + }), + .ToolUse => |tu| { + const id = try self.allocator.dupe(u8, tu.id); + errdefer self.allocator.free(id); + const name = try self.allocator.dupe(u8, tu.name); + errdefer self.allocator.free(name); + var input_buf: conversation.TextualBlock = .empty; + errdefer input_buf.deinit(self.allocator); + try input_buf.appendSlice(self.allocator, tu.input); + try blocks.append(self.allocator, .{ .ToolUse = .{ + .id = id, + .name = name, + .input = input_buf, + } }); + }, + } + } + const moved = try blocks.toOwnedSlice(self.allocator); + defer self.allocator.free(moved); + try self.conv.addAssistantMessageWithUsage(moved, self.turn.usage); + + const msg = self.conv.messages.items[self.conv.messages.items.len - 1]; + try out.push(.{ .message_complete = .{ .message = msg, .usage = self.turn.usage } }); + return .response_complete; + } + + fn deinitVT(ptr: *anyopaque) void { + const self: *StubResponse = @ptrCast(@alignCast(ptr)); + self.allocator.destroy(self); + } +}; + +fn stubOpenStream( allocator: Allocator, _: Io, _: *const config_mod.Config, conv: *conversation.Conversation, - _: *provider_mod.Receiver, diag: ?*provider_mod.ProviderDiagnostic, -) anyerror!void { +) anyerror!provider_mod.ProviderStream { const self = stub_active orelse return error.NoStubInstalled; - _ = allocator; self.calls_made += 1; if (self.error_idx < self.scripted_errors.len) { const e = self.scripted_errors[self.error_idx]; @@ -1342,38 +1593,7 @@ fn stubStreamStep( if (self.next >= self.scripted.len) return error.NoMoreScriptedTurns; const turn = self.scripted[self.next]; self.next += 1; - - var blocks: std.ArrayList(conversation.ContentBlock) = .empty; - errdefer { - for (blocks.items) |*b| b.deinit(self.allocator); - blocks.deinit(self.allocator); - } - for (turn.blocks) |tb| { - switch (tb) { - .Text => |s| { - try blocks.append(self.allocator, .{ - .Text = try conversation.textualBlockFromSlice(self.allocator, s), - }); - }, - .ToolUse => |tu| { - const id = try self.allocator.dupe(u8, tu.id); - errdefer self.allocator.free(id); - const name = try self.allocator.dupe(u8, tu.name); - errdefer self.allocator.free(name); - var input_buf: conversation.TextualBlock = .empty; - errdefer input_buf.deinit(self.allocator); - try input_buf.appendSlice(self.allocator, tu.input); - try blocks.append(self.allocator, .{ .ToolUse = .{ - .id = id, - .name = name, - .input = input_buf, - } }); - }, - } - } - const moved = try blocks.toOwnedSlice(self.allocator); - defer self.allocator.free(moved); - try conv.addAssistantMessageWithUsage(moved, turn.usage); + return StubResponse.create(allocator, conv, turn); } /// Build a stack registry + active `Config` snapshot wired together, for @@ -1551,31 +1771,6 @@ const HardFailTool = struct { } }; -const NoopReceiver = struct { - fn make() provider_mod.Receiver { - return .{ .ptr = @ptrCast(@constCast(&dummy)), .vtable = &vt }; - } - var dummy: u8 = 0; - const vt: provider_mod.ReceiverVTable = .{ - .onMessageStart = noop1, - .onBlockStart = noop2, - .onToolDetails = noopToolDetails, - .onContentDelta = noop3, - .onBlockComplete = noop4, - .onMessageComplete = noop5, - .onError = noop6, - .onProviderRetry = noop7, - }; - fn noop1(_: *anyopaque, _: conversation.MessageRole) anyerror!void {} - fn noop2(_: *anyopaque, _: provider_mod.ContentBlockType, _: usize) anyerror!void {} - fn noopToolDetails(_: *anyopaque, _: usize, _: []const u8, _: []const u8) anyerror!void {} - fn noop3(_: *anyopaque, _: usize, _: []const u8) anyerror!void {} - fn noop4(_: *anyopaque, _: usize, _: conversation.ContentBlock) anyerror!void {} - fn noop5(_: *anyopaque, _: conversation.Message, _: ?provider_mod.Usage) anyerror!void {} - fn noop6(_: *anyopaque, _: anyerror) void {} - fn noop7(_: *anyopaque, _: provider_mod.ProviderRetryInfo) void {} -}; - /// An in-memory `SessionStore` test double: records every appended /// `DiskMessage` (role + provider/model stamp) so tests can assert the /// agent persisted the right turn without touching disk. Honors the store @@ -1661,13 +1856,11 @@ test "agent persists user, assistant, and tool-result messages of a turn" { defer cap.deinit(); var agent = Agent.init(allocator, io, &h.config, cap.store(), null); defer agent.deinit(); - agent.stream_fn = stub.install(); + agent.open_stream_fn = stub.install(); agent.persist_provider = "openai"; agent.persist_model = "gpt-4o"; - var recv = NoopReceiver.make(); - try agent.submitUserMessage("call a tool"); - try agent.runStep(&recv); + try drainTurn(&agent, "call a tool"); // Persisted, in order: user prompt, assistant(ToolUse), user(ToolResult), // assistant(text). @@ -1701,11 +1894,9 @@ test "agent runs a turn against NullStore without persisting or erroring" { var ns = null_store_mod.NullStore.init(allocator); var agent = Agent.init(allocator, io, &h.config, ns.store(), null); defer agent.deinit(); - agent.stream_fn = stub.install(); + agent.open_stream_fn = stub.install(); - var recv = NoopReceiver.make(); - try agent.submitUserMessage("hello"); - try agent.runStep(&recv); + try drainTurn(&agent, "hello"); // Nothing crashed; the conversation has the user + assistant messages. try testing.expectEqual(@as(usize, 2), agent.conversation.messages.items.len); @@ -1992,13 +2183,11 @@ test "runStep dispatches a tool call and loops to a final text turn" { var ns = null_store_mod.NullStore.init(allocator); var agent = Agent.init(allocator, io, &h.config, ns.store(), null); defer agent.deinit(); - agent.stream_fn = stub.install(); + agent.open_stream_fn = stub.install(); const conv = &agent.conversation; - try conv.addUserMessage("call a tool"); - var recv = NoopReceiver.make(); - try agent.runStep(&recv); + try drainTurn(&agent, "call a tool"); try testing.expectEqual(@as(usize, 4), conv.messages.items.len); @@ -2044,13 +2233,11 @@ test "runStep dispatches multiple tool calls in parallel" { var ns = null_store_mod.NullStore.init(allocator); var agent = Agent.init(allocator, io, &h.config, ns.store(), null); defer agent.deinit(); - agent.stream_fn = stub.install(); + agent.open_stream_fn = stub.install(); const conv = &agent.conversation; - try conv.addUserMessage("go"); - var recv = NoopReceiver.make(); - try agent.runStep(&recv); + try drainTurn(&agent, "go"); const tr_msg = conv.messages.items[2]; try testing.expectEqual(@as(usize, 3), tr_msg.content.items.len); @@ -2085,13 +2272,11 @@ test "runStep: native tool handler error becomes an error result and the model g var ns = null_store_mod.NullStore.init(allocator); var agent = Agent.init(allocator, io, &h.config, ns.store(), null); defer agent.deinit(); - agent.stream_fn = stub.install(); + agent.open_stream_fn = stub.install(); const conv = &agent.conversation; - try conv.addUserMessage("break it"); - var recv = NoopReceiver.make(); - try agent.runStep(&recv); + try drainTurn(&agent, "break it"); // user, assistant(tool_use), user(tool_result), assistant(text) try testing.expectEqual(@as(usize, 4), conv.messages.items.len); @@ -2120,13 +2305,11 @@ test "runStep: unknown tool becomes an error tool result and the loop continues" var ns = null_store_mod.NullStore.init(allocator); var agent = Agent.init(allocator, io, &h.config, ns.store(), null); defer agent.deinit(); - agent.stream_fn = stub.install(); + agent.open_stream_fn = stub.install(); const conv = &agent.conversation; - try conv.addUserMessage("call a ghost"); - var recv = NoopReceiver.make(); - try agent.runStep(&recv); + try drainTurn(&agent, "call a ghost"); // messages: user, assistant(tool_use), user(tool_result), assistant(text) try testing.expectEqual(@as(usize, 4), conv.messages.items.len); @@ -2154,13 +2337,11 @@ test "runStep with no tool calls returns after one provider step" { var ns = null_store_mod.NullStore.init(allocator); var agent = Agent.init(allocator, io, &h.config, ns.store(), null); defer agent.deinit(); - agent.stream_fn = stub.install(); + agent.open_stream_fn = stub.install(); const conv = &agent.conversation; - try conv.addUserMessage("hello"); - var recv = NoopReceiver.make(); - try agent.runStep(&recv); + try drainTurn(&agent, "hello"); try testing.expectEqual(@as(usize, 2), conv.messages.items.len); try testing.expectEqualStrings("hi", conv.messages.items[1].content.items[0].Text.items); @@ -2182,13 +2363,10 @@ test "runStep surfaces EmptyAssistantResponse when provider commits an empty mes var ns = null_store_mod.NullStore.init(allocator); var agent = Agent.init(allocator, io, &h.config, ns.store(), null); defer agent.deinit(); - agent.stream_fn = stub.install(); + agent.open_stream_fn = stub.install(); - const conv = &agent.conversation; - try conv.addUserMessage("hi"); - var recv = NoopReceiver.make(); - try testing.expectError(error.EmptyAssistantResponse, agent.runStep(&recv)); + try testing.expectError(error.EmptyAssistantResponse, drainTurn(&agent, "hi")); } // ------------ ToolSource tests ------------ @@ -2215,13 +2393,11 @@ test "runStep delivers all source-backed calls in one batch on one thread" { var ns = null_store_mod.NullStore.init(allocator); var agent = Agent.init(allocator, io, &h.config, ns.store(), null); defer agent.deinit(); - agent.stream_fn = stub.install(); + agent.open_stream_fn = stub.install(); const conv = &agent.conversation; - try conv.addUserMessage("go"); - var recv = NoopReceiver.make(); - try agent.runStep(&recv); + try drainTurn(&agent, "go"); // Locate the source and inspect its observed batches. const view = h.registry.lookup("lua_x") orelse return error.NotFound; @@ -2268,13 +2444,10 @@ test "runStep: distinct sources run on distinct threads in parallel" { var ns = null_store_mod.NullStore.init(allocator); var agent = Agent.init(allocator, io, &h.config, ns.store(), null); defer agent.deinit(); - agent.stream_fn = stub.install(); + agent.open_stream_fn = stub.install(); - const conv = &agent.conversation; - try conv.addUserMessage("go"); - var recv = NoopReceiver.make(); - try agent.runStep(&recv); + try drainTurn(&agent, "go"); const view_a = h.registry.lookup("src_a_t") orelse return error.NotFound; const view_b = h.registry.lookup("src_b_t") orelse return error.NotFound; @@ -2308,13 +2481,11 @@ test "runStep: source whole-batch error becomes per-call error results and conti var ns = null_store_mod.NullStore.init(allocator); var agent = Agent.init(allocator, io, &h.config, ns.store(), null); defer agent.deinit(); - agent.stream_fn = stub.install(); + agent.open_stream_fn = stub.install(); const conv = &agent.conversation; - try conv.addUserMessage("kaboom"); - var recv = NoopReceiver.make(); - try agent.runStep(&recv); + try drainTurn(&agent, "kaboom"); // user, assistant(tool_use x2), user(tool_result x2), assistant(text) try testing.expectEqual(@as(usize, 4), conv.messages.items.len); @@ -2353,13 +2524,11 @@ test "runStep: mixed single Tools and source-backed tools coexist in one turn" { var ns = null_store_mod.NullStore.init(allocator); var agent = Agent.init(allocator, io, &h.config, ns.store(), null); defer agent.deinit(); - agent.stream_fn = stub.install(); + agent.open_stream_fn = stub.install(); const conv = &agent.conversation; - try conv.addUserMessage("go"); - var recv = NoopReceiver.make(); - try agent.runStep(&recv); + try drainTurn(&agent, "go"); const tr_msg = conv.messages.items[2]; try testing.expectEqual(@as(usize, 3), tr_msg.content.items.len); @@ -2406,7 +2575,7 @@ test "setConfig swaps the visible tool set between turns" { var ns = null_store_mod.NullStore.init(allocator); var agent = Agent.init(allocator, io, &cfg_a, ns.store(), null); defer agent.deinit(); - agent.stream_fn = stub.install(); + agent.open_stream_fn = stub.install(); // Under A: `echo` visible, `late` not. try testing.expect(agent.config.registry.lookup("echo") != null); @@ -2420,9 +2589,7 @@ test "setConfig swaps the visible tool set between turns" { // A real turn under B resolves `late` (which would have been // UnknownTool under A), then loops to the final text turn. const conv = &agent.conversation; - try conv.addUserMessage("go"); - var recv = NoopReceiver.make(); - try agent.runStep(&recv); + try drainTurn(&agent, "go"); const tr = conv.messages.items[2].content.items[0].ToolResult; try testing.expectEqualStrings("2", tr.tool_use_id); @@ -2450,7 +2617,7 @@ test "compact: summarizes prefix, keeps suffix, system survives" { var ns = null_store_mod.NullStore.init(allocator); var agent = Agent.init(allocator, io, &h.config, ns.store(), null); defer agent.deinit(); - agent.stream_fn = stub.install(); + agent.open_stream_fn = stub.install(); const conv = &agent.conversation; try conv.addSystemMessage("you are helpful"); @@ -2512,7 +2679,7 @@ test "compact: restated suffix usage reconstructs a fresh cumulative chain" { var ns = null_store_mod.NullStore.init(allocator); var agent = Agent.init(allocator, io, &h.config, ns.store(), null); defer agent.deinit(); - agent.stream_fn = stub.install(); + agent.open_stream_fn = stub.install(); const conv = &agent.conversation; // Prefix turn (will be summarized). Cumulative footprint = 500+40+10+50 @@ -2585,7 +2752,7 @@ test "compact: no-op when conversation already fits the budget" { var ns = null_store_mod.NullStore.init(allocator); var agent = Agent.init(allocator, io, &h.config, ns.store(), null); defer agent.deinit(); - agent.stream_fn = stub.install(); + agent.open_stream_fn = stub.install(); const conv = &agent.conversation; try conv.addSystemMessage("sys"); @@ -2622,7 +2789,7 @@ test "compact: extra instructions are appended to the system prompt" { var ns = null_store_mod.NullStore.init(allocator); var agent = Agent.init(allocator, io, &h.config, ns.store(), null); defer agent.deinit(); - agent.stream_fn = stub.install(); + agent.open_stream_fn = stub.install(); const conv = &agent.conversation; try conv.addUserMessage("question one two three"); @@ -2662,7 +2829,7 @@ test "runStep: auto-compacts on context overflow and retries once" { var ns = null_store_mod.NullStore.init(allocator); var agent = Agent.init(allocator, io, &h.config, ns.store(), null); defer agent.deinit(); - agent.stream_fn = stub.install(); + agent.open_stream_fn = stub.install(); agent.compaction_system_prompt = "Summarize the conversation."; const conv = &agent.conversation; @@ -2671,10 +2838,8 @@ test "runStep: auto-compacts on context overflow and retries once" { try conv.addAssistantMessage(&.{ .{ .Text = try conversation.textualBlockFromSlice(allocator, "first answer with several words") }, }); - try conv.addUserMessage("second recent question"); - var recv = NoopReceiver.make(); - try agent.runStep(&recv); + try drainTurn(&agent, "second recent question"); try testing.expect(agent.auto_compacted); // After compaction + retry: [system, summary, user q2, assistant final]. @@ -2708,55 +2873,39 @@ test "runStep: context overflow without compaction prompt propagates" { var ns = null_store_mod.NullStore.init(allocator); var agent = Agent.init(allocator, io, &h.config, ns.store(), null); defer agent.deinit(); - agent.stream_fn = stub.install(); + agent.open_stream_fn = stub.install(); // No compaction_system_prompt set -> overflow propagates. - const conv = &agent.conversation; - try conv.addUserMessage("hi"); - var recv = NoopReceiver.make(); - try testing.expectError(error.ContextOverflow, agent.runStep(&recv)); + try testing.expectError(error.ContextOverflow, drainTurn(&agent, "hi")); } // ----------------------------------------------------------------------------- // Phase 6: provider retry + tool-error holistic tests // ----------------------------------------------------------------------------- -/// Receiver that records `onProviderRetry` notifications (and nothing else), -/// so tests can assert retry scheduling without a live provider. -const RetryRecordingReceiver = struct { +/// Records the `provider_retry` events a turn emits, so tests can assert +/// retry scheduling without a live provider. +const RetryRecorder = struct { infos: std.ArrayList(provider_mod.ProviderRetryInfo) = .empty, allocator: Allocator, - fn make(self: *RetryRecordingReceiver) provider_mod.Receiver { - return .{ .ptr = self, .vtable = &vt }; - } - fn deinit(self: *RetryRecordingReceiver) void { + fn deinit(self: *RetryRecorder) void { self.infos.deinit(self.allocator); } - const vt: provider_mod.ReceiverVTable = .{ - .onMessageStart = onMessageStart, - .onBlockStart = onBlockStart, - .onToolDetails = onToolDetails, - .onContentDelta = onContentDelta, - .onBlockComplete = onBlockComplete, - .onMessageComplete = onMessageComplete, - .onError = onError, - .onProviderRetry = onProviderRetry, - }; - fn onMessageStart(_: *anyopaque, _: conversation.MessageRole) anyerror!void {} - fn onBlockStart(_: *anyopaque, _: provider_mod.ContentBlockType, _: usize) anyerror!void {} - fn onToolDetails(_: *anyopaque, _: usize, _: []const u8, _: []const u8) anyerror!void {} - fn onContentDelta(_: *anyopaque, _: usize, _: []const u8) anyerror!void {} - fn onBlockComplete(_: *anyopaque, _: usize, _: conversation.ContentBlock) anyerror!void {} - fn onMessageComplete(_: *anyopaque, _: conversation.Message, _: ?provider_mod.Usage) anyerror!void {} - fn onError(_: *anyopaque, _: anyerror) void {} - fn onProviderRetry(ptr: *anyopaque, info: provider_mod.ProviderRetryInfo) void { - const self: *RetryRecordingReceiver = @ptrCast(@alignCast(ptr)); - self.infos.append(self.allocator, info) catch {}; - } }; +/// Drive a whole turn via the pull `Stream`, recording every +/// `provider_retry` event into `rr` and discarding the rest. Returns the +/// same terminal error the turn would raise. +fn drainTurnRecording(agent: *Agent, text: []const u8, rr: *RetryRecorder) !void { + var s = try agent.run(.{ .text = text }); + defer s.deinit(); + while (try s.next()) |ev| { + if (ev == .provider_retry) try rr.infos.append(rr.allocator, ev.provider_retry); + } +} + /// Build an agent + harness with near-zero backoff so retry tests don't /// actually sleep. Caller owns the harness and must keep it alive. fn fastRetryHarness(h: *TestHarness) void { @@ -2795,15 +2944,13 @@ test "runStep: provider 429 retries then succeeds without duplicate messages" { var ns = null_store_mod.NullStore.init(allocator); var agent = Agent.init(allocator, io, &h.config, ns.store(), null); defer agent.deinit(); - agent.stream_fn = stub.install(); + agent.open_stream_fn = stub.install(); const conv = &agent.conversation; - try conv.addUserMessage("hi"); - var rr = RetryRecordingReceiver{ .allocator = allocator }; + var rr = RetryRecorder{ .allocator = allocator }; defer rr.deinit(); - var recv = rr.make(); - try agent.runStep(&recv); + try drainTurnRecording(&agent, "hi", &rr); // Two failures + one success. try testing.expectEqual(@as(usize, 3), stub.calls_made); @@ -2840,15 +2987,12 @@ test "runStep: provider 500 retries with backoff notification" { var ns = null_store_mod.NullStore.init(allocator); var agent = Agent.init(allocator, io, &h.config, ns.store(), null); defer agent.deinit(); - agent.stream_fn = stub.install(); + agent.open_stream_fn = stub.install(); - const conv = &agent.conversation; - try conv.addUserMessage("hi"); - var rr = RetryRecordingReceiver{ .allocator = allocator }; + var rr = RetryRecorder{ .allocator = allocator }; defer rr.deinit(); - var recv = rr.make(); - try agent.runStep(&recv); + try drainTurnRecording(&agent, "hi", &rr); try testing.expectEqual(@as(usize, 2), stub.calls_made); try testing.expectEqual(@as(usize, 1), rr.infos.items.len); @@ -2880,15 +3024,12 @@ test "runStep: provider auth failure does not retry" { var ns = null_store_mod.NullStore.init(allocator); var agent = Agent.init(allocator, io, &h.config, ns.store(), null); defer agent.deinit(); - agent.stream_fn = stub.install(); + agent.open_stream_fn = stub.install(); - const conv = &agent.conversation; - try conv.addUserMessage("hi"); - var rr = RetryRecordingReceiver{ .allocator = allocator }; + var rr = RetryRecorder{ .allocator = allocator }; defer rr.deinit(); - var recv = rr.make(); - try testing.expectError(error.ProviderAuthFailed, agent.runStep(&recv)); + try testing.expectError(error.ProviderAuthFailed, drainTurnRecording(&agent, "hi", &rr)); // Exactly one attempt, no retry notification. try testing.expectEqual(@as(usize, 1), stub.calls_made); @@ -2921,15 +3062,12 @@ test "runStep: retries exhaust and hard-fail after max_attempts" { var ns = null_store_mod.NullStore.init(allocator); var agent = Agent.init(allocator, io, &h.config, ns.store(), null); defer agent.deinit(); - agent.stream_fn = stub.install(); + agent.open_stream_fn = stub.install(); - const conv = &agent.conversation; - try conv.addUserMessage("hi"); - var rr = RetryRecordingReceiver{ .allocator = allocator }; + var rr = RetryRecorder{ .allocator = allocator }; defer rr.deinit(); - var recv = rr.make(); - try testing.expectError(error.ProviderUnavailable, agent.runStep(&recv)); + try testing.expectError(error.ProviderUnavailable, drainTurnRecording(&agent, "hi", &rr)); // 4 attempts total (max_attempts), 3 retry notifications. try testing.expectEqual(@as(usize, 4), stub.calls_made); @@ -2961,15 +3099,12 @@ test "runStep: Retry-After is honored and reported" { var ns = null_store_mod.NullStore.init(allocator); var agent = Agent.init(allocator, io, &h.config, ns.store(), null); defer agent.deinit(); - agent.stream_fn = stub.install(); + agent.open_stream_fn = stub.install(); - const conv = &agent.conversation; - try conv.addUserMessage("hi"); - var rr = RetryRecordingReceiver{ .allocator = allocator }; + var rr = RetryRecorder{ .allocator = allocator }; defer rr.deinit(); - var recv = rr.make(); - try agent.runStep(&recv); + try drainTurnRecording(&agent, "hi", &rr); try testing.expectEqual(@as(usize, 1), rr.infos.items.len); // Reported Retry-After is the raw provider value... @@ -2998,13 +3133,11 @@ test "runStep: cancellation from a tool still hard-fails" { var ns = null_store_mod.NullStore.init(allocator); var agent = Agent.init(allocator, io, &h.config, ns.store(), null); defer agent.deinit(); - agent.stream_fn = stub.install(); + agent.open_stream_fn = stub.install(); const conv = &agent.conversation; - try conv.addUserMessage("go"); - var recv = NoopReceiver.make(); - try testing.expectError(error.Canceled, agent.runStep(&recv)); + try testing.expectError(error.Canceled, drainTurn(&agent, "go")); // Turn aborts: no tool result appended (user + assistant only). try testing.expectEqual(@as(usize, 2), conv.messages.items.len); } @@ -3030,13 +3163,11 @@ test "runStep: source per-call error produces a per-call error result and contin var ns = null_store_mod.NullStore.init(allocator); var agent = Agent.init(allocator, io, &h.config, ns.store(), null); defer agent.deinit(); - agent.stream_fn = stub.install(); + agent.open_stream_fn = stub.install(); const conv = &agent.conversation; - try conv.addUserMessage("go"); - var recv = NoopReceiver.make(); - try agent.runStep(&recv); + try drainTurn(&agent, "go"); try testing.expectEqual(@as(usize, 4), conv.messages.items.len); const tr_msg = conv.messages.items[2]; @@ -3070,7 +3201,7 @@ test "runStep: context-overflow compaction fires a compaction retry notification var ns = null_store_mod.NullStore.init(allocator); var agent = Agent.init(allocator, io, &h.config, ns.store(), null); defer agent.deinit(); - agent.stream_fn = stub.install(); + agent.open_stream_fn = stub.install(); agent.compaction_system_prompt = "Summarize the conversation."; const conv = &agent.conversation; @@ -3079,12 +3210,10 @@ test "runStep: context-overflow compaction fires a compaction retry notification try conv.addAssistantMessage(&.{ .{ .Text = try conversation.textualBlockFromSlice(allocator, "first answer with several words") }, }); - try conv.addUserMessage("second recent question"); - var rr = RetryRecordingReceiver{ .allocator = allocator }; + var rr = RetryRecorder{ .allocator = allocator }; defer rr.deinit(); - var recv = rr.make(); - try agent.runStep(&recv); + try drainTurnRecording(&agent, "second recent question", &rr); try testing.expect(agent.auto_compacted); // Exactly one notification, flagged as a compaction retry with no delay. 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; diff --git a/libpanto/src/provider_anthropic_messages.zig b/libpanto/src/provider_anthropic_messages.zig index 55efb5f..4bd146a 100644 --- a/libpanto/src/provider_anthropic_messages.zig +++ b/libpanto/src/provider_anthropic_messages.zig @@ -20,11 +20,15 @@ const Uri = std.Uri; const conversation = @import("conversation.zig"); const provider_mod = @import("provider.zig"); +const stream_mod = @import("stream.zig"); const sse_mod = @import("sse.zig"); const json_mod = @import("anthropic_messages_json.zig"); const config_mod = @import("config.zig"); const tool_registry_mod = @import("tool_registry.zig"); +const Event = stream_mod.Event; +const EventQueue = stream_mod.EventQueue; + /// A single Anthropic Messages streaming request. Transient: constructed /// per `streamStep`, holds only borrowed state (allocator, io, the global /// HTTP client, and the active config). Carries nothing across requests. @@ -36,26 +40,29 @@ pub const AnthropicMessagesRequest = struct { /// Optional diagnostic side-channel; see `OpenAIChatRequest.diag`. diag: ?*provider_mod.ProviderDiagnostic = null, - pub fn streamStep( + /// Open the streaming HTTP request and return a heap-allocated resumable + /// response. Reads response headers (classifying any >=400 status) but + /// does not pump the body — that happens lazily in + /// `ResumableResponse.produce`. On success the caller owns the returned + /// `*ResumableResponse` and must `deinit` it. + pub fn open( self: *AnthropicMessagesRequest, conv: *conversation.Conversation, tools: *const provider_mod.ToolRegistry, - receiver: *provider_mod.Receiver, - ) !void { - // Outer wrapper guarantees `onError` is called exactly once if - // anything fails. - self.streamStepInner(conv, tools, receiver) catch |err| { - receiver.onError(err); - return err; + ) !*ResumableResponse { + const rr = try self.allocator.create(ResumableResponse); + errdefer self.allocator.destroy(rr); + rr.* = .{ + .allocator = self.allocator, + .conv = conv, + .parser = sse_mod.SSEParser.init(self.allocator), + .state = .init(self.allocator), }; - } + errdefer { + rr.parser.deinit(); + rr.state.deinit(); + } - fn streamStepInner( - self: *AnthropicMessagesRequest, - conv: *conversation.Conversation, - tools: *const provider_mod.ToolRegistry, - receiver: *provider_mod.Receiver, - ) !void { const url = try std.fmt.allocPrint( self.allocator, "{s}/v1/messages", @@ -75,7 +82,7 @@ pub const AnthropicMessagesRequest = struct { .{ .name = "anthropic-version", .value = self.config.api_version }, }; - var req = try self.http_client.request(.POST, uri, .{ + rr.req = try self.http_client.request(.POST, uri, .{ .extra_headers = &extra_headers, // Disable compression: gzip buffers small SSE frames, defeating // the streaming property we paid for `stream: true` to get. @@ -83,22 +90,25 @@ pub const AnthropicMessagesRequest = struct { .keep_alive = false, .redirect_behavior = .not_allowed, }); - defer req.deinit(); + rr.req_open = true; + errdefer { + rr.req.deinit(); + rr.req_open = false; + } - req.transfer_encoding = .{ .content_length = body.len }; + rr.req.transfer_encoding = .{ .content_length = body.len }; var send_buf: [4096]u8 = undefined; - var bw = try req.sendBodyUnflushed(&send_buf); + var bw = try rr.req.sendBodyUnflushed(&send_buf); try bw.writer.writeAll(body); try bw.end(); - try req.connection.?.flush(); + try rr.req.connection.?.flush(); var redirect_buf: [1024]u8 = undefined; - var response = try req.receiveHead(&redirect_buf); + rr.response = try rr.req.receiveHead(&redirect_buf); - if (@intFromEnum(response.head.status) >= 400) { - var transfer_buf: [4096]u8 = undefined; - const body_reader = response.reader(&transfer_buf); + if (@intFromEnum(rr.response.head.status) >= 400) { + const body_reader = rr.response.reader(&rr.transfer_buf); var err_buf: std.ArrayList(u8) = .empty; defer err_buf.deinit(self.allocator); var tmp: [1024]u8 = undefined; @@ -108,7 +118,7 @@ pub const AnthropicMessagesRequest = struct { try err_buf.appendSlice(self.allocator, tmp[0..n]); if (err_buf.items.len > 16 * 1024) break; } - const status: u16 = @intFromEnum(response.head.status); + const status: u16 = @intFromEnum(rr.response.head.status); std.log.err("anthropic_messages HTTP {d}: {s}", .{ status, err_buf.items }); // Anthropic rejects oversized requests with HTTP 400 and a // "prompt is too long" message; `classifyHttpStatus` maps that to @@ -117,52 +127,104 @@ pub const AnthropicMessagesRequest = struct { const classified = provider_mod.classifyHttpStatus(status, err_buf.items); if (self.diag) |d| { d.status_code = status; - d.retry_after_ms = provider_mod.retryAfterFromHead(response.head); + d.retry_after_ms = provider_mod.retryAfterFromHead(rr.response.head); } return classified; } - var transfer_buf: [4096]u8 = undefined; - const body_reader = response.reader(&transfer_buf); + rr.body_reader = rr.response.reader(&rr.transfer_buf); + return rr; + } +}; + +/// A resumable Anthropic Messages streaming response. Owns the pinned HTTP +/// request/response, the body reader's transfer buffer, the `SSEParser`, and +/// the block-assembly `StreamState`. Must be heap-allocated and never moved: +/// `body_reader` borrows `&self.response`. +pub const ResumableResponse = struct { + allocator: Allocator, + conv: *conversation.Conversation, + parser: sse_mod.SSEParser, + state: StreamState, - var parser = sse_mod.SSEParser.init(self.allocator); - defer parser.deinit(); + req: http.Client.Request = undefined, + response: http.Client.Response = undefined, + transfer_buf: [4096]u8 = undefined, + body_reader: *std.Io.Reader = undefined, + chunk: [4096]u8 = undefined, - var state: StreamState = .init(self.allocator); - defer state.deinit(); + req_open: bool = false, + done: bool = false, - // Use `readVec` so we return to the event loop as soon as *any* - // bytes arrive, rather than waiting for the buffer to fill. - // `readSliceShort` blocks until EOF or full, which defeats streaming. - // - // Per `std.Io.Reader.readVec` docs: `n == 0` does NOT mean EOF; - // EOF is signalled only via `error.EndOfStream`. Breaking on - // `n == 0` truncates the response mid-stream. - var chunk: [4096]u8 = undefined; - var vecs: [1][]u8 = .{&chunk}; - while (true) { - const n = body_reader.readVec(&vecs) catch |err| switch (err) { - error.EndOfStream => break, - // Transport failure before the message completed: retryable. - else => return error.ProviderStreamMalformed, - }; - if (n == 0) continue; + pub const ProduceStatus = provider_mod.ProviderStream.ProduceStatus; - const events = try parser.feed(chunk[0..n]); - defer parser.freeEvents(events); + /// Wrap this response in the provider-agnostic `ProviderStream` the agent + /// loop drives. + pub fn providerStream(self: *ResumableResponse) provider_mod.ProviderStream { + return .{ .ptr = self, .vtable = &vtable }; + } - for (events) |ev_payload| { - std.log.debug("anthropic_messages <= {s}", .{ev_payload}); - try handleEvent(self.allocator, ev_payload, &state, receiver); - if (state.end_of_stream) { - try state.finalize(receiver, conv); - return; - } + const vtable: provider_mod.ProviderStream.VTable = .{ + .produce = produceVT, + .deinit = deinitVT, + }; + + fn produceVT(ptr: *anyopaque, out: *EventQueue) anyerror!ProduceStatus { + const self: *ResumableResponse = @ptrCast(@alignCast(ptr)); + return self.produce(out); + } + + fn deinitVT(ptr: *anyopaque) void { + const self: *ResumableResponse = @ptrCast(@alignCast(ptr)); + self.deinit(); + } + + pub fn deinit(self: *ResumableResponse) void { + if (self.req_open) self.req.deinit(); + self.parser.deinit(); + self.state.deinit(); + self.allocator.destroy(self); + } + + /// Pump the response: read one chunk, feed it through the SSE parser, and + /// decode each SSE event into zero or more `Event`s appended to `out`. + /// Returns `.more` if the caller should pump again, or + /// `.response_complete` once `message_stop` (or EOF) is reached and the + /// assistant message has been committed + a final `message_complete` + /// pushed. + pub fn produce(self: *ResumableResponse, out: *EventQueue) !ProduceStatus { + if (self.done) return .response_complete; + + var vecs: [1][]u8 = .{&self.chunk}; + const n = self.body_reader.readVec(&vecs) catch |err| switch (err) { + // Stream ended without an explicit message_stop. Finalize anyway. + error.EndOfStream => { + try self.finishStream(out); + return .response_complete; + }, + // Transport failure before the message completed: retryable. + else => return error.ProviderStreamMalformed, + }; + if (n == 0) return .more; + + const events = try self.parser.feed(self.chunk[0..n]); + defer self.parser.freeEvents(events); + + for (events) |ev_payload| { + std.log.debug("anthropic_messages <= {s}", .{ev_payload}); + try handleEvent(self.allocator, ev_payload, &self.state, out); + if (self.state.end_of_stream) { + try self.finishStream(out); + return .response_complete; } } + return .more; + } - // Stream ended without an explicit message_stop. Finalize anyway. - try state.finalize(receiver, conv); + fn finishStream(self: *ResumableResponse, out: *EventQueue) !void { + if (self.done) return; + self.done = true; + try self.state.finalize(out, self.conv); } }; @@ -224,10 +286,10 @@ const StreamState = struct { if (self.stop_reason) |s| self.allocator.free(s); } - fn ensureStarted(self: *StreamState, receiver: *provider_mod.Receiver) !void { + fn ensureStarted(self: *StreamState, out: *EventQueue) !void { if (self.started) return; self.started = true; - try receiver.onMessageStart(.assistant); + try out.push(.{ .message_start = .assistant }); } /// Merge a wire-level usage snapshot into the accumulated counts. @@ -248,7 +310,7 @@ const StreamState = struct { fn openBlock( self: *StreamState, - receiver: *provider_mod.Receiver, + out: *EventQueue, wire_index: usize, kind: BlockKind, tool_id: ?[]const u8, @@ -291,14 +353,20 @@ const StreamState = struct { .unsupported => null, }; if (block_type) |bt| { - try receiver.onBlockStart(bt, wire_index); + try out.push(.{ .block_start = .{ .block_type = bt, .index = wire_index } }); // Anthropic delivers tool id+name whole on content_block_start, - // so we can fire onToolDetails immediately — before any arg + // so we can fire tool_details immediately — before any arg // deltas. If the wire was malformed and either field is // missing, skip: closeBlock will drop the block defensively. + // id/name are owned by `ab` (stable) but we dupe into the queue + // arena for a uniform borrow lifetime. if (kind == .tool_use) { if (ab.tool_id != null and ab.tool_name != null) { - try receiver.onToolDetails(wire_index, ab.tool_id.?, ab.tool_name.?); + try out.push(.{ .tool_details = .{ + .index = wire_index, + .id = try out.dupeBytes(ab.tool_id.?), + .name = try out.dupeBytes(ab.tool_name.?), + } }); } } } @@ -306,26 +374,34 @@ const StreamState = struct { fn appendTextDelta( self: *StreamState, - receiver: *provider_mod.Receiver, + out: *EventQueue, delta: []const u8, ) !void { const a = &(self.active orelse return); if (a.kind == .unsupported) return; try a.text_buf.appendSlice(self.allocator, delta); - try receiver.onContentDelta(a.wire_index, delta); + // Dupe into the queue arena: `delta` borrows the transient SSE/JSON + // payload that `produce` frees before `next()` reads the queue. + try out.push(.{ .content_delta = .{ + .index = a.wire_index, + .delta = try out.dupeBytes(delta), + } }); } /// Append a chunk of the streamed JSON arguments for the active /// tool_use block. No-op if the active block isn't a tool_use. fn appendInputJsonDelta( self: *StreamState, - receiver: *provider_mod.Receiver, + out: *EventQueue, delta: []const u8, ) !void { const a = &(self.active orelse return); if (a.kind != .tool_use) return; try a.text_buf.appendSlice(self.allocator, delta); - try receiver.onContentDelta(a.wire_index, delta); + try out.push(.{ .content_delta = .{ + .index = a.wire_index, + .delta = try out.dupeBytes(delta), + } }); } fn setSignature(self: *StreamState, sig: []const u8) !void { @@ -339,10 +415,10 @@ const StreamState = struct { self.stop_reason = if (reason) |r| try self.allocator.dupe(u8, r) else null; } - /// Close the active block: append it to `blocks` and emit onBlockComplete. + /// Close the active block: append it to `blocks` and emit block_complete. fn closeBlock( self: *StreamState, - receiver: *provider_mod.Receiver, + out: *EventQueue, ) !void { var a = self.active orelse return; self.active = null; @@ -389,7 +465,10 @@ const StreamState = struct { }; try self.blocks.append(self.allocator, block); - try receiver.onBlockComplete(a.wire_index, self.blocks.items[self.blocks.items.len - 1]); + try out.push(.{ .block_complete = .{ + .index = a.wire_index, + .block = self.blocks.items[self.blocks.items.len - 1], + } }); } /// Drop the active block without emitting a completion callback. @@ -407,7 +486,7 @@ const StreamState = struct { fn finalize( self: *StreamState, - receiver: *provider_mod.Receiver, + out: *EventQueue, conv: *conversation.Conversation, ) !void { if (self.finalized) return; @@ -416,7 +495,7 @@ const StreamState = struct { if (self.active != null) { // Preserve an interrupted tool call so the agent can answer it // with a synthetic error ToolResult instead of invoking it. - try self.closeBlock(receiver); + try self.closeBlock(out); } const moved_blocks = try self.blocks.toOwnedSlice(self.allocator); @@ -426,7 +505,7 @@ const StreamState = struct { try conv.addAssistantMessageWithUsage(moved_blocks, usage); const msg = conv.messages.items[conv.messages.items.len - 1]; - try receiver.onMessageComplete(msg, usage); + try out.push(.{ .message_complete = .{ .message = msg, .usage = usage } }); } }; @@ -434,35 +513,35 @@ fn handleEvent( allocator: Allocator, payload: []const u8, state: *StreamState, - receiver: *provider_mod.Receiver, + out: *EventQueue, ) !void { var parsed = try json_mod.parseStreamEvent(allocator, payload); defer parsed.deinit(); switch (parsed.event) { .message_start => |s| { - try state.ensureStarted(receiver); + try state.ensureStarted(out); state.mergeUsage(s.usage); }, .content_block_start => |s| { - try state.ensureStarted(receiver); + try state.ensureStarted(out); const kind: StreamState.BlockKind = switch (s.kind) { .text => .text, .thinking => .thinking, .tool_use => .tool_use, .unknown => .unsupported, }; - try state.openBlock(receiver, s.index, kind, s.tool_id, s.tool_name); + try state.openBlock(out, s.index, kind, s.tool_id, s.tool_name); }, .content_block_delta => |d| { - if (d.text_delta) |t| try state.appendTextDelta(receiver, t); - if (d.thinking_delta) |t| try state.appendTextDelta(receiver, t); + if (d.text_delta) |t| try state.appendTextDelta(out, t); + if (d.thinking_delta) |t| try state.appendTextDelta(out, t); if (d.signature_delta) |sig| try state.setSignature(sig); - if (d.input_json_delta) |j| try state.appendInputJsonDelta(receiver, j); + if (d.input_json_delta) |j| try state.appendInputJsonDelta(out, j); }, .content_block_stop => |s| { if (state.active) |a| { - if (a.wire_index == s.index) try state.closeBlock(receiver); + if (a.wire_index == s.index) try state.closeBlock(out); } }, .message_delta => |d| { @@ -494,12 +573,16 @@ fn handleEvent( const testing = std.testing; -/// Recording Receiver that captures the callback sequence for assertions. -const RecordingReceiver = struct { +/// Records the decoded pull `Event`s for assertions. Keeps the same typed +/// schema the old RecordingReceiver exposed (message_start / block_start / +/// delta / block_complete / message_complete), so the existing assertions +/// are preserved verbatim. The recorder owns copies of all byte payloads +/// (the queue arena is reset on full drain). +const EventRecorder = struct { allocator: Allocator, - events: std.ArrayList(Event) = .empty, + events: std.ArrayList(Rec) = .empty, - const Event = union(enum) { + const Rec = union(enum) { message_start: conversation.MessageRole, block_start: struct { kind: provider_mod.ContentBlockType, @@ -516,14 +599,13 @@ const RecordingReceiver = struct { signature: ?[]const u8 = null, // owned copy when present }, message_complete: ?provider_mod.Usage, - err: anyerror, }; - fn init(allocator: Allocator) RecordingReceiver { + fn init(allocator: Allocator) EventRecorder { return .{ .allocator = allocator }; } - fn deinit(self: *RecordingReceiver) void { + fn deinit(self: *EventRecorder) void { for (self.events.items) |ev| { switch (ev) { .delta => |d| self.allocator.free(d.bytes), @@ -537,102 +619,76 @@ const RecordingReceiver = struct { self.events.deinit(self.allocator); } - fn receiver(self: *RecordingReceiver) provider_mod.Receiver { - return .{ .ptr = self, .vtable = &vt }; - } - - const vt: provider_mod.ReceiverVTable = .{ - .onMessageStart = onMessageStart, - .onBlockStart = onBlockStart, - .onToolDetails = onToolDetails, - .onContentDelta = onContentDelta, - .onBlockComplete = onBlockComplete, - .onMessageComplete = onMessageComplete, - .onError = onError, - .onProviderRetry = onProviderRetry, - }; - - fn onMessageStart(ptr: *anyopaque, role: conversation.MessageRole) anyerror!void { - const self: *RecordingReceiver = @ptrCast(@alignCast(ptr)); - try self.events.append(self.allocator, .{ .message_start = role }); - } - fn onProviderRetry(_: *anyopaque, _: provider_mod.ProviderRetryInfo) void {} - fn onBlockStart( - ptr: *anyopaque, - bt: provider_mod.ContentBlockType, - idx: usize, - ) anyerror!void { - const self: *RecordingReceiver = @ptrCast(@alignCast(ptr)); - try self.events.append(self.allocator, .{ .block_start = .{ .kind = bt, .index = idx } }); - } - fn onToolDetails( - _: *anyopaque, - _: usize, - _: []const u8, - _: []const u8, - ) anyerror!void { - // Anthropic delivers identity at block_start time; the existing - // tests assert tool-use blocks via the ContentBlock in conv after - // finalize, not via the event stream. We accept and drop these - // here to keep the test recorder schema stable. - } - fn onContentDelta(ptr: *anyopaque, idx: usize, delta: []const u8) anyerror!void { - const self: *RecordingReceiver = @ptrCast(@alignCast(ptr)); - const copy = try self.allocator.dupe(u8, delta); - try self.events.append(self.allocator, .{ .delta = .{ .index = idx, .bytes = copy } }); - } - fn onBlockComplete( - ptr: *anyopaque, - idx: usize, - block: conversation.ContentBlock, - ) anyerror!void { - const self: *RecordingReceiver = @ptrCast(@alignCast(ptr)); - switch (block) { - .Text => |tb| { - const txt = try self.allocator.dupe(u8, tb.items); - try self.events.append(self.allocator, .{ .block_complete = .{ - .kind = .Text, - .index = idx, - .text = txt, - } }); + /// Translate one pull `Event` into the recorder's schema. Tool identity + /// (`tool_details`) is dropped: the anthropic tests assert tool-use + /// blocks via the ContentBlock in conv after finalize, not via the + /// event stream. Tool-arg `content_delta`s are also dropped here because + /// the old recorder only recorded text/thinking deltas (it routed tool + /// args through a separate path that didn't call onContentDelta in a way + /// these tests observe) — we preserve that by only recording deltas for + /// the currently text/thinking block. Since the recorder can't see block + /// kind from a bare delta, we record every delta; the existing tests + /// only assert delta bytes for text/thinking turns, so this is + /// equivalent for them. + fn record(self: *EventRecorder, ev: Event) !void { + switch (ev) { + .message_start => |role| try self.events.append(self.allocator, .{ .message_start = role }), + .block_start => |b| try self.events.append(self.allocator, .{ .block_start = .{ + .kind = b.block_type, + .index = b.index, + } }), + .content_delta => |d| { + const copy = try self.allocator.dupe(u8, d.delta); + try self.events.append(self.allocator, .{ .delta = .{ .index = d.index, .bytes = copy } }); }, - .Thinking => |tb| { - const txt = try self.allocator.dupe(u8, tb.text.items); - const sig = if (tb.signature) |s| try self.allocator.dupe(u8, s) else null; - try self.events.append(self.allocator, .{ .block_complete = .{ - .kind = .Thinking, - .index = idx, - .text = txt, - .signature = sig, - } }); + .block_complete => |bc| switch (bc.block) { + .Text => |tb| { + const txt = try self.allocator.dupe(u8, tb.items); + try self.events.append(self.allocator, .{ .block_complete = .{ + .kind = .Text, + .index = bc.index, + .text = txt, + } }); + }, + .Thinking => |tb| { + const txt = try self.allocator.dupe(u8, tb.text.items); + const sig = if (tb.signature) |s| try self.allocator.dupe(u8, s) else null; + try self.events.append(self.allocator, .{ .block_complete = .{ + .kind = .Thinking, + .index = bc.index, + .text = txt, + .signature = sig, + } }); + }, + else => {}, }, + .message_complete => |m| try self.events.append(self.allocator, .{ .message_complete = m.usage }), else => {}, } } - fn onMessageComplete(ptr: *anyopaque, _: conversation.Message, usage: ?provider_mod.Usage) anyerror!void { - const self: *RecordingReceiver = @ptrCast(@alignCast(ptr)); - try self.events.append(self.allocator, .{ .message_complete = usage }); - } - fn onError(ptr: *anyopaque, err: anyerror) void { - const self: *RecordingReceiver = @ptrCast(@alignCast(ptr)); - self.events.append(self.allocator, .{ .err = err }) catch {}; - } }; fn runStreamedTurn( allocator: Allocator, conv: *conversation.Conversation, - receiver: *provider_mod.Receiver, + rec: ?*EventRecorder, events: []const []const u8, ) !void { var state: StreamState = .init(allocator); defer state.deinit(); + var queue = EventQueue.init(allocator); + defer queue.deinit(); + for (events) |payload| { - try handleEvent(allocator, payload, &state, receiver); + try handleEvent(allocator, payload, &state, &queue); if (state.end_of_stream) break; } - try state.finalize(receiver, conv); + try state.finalize(&queue, conv); + + while (queue.pop()) |ev| { + if (rec) |r| try r.record(ev); + } } test "streams a text-only turn end-to-end" { @@ -642,9 +698,8 @@ test "streams a text-only turn end-to-end" { defer conv.deinit(); try conv.addUserMessage("hello"); - var rec = RecordingReceiver.init(allocator); + var rec = EventRecorder.init(allocator); defer rec.deinit(); - var recv = rec.receiver(); const events = [_][]const u8{ \\{"type":"message_start","message":{"id":"msg_1","type":"message","role":"assistant","content":[],"model":"claude"}} @@ -663,7 +718,7 @@ test "streams a text-only turn end-to-end" { , }; - try runStreamedTurn(allocator, &conv, &recv, &events); + try runStreamedTurn(allocator, &conv, &rec, &events); // Conversation now holds the assistant reply. try testing.expectEqual(@as(usize, 2), conv.messages.items.len); @@ -693,9 +748,8 @@ test "anthropic: captures usage from message_start and message_delta on message_ defer conv.deinit(); try conv.addUserMessage("hi"); - var rec = RecordingReceiver.init(allocator); + var rec = EventRecorder.init(allocator); defer rec.deinit(); - var recv = rec.receiver(); const events = [_][]const u8{ // Initial input-side counts on message_start. @@ -714,7 +768,7 @@ test "anthropic: captures usage from message_start and message_delta on message_ , }; - try runStreamedTurn(allocator, &conv, &recv, &events); + try runStreamedTurn(allocator, &conv, &rec, &events); // Find the message_complete event and check its usage payload. var found: ?provider_mod.Usage = null; @@ -737,9 +791,8 @@ test "anthropic: message_complete carries null usage when wire omits it" { defer conv.deinit(); try conv.addUserMessage("hi"); - var rec = RecordingReceiver.init(allocator); + var rec = EventRecorder.init(allocator); defer rec.deinit(); - var recv = rec.receiver(); const events = [_][]const u8{ \\{"type":"message_start","message":{"id":"m","role":"assistant","content":[],"model":"claude"}} @@ -756,7 +809,7 @@ test "anthropic: message_complete carries null usage when wire omits it" { , }; - try runStreamedTurn(allocator, &conv, &recv, &events); + try runStreamedTurn(allocator, &conv, &rec, &events); var saw_complete = false; for (rec.events.items) |ev| { @@ -775,9 +828,8 @@ test "captures thinking signature for round-trip" { defer conv.deinit(); try conv.addUserMessage("solve"); - var rec = RecordingReceiver.init(allocator); + var rec = EventRecorder.init(allocator); defer rec.deinit(); - var recv = rec.receiver(); const events = [_][]const u8{ \\{"type":"message_start","message":{"role":"assistant"}} @@ -802,7 +854,7 @@ test "captures thinking signature for round-trip" { , }; - try runStreamedTurn(allocator, &conv, &recv, &events); + try runStreamedTurn(allocator, &conv, &rec, &events); // The assistant message has Thinking + Text, with signature on the Thinking. const asst = conv.messages.items[1]; @@ -822,9 +874,8 @@ test "signature-only thinking block (display omitted)" { defer conv.deinit(); try conv.addUserMessage("hi"); - var rec = RecordingReceiver.init(allocator); + var rec = EventRecorder.init(allocator); defer rec.deinit(); - var recv = rec.receiver(); const events = [_][]const u8{ \\{"type":"message_start","message":{"role":"assistant"}} @@ -845,7 +896,7 @@ test "signature-only thinking block (display omitted)" { , }; - try runStreamedTurn(allocator, &conv, &recv, &events); + try runStreamedTurn(allocator, &conv, &rec, &events); const asst = conv.messages.items[1]; try testing.expectEqual(@as(usize, 2), asst.content.items.len); @@ -860,9 +911,8 @@ test "ping and unknown events are ignored" { defer conv.deinit(); try conv.addUserMessage("hi"); - var rec = RecordingReceiver.init(allocator); + var rec = EventRecorder.init(allocator); defer rec.deinit(); - var recv = rec.receiver(); const events = [_][]const u8{ \\{"type":"message_start","message":{"role":"assistant"}} @@ -883,7 +933,7 @@ test "ping and unknown events are ignored" { , }; - try runStreamedTurn(allocator, &conv, &recv, &events); + try runStreamedTurn(allocator, &conv, &rec, &events); try testing.expectEqualStrings( "ok", @@ -898,9 +948,8 @@ test "tool_use blocks are captured with id, name, and assembled input" { defer conv.deinit(); try conv.addUserMessage("use a tool"); - var rec = RecordingReceiver.init(allocator); + var rec = EventRecorder.init(allocator); defer rec.deinit(); - var recv = rec.receiver(); const events = [_][]const u8{ \\{"type":"message_start","message":{"role":"assistant"}} @@ -923,7 +972,7 @@ test "tool_use blocks are captured with id, name, and assembled input" { , }; - try runStreamedTurn(allocator, &conv, &recv, &events); + try runStreamedTurn(allocator, &conv, &rec, &events); const asst = conv.messages.items[1]; try testing.expectEqual(@as(usize, 2), asst.content.items.len); @@ -946,9 +995,8 @@ test "inbound wire tool name is decoded to dotted form" { defer conv.deinit(); try conv.addUserMessage("use a tool"); - var rec = RecordingReceiver.init(allocator); + var rec = EventRecorder.init(allocator); defer rec.deinit(); - var recv = rec.receiver(); const events = [_][]const u8{ \\{"type":"message_start","message":{"role":"assistant"}} @@ -961,7 +1009,7 @@ test "inbound wire tool name is decoded to dotted form" { , }; - try runStreamedTurn(allocator, &conv, &recv, &events); + try runStreamedTurn(allocator, &conv, &rec, &events); const tu = conv.messages.items[1].content.items[0].ToolUse; try testing.expectEqualStrings("calc.sum", tu.name); @@ -974,9 +1022,8 @@ test "error event propagates as Zig error" { defer conv.deinit(); try conv.addUserMessage("hi"); - var rec = RecordingReceiver.init(allocator); - defer rec.deinit(); - var recv = rec.receiver(); + var queue = EventQueue.init(allocator); + defer queue.deinit(); var state: StreamState = .init(allocator); defer state.deinit(); @@ -986,7 +1033,7 @@ test "error event propagates as Zig error" { \\{"type":"message_start","message":{"role":"assistant"}} , &state, - &recv, + &queue, ); const result = handleEvent( @@ -994,7 +1041,7 @@ test "error event propagates as Zig error" { \\{"type":"error","error":{"type":"overloaded_error","message":"too busy"}} , &state, - &recv, + &queue, ); try testing.expectError(error.ProviderStreamMalformed, result); } @@ -1008,9 +1055,8 @@ test "two streamed turns persist assistant replies in the conversation" { try conv.addSystemMessage("Be brief."); try conv.addUserMessage("hi"); - var rec = RecordingReceiver.init(allocator); + var rec = EventRecorder.init(allocator); defer rec.deinit(); - var recv = rec.receiver(); const turn1 = [_][]const u8{ \\{"type":"message_start","message":{"role":"assistant"}} @@ -1024,7 +1070,7 @@ test "two streamed turns persist assistant replies in the conversation" { \\{"type":"message_stop"} , }; - try runStreamedTurn(allocator, &conv, &recv, &turn1); + try runStreamedTurn(allocator, &conv, &rec, &turn1); try conv.addUserMessage("what did you say?"); @@ -1040,7 +1086,7 @@ test "two streamed turns persist assistant replies in the conversation" { \\{"type":"message_stop"} , }; - try runStreamedTurn(allocator, &conv, &recv, &turn2); + try runStreamedTurn(allocator, &conv, &rec, &turn2); // system + user + assistant + user + assistant = 5 try testing.expectEqual(@as(usize, 5), conv.messages.items.len); diff --git a/libpanto/src/provider_openai_chat.zig b/libpanto/src/provider_openai_chat.zig index ca01d7c..30bccb6 100644 --- a/libpanto/src/provider_openai_chat.zig +++ b/libpanto/src/provider_openai_chat.zig @@ -18,11 +18,15 @@ const Uri = std.Uri; const conversation = @import("conversation.zig"); const provider_mod = @import("provider.zig"); +const stream_mod = @import("stream.zig"); const sse_mod = @import("sse.zig"); const json_mod = @import("openai_chat_json.zig"); const config_mod = @import("config.zig"); const tool_registry_mod = @import("tool_registry.zig"); +const Event = stream_mod.Event; +const EventQueue = stream_mod.EventQueue; + /// Decode a wire tool name (`__` -> `.`) in place within an assembled /// name buffer. Decoding only ever shrinks the buffer (reads stay ahead /// of writes), so aliasing src/dst is safe; we then truncate to the @@ -52,27 +56,31 @@ pub const OpenAIChatRequest = struct { /// they live only as long as this request object. diag: ?*provider_mod.ProviderDiagnostic = null, - pub fn streamStep( + /// Open the streaming HTTP request and return a heap-allocated + /// resumable response. Performs the POST and reads response headers + /// (classifying any >=400 status into a provider error), but does NOT + /// pump the body — that happens lazily in `ResumableResponse.produce`. + /// + /// On success the caller owns the returned `*ResumableResponse` and must + /// `deinit` it. On failure nothing is allocated. + pub fn open( self: *OpenAIChatRequest, conv: *conversation.Conversation, tools: *const provider_mod.ToolRegistry, - receiver: *provider_mod.Receiver, - ) !void { - // Outer wrapper guarantees `onError` is called exactly once if - // anything fails — whether the receiver, the HTTP transport, or the - // SSE/JSON parsers. The inner `streamStepInner` does the real work. - self.streamStepInner(conv, tools, receiver) catch |err| { - receiver.onError(err); - return err; + ) !*ResumableResponse { + const rr = try self.allocator.create(ResumableResponse); + errdefer self.allocator.destroy(rr); + rr.* = .{ + .allocator = self.allocator, + .conv = conv, + .parser = sse_mod.SSEParser.init(self.allocator), + .state = .init(self.allocator), }; - } + errdefer { + rr.parser.deinit(); + rr.state.deinit(); + } - fn streamStepInner( - self: *OpenAIChatRequest, - conv: *conversation.Conversation, - tools: *const provider_mod.ToolRegistry, - receiver: *provider_mod.Receiver, - ) !void { // Build URL: "{base_url}/chat/completions" const url = try std.fmt.allocPrint( self.allocator, @@ -102,8 +110,10 @@ pub const OpenAIChatRequest = struct { }; // Open the request. We can't use `fetch()` because it buffers the - // response; we want to stream the body as it arrives. - var req = try self.http_client.request(.POST, uri, .{ + // response; we want to stream the body as it arrives. The request + // is moved into the heap struct so the body reader (which borrows + // `&rr.response`) stays valid across `produce` calls. + rr.req = try self.http_client.request(.POST, uri, .{ .extra_headers = &extra_headers, // Disable compression: gzip buffers small SSE frames, defeating // the streaming property we paid for `stream: true` to get. @@ -111,24 +121,27 @@ pub const OpenAIChatRequest = struct { .keep_alive = false, .redirect_behavior = .not_allowed, }); - defer req.deinit(); + rr.req_open = true; + errdefer { + rr.req.deinit(); + rr.req_open = false; + } - req.transfer_encoding = .{ .content_length = body.len }; + rr.req.transfer_encoding = .{ .content_length = body.len }; var send_buf: [4096]u8 = undefined; - var bw = try req.sendBodyUnflushed(&send_buf); + var bw = try rr.req.sendBodyUnflushed(&send_buf); try bw.writer.writeAll(body); try bw.end(); - try req.connection.?.flush(); + try rr.req.connection.?.flush(); // Receive response headers. var redirect_buf: [1024]u8 = undefined; - var response = try req.receiveHead(&redirect_buf); + rr.response = try rr.req.receiveHead(&redirect_buf); - if (@intFromEnum(response.head.status) >= 400) { + if (@intFromEnum(rr.response.head.status) >= 400) { // Drain body for diagnostics. - var transfer_buf: [4096]u8 = undefined; - const body_reader = response.reader(&transfer_buf); + const body_reader = rr.response.reader(&rr.transfer_buf); var err_buf: std.ArrayList(u8) = .empty; defer err_buf.deinit(self.allocator); var tmp: [1024]u8 = undefined; @@ -138,7 +151,7 @@ pub const OpenAIChatRequest = struct { try err_buf.appendSlice(self.allocator, tmp[0..n]); if (err_buf.items.len > 16 * 1024) break; } - const status: u16 = @intFromEnum(response.head.status); + const status: u16 = @intFromEnum(rr.response.head.status); std.log.err("openai_chat HTTP {d}: {s}", .{ status, err_buf.items }); // Classify the status into a retryable/terminal provider error. // HTTP 400 with a context marker becomes `ContextOverflow` so the @@ -146,64 +159,127 @@ pub const OpenAIChatRequest = struct { const classified = provider_mod.classifyHttpStatus(status, err_buf.items); if (self.diag) |d| { d.status_code = status; - d.retry_after_ms = provider_mod.retryAfterFromHead(response.head); + d.retry_after_ms = provider_mod.retryAfterFromHead(rr.response.head); } return classified; } - // Stream the body through the SSE parser and event handler. - var transfer_buf: [4096]u8 = undefined; - const body_reader = response.reader(&transfer_buf); + // Bind the streaming body reader. Valid for the lifetime of `rr` + // (it borrows `&rr.response` and `&rr.transfer_buf`, both pinned). + rr.body_reader = rr.response.reader(&rr.transfer_buf); + return rr; + } +}; - var parser = sse_mod.SSEParser.init(self.allocator); - defer parser.deinit(); +/// A resumable OpenAI Chat streaming response. Owns the pinned HTTP +/// request/response, the body reader's transfer buffer, the `SSEParser`, +/// and the block-assembly `StreamState`. `produce` pumps just enough bytes +/// to emit one or more events into the queue, or reports the response is +/// complete (its assistant message committed to the conversation). +/// +/// Must be heap-allocated and never moved: `body_reader` borrows +/// `&self.response`. +pub const ResumableResponse = struct { + allocator: Allocator, + conv: *conversation.Conversation, + parser: sse_mod.SSEParser, + state: StreamState, - var state: StreamState = .init(self.allocator); - defer state.deinit(); + req: http.Client.Request = undefined, + response: http.Client.Response = undefined, + /// Transfer buffer backing `body_reader`. Pinned in the heap struct. + transfer_buf: [4096]u8 = undefined, + /// The streaming body reader, bound in `open` after a 2xx response. + body_reader: *std.Io.Reader = undefined, + /// Chunk scratch for `readVec`. + chunk: [4096]u8 = undefined, - // Use `readVec` so we return to the event loop as soon as *any* - // bytes arrive, rather than waiting for the buffer to fill. - // `readSliceShort` blocks until EOF or full, which defeats streaming. - // - // Per `std.Io.Reader.readVec` docs: a return value of 0 does NOT - // signal end-of-stream — it just means no new bytes were available - // this call, and the caller should try again. EOF is reported only - // via `error.EndOfStream`. Breaking on `n == 0` truncates the - // response and silently cuts off mid-stream. - var chunk: [4096]u8 = undefined; - var vecs: [1][]u8 = .{&chunk}; - while (true) { - const n = body_reader.readVec(&vecs) catch |err| switch (err) { - error.EndOfStream => break, - // A transport read failure mid-stream (reset, TLS, timeout) - // before `[DONE]` means no assistant message was committed. - // Surface it as a retryable malformed-stream error. - else => return error.ProviderStreamMalformed, - }; - if (n == 0) continue; + /// True once `req` has been initialized (so `deinit` knows to free it). + req_open: bool = false, + /// Set once the response is fully decoded and finalized. + done: bool = false, - const events = try parser.feed(chunk[0..n]); - defer parser.freeEvents(events); + pub const ProduceStatus = provider_mod.ProviderStream.ProduceStatus; - for (events) |ev_payload| { - std.log.debug("openai_chat <= {s}", .{ev_payload}); - if (std.mem.eql(u8, ev_payload, "[DONE]")) { - try state.finalize(receiver, conv); - return; - } - try handleEvent(self.allocator, ev_payload, &state, receiver); - // Note: we do NOT bail when state.end_of_stream is set. - // OpenAI emits the terminating `usage` chunk *after* the - // chunk carrying finish_reason, then sends `[DONE]`. If - // we returned on finish_reason we'd never capture usage. - // `[DONE]` is the authoritative end-of-stream marker. + /// Wrap this response in the provider-agnostic `ProviderStream` the agent + /// loop drives. + pub fn providerStream(self: *ResumableResponse) provider_mod.ProviderStream { + return .{ .ptr = self, .vtable = &vtable }; + } + + const vtable: provider_mod.ProviderStream.VTable = .{ + .produce = produceVT, + .deinit = deinitVT, + }; + + fn produceVT(ptr: *anyopaque, out: *EventQueue) anyerror!ProduceStatus { + const self: *ResumableResponse = @ptrCast(@alignCast(ptr)); + return self.produce(out); + } + + fn deinitVT(ptr: *anyopaque) void { + const self: *ResumableResponse = @ptrCast(@alignCast(ptr)); + self.deinit(); + } + + pub fn deinit(self: *ResumableResponse) void { + if (self.req_open) self.req.deinit(); + self.parser.deinit(); + self.state.deinit(); + self.allocator.destroy(self); + } + + /// Pump the response: read one chunk, feed it through the SSE parser, + /// and decode each SSE event into zero or more `Event`s appended to + /// `out`. Returns `.more` if the caller should pump again, or + /// `.response_complete` once the terminal (`[DONE]` or EOF) has been + /// reached and the assistant message has been committed + a final + /// `message_complete` pushed. + /// + /// Reading and finalizing here means a single `produce` call may push + /// several events; the `Stream` drains the queue before pumping again. + pub fn produce(self: *ResumableResponse, out: *EventQueue) !ProduceStatus { + if (self.done) return .response_complete; + + var vecs: [1][]u8 = .{&self.chunk}; + const n = self.body_reader.readVec(&vecs) catch |err| switch (err) { + // Stream ended without [DONE]. Some servers and proxies omit it + // (or drop the trailing usage chunk). Finalize with whatever + // we've got — usage will be null in that case, which is fine. + error.EndOfStream => { + try self.finishStream(out); + return .response_complete; + }, + // A transport read failure mid-stream (reset, TLS, timeout) + // before `[DONE]` means no assistant message was committed. + // Surface it as a retryable malformed-stream error. + else => return error.ProviderStreamMalformed, + }; + if (n == 0) return .more; + + const events = try self.parser.feed(self.chunk[0..n]); + defer self.parser.freeEvents(events); + + for (events) |ev_payload| { + std.log.debug("openai_chat <= {s}", .{ev_payload}); + if (std.mem.eql(u8, ev_payload, "[DONE]")) { + try self.finishStream(out); + return .response_complete; } + try handleEvent(self.allocator, ev_payload, &self.state, out); + // Note: we do NOT bail when state.end_of_stream is set. + // OpenAI emits the terminating `usage` chunk *after* the + // chunk carrying finish_reason, then sends `[DONE]`. If + // we returned on finish_reason we'd never capture usage. + // `[DONE]` is the authoritative end-of-stream marker. } + return .more; + } - // Stream ended without [DONE]. Some servers and proxies omit it - // (or drop the trailing usage chunk). Finalize with whatever we've - // got — usage will be null in that case, which is fine. - try state.finalize(receiver, conv); + fn finishStream(self: *ResumableResponse, out: *EventQueue) !void { + if (self.done) return; + self.done = true; + try self.state.finalize(out, self.conv); } }; @@ -310,7 +386,7 @@ const StreamState = struct { /// Close the active text/thinking block (if any) and emit /// onBlockComplete. Ownership of `current_buf` transfers into the /// appended block. - fn closeActive(self: *StreamState, receiver: *provider_mod.Receiver) !void { + fn closeActive(self: *StreamState, out: *EventQueue) !void { if (self.active == .none) return; const block: conversation.ContentBlock = switch (self.active) { @@ -321,7 +397,10 @@ const StreamState = struct { self.current_buf = .empty; try self.blocks.append(self.allocator, block); - try receiver.onBlockComplete(self.block_index, self.blocks.items[self.blocks.items.len - 1]); + try out.push(.{ .block_complete = .{ + .index = self.block_index, + .block = self.blocks.items[self.blocks.items.len - 1], + } }); self.active = .none; } @@ -330,12 +409,12 @@ const StreamState = struct { fn openBlock( self: *StreamState, new_active: ActiveBlock, - receiver: *provider_mod.Receiver, + out: *EventQueue, ) !void { std.debug.assert(new_active == .text or new_active == .thinking); if (self.active == new_active) return; if (self.active != .none) { - try self.closeActive(receiver); + try self.closeActive(out); self.block_index += 1; } self.active = new_active; @@ -344,16 +423,21 @@ const StreamState = struct { .thinking => .Thinking, .tool_use, .none => unreachable, }; - try receiver.onBlockStart(block_type, self.block_index); + try out.push(.{ .block_start = .{ .block_type = block_type, .index = self.block_index } }); } fn appendDelta( self: *StreamState, - receiver: *provider_mod.Receiver, + out: *EventQueue, delta: []const u8, ) !void { try self.current_buf.appendSlice(self.allocator, delta); - try receiver.onContentDelta(self.block_index, delta); + // Dupe into the queue arena: the raw `delta` borrows the transient + // SSE payload that `produce` frees before `next()` reads the queue. + try out.push(.{ .content_delta = .{ + .index = self.block_index, + .delta = try out.dupeBytes(delta), + } }); } /// Apply one streaming tool_call delta. Opens a new tool_use on the @@ -362,7 +446,7 @@ const StreamState = struct { /// a malformed stream — we log and drop it. fn applyToolCallDelta( self: *StreamState, - receiver: *provider_mod.Receiver, + out: *EventQueue, d: json_mod.ToolCallDelta, ) !void { // Degenerate backend: a delta arrived for an index whose block we @@ -382,14 +466,14 @@ const StreamState = struct { // the only signal openai_chat gives us for mid-stream tool_use // boundaries; see the StreamState doc-comment for the rationale. if (self.current_tool_index) |cur| { - if (cur != d.index) try self.closeActiveTool(receiver); + if (cur != d.index) try self.closeActiveTool(out); } if (self.active_tool == null) { // Opening a new tool_use. First close any open text/thinking // block so the tool_use gets its own block_index. if (self.active != .none) { - try self.closeActive(receiver); + try self.closeActive(out); self.block_index += 1; } self.active_tool = .{ .block_index = self.block_index }; @@ -410,32 +494,39 @@ const StreamState = struct { // to render. If the block closes before any args arrive (zero-arg // tool), `closeActiveTool` emits the start there. if (d.arguments) |a| { - try self.emitStartIfNeeded(receiver, tu); - // Fire `onToolDetails` as soon as both id and name are + try self.emitStartIfNeeded(out, tu); + // Fire `tool_details` as soon as both id and name are // known. We can't know identity is *final* until the block // closes (a later delta could append more bytes), but in // practice OpenAI sends each whole on the first delta. A // pathological backend that streams id/name across many // chunks would have us emit a truncated value here. We - // accept that trade-off: receivers that need the canonical + // accept that trade-off: consumers that need the canonical // value can read it from the assembled ContentBlock at - // onBlockComplete. - try self.emitDetailsIfReady(receiver, tu); + // block_complete. + try self.emitDetailsIfReady(out, tu); try tu.arguments.appendSlice(self.allocator, a); - try receiver.onContentDelta(tu.block_index, a); + // Dupe into the queue arena (the SSE payload is freed before + // `next()` reads the queue). + try out.push(.{ .content_delta = .{ + .index = tu.block_index, + .delta = try out.dupeBytes(a), + } }); } else { // Identity-only chunk (no args yet). Still try to emit // details, in case both fields are now populated. - if (tu.started) try self.emitDetailsIfReady(receiver, tu); + if (tu.started) try self.emitDetailsIfReady(out, tu); } } - /// Fire `onToolDetails` once both id and name are non-empty. No-op if + /// Fire `tool_details` once both id and name are non-empty. No-op if /// already fired or if either field is still empty. Requires that - /// `onBlockStart` has already been emitted. + /// `block_start` has already been emitted. Slices are duped into the + /// queue arena because `id_buf`/`name_buf` may still grow (and realloc) + /// on later fragments. fn emitDetailsIfReady( self: *StreamState, - receiver: *provider_mod.Receiver, + out: *EventQueue, tu: *ToolUseInProgress, ) !void { _ = self; @@ -443,13 +534,17 @@ const StreamState = struct { if (!tu.started) return; if (tu.id_buf.items.len == 0 or tu.name_buf.items.len == 0) return; tu.details_emitted = true; - try receiver.onToolDetails(tu.block_index, tu.id_buf.items, tu.name_buf.items); + try out.push(.{ .tool_details = .{ + .index = tu.block_index, + .id = try out.dupeBytes(tu.id_buf.items), + .name = try out.dupeBytes(tu.name_buf.items), + } }); } - /// Close the currently-active tool_use (if any), emitting onBlockStart - /// (if it wasn't already), onBlockComplete, and recording the wire + /// Close the currently-active tool_use (if any), emitting block_start + /// (if it wasn't already), block_complete, and recording the wire /// index as closed. No-op if there's no active tool_use. - fn closeActiveTool(self: *StreamState, receiver: *provider_mod.Receiver) !void { + fn closeActiveTool(self: *StreamState, out: *EventQueue) !void { var tu = self.active_tool orelse return; self.active_tool = null; const wire_index = self.current_tool_index.?; @@ -481,12 +576,12 @@ const StreamState = struct { // internal (dotted) name. Decoding never grows the buffer. decodeNameInPlace(&tu.name_buf); - // If no arguments ever arrived, we haven't emitted onBlockStart - // yet — do it now so the receiver sees a balanced start/complete. - try self.emitStartIfNeeded(receiver, &tu); + // If no arguments ever arrived, we haven't emitted block_start + // yet — do it now so the consumer sees a balanced start/complete. + try self.emitStartIfNeeded(out, &tu); // Last chance to fire details if a fragmented-identity provider // only finished id/name accumulation at the very end. - try self.emitDetailsIfReady(receiver, &tu); + try self.emitDetailsIfReady(out, &tu); const id_owned = try tu.id_buf.toOwnedSlice(self.allocator); const name_owned = try tu.name_buf.toOwnedSlice(self.allocator); @@ -500,39 +595,41 @@ const StreamState = struct { tu.arguments = .empty; try self.blocks.append(self.allocator, block); - try receiver.onBlockComplete(tu.block_index, self.blocks.items[self.blocks.items.len - 1]); + try out.push(.{ .block_complete = .{ + .index = tu.block_index, + .block = self.blocks.items[self.blocks.items.len - 1], + } }); } - /// Emit `onBlockStart(.ToolUse, ...)` once per in-progress tool use. - /// Callers must invoke this before the first `onContentDelta` or - /// `onBlockComplete` for the block. Identity (id/name) is *not* - /// passed at start — see provider.zig's ReceiverVTable docs for the - /// rationale. Receivers get identity from the assembled ContentBlock - /// at onBlockComplete time. + /// Emit `block_start(.ToolUse, ...)` once per in-progress tool use. + /// Callers must invoke this before the first `content_delta` or + /// `block_complete` for the block. Identity (id/name) is *not* passed at + /// start — consumers get identity from `tool_details` or the assembled + /// ContentBlock at block_complete time. fn emitStartIfNeeded( self: *StreamState, - receiver: *provider_mod.Receiver, + out: *EventQueue, tu: *ToolUseInProgress, ) !void { _ = self; if (tu.started) return; tu.started = true; - try receiver.onBlockStart(.ToolUse, tu.block_index); + try out.push(.{ .block_start = .{ .block_type = .ToolUse, .index = tu.block_index } }); } /// End the stream: close any open text/thinking block, close the still- /// active tool_use (if any), then commit the assembled assistant - /// Message to the conversation. + /// Message to the conversation and push the terminal `message_complete`. fn finalize( self: *StreamState, - receiver: *provider_mod.Receiver, + out: *EventQueue, conv: *conversation.Conversation, ) !void { if (self.finalized) return; self.finalized = true; - try self.closeActive(receiver); - try self.closeActiveTool(receiver); + try self.closeActive(out); + try self.closeActiveTool(out); // Move blocks into a fresh conversation message. const moved_blocks = try self.blocks.toOwnedSlice(self.allocator); @@ -541,7 +638,7 @@ const StreamState = struct { try conv.addAssistantMessageWithUsage(moved_blocks, self.usage); const msg = conv.messages.items[conv.messages.items.len - 1]; - try receiver.onMessageComplete(msg, self.usage); + try out.push(.{ .message_complete = .{ .message = msg, .usage = self.usage } }); } }; @@ -549,7 +646,7 @@ fn handleEvent( allocator: Allocator, payload: []const u8, state: *StreamState, - receiver: *provider_mod.Receiver, + out: *EventQueue, ) !void { var parsed = try json_mod.parseStreamEvent(allocator, payload); defer parsed.deinit(); @@ -587,33 +684,33 @@ fn handleEvent( if (!state.started and d.role != null) { state.started = true; - try receiver.onMessageStart(.assistant); + try out.push(.{ .message_start = .assistant }); } if (d.reasoning_content) |rc| { if (!state.started) { state.started = true; - try receiver.onMessageStart(.assistant); + try out.push(.{ .message_start = .assistant }); } - try state.openBlock(.thinking, receiver); - try state.appendDelta(receiver, rc); + try state.openBlock(.thinking, out); + try state.appendDelta(out, rc); } if (d.content) |c| { if (!state.started) { state.started = true; - try receiver.onMessageStart(.assistant); + try out.push(.{ .message_start = .assistant }); } - try state.openBlock(.text, receiver); - try state.appendDelta(receiver, c); + try state.openBlock(.text, out); + try state.appendDelta(out, c); } if (d.tool_calls.len > 0) { if (!state.started) { state.started = true; - try receiver.onMessageStart(.assistant); + try out.push(.{ .message_start = .assistant }); } - for (d.tool_calls) |tc| try state.applyToolCallDelta(receiver, tc); + for (d.tool_calls) |tc| try state.applyToolCallDelta(out, tc); } if (d.finish_reason) |_| { @@ -627,54 +724,76 @@ fn handleEvent( const testing = std.testing; -/// A no-op Receiver that drops every callback. Useful when the test cares -/// about post-stream conversation state rather than callback observability. -const NoopReceiver = struct { - fn make() provider_mod.Receiver { - return .{ .ptr = @ptrCast(@constCast(&dummy)), .vtable = &vt }; - } - var dummy: u8 = 0; - const vt: provider_mod.ReceiverVTable = .{ - .onMessageStart = noopMsgStart, - .onBlockStart = noopBlockStart, - .onToolDetails = noopToolDetails, - .onContentDelta = noopDelta, - .onBlockComplete = noopBlockComplete, - .onMessageComplete = noopMsgComplete, - .onError = noopErr, - .onProviderRetry = noopRetry, - }; - fn noopMsgStart(_: *anyopaque, _: conversation.MessageRole) anyerror!void {} - fn noopBlockStart(_: *anyopaque, _: provider_mod.ContentBlockType, _: usize) anyerror!void {} - fn noopToolDetails(_: *anyopaque, _: usize, _: []const u8, _: []const u8) anyerror!void {} - fn noopDelta(_: *anyopaque, _: usize, _: []const u8) anyerror!void {} - fn noopBlockComplete(_: *anyopaque, _: usize, _: conversation.ContentBlock) anyerror!void {} - fn noopMsgComplete(_: *anyopaque, _: conversation.Message, _: ?provider_mod.Usage) anyerror!void {} - fn noopErr(_: *anyopaque, _: anyerror) void {} - fn noopRetry(_: *anyopaque, _: provider_mod.ProviderRetryInfo) void {} -}; - /// Feed a sequence of SSE event payloads through the state machine as if -/// they had been delivered by the wire, finalizing into `conv`. +/// they had been delivered by the wire, finalizing into `conv`. The decoded +/// `Event`s are recorded as compact strings (the same schema the old +/// RecordingReceiver used) so callback-ordering assertions are preserved. fn runStreamedTurn( allocator: Allocator, conv: *conversation.Conversation, - receiver: *provider_mod.Receiver, + rec: ?*EventRecorder, events: []const []const u8, ) !void { var state: StreamState = .init(allocator); defer state.deinit(); + var queue = EventQueue.init(allocator); + defer queue.deinit(); + for (events) |payload| { if (std.mem.eql(u8, payload, "[DONE]")) break; // Process every chunk through to [DONE], including the - // post-finish_reason usage chunk. Mirrors the production loop - // in OpenAIChatRequest.streamStep. - try handleEvent(allocator, payload, &state, receiver); + // post-finish_reason usage chunk. Mirrors the production pump in + // ResumableResponse.produce. + try handleEvent(allocator, payload, &state, &queue); + } + try state.finalize(&queue, conv); + + // Drain into the recorder before the arena resets. The queue holds all + // events from this turn; popping records each, and the final null-pop + // resets the arena. + while (queue.pop()) |ev| { + if (rec) |r| try r.record(ev); } - try state.finalize(receiver, conv); } +/// Records decoded `Event`s as compact strings for ordering assertions. +const EventRecorder = struct { + allocator: Allocator, + events: std.ArrayList([]const u8) = .empty, + + fn deinit(self: *EventRecorder) void { + for (self.events.items) |e| self.allocator.free(e); + self.events.deinit(self.allocator); + } + + fn push(self: *EventRecorder, comptime fmt: []const u8, args: anytype) !void { + const owned = try std.fmt.allocPrint(self.allocator, fmt, args); + try self.events.append(self.allocator, owned); + } + + fn record(self: *EventRecorder, ev: Event) !void { + switch (ev) { + .message_start => try self.push("msg_start", .{}), + .block_start => |b| try self.push("block_start[{d}]:{s}", .{ b.index, @tagName(b.block_type) }), + .tool_details => |t| try self.push("tool_details[{d}]:{s}:{s}", .{ t.index, t.id, t.name }), + .content_delta => |d| try self.push("delta[{d}]:{s}", .{ d.index, d.delta }), + .block_complete => |b| try self.push("block_complete[{d}]", .{b.index}), + .message_complete => |m| { + if (m.usage) |u| { + try self.push( + "msg_complete[usage:in={d},out={d},cr={d},cw={d},rsn={d}]", + .{ u.input, u.output, u.cache_read, u.cache_write, u.reasoning }, + ); + } else { + try self.push("msg_complete[usage:null]", .{}); + } + }, + else => {}, + } + } +}; + test "two streamed turns persist assistant replies in the conversation" { // Regression test for the bug where `finish_reason` arrived before // `[DONE]` and `finalize` early-returned without appending the assistant @@ -688,8 +807,6 @@ test "two streamed turns persist assistant replies in the conversation" { try conv.addSystemMessage("You are a helpful assistant."); try conv.addUserMessage("hello!"); - var recv = NoopReceiver.make(); - const turn1 = [_][]const u8{ \\{"choices":[{"delta":{"role":"assistant"}}]} , @@ -701,7 +818,7 @@ test "two streamed turns persist assistant replies in the conversation" { , "[DONE]", }; - try runStreamedTurn(allocator, &conv, &recv, &turn1); + try runStreamedTurn(allocator, &conv, null, &turn1); try testing.expectEqual(@as(usize, 3), conv.messages.items.len); try testing.expectEqual(conversation.MessageRole.assistant, conv.messages.items[2].role); @@ -722,7 +839,7 @@ test "two streamed turns persist assistant replies in the conversation" { , "[DONE]", }; - try runStreamedTurn(allocator, &conv, &recv, &turn2); + try runStreamedTurn(allocator, &conv, null, &turn2); // System + user + assistant + user + assistant = 5 messages. try testing.expectEqual(@as(usize, 5), conv.messages.items.len); @@ -740,12 +857,8 @@ test "openai_chat: terminating usage chunk lands on message_complete with split defer conv.deinit(); try conv.addUserMessage("hi"); - var rec = RecordingReceiver{ .allocator = allocator }; - defer { - for (rec.events.items) |s| allocator.free(s); - rec.events.deinit(allocator); - } - var recv = rec.receiver(); + var rec = EventRecorder{ .allocator = allocator }; + defer rec.deinit(); const events = [_][]const u8{ \\{"choices":[{"delta":{"role":"assistant"}}]} @@ -759,7 +872,7 @@ test "openai_chat: terminating usage chunk lands on message_complete with split , "[DONE]", }; - try runStreamedTurn(allocator, &conv, &recv, &events); + try runStreamedTurn(allocator, &conv, &rec, &events); var found: ?[]const u8 = null; for (rec.events.items) |s| { @@ -777,12 +890,8 @@ test "openai_chat: omitted stream usage yields null on message_complete" { defer conv.deinit(); try conv.addUserMessage("hi"); - var rec = RecordingReceiver{ .allocator = allocator }; - defer { - for (rec.events.items) |s| allocator.free(s); - rec.events.deinit(allocator); - } - var recv = rec.receiver(); + var rec = EventRecorder{ .allocator = allocator }; + defer rec.deinit(); const events = [_][]const u8{ \\{"choices":[{"delta":{"role":"assistant"}}]} @@ -793,7 +902,7 @@ test "openai_chat: omitted stream usage yields null on message_complete" { , "[DONE]", }; - try runStreamedTurn(allocator, &conv, &recv, &events); + try runStreamedTurn(allocator, &conv, &rec, &events); var found: ?[]const u8 = null; for (rec.events.items) |s| { @@ -814,8 +923,6 @@ test "fragmented tool_call id and name are reassembled" { defer conv.deinit(); try conv.addUserMessage("call something"); - var recv = NoopReceiver.make(); - const events = [_][]const u8{ \\{"choices":[{"delta":{"role":"assistant"}}]} , @@ -830,7 +937,7 @@ test "fragmented tool_call id and name are reassembled" { "[DONE]", }; - try runStreamedTurn(allocator, &conv, &recv, &events); + try runStreamedTurn(allocator, &conv, null, &events); const asst = conv.messages.items[1]; try testing.expectEqual(@as(usize, 1), asst.content.items.len); @@ -851,8 +958,6 @@ test "inbound wire tool name is decoded to dotted form (even split across __)" { defer conv.deinit(); try conv.addUserMessage("read a file"); - var recv = NoopReceiver.make(); - const events = [_][]const u8{ \\{"choices":[{"delta":{"role":"assistant"}}]} , @@ -867,96 +972,12 @@ test "inbound wire tool name is decoded to dotted form (even split across __)" { "[DONE]", }; - try runStreamedTurn(allocator, &conv, &recv, &events); + try runStreamedTurn(allocator, &conv, null, &events); const tu = conv.messages.items[1].content.items[0].ToolUse; try testing.expectEqualStrings("std.read", tu.name); } -/// A Receiver that records the sequence of callback events as compact -/// strings. Useful for asserting per-block start/complete ordering. -const RecordingReceiver = struct { - allocator: Allocator, - events: std.ArrayList([]const u8) = .empty, - - fn receiver(self: *RecordingReceiver) provider_mod.Receiver { - return .{ .ptr = self, .vtable = &vt }; - } - - const vt: provider_mod.ReceiverVTable = .{ - .onMessageStart = onMessageStart, - .onBlockStart = onBlockStart, - .onToolDetails = onToolDetails, - .onContentDelta = onContentDelta, - .onBlockComplete = onBlockComplete, - .onMessageComplete = onMessageComplete, - .onError = onError, - .onProviderRetry = onProviderRetry, - }; - - fn onProviderRetry(_: *anyopaque, _: provider_mod.ProviderRetryInfo) void {} - - fn record(self: *RecordingReceiver, s: []const u8) !void { - const owned = try self.allocator.dupe(u8, s); - try self.events.append(self.allocator, owned); - } - - fn recordFmt(self: *RecordingReceiver, comptime fmt: []const u8, args: anytype) !void { - const owned = try std.fmt.allocPrint(self.allocator, fmt, args); - try self.events.append(self.allocator, owned); - } - - fn deinit(self: *RecordingReceiver) void { - for (self.events.items) |e| self.allocator.free(e); - self.events.deinit(self.allocator); - } - - fn onMessageStart(ptr: *anyopaque, _: conversation.MessageRole) anyerror!void { - const self: *RecordingReceiver = @ptrCast(@alignCast(ptr)); - try self.record("msg_start"); - } - fn onBlockStart( - ptr: *anyopaque, - bt: provider_mod.ContentBlockType, - idx: usize, - ) anyerror!void { - const self: *RecordingReceiver = @ptrCast(@alignCast(ptr)); - try self.recordFmt("block_start[{d}]:{s}", .{ idx, @tagName(bt) }); - } - fn onToolDetails( - ptr: *anyopaque, - idx: usize, - id: []const u8, - name: []const u8, - ) anyerror!void { - const self: *RecordingReceiver = @ptrCast(@alignCast(ptr)); - try self.recordFmt("tool_details[{d}]:{s}:{s}", .{ idx, id, name }); - } - fn onContentDelta(ptr: *anyopaque, idx: usize, delta: []const u8) anyerror!void { - const self: *RecordingReceiver = @ptrCast(@alignCast(ptr)); - try self.recordFmt("delta[{d}]:{s}", .{ idx, delta }); - } - fn onBlockComplete( - ptr: *anyopaque, - idx: usize, - _: conversation.ContentBlock, - ) anyerror!void { - const self: *RecordingReceiver = @ptrCast(@alignCast(ptr)); - try self.recordFmt("block_complete[{d}]", .{idx}); - } - fn onMessageComplete(ptr: *anyopaque, _: conversation.Message, usage: ?provider_mod.Usage) anyerror!void { - const self: *RecordingReceiver = @ptrCast(@alignCast(ptr)); - if (usage) |u| { - try self.recordFmt( - "msg_complete[usage:in={d},out={d},cr={d},cw={d},rsn={d}]", - .{ u.input, u.output, u.cache_read, u.cache_write, u.reasoning }, - ); - } else { - try self.record("msg_complete[usage:null]"); - } - } - fn onError(_: *anyopaque, _: anyerror) void {} -}; test "parallel tool_calls emit one complete start/delta/complete cycle per block" { // Regression test: previously, the OpenAI provider deferred ALL @@ -972,9 +993,8 @@ test "parallel tool_calls emit one complete start/delta/complete cycle per block defer conv.deinit(); try conv.addUserMessage("ping four hosts"); - var rec: RecordingReceiver = .{ .allocator = allocator }; + var rec: EventRecorder = .{ .allocator = allocator }; defer rec.deinit(); - var recv = rec.receiver(); const events = [_][]const u8{ \\{"choices":[{"delta":{"role":"assistant"}}]} @@ -992,7 +1012,7 @@ test "parallel tool_calls emit one complete start/delta/complete cycle per block "[DONE]", }; - try runStreamedTurn(allocator, &conv, &recv, &events); + try runStreamedTurn(allocator, &conv, &rec, &events); const expected = [_][]const u8{ "msg_start", @@ -1039,9 +1059,8 @@ test "non-contiguous tool_call deltas: re-emission of a closed index is dropped" defer conv.deinit(); try conv.addUserMessage("go"); - var rec: RecordingReceiver = .{ .allocator = allocator }; + var rec: EventRecorder = .{ .allocator = allocator }; defer rec.deinit(); - var recv = rec.receiver(); const events = [_][]const u8{ \\{"choices":[{"delta":{"role":"assistant"}}]} @@ -1058,7 +1077,7 @@ test "non-contiguous tool_call deltas: re-emission of a closed index is dropped" "[DONE]", }; - try runStreamedTurn(allocator, &conv, &recv, &events); + try runStreamedTurn(allocator, &conv, &rec, &events); // Two well-formed tool_use blocks in the final message, args unaffected // by the dropped fragment. @@ -1088,9 +1107,8 @@ test "onToolDetails fires after id+name complete, even mid-arg-stream" { defer conv.deinit(); try conv.addUserMessage("go"); - var rec: RecordingReceiver = .{ .allocator = allocator }; + var rec: EventRecorder = .{ .allocator = allocator }; defer rec.deinit(); - var recv = rec.receiver(); const events = [_][]const u8{ \\{"choices":[{"delta":{"role":"assistant"}}]} @@ -1114,7 +1132,7 @@ test "onToolDetails fires after id+name complete, even mid-arg-stream" { "[DONE]", }; - try runStreamedTurn(allocator, &conv, &recv, &events); + try runStreamedTurn(allocator, &conv, &rec, &events); // Exactly one tool_details event, fired with the id-prefix that was // current at first-args-arrival, and ordered between block_start and @@ -1155,8 +1173,6 @@ test "tool_call with no arguments still finalizes a well-formed ToolUse" { defer conv.deinit(); try conv.addUserMessage("ring it"); - var recv = NoopReceiver.make(); - const events = [_][]const u8{ \\{"choices":[{"delta":{"role":"assistant"}}]} , @@ -1167,7 +1183,7 @@ test "tool_call with no arguments still finalizes a well-formed ToolUse" { "[DONE]", }; - try runStreamedTurn(allocator, &conv, &recv, &events); + try runStreamedTurn(allocator, &conv, null, &events); const asst = conv.messages.items[1]; try testing.expectEqual(@as(usize, 1), asst.content.items.len); diff --git a/libpanto/src/root.zig b/libpanto/src/root.zig index a49a428..b5fab1a 100644 --- a/libpanto/src/root.zig +++ b/libpanto/src/root.zig @@ -2,6 +2,7 @@ const std = @import("std"); pub const conversation = @import("conversation.zig"); pub const provider = @import("provider.zig"); +pub const stream = @import("stream.zig"); pub const agent = @import("agent.zig"); pub const config = @import("config.zig"); pub const sse = @import("sse.zig"); @@ -32,12 +33,17 @@ pub const ToolRegistry = tool_registry.ToolRegistry; pub const Config = config.Config; pub const ProviderConfig = config.ProviderConfig; +// Re-export the pull-streaming surface for embedders. +pub const Event = stream.Event; +pub const Stream = agent.Stream; +pub const Agent = agent.Agent; + // Internal modules. Not part of the public API — callers drive turns via -// `provider.streamStep(allocator, io, &config, conv, receiver)` (or via the -// `Agent`, which holds a swappable `*const Config`). The process-global HTTP -// client is initialized with `config.initHttp` / torn down with -// `config.deinitHttp`. These impls are exposed here only so `refAllDecls` -// picks up their tests. +// the `Agent` (which holds a swappable `*const Config`): `agent.run()` +// returns a `*Stream` whose `next()` pulls one `Event` at a time. The +// process-global HTTP client is initialized with `config.initHttp` / torn +// down with `config.deinitHttp`. These impls are exposed here only so +// `refAllDecls` picks up their tests. const openai_chat_json = @import("openai_chat_json.zig"); const provider_openai_chat = @import("provider_openai_chat.zig"); const anthropic_messages_json = @import("anthropic_messages_json.zig"); diff --git a/libpanto/src/stream.zig b/libpanto/src/stream.zig new file mode 100644 index 0000000..e626f31 --- /dev/null +++ b/libpanto/src/stream.zig @@ -0,0 +1,174 @@ +//! 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, + + /// 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; + } +}; |