summaryrefslogtreecommitdiff
path: root/libpanto/src/agent.zig
diff options
context:
space:
mode:
Diffstat (limited to 'libpanto/src/agent.zig')
-rw-r--r--libpanto/src/agent.zig787
1 files changed, 458 insertions, 329 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.
generated by cgit v1.3 (git 2.53.0) at 2026-07-09 08:38:35 +0000