diff options
Diffstat (limited to 'libpanto/src/agent.zig')
| -rw-r--r-- | libpanto/src/agent.zig | 3925 |
1 files changed, 0 insertions, 3925 deletions
diff --git a/libpanto/src/agent.zig b/libpanto/src/agent.zig deleted file mode 100644 index 09b0994..0000000 --- a/libpanto/src/agent.zig +++ /dev/null @@ -1,3925 +0,0 @@ -//! The Agent owns the conversation-driving loop: provider streaming + -//! tool dispatch. -//! -//! On each turn, after the provider streams an assistant message, the -//! agent inspects it for ToolUse blocks. If any are present, the agent: -//! -//! 1. Groups them by their *owning registration* in the registry — a -//! single `Tool` is its own group; every `ToolSource`-backed tool -//! whose name maps to the same source forms one group. -//! 2. Spawns one concurrent task per group via `std.Io.Group`. -//! A single-`Tool` group runs the tool's `invoke` once; a -//! `ToolSource` group calls the source's `invoke_batch` with all -//! of its calls at once. We use `Group.concurrent` (not `async`) -//! because tool invocations may block on I/O and we need real -//! concurrency, not just expressed asynchrony. -//! 3. Awaits the group. ToolResult blocks are assembled in the -//! *original* call order (i.e. the order the LLM emitted them). -//! 4. Appends a user message containing the ToolResult blocks back -//! into the conversation and loops. -//! -//! The "thread-safe" promise for single `Tool` registrations is -//! unchanged. For `ToolSource`-backed tools, the source's runtime -//! receives all of its calls on one thread per turn, so it can keep a -//! single-threaded interpreter (Lua, Python, ...) without further -//! synchronization. - -const std = @import("std"); -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"); -const tool_mod = @import("tool.zig"); -const image_mod = @import("image.zig"); -const tool_source_mod = @import("tool_source.zig"); -const tool_registry_mod = @import("tool_registry.zig"); -const session_store_mod = @import("session_store.zig"); -const null_store_mod = @import("null_store.zig"); -const turn_persist = @import("turn_persist.zig"); - -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; - -/// Re-export for the `compact` usages parameter (provider-reported token -/// usage per message, used for retention sizing). -pub const conversation_Usage = @import("session.zig").Usage; - -/// Append a single-text user message. `Conversation.addUserMessage` now -/// takes a block slice (symmetric with `addAssistantMessage`); this wraps -/// the common plain-text case used by the agent's interactive turn and the -/// compaction prompt. -fn addUserText(conv: *conversation.Conversation, text: []const u8) !void { - const tb = try conversation.textualBlockFromSlice(conv.allocator, text); - var block: conversation.ContentBlock = .{ .Text = tb }; - errdefer block.deinit(conv.allocator); - try conv.addUserMessage(&.{block}); -} - -/// Deep-copy a message (role + all content blocks) into fresh owned -/// allocations. Used when rebuilding the conversation after compaction. -fn cloneMessage(alloc: Allocator, msg: conversation.Message) !conversation.Message { - var content: std.ArrayList(conversation.ContentBlock) = .empty; - errdefer { - for (content.items) |*b| b.deinit(alloc); - content.deinit(alloc); - } - try content.ensureTotalCapacity(alloc, msg.content.items.len); - for (msg.content.items) |block| { - content.appendAssumeCapacity(try cloneBlock(alloc, block)); - } - return .{ - .role = msg.role, - .content = content, - .usage = msg.usage, - .metadata = if (msg.metadata) |m| try alloc.dupe(u8, m) else null, - // Preserve the producing identity so a kept-verbatim turn isn't - // re-stamped with the compaction model on persist. - .identity = if (msg.identity) |id| try conversation.dupeWireIdentity(alloc, id) else null, - }; -} - -fn cloneBlock(alloc: Allocator, block: conversation.ContentBlock) !conversation.ContentBlock { - return switch (block) { - .Text => |b| .{ .Text = try conversation.textualBlockFromSlice(alloc, b.items) }, - .Thinking => |b| blk: { - const tb = try conversation.textualBlockFromSlice(alloc, b.text.items); - errdefer { - var mut = tb; - mut.deinit(alloc); - } - const sig: ?[]const u8 = if (b.signature) |s| try alloc.dupe(u8, s) else null; - errdefer if (sig) |s| alloc.free(s); - const origin: ?conversation.SignatureOrigin = if (b.signature_origin) |o| try o.dupe(alloc) else null; - break :blk .{ .Thinking = .{ .text = tb, .signature = sig, .signature_origin = origin } }; - }, - .ToolUse => |b| blk: { - const id = try alloc.dupe(u8, b.id); - errdefer alloc.free(id); - const name = try alloc.dupe(u8, b.name); - errdefer alloc.free(name); - const input = try conversation.textualBlockFromSlice(alloc, b.input.items); - break :blk .{ .ToolUse = .{ .id = id, .name = name, .input = input } }; - }, - .ToolResult => |b| blk: { - const tuid = try alloc.dupe(u8, b.tool_use_id); - errdefer alloc.free(tuid); - var parts: std.ArrayList(conversation.ResultPartStored) = .empty; - errdefer { - for (parts.items) |*p| p.deinit(alloc); - parts.deinit(alloc); - } - try parts.ensureTotalCapacity(alloc, b.parts.items.len); - for (b.parts.items) |src| { - switch (src) { - .text => |tb| { - const t = try conversation.textualBlockFromSlice(alloc, tb.items); - parts.appendAssumeCapacity(.{ .text = t }); - }, - .media => |m| { - const mt = try alloc.dupe(u8, m.media_type); - errdefer alloc.free(mt); - const data = try conversation.textualBlockFromSlice(alloc, m.data.items); - parts.appendAssumeCapacity(.{ .media = .{ .media_type = mt, .data = data } }); - }, - } - } - break :blk .{ .ToolResult = .{ .tool_use_id = tuid, .parts = parts, .is_error = b.is_error } }; - }, - .System => |b| .{ .System = .{ - .text = try conversation.textualBlockFromSlice(alloc, b.text.items), - .mode = b.mode, - } }, - .CompactionSummary => |b| .{ .CompactionSummary = .{ - .text = try conversation.textualBlockFromSlice(alloc, b.text.items), - } }, - }; -} - -fn isValidToolInput(input: []const u8) bool { - if (input.len == 0) return true; - if (input[0] != '{') return true; // legacy tests/tools may use opaque bytes - var parsed = std.json.parseFromSlice(std.json.Value, std.heap.page_allocator, input, .{}) catch return false; - defer parsed.deinit(); - return parsed.value == .object; -} - -fn invalidInputResult(allocator: Allocator, input: []const u8) !tool_mod.ResultParts { - const msg = try std.fmt.allocPrint( - allocator, - "Tool call was not executed: tool input was incomplete or invalid JSON. Partial input: {s}", - .{input}, - ); - return tool_mod.ResultParts.fromTextOwned(allocator, msg); -} - -/// What to do with an error returned by tool dispatch. -const ToolErrorAction = enum { - /// Surface the failure to the model as an error `ToolResult`, then let - /// the agent loop continue. - tool_result, - /// Abort the whole turn and propagate to the embedder. Reserved for - /// failures that belong to the host, not the model/provider exchange. - hard_fail, -}; - -/// Decide how to handle a tool dispatch error. Only genuine host failures -/// abort the turn; everything else becomes a model-visible tool result so -/// the model can correct course (and so every `ToolUse` keeps its matching -/// `ToolResult`, which providers require). -fn classifyToolError(err: anyerror) ToolErrorAction { - return switch (err) { - error.Canceled, error.OutOfMemory => .hard_fail, - else => .tool_result, - }; -} - -/// Build an error `ResultPart` describing a failed tool call, in the -/// model-readable form the plan specifies. -fn toolErrorResult( - allocator: Allocator, - tool_name: []const u8, - err: anyerror, -) !tool_mod.ResultParts { - const msg = try std.fmt.allocPrint( - allocator, - "Tool execution failed for `{s}`: {s}\n" ++ - "You may fix the arguments, try a different tool, or explain the failure to the user.", - .{ tool_name, @errorName(err) }, - ); - return tool_mod.ResultParts.fromTextOwned(allocator, msg); -} - -/// The user's submission that opens a turn: an ordered list of content -/// blocks (text, tool results, images, …), the same shape used to rebuild a -/// persisted user message. Ownership of `blocks` transfers to the agent's -/// conversation on `run`; the caller must not deinit them afterwards. A plain -/// chat turn is one `.Text` block; a turn that resumes after the embedder -/// handled tool calls is one or more `.ToolResult` blocks. -pub const UserMessage = struct { - blocks: []const conversation.ContentBlock, -}; - -/// Outcome of a compaction attempt. -pub const CompactionResult = struct { - /// Whether the conversation was actually compacted. False means the - /// active conversation already fit within the keep-verbatim budget - /// (nothing to summarize) — the conversation is unchanged. - compacted: bool, - /// Number of whole turns kept verbatim after the summary. - kept_turns: usize = 0, - /// Number of conversation messages folded into the summary. - summarized_messages: usize = 0, -}; - -pub const Agent = struct { - _allocator: Allocator, - _io: Io, - /// The active configuration snapshot, consulted fresh at the top of - /// every turn. Immutable while a turn is in flight; swap this pointer - /// (`setConfig`) between turns to change provider/model/base_url - /// atomically. The pointee is owned by the embedder, not the agent. The - /// tool set is no longer part of the snapshot — it lives on `_registry`. - _config: *const Config, - /// The tool set this agent exposes. Owned by the agent: created empty at - /// `init`, populated via `registerTool`/`registerToolSource`, torn down - /// in `deinit`. Read fresh by the agent loop each turn, so a - /// registration between turns is visible at the next turn boundary. - _registry: ToolRegistry, - /// The live conversation the agent drives. Owned by the agent (adopted - /// at `init`); torn down in `deinit`. Turn-driving methods operate on - /// this directly rather than taking a `*Conversation` parameter. - /// - /// This is the one intended-public field (every other field is - /// underscore-prefixed internal state): a borrowed handle on the live - /// conversation, for in-place context-management surgery. Valid for the - /// agent's lifetime; do not retain past it. - conversation: conversation.Conversation, - /// The session this agent appends to. Minted from the store at `init` - /// (fresh: `store.create()`) or supplied by the embedder on resume - /// (`resolve`/`latest`). `Session.append` proxies to the store and - /// updates the session's last-used wire identity. The embedder owns the - /// underlying store, which must outlive the agent. - _session: session_store_mod.Session, - /// Injectable streaming seam. Defaults to the real provider dispatch - /// (`provider_mod.openStream`); tests override it with a stub. - _open_stream_fn: provider_mod.OpenStreamFn = provider_mod.openStream, - /// Set by the embedder after `runStep` returns to learn whether an - /// automatic compaction occurred this turn (so it can persist the - /// rewritten conversation). Reset at the top of each `runStep`. - _auto_compacted: bool = false, - /// PRNG state for backoff jitter. Seeded lazily on first retry. Only - /// touched from the single agent-loop thread (retries are serial), so - /// no synchronization is needed. - _retry_prng: ?std.Random.DefaultPrng = null, - /// High-water mark of messages durably handed to the store: the count of - /// conversation messages already persisted. `flushPersist` appends only - /// `[_persisted_through .. coherent_end)` and advances this, so the store - /// stays current at every `Stream.next()` boundary without re-appending. - /// Initialized to the loaded history length (already persisted) and - /// rewound by `rewriteWithSummary` so a compaction re-persists its summary - /// + restated suffix. - _persisted_through: usize = 0, - /// User-message blocks queued by an embedder seam (the CLI's - /// `panto.ext.agent:submit`) for the host front end to open the next - /// turn with, so the turn runs on the host's own driver (rendering, - /// interrupts) rather than inside the queuing callback. Blocks and the - /// list storage are owned by `conversation.allocator` — the same - /// ownership `run` adopts. - _pending_submission: std.ArrayList(conversation.ContentBlock) = .empty, - - /// Construct an agent. - /// - /// `store` is the persistence backend (use `null_store.store()` to opt - /// out). `maybe_conversation` is adopted (ownership transferred) when - /// non-null — the resume path: open a store, ask it for the - /// conversation, hand it here. When null, a fresh empty conversation is - /// created. Either way the agent owns and tears down the conversation. - /// The inner is heap-pinned at `init`, so the returned `*Agent` is a - /// cheap, movable handle ("don't move the Agent" stops being a rule - /// anyone can violate). The caller owns it and must `deinit` it. - pub fn init( - allocator: Allocator, - io: Io, - config: *const Config, - session: session_store_mod.Session, - maybe_conversation: ?conversation.Conversation, - ) !*Agent { - const self = try allocator.create(Agent); - self.* = .{ - ._allocator = allocator, - ._io = io, - ._config = config, - ._registry = ToolRegistry.init(allocator), - .conversation = maybe_conversation orelse conversation.Conversation.init(allocator), - ._session = session, - }; - // Loaded history is already in the store; only messages produced from - // here on need persisting. - self._persisted_through = self.conversation.messages.items.len; - return self; - } - - pub fn deinit(self: *Agent) void { - // The agent owns the conversation, the tool registry, and the - // session handle's `info` (minted by `store.create()` or resolved - // by the embedder and handed in). It borrows the config snapshot - // and the underlying store, which the embedder tears down. It is - // self-heap-pinned (`init` allocated it), so it frees itself last. - const allocator = self._allocator; - self._registry.deinit(); - for (self._pending_submission.items) |*b| b.deinit(self.conversation.allocator); - self._pending_submission.deinit(self.conversation.allocator); - self.conversation.deinit(); - self._session.info.deinit(allocator); - allocator.destroy(self); - } - - /// The id of the session this agent appends to. - pub fn sessionId(self: *const Agent) []const u8 { - return self._session.info.id; - } - - /// Add a single tool to this agent's tool set. Visible at the next turn. - pub fn registerTool(self: *Agent, tool: Tool) !void { - try self._registry.register(tool); - } - - /// Add a tool source (a dynamic group of tools) to this agent's tool - /// set. Visible at the next turn. - pub fn registerToolSource(self: *Agent, src: ToolSource) !void { - try self._registry.registerSource(src); - } - - /// The wire-format provider identity stamped on persisted entries, - /// derived from the active config snapshot. Ground truth: never a CLI - /// alias, never any `api_key` material. - fn wireIdentity(self: *const Agent) session_store_mod.WireIdentity { - return self._config.provider.wireIdentity(); - } - - /// Swap the active configuration snapshot. Takes effect at the start of - /// the next turn. Safe to call between `runStep` invocations or from a - /// tool handler that runs between provider steps; never mutates a - /// snapshot a turn is currently reading. - pub fn setConfig(self: *Agent, config: *const Config) void { - self._config = config; - } - - /// 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. - /// Append a system message (`.append` mode) and persist it. Adds to the - /// effective system prompt. - pub fn addSystemMessage(self: *Agent, text: []const u8) !void { - return self._persistSystemMessage(text, .append); - } - - /// Replace the effective system prompt (`.replace` mode) and persist it. - /// Discards all prior system text; replay reconstructs the same prompt. - pub fn setSystemPrompt(self: *Agent, text: []const u8) !void { - return self._persistSystemMessage(text, .replace); - } - - fn _persistSystemMessage( - self: *Agent, - text: []const u8, - mode: conversation.SystemMode, - ) !void { - switch (mode) { - .append => try self.conversation.addSystemMessage(text), - .replace => try self.conversation.replaceSystemMessage(text), - } - self.flushPersist(); - } - - /// Replace the text of the ToolResult block matching `tool_use_id` in - /// the conversation's LAST message. Media parts are preserved (the new - /// text becomes the single leading text part); `is_error` is untouched. - /// Returns false when the last message holds no such block. - /// - /// This is the application half of the embedder's "tool_result output - /// override" (a `tool_result` event handler assigning `ev.output`): - /// called between tool dispatch and the next provider request, the - /// override becomes canonical — wire, persistence, and compaction all - /// see the replaced text. - pub fn overrideToolResultOutput(self: *Agent, tool_use_id: []const u8, text: []const u8) !bool { - const msgs = self.conversation.messages.items; - if (msgs.len == 0) return false; - const last = &msgs[msgs.len - 1]; - if (last.role != .user) return false; - for (last.content.items) |*block| { - if (block.* != .ToolResult) continue; - const tr = &block.ToolResult; - if (!std.mem.eql(u8, tr.tool_use_id, tool_use_id)) continue; - - var new_parts: std.ArrayList(conversation.ResultPartStored) = .empty; - errdefer new_parts.deinit(self._allocator); - try new_parts.ensureTotalCapacity(self._allocator, tr.parts.items.len + 1); - var new_text: conversation.TextualBlock = .empty; - try new_text.appendSlice(self._allocator, text); - errdefer new_text.deinit(self._allocator); - new_parts.appendAssumeCapacity(.{ .text = new_text }); - for (tr.parts.items) |*p| switch (p.*) { - .text => |*t| t.deinit(self._allocator), - .media => |m| new_parts.appendAssumeCapacity(.{ .media = m }), - }; - tr.parts.deinit(self._allocator); - tr.parts = new_parts; - return true; - } - return false; - } - - /// Replace the input JSON of the ToolUse block matching `tool_use_id` - /// in the conversation's LAST message (the committed assistant message, - /// between response completion and tool dispatch). Returns false when - /// no such block exists there. - /// - /// Application half of the "tool_call_complete input override": the - /// rewritten input is what the tool executes with AND what the model - /// sees of its own call in future requests. - pub fn overrideToolUseInput(self: *Agent, tool_use_id: []const u8, input_json: []const u8) !bool { - const msgs = self.conversation.messages.items; - if (msgs.len == 0) return false; - const last = &msgs[msgs.len - 1]; - if (last.role != .assistant) return false; - for (last.content.items) |*block| { - if (block.* != .ToolUse) continue; - const tu = &block.ToolUse; - if (!std.mem.eql(u8, tu.id, tool_use_id)) continue; - tu.input.clearRetainingCapacity(); - try tu.input.appendSlice(self._allocator, input_json); - return true; - } - return false; - } - - /// Submit a user message and begin a turn, returning a resumable pull - /// `Stream`. - /// - /// 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. - /// Append blocks to the pending submission (see `_pending_submission`). - /// Adopts the block *contents*; the `blocks` slice itself stays the - /// caller's. Multiple calls accumulate into one pending user message. - pub fn queueSubmission(self: *Agent, blocks: []const conversation.ContentBlock) !void { - try self._pending_submission.appendSlice(self.conversation.allocator, blocks); - } - - /// Take the pending submission, or null when none is queued. The caller - /// owns the returned slice: typically hand it to `run` (which adopts the - /// block contents) and free the slice itself with - /// `conversation.allocator`. - pub fn takeSubmission(self: *Agent) !?[]conversation.ContentBlock { - if (self._pending_submission.items.len == 0) return null; - return try self._pending_submission.toOwnedSlice(self.conversation.allocator); - } - - pub fn run(self: *Agent, message: UserMessage) !*Stream { - self._auto_compacted = false; - - // Append + persist the user prompt up front (the dangling-prompt - // recovery guarantee). `addMessage` adopts the blocks; persistence is - // brought current through the shared high-water flush. - try self.conversation.addMessage(.user, message.blocks, null); - self.flushPersist(); - - const s = try self._allocator.create(Stream); - s.* = Stream.init(self); - return s; - } - - /// Bring the session store current with the conversation: append every - /// message in `[_persisted_through .. coherent_end)` as a single batch and - /// advance the high-water mark. `coherent_end` excludes a trailing - /// assistant message whose ToolUse blocks have no following results — a - /// turn interrupted between the assistant's tool call and dispatch. Leaving - /// it unpersisted keeps the log replayable and the live conversation in a - /// state `compact()` (or a tool-result `run`) can resume from. Called at - /// every `Stream.next()` boundary, so after any `next()` the store reflects - /// all committed, coherent messages. - pub fn flushPersist(self: *Agent) void { - const msgs = self.conversation.messages.items; - // An operation that rewrote the prefix (replace-system, cancel) can - // leave the mark past the new length; clamp before comparing. - if (self._persisted_through > msgs.len) self._persisted_through = msgs.len; - var end = msgs.len; - if (end > self._persisted_through and - msgs[end - 1].role == .assistant and - turn_persist.hasToolUseWithoutFollowingResults(&self.conversation, end - 1)) - { - end -= 1; - } - if (end <= self._persisted_through) return; - turn_persist.persistRange( - self._allocator, - &self._session, - &self.conversation, - self._persisted_through, - end, - self.wireIdentity(), - &.{}, - ) catch |e| { - std.log.err("session: failed to persist turn: {t}", .{e}); - return; - }; - self._persisted_through = end; - } - - fn hasToolUseBlock(msg: conversation.Message) bool { - return toolUseCount(msg) > 0; - } - - /// 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 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). - /// - retryable provider error (rate limit, server, transport, - /// malformed stream): sleep with exponential backoff + jitter - /// (honoring `Retry-After` when present) and retry, up to - /// `retry.max_attempts` total attempts. - /// - anything else (auth, bad request, cancellation, local errors): - /// propagate immediately. - /// - /// A failed open never mutates the conversation (providers commit the - /// assistant message only on success), so each retry runs against the - /// 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, - out: *stream_mod.EventQueue, - ) !provider_mod.ProviderStream { - const policy = cfg.retry; - var attempt: usize = 1; - while (true) { - var diag: provider_mod.ProviderDiagnostic = .{}; - const ps = self._open_stream_fn(self._allocator, self._io, cfg, &self._registry, &self.conversation, &diag) catch |err| { - if (err == error.ContextOverflow) { - 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); - try out.push(.{ .provider_retry = .{ - .attempt = attempt, - .max_attempts = policy.max_attempts, - .delay_ms = delay_ms, - .err = err, - .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; - } - attempt += 1; - continue; - }; - return ps; - } - } - - /// 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, - out: *stream_mod.EventQueue, - err: anyerror, - ) !provider_mod.ProviderStream { - if (self._auto_compacted) return err; // already retried once this turn - const sys = self._config.compaction.compaction_prompt orelse return err; - const res = try self._compactInPlace(sys, null); - if (!res.compacted) return err; // nothing to shed; give up - self._auto_compacted = true; - 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. - var diag: provider_mod.ProviderDiagnostic = .{}; - return self._open_stream_fn(self._allocator, self._io, cfg, &self._registry, &self.conversation, &diag); - } - - /// Compute the backoff delay (ms) for the just-failed `attempt` - /// (1-based). Prefers a provider `Retry-After` (capped by policy); - /// otherwise exponential `initial * multiplier^(attempt-1)`, capped, - /// with optional full jitter in `[0, delay)`. - fn backoffDelayMs( - self: *Agent, - policy: config_mod.RetryConfig, - attempt: usize, - retry_after_ms: ?u64, - ) u64 { - if (retry_after_ms) |ra| { - return @min(ra, policy.max_delay_ms); - } - const exp: f64 = @floatFromInt(attempt - 1); - const base: f64 = @as(f64, @floatFromInt(policy.initial_delay_ms)) * - std.math.pow(f64, policy.multiplier, exp); - const capped: f64 = @min(base, @as(f64, @floatFromInt(policy.max_delay_ms))); - var delay: u64 = @intFromFloat(capped); - if (policy.jitter and delay > 0) { - if (self._retry_prng == null) { - const ns = std.Io.Clock.now(.real, self._io).nanoseconds; - const seed: u64 = @truncate(@as(u128, @bitCast(@as(i128, ns)))); - self._retry_prng = std.Random.DefaultPrng.init(seed); - } - const r = self._retry_prng.?.random(); - delay = r.intRangeLessThan(u64, 0, delay + 1); - } - return delay; - } - - /// Compact the conversation: summarize an older prefix into a single - /// `.CompactionSummary` block and keep a recent suffix of whole turns - /// verbatim. Mutates `self.conversation` in place. - /// - /// This is the pure transform — it does **not** persist. The explicit - /// `/compact` entry point is the public `compact` (which persists); the - /// automatic (context-overflow) path persists via `runStep`'s turn tail - /// (the rewritten window is logged as a fresh compaction window). - /// - /// The system prompt survives untouched: all `.system`-role messages - /// are preserved in order, and no `replace` block is written. Only the - /// conversation (user/assistant) prefix is summarized. - /// - /// Per-message provider usage is read directly off the conversation - /// (`Message.usage`, set live by the provider and on replay from disk). - /// `computeSplit` uses it to size the retention window; messages - /// lacking usage fall back to word counting. - /// - /// `extra_instructions`, when non-null, is appended to the compaction - /// system prompt for this run (the `/compact $ARGUMENTS` path). - /// - /// `system_prompt` is the compaction system prompt (resolved by the - /// embedder from its `COMPACTION.md` layers, or a built-in default). - fn _compactInPlace( - self: *Agent, - system_prompt: []const u8, - extra_instructions: ?[]const u8, - ) !CompactionResult { - const conv = &self.conversation; - const messages = conv.messages.items; - - // Project per-message usage off the conversation for sizing. - const usages = try self._allocator.alloc(?conversation_Usage, messages.len); - defer self._allocator.free(usages); - for (messages, 0..) |m, i| usages[i] = m.usage; - - const split = compaction_mod.computeSplit(messages, usages, self._config.compaction.keep_verbatim); - - // Determine the active conversation start (after any prior summary). - const active_start: usize = if (conversation.latestCompactionIndex(messages)) |a| a + 1 else 0; - - // Nothing to summarize: the active conversation already fits, or the - // prefix boundary is at/under the first active turn. - if (split.prefix_end <= active_start) { - return .{ .compacted = false }; - } - // Count how many *conversation* (non-system) messages are in the - // summarized prefix. If none, this is also a no-op. - var summarized: usize = 0; - for (messages[active_start..split.prefix_end]) |m| { - if (m.role != .system) summarized += 1; - } - if (summarized == 0) return .{ .compacted = false }; - - // Serialize the prefix transcript and carry forward the latest - // existing summary (chained-compaction invariant). - const transcript = try compaction_mod.serializeTranscript( - self._allocator, - messages[active_start..split.prefix_end], - ); - defer self._allocator.free(transcript); - - const previous_summary = compaction_mod.latestSummaryText(messages); - const body = try compaction_mod.buildRequestBody(self._allocator, transcript, previous_summary); - defer self._allocator.free(body); - - const summary = try self.runCompactionRequest(system_prompt, body, extra_instructions); - defer self._allocator.free(summary.text); - - try self.rewriteWithSummary(conv, split.prefix_end, summary.text, summary.size); - - return .{ - .compacted = true, - .kept_turns = split.kept_turns, - .summarized_messages = summarized, - }; - } - - /// Compact and persist the result to the session store. This is the - /// explicit `/compact` entry point: it summarizes (via the private - /// `_compactInPlace` transform) and, if anything was compacted, appends - /// the new compaction window (summary + restated kept suffix) to the - /// store. Returns the `CompactionResult` for the embedder to report. - /// - /// `override_system_prompt`, when non-null, is the compaction system - /// prompt for this run; when null it falls back to - /// `config.compaction.compaction_prompt`. With neither set, this errors - /// (`error.NoCompactionPrompt`). - pub fn compact( - self: *Agent, - override_system_prompt: ?[]const u8, - extra_instructions: ?[]const u8, - ) !CompactionResult { - const system_prompt = override_system_prompt orelse - self._config.compaction.compaction_prompt orelse - return error.NoCompactionPrompt; - const res = try self._compactInPlace(system_prompt, extra_instructions); - // `rewriteWithSummary` rewound the high-water mark to the summary; - // flush appends the new compaction window (summary + restated suffix). - if (res.compacted) self.flushPersist(); - return res; - } - - /// Rewrite `conv.messages` to `[all system messages..., summary, - /// kept-suffix...]`. The summarized conversation prefix (everything - /// before `prefix_end` that isn't a system message) is dropped; system - /// messages survive in order; a `.CompactionSummary` user message is - /// inserted; the kept suffix (`messages[prefix_end..]`) is preserved. - fn rewriteWithSummary( - self: *Agent, - conv: *conversation.Conversation, - prefix_end: usize, - summary: []const u8, - summary_size: u64, - ) !void { - const alloc = self._allocator; - const old = conv.messages.items; - - var rebuilt: std.ArrayList(conversation.Message) = .empty; - errdefer { - for (rebuilt.items) |*m| m.deinit(alloc); - rebuilt.deinit(alloc); - } - - // 1. All system messages from the summarized prefix survive, in - // order. (System messages in the kept suffix come along with it - // below, so only scan the prefix here.) - for (old[0..prefix_end]) |*m| { - if (m.role != .system) continue; - try rebuilt.append(alloc, try cloneMessage(alloc, m.*)); - } - - // 2. The compaction summary, alone in a user message. - { - const tb = try conversation.textualBlockFromSlice(alloc, summary); - var content: std.ArrayList(conversation.ContentBlock) = .empty; - errdefer { - for (content.items) |*b| b.deinit(alloc); - content.deinit(alloc); - } - try content.append(alloc, .{ .CompactionSummary = .{ .text = tb } }); - try rebuilt.append(alloc, .{ .role = .user, .content = content }); - } - - // 3. The kept verbatim suffix, with usage recomputed so the - // restated window reads like a fresh conversation anchored at - // the summary. - // - // The post-compaction window is `[summary(user), kept_user, - // kept_assistant, ...]`. `usage.input` is *cumulative* (the whole - // prior prompt). We walk forward maintaining `running_input` — - // the synthetic cumulative prompt size as of "just before the - // next assistant output" — seeded with the summary's size: - // - // - non-assistant message → it occupies context; add its size - // (word-count heuristic, the same `messageTokenEstimate` the - // splitter uses for usage-less messages). - // - assistant message → its synthetic prompt is `running_input`. - // The whole prompt total collapses into `input` (the rewrite - // busts any provider prefix cache, so cache_read/cache_write - // are zeroed). `output`/`reasoning` are copied verbatim (real - // generation). Then `running_input += output` for the next - // turn. - // - // Assistants that had no usage stay null (we can't invent an - // output we never measured; the splitter already tolerates null). - var running_input: u64 = summary_size; - for (old[prefix_end..]) |*m| { - var cloned = try cloneMessage(alloc, m.*); - if (cloned.role == .assistant) { - if (m.usage) |u| { - cloned.usage = .{ - .input = running_input, - .output = u.output, - .cache_read = 0, - .cache_write = 0, - .reasoning = u.reasoning, - }; - running_input += u.output; - } - } else { - running_input += compaction_mod.messageTokenEstimate(m.*, null); - } - try rebuilt.append(alloc, cloned); - } - - // Swap in the rebuilt list and free the old one. - for (conv.messages.items) |*m| m.deinit(alloc); - conv.messages.deinit(alloc); - conv.messages = rebuilt; - - // The rewrite invalidated the old message indices. Rewind the - // persistence high-water mark to the inserted summary so the next - // `flushPersist` re-appends the new compaction window (summary + - // restated kept suffix) as fresh entries. - self._persisted_through = - conversation.latestCompactionIndex(conv.messages.items) orelse 0; - } - - /// Run a single compaction provider call against a throwaway - /// conversation. Returns the assistant's summary text (caller owns). - /// - /// Model selection: try `config.compaction.model` if set; on failure, - /// fall back to the active chat model. Compaction runs with an empty - /// tool registry and a single user message (the request body); no tools - /// are exposed and no session logging occurs. - fn runCompactionRequest( - self: *Agent, - system_prompt: []const u8, - body: []const u8, - extra_instructions: ?[]const u8, - ) !CompactionSummary { - const alloc = self._allocator; - - // Assemble the effective compaction system prompt (+ extra - // instructions for a `/compact $ARGUMENTS` run). - var sys_text: []const u8 = system_prompt; - var sys_owned: ?[]u8 = null; - defer if (sys_owned) |s| alloc.free(s); - if (extra_instructions) |extra| { - if (extra.len > 0) { - const combined = try std.fmt.allocPrint( - alloc, - "{s}\n\n## Additional instructions for this compaction run\n\n{s}", - .{ system_prompt, extra }, - ); - sys_owned = combined; - sys_text = combined; - } - } - - var empty_registry = ToolRegistry.init(alloc); - defer empty_registry.deinit(); - - // Try the configured compaction model first, then fall back to the - // active chat model on any failure. - if (self._config.compaction.model) |comp_provider| { - const cfg: config_mod.Config = .{ - .provider = comp_provider, - .compaction = self._config.compaction, - }; - if (self.runSingleCompactionTurn(&cfg, &empty_registry, sys_text, body)) |summary| { - return summary; - } else |err| { - std.log.warn("compaction model failed ({t}); falling back to active model", .{err}); - } - } - - const cfg: config_mod.Config = .{ - .provider = self._config.provider, - .compaction = self._config.compaction, - }; - return self.runSingleCompactionTurn(&cfg, &empty_registry, sys_text, body); - } - - /// A generated compaction summary plus the token size to attribute to - /// it when it becomes the new conversation anchor. `size` is the - /// provider-reported `output` token count for the summary turn (the - /// real generated length), falling back to the word-count heuristic - /// when the provider emitted no usage. `text` is caller-owned. - const CompactionSummary = struct { - text: []u8, - size: u64, - }; - - /// One provider call for compaction. Builds a throwaway conversation - /// (system prompt + one user message), streams a single turn through a - /// capturing receiver, and returns the assembled assistant text plus - /// the summary's token size (for usage anchoring on the rewrite). - fn runSingleCompactionTurn( - self: *Agent, - cfg: *const config_mod.Config, - registry: *const ToolRegistry, - system_prompt: []const u8, - body: []const u8, - ) !CompactionSummary { - const alloc = self._allocator; - var conv = conversation.Conversation.init(alloc); - defer conv.deinit(); - try conv.addSystemMessage(system_prompt); - try addUserText(&conv, body); - - // 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, registry, &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]; - if (last.role != .assistant) return error.CompactionNoResponse; - var out: std.ArrayList(u8) = .empty; - errdefer out.deinit(alloc); - for (last.content.items) |block| { - if (block == .Text) try out.appendSlice(alloc, block.Text.items); - } - if (out.items.len == 0) return error.CompactionEmptySummary; - - // Summary size: prefer the provider's reported output token count - // for this turn; otherwise word-count the assembled summary text - // (the same fallback the splitter uses for usage-less messages). - const size: u64 = if (last.usage) |u| - u.output - else - compaction_mod.messageTokenEstimate(last, null); - - return .{ .text = try out.toOwnedSlice(alloc), .size = size }; - } - - /// Dispatch every ToolUse block in `assistant_msg`. Groups by owning - /// registration; one OS thread per group; results assembled in the - /// original call order. - fn dispatchToolCalls( - self: *Agent, - assistant_msg: conversation.Message, - ) !void { - const conv = &self.conversation; - // Build the flat call list (in original order) and group calls - // by owning registration. - var calls: std.array_list.Managed(FlatCall) = .init(self._allocator); - defer calls.deinit(); - - for (assistant_msg.content.items) |block| { - if (block != .ToolUse) continue; - const tu = block.ToolUse; - if (!isValidToolInput(tu.input.items)) { - try calls.append(.{ - .tool_use_id = tu.id, - .tool_name = tu.name, - .input = tu.input.items, - .entry = null, - .result = try invalidInputResult(self._allocator, tu.input.items), - .err = null, - .is_error = true, - }); - continue; - } - const entry = self._registry.lookup(tu.name) orelse { - // Unknown tool: don't abort. Synthesize an error result so - // the model can correct, and so this ToolUse still gets its - // matching ToolResult (providers reject a follow-up request - // otherwise). - try calls.append(.{ - .tool_use_id = tu.id, - .tool_name = tu.name, - .input = tu.input.items, - .entry = null, - .result = try toolErrorResult(self._allocator, tu.name, error.UnknownTool), - .err = null, - .is_error = true, - }); - continue; - }; - try calls.append(.{ - .tool_use_id = tu.id, - .tool_name = tu.name, - .input = tu.input.items, - .entry = entry.entry, - .result = null, - .err = null, - }); - } - std.debug.assert(calls.items.len > 0); - - // Partition into groups. A group's `kind` determines how it - // runs; the `member_indices` are positions into `calls` (the - // original call order) so we can write back results without - // re-ordering. - var groups: std.array_list.Managed(Group) = .init(self._allocator); - defer { - for (groups.items) |*g| g.deinit(self._allocator); - groups.deinit(); - } - try buildGroups(self._allocator, calls.items, &groups); - - // Spawn one concurrent task per group via `std.Io.Group`. - // Single-tool groups run the tool's vtable; source groups run - // the source's `invoke_batch`. We use `concurrent` rather than - // `async` because tool work may block on I/O — under a - // single-threaded `Io` `async` would deadlock; `concurrent` - // forces real concurrency (or `error.ConcurrencyUnavailable`). - var task_group: Io.Group = .init; - // `cancel` is idempotent with `await`; if anything below this - // point errors before we successfully `await`, this releases - // the group's resources. - defer task_group.cancel(self._io); - errdefer { - for (calls.items) |*c| { - if (c.result) |r| r.deinit(self._allocator); - } - } - - // Try real concurrency first. If the `Io` implementation can't - // provide it (`error.ConcurrencyUnavailable`), fall back to running - // every group sequentially on this thread — tool batches are small - // (rarely more than a handful of calls) so the serial path is a fine - // safety net rather than a hard failure. - var ran_concurrently = true; - for (groups.items) |*g| { - task_group.concurrent(self._io, runGroup, .{ self, g, calls.items }) catch |e| { - if (e == error.ConcurrencyUnavailable) { - ran_concurrently = false; - break; - } - return e; - }; - } - if (ran_concurrently) { - // `error.Canceled` here means cancellation propagated into this - // dispatch from above; surface it like any other error. - try task_group.await(self._io); - } else { - // Cancel any tasks that were spawned before the failure, then - // run all groups serially. Only entry-bearing calls are touched - // by `runGroup`; the pre-seeded error results (unknown tool, - // invalid input) have `entry == null` and must be left intact. - task_group.cancel(self._io); - for (calls.items) |*c| { - if (c.entry == null) continue; - if (c.result) |r| r.deinit(self._allocator); - c.result = null; - c.err = null; - } - for (groups.items) |*g| runGroup(self, g, calls.items); - } - - // Pre-pass: resolve worker-reported errors. A hard host failure - // (cancellation, OOM) aborts the whole turn. Every other failure is - // converted into a model-visible error `ToolResult` so the model can - // recover and so each `ToolUse` keeps its matching `ToolResult` - // (providers reject the next request otherwise). - for (calls.items) |*c| { - const e = c.err orelse continue; - if (classifyToolError(e) == .hard_fail) return e; - // Replace any partial result with a synthesized error result. - if (c.result) |r| { - r.deinit(self._allocator); - c.result = null; - } - c.result = try toolErrorResult(self._allocator, c.tool_name, e); - c.err = null; - c.is_error = true; - } - - // Assemble ToolResult blocks in original call order. - var content: std.ArrayList(conversation.ContentBlock) = .empty; - errdefer { - for (content.items) |*b| b.deinit(self._allocator); - content.deinit(self._allocator); - } - try content.ensureTotalCapacity(self._allocator, calls.items.len); - - for (calls.items) |*c| { - const result_parts = c.result orelse { - // Internal invariant: every call should now have a result - // (success, synthesized error, or pre-seeded error). - return error.MissingToolResult; - }; - c.result = null; // ownership transferred below - defer result_parts.deinit(self._allocator); - - const id_copy = try self._allocator.dupe(u8, c.tool_use_id); - errdefer self._allocator.free(id_copy); - - var stored: std.ArrayList(conversation.ResultPartStored) = .empty; - errdefer { - for (stored.items) |*p| p.deinit(self._allocator); - stored.deinit(self._allocator); - } - try stored.ensureTotalCapacity(self._allocator, result_parts.items.len); - for (result_parts.items) |part| { - switch (part) { - .text => |t| { - var buf: conversation.TextualBlock = .empty; - errdefer buf.deinit(self._allocator); - try buf.appendSlice(self._allocator, t); - stored.appendAssumeCapacity(.{ .text = buf }); - }, - .media => |m| { - // libpanto owns the heavy lifting: detect the type - // (when the tool gave no hint), resize large - // rasters, then base64-encode for storage. Tools - // hand over raw bytes only. - const processed = image_mod.process(self._allocator, m.data, m.media_type) catch |e| { - // Media processing failure: keep the turn alive by - // dropping the attachment and noting it as text, - // rather than aborting. `UnknownMediaType` gets a - // friendly note; other failures name the error. - var note: conversation.TextualBlock = .empty; - errdefer note.deinit(self._allocator); - if (e == error.UnknownMediaType) { - try note.appendSlice(self._allocator, "[unrecognized binary attachment dropped]"); - } else { - const txt = try std.fmt.allocPrint( - self._allocator, - "[media attachment dropped: {s}]", - .{@errorName(e)}, - ); - defer self._allocator.free(txt); - try note.appendSlice(self._allocator, txt); - } - stored.appendAssumeCapacity(.{ .text = note }); - continue; - }; - defer self._allocator.free(processed.data); - - const mt = try self._allocator.dupe(u8, processed.media_type); - errdefer self._allocator.free(mt); - - const enc = std.base64.standard.Encoder; - var buf: conversation.TextualBlock = .empty; - errdefer buf.deinit(self._allocator); - try buf.resize(self._allocator, enc.calcSize(processed.data.len)); - _ = enc.encode(buf.items, processed.data); - - stored.appendAssumeCapacity(.{ .media = .{ .media_type = mt, .data = buf } }); - }, - } - } - - content.appendAssumeCapacity(.{ .ToolResult = .{ - .tool_use_id = id_copy, - .parts = stored, - .is_error = c.is_error, - } }); - } - - try conv.messages.append(self._allocator, .{ - .role = .user, - .content = content, - }); - } -}; - -/// 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 { - // Internal state (underscore-prefixed by convention: not part of the - // public surface, even though Zig fields are always reachable). The only - // intended-public field is `state`, the transparent turn state machine. - _agent: *Agent, - _queue: stream_mod.EventQueue, - state: State, - /// The active provider response, when in `.streaming`. - _response: ?provider_mod.ProviderStream = null, - /// First message index of this turn (the boundary `cancel` rolls back to). - _start: usize, - /// 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, - /// Count of mid-stream retries attempted for the CURRENT response. Unlike - /// `openWithRetries` (which loops internally with a local counter), a - /// mid-stream failure unwinds back to `.turn_start` and re-opens, so the - /// attempt count must live on the stream to survive across re-opens. - /// Reset to 0 each time a response completes. A persistently-broken - /// stream would otherwise re-open forever without ever exhausting - /// `retry.max_attempts`. - _stream_retries: usize = 0, - /// Owned copy of the provider's diagnostic message for the in-flight - /// mid-stream retry notice. Borrowed by the queued `provider_retry` - /// event, so it must outlive that event: freed on the next mid-stream - /// failure and on `deinit`. - _retry_message: ?[]u8 = null, - - pub const State = 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, - /// Tool dispatch has been announced to callers; run the blocking - /// dispatch on the next pull so `tool_dispatch_start` is observable - /// before long-running tools complete. - dispatching_tools, - /// 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), - .state = .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. - // `next()` already flushes at every boundary, so this is usually a - // no-op; it catches a stream dropped without a final `next()`. - self._agent.flushPersist(); - if (self._response) |ps| ps.deinit(); - if (self._retry_message) |m| self._agent._allocator.free(m); - self._queue.deinit(); - self._agent._allocator.destroy(self); - } - - /// Abort this turn from the embedder side (for example, Escape in a TUI). - /// Provider/tool work that is currently inside `next()` cannot be unwound - /// until that call returns, but once control reaches the embedder this - /// drops every assistant/tool-result message produced by this stream before - /// `deinit()` has a chance to persist them. The already-persisted user - /// prompt (written by `Agent.run`) is intentionally kept, leaving a valid - /// conversation prefix with no dangling ToolUse blocks. - pub fn cancel(self: *Stream) void { - if (self._response) |ps| { - ps.deinit(); - self._response = null; - } - const conv = &self._agent.conversation; - while (conv.messages.items.len > self._start) { - var msg = conv.messages.pop().?; - msg.deinit(conv.allocator); - } - // Rolled the conversation back; clamp the persistence mark to match. - self._agent.flushPersist(); - self.state = .done; - } - - /// Reset a `.failed` stream back to `.turn_start` so the caller can resume - /// `next()` after changing the agent config (e.g. an embedder that catches - /// a terminal `ProviderBadRequest`, rewrites the provider config, and - /// wants to retry the SAME turn without re-appending the user message). - /// - /// Safe only at the OPEN boundary: a failed open never mutates the - /// conversation, so the already-committed user prompt at `_start` stays - /// intact and the re-open re-reads the (now-swapped) config snapshot. - /// Returns `error.StreamNotFailed` if the stream is not in `.failed`. - pub fn reopen(self: *Stream) !void { - if (self.state != .failed) return error.StreamNotFailed; - if (self._response) |ps| { - ps.deinit(); - self._response = null; - } - self._pending_error = null; - self.state = .turn_start; - } - - /// Pull the next event, or null past the terminal. See the contract - /// above. - pub fn next(self: *Stream) !?Event { - // Bring persistence current on every return path (events, terminal, - // errors). This is the interruptibility guarantee: after any `next()` - // the store reflects all committed, coherent messages, so the caller - // can `break` and immediately `compact()` or resume with a new turn. - defer self._agent.flushPersist(); - // 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.state = .failed; - return err; - } - - while (true) { - switch (self.state) { - .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.state = .failed; - return err; - }; - self._response = ps; - self.state = .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. - // - // Capture the provider's diagnostic message (e.g. an - // Anthropic `overloaded_error`) BEFORE deinit frees - // it, dup'ing it so it outlives the response. Owned - // here and freed after the retry notice is queued. - if (self._retry_message) |old| { - self._agent._allocator.free(old); - self._retry_message = null; - } - if (ps.lastError()) |m| { - self._retry_message = self._agent._allocator.dupe(u8, m) catch null; - } - ps.deinit(); - self._response = null; - if (!provider_mod.isRetryableProviderError(err)) { - self.state = .failed; - return err; - } - // Out of attempts: hard-fail with the last error - // instead of re-opening forever. `_stream_retries` - // counts mid-stream retries already made for this - // response; the initial attempt is the open itself, - // so the Nth retry is attempt N+1. - const cfg = self._agent._config; - self._stream_retries += 1; - if (self._stream_retries >= cfg.retry.max_attempts) { - self.state = .failed; - return err; - } - self.state = .turn_start; - // Emit a retry notice so consumers see the stall. - const delay_ms = self._agent.backoffDelayMs(cfg.retry, self._stream_retries, null); - self._queue.push(.{ .provider_retry = .{ - .attempt = self._stream_retries, - .max_attempts = cfg.retry.max_attempts, - .delay_ms = delay_ms, - .err = err, - .message = self._retry_message, - } }) catch |e| { - self.state = .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.state = .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; - // A response completed: reset the mid-stream retry - // budget so a later turn starts fresh. - self._stream_retries = 0; - self.state = .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.state = .failed; - return error.EmptyAssistantResponse; - } - - if (!Agent.hasToolUseBlock(last)) { - self.state = .done; - return .turn_complete; - } - - // Announce tool dispatch and return immediately. The - // dispatch itself can block for a long time (especially - // with parallel tools); yielding this boundary event first - // lets UIs/renderers show all tool calls as running instead - // of appearing frozen until the slowest tool completes. - const count = toolUseCount(last); - self._queue.push(.{ .tool_dispatch_start = .{ .count = count } }) catch |e| { - self.state = .failed; - return e; - }; - self.state = .dispatching_tools; - if (self._queue.pop()) |ev| return ev; - }, - .dispatching_tools => { - const conv = &self._agent.conversation; - const last = conv.messages.items[conv.messages.items.len - 1]; - std.debug.assert(last.role == .assistant); - std.debug.assert(Agent.hasToolUseBlock(last)); - - self._agent.dispatchToolCalls(last) catch |err| { - self.state = .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.state = .failed; - return e; - }; - self.state = .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. -const FlatCall = struct { - tool_use_id: []const u8, // borrowed from assistant_msg - tool_name: []const u8, // borrowed from assistant_msg - input: []const u8, // borrowed from assistant_msg - entry: ?Entry, - - /// Owned result parts from `Tool.invoke` or `ToolSource.invoke_batch`. - /// Allocated with the agent's allocator. Transferred into a - /// ToolResultBlock on success. - result: ?tool_mod.ResultParts, - - /// If non-null, the worker reported a failure for this call. After - /// dispatch it is classified: host failures abort the turn, everything - /// else is converted into an error `ToolResult`. - err: ?anyerror, - - /// True when `result` already holds a synthesized error result (unknown - /// tool, invalid input). Worker-reported `err`s are folded into this - /// during assembly. - is_error: bool = false, -}; - -/// One dispatch group. Either a single Tool invocation, or a batch of -/// calls headed to one ToolSource. -const Group = union(enum) { - single: SingleGroup, - source: SourceGroup, - - pub const SingleGroup = struct { - tool: Tool, - /// Index into the flat calls array. - call_index: usize, - }; - - pub const SourceGroup = struct { - source: *ToolSource, - /// Indices into the flat calls array. Owned by the group. - member_indices: []usize, - }; - - fn deinit(self: *Group, allocator: Allocator) void { - switch (self.*) { - .single => {}, - .source => |sg| allocator.free(sg.member_indices), - } - } -}; - -/// Partition the flat call list into groups. Order of groups is -/// arbitrary; order within a `source` group preserves the original -/// call order so that batch results can be written back positionally. -fn buildGroups( - allocator: Allocator, - calls: []const FlatCall, - out: *std.array_list.Managed(Group), -) !void { - // Map from source pointer to the index of its group in `out`. - // Buffers per source, accumulated then frozen into slices. - var pending: std.AutoHashMap(*ToolSource, std.array_list.Managed(usize)) = - .init(allocator); - defer { - var it = pending.valueIterator(); - while (it.next()) |l| l.deinit(); - pending.deinit(); - } - - for (calls, 0..) |c, i| { - const ent = c.entry orelse continue; - switch (ent) { - .single => |t| try out.append(.{ .single = .{ .tool = t, .call_index = i } }), - .source => |sr| { - const gop = try pending.getOrPut(sr.source); - if (!gop.found_existing) { - gop.value_ptr.* = std.array_list.Managed(usize).init(allocator); - } - try gop.value_ptr.append(i); - }, - } - } - - // Freeze each pending list into a source-group entry. We move - // ownership of the indices into `Group.source.member_indices`. - var pit = pending.iterator(); - while (pit.next()) |entry| { - const src = entry.key_ptr.*; - const indices = try entry.value_ptr.toOwnedSlice(); - try out.append(.{ .source = .{ .source = src, .member_indices = indices } }); - } -} - -/// Worker entry point. Runs one group to completion, populating -/// `calls[i].result` or `calls[i].err` for each member call. -/// -/// Return type is `void`, which coerces to `Io.Cancelable!void` as -/// required by `Group.concurrent`. Tool errors are reported via -/// `FlatCall.err`, not by returning from this function. -fn runGroup(agent: *Agent, group: *Group, calls: []FlatCall) void { - switch (group.*) { - .single => |sg| { - const i = sg.call_index; - const c = &calls[i]; - const out = sg.tool.vtable.invoke(sg.tool.ctx, c.input, agent._allocator) catch |e| { - c.err = e; - return; - }; - c.result = out; - }, - .source => |sg| runSourceGroup(agent, sg, calls), - } -} - -fn runSourceGroup(agent: *Agent, sg: Group.SourceGroup, calls: []FlatCall) void { - const n = sg.member_indices.len; - - const batch_calls = agent._allocator.alloc(tool_source_mod.Call, n) catch |e| { - for (sg.member_indices) |i| calls[i].err = e; - return; - }; - defer agent._allocator.free(batch_calls); - - const batch_results = agent._allocator.alloc(tool_source_mod.CallResult, n) catch |e| { - for (sg.member_indices) |i| calls[i].err = e; - return; - }; - defer agent._allocator.free(batch_results); - - for (sg.member_indices, 0..) |idx, j| { - batch_calls[j] = .{ - .tool_name = calls[idx].tool_name, - .input = calls[idx].input, - }; - batch_results[j] = .{ .err = error.SourceDroppedCall }; - } - - sg.source.vtable.invoke_batch( - sg.source.ctx, - batch_calls, - batch_results, - agent._allocator, - ) catch |e| { - // Whole-batch failure: free any partial successes the source - // already wrote, then mark every member as failed. - for (batch_results) |r| switch (r) { - .ok => |b| b.deinit(agent._allocator), - .err => {}, - }; - for (sg.member_indices) |i| calls[i].err = e; - return; - }; - - // Per-call success/error. - for (sg.member_indices, 0..) |i, j| { - switch (batch_results[j]) { - .ok => |b| calls[i].result = b, - .err => |e| calls[i].err = e, - } - } -} - -// ----------------------------------------------------------------------------- -// Tests -// ----------------------------------------------------------------------------- - -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 runUserText(agent, text); - defer s.deinit(); - while (try s.next()) |_| {} -} - -/// Test helper: open a turn from a single user `.Text` block, mirroring the -/// old text-only `run(.{ .text = ... })`. The block is adopted by the agent's -/// conversation. -fn runUserText(agent: *Agent, text: []const u8) !*Stream { - var blocks = [_]conversation.ContentBlock{ - .{ .Text = try conversation.textualBlockFromSlice(agent._allocator, text) }, - }; - return agent.run(.{ .blocks = &blocks }); -} - -/// Test helper: the items of a ToolResultBlock's first text part. -fn trText(tr: conversation.ToolResultBlock) []const u8 { - for (tr.parts.items) |p| { - if (p == .text) return p.text.items; - } - return ""; -} - -/// Test harness for the injectable `_open_stream_fn` seam. -/// -/// `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 `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 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 open. Used to drive the - /// transient-retry path. - scripted_errors: []const ScriptedError = &.{}, - error_idx: usize = 0, - /// A queue of MID-STREAM errors. Unlike `scripted_errors` (which fail at - /// open time), each entry here lets the open succeed but makes the - /// returned response fail on its first `produce`, exercising the - /// `.streaming` retry path. Consumed in order, one per open. - stream_errors: []const StreamFailure = &.{}, - stream_error_idx: usize = 0, - /// Count of opens observed (failed + succeeded). Lets tests assert the - /// exact number of attempts. - calls_made: usize = 0, - - const ScriptedError = struct { - err: anyerror, - status_code: ?u16 = null, - retry_after_ms: ?u64 = null, - }; - - const StreamFailure = struct { - err: anyerror, - /// Optional provider diagnostic surfaced via `ProviderStream.lastError`. - message: ?[]const u8 = null, - }; - - const ScriptedTurn = struct { - blocks: []const TestBlock, - /// Optional provider usage to stamp on the committed assistant - /// message (mirrors a real provider's terminal usage). - usage: ?conversation.Usage = null, - }; - - const TestBlock = union(enum) { - Text: []const u8, - ToolUse: struct { - id: []const u8, - name: []const u8, - input: []const u8, - }, - }; - - /// Point the global seam at this stub and return the function to assign - /// to `agent._open_stream_fn`. Call once per test, after constructing the - /// stub on the stack. - fn install(self: *StubProvider) provider_mod.OpenStreamFn { - stub_active = self; - return stubOpenStream; - } -}; - -/// 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, - /// If set, the first `produce` returns this error instead of committing - /// (simulates a mid-stream provider failure after a successful open). - produce_error: ?anyerror = null, - /// Optional provider diagnostic surfaced via `lastError` after the - /// mid-stream failure (borrowed; static test strings). - error_message: ?[]const u8 = null, - - 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 }; - } - - fn createFailing( - allocator: Allocator, - conv: *conversation.Conversation, - err: anyerror, - message: ?[]const u8, - ) !provider_mod.ProviderStream { - const self = try allocator.create(StubResponse); - self.* = .{ - .allocator = allocator, - .conv = conv, - .turn = .{ .blocks = &.{} }, - .produce_error = err, - .error_message = message, - }; - return .{ .ptr = self, .vtable = &vtable }; - } - - const vtable: provider_mod.ProviderStream.VTable = .{ - .produce = produceVT, - .deinit = deinitVT, - .last_error = lastErrorVT, - }; - - fn lastErrorVT(ptr: *anyopaque) ?[]const u8 { - const self: *StubResponse = @ptrCast(@alignCast(ptr)); - return self.error_message; - } - - fn produceVT(ptr: *anyopaque, out: *stream_mod.EventQueue) anyerror!provider_mod.ProviderStream.ProduceStatus { - const self: *StubResponse = @ptrCast(@alignCast(ptr)); - if (self.produce_error) |err| return err; - 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.addAssistantMessage(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, - _: *const ToolRegistry, - conv: *conversation.Conversation, - diag: ?*provider_mod.ProviderDiagnostic, -) anyerror!provider_mod.ProviderStream { - const self = stub_active orelse return error.NoStubInstalled; - self.calls_made += 1; - if (self.error_idx < self.scripted_errors.len) { - const e = self.scripted_errors[self.error_idx]; - self.error_idx += 1; - if (diag) |d| { - d.status_code = e.status_code; - d.retry_after_ms = e.retry_after_ms; - } - return e.err; - } - if (self.stream_error_idx < self.stream_errors.len) { - const f = self.stream_errors[self.stream_error_idx]; - self.stream_error_idx += 1; - return StubResponse.createFailing(allocator, conv, f.err, f.message); - } - if (self.overflow_calls > 0) { - self.overflow_calls -= 1; - return error.ContextOverflow; - } - if (self.next >= self.scripted.len) return error.NoMoreScriptedTurns; - const turn = self.scripted[self.next]; - self.next += 1; - return StubResponse.create(allocator, conv, turn); -} - -/// Build a stack registry + active `Config` snapshot for tests that drive -/// the agent. Post-R1 the registry no longer lives on `Config` — it lives -/// on the `Agent`. The harness still owns a registry so tests can pre-stage -/// tools and copy them onto the agent (`seed`) after `init`. The caller -/// owns both and must keep them alive for the agent's lifetime. -const TestHarness = struct { - registry: ToolRegistry, - config: config_mod.Config, - - fn init(allocator: Allocator) TestHarness { - return .{ .registry = ToolRegistry.init(allocator), .config = undefined }; - } - - /// Finalize the config snapshot. Must be called after `init` and before - /// constructing the agent, once the harness has a stable address. - fn activate(self: *TestHarness) void { - self.config = .{ - .provider = .{ .openai_chat = .{ .api_key = "k", .base_url = "u", .model = "m" } }, - }; - } - - /// Move the tools pre-staged on the harness registry onto a freshly - /// `init`ed agent's own registry. Post-R1 the agent owns its tool set, - /// so tests stage tools on the harness then transplant them here. The - /// agent's empty registry is freed and replaced; the harness registry - /// is left empty (its `deinit` becomes a no-op free). - fn seedInto(self: *TestHarness, agent: *Agent) void { - agent._registry.deinit(); - agent._registry = self.registry; - self.registry = ToolRegistry.init(self.registry.allocator); - } - - fn deinit(self: *TestHarness) void { - self.registry.deinit(); - } -}; - -const EchoTool = struct { - prefix_owned: []u8, - name_owned: []u8, - - fn create(allocator: Allocator, name: []const u8, prefix: []const u8) !Tool { - const self = try allocator.create(EchoTool); - errdefer allocator.destroy(self); - self.name_owned = try allocator.dupe(u8, name); - errdefer allocator.free(self.name_owned); - self.prefix_owned = try allocator.dupe(u8, prefix); - return .{ - .decl = .{ - .name = self.name_owned, - .description = "echo", - .schema_json = "{}", - }, - .ctx = self, - .vtable = &vt, - }; - } - - const vt: Tool.VTable = .{ .invoke = invoke, .deinit = deinit }; - - fn invoke(ctx: *anyopaque, input: []const u8, allocator: Allocator) anyerror!tool_mod.ResultParts { - const self: *EchoTool = @ptrCast(@alignCast(ctx)); - const msg = try std.fmt.allocPrint(allocator, "{s}{s}", .{ self.prefix_owned, input }); - return tool_mod.ResultParts.fromTextOwned(allocator, msg); - } - - fn deinit(ctx: *anyopaque, allocator: Allocator) void { - const self: *EchoTool = @ptrCast(@alignCast(ctx)); - allocator.free(self.name_owned); - allocator.free(self.prefix_owned); - allocator.destroy(self); - } -}; - -const BarrierTool = struct { - name_owned: []u8, - barrier: *Barrier, - - const Barrier = struct { - target: u32, - arrived: std.atomic.Value(u32) = .init(0), - thread_ids: [4]std.atomic.Value(u64) = .{ - .init(0), .init(0), .init(0), .init(0), - }, - }; - - fn create(allocator: Allocator, name: []const u8, barrier: *Barrier) !Tool { - const self = try allocator.create(BarrierTool); - errdefer allocator.destroy(self); - self.name_owned = try allocator.dupe(u8, name); - self.barrier = barrier; - return .{ - .decl = .{ - .name = self.name_owned, - .description = "barrier", - .schema_json = "{}", - }, - .ctx = self, - .vtable = &vt, - }; - } - - const vt: Tool.VTable = .{ .invoke = invoke, .deinit = deinit }; - - fn invoke(ctx: *anyopaque, _: []const u8, allocator: Allocator) anyerror!tool_mod.ResultParts { - const self: *BarrierTool = @ptrCast(@alignCast(ctx)); - const arrived = self.barrier.arrived.fetchAdd(1, .acq_rel); - if (arrived < self.barrier.thread_ids.len) { - self.barrier.thread_ids[arrived].store(std.Thread.getCurrentId(), .release); - } - - var i: usize = 0; - while (self.barrier.arrived.load(.acquire) < self.barrier.target) : (i += 1) { - if (i > 50_000) return error.BarrierTimeout; - std.Thread.yield() catch {}; - } - return tool_mod.ResultParts.fromText(allocator, "done"); - } - - fn deinit(ctx: *anyopaque, allocator: Allocator) void { - const self: *BarrierTool = @ptrCast(@alignCast(ctx)); - allocator.free(self.name_owned); - allocator.destroy(self); - } -}; - -const FailingTool = struct { - name_owned: []u8, - - fn create(allocator: Allocator, name: []const u8) !Tool { - const self = try allocator.create(FailingTool); - errdefer allocator.destroy(self); - self.name_owned = try allocator.dupe(u8, name); - return .{ - .decl = .{ - .name = self.name_owned, - .description = "fails", - .schema_json = "{}", - }, - .ctx = self, - .vtable = &vt, - }; - } - - const vt: Tool.VTable = .{ .invoke = invoke, .deinit = deinit }; - - fn invoke(_: *anyopaque, _: []const u8, _: Allocator) anyerror!tool_mod.ResultParts { - return error.ToolExploded; - } - - fn deinit(ctx: *anyopaque, allocator: Allocator) void { - const self: *FailingTool = @ptrCast(@alignCast(ctx)); - allocator.free(self.name_owned); - allocator.destroy(self); - } -}; - -/// A tool that returns a hard host failure (`error.Canceled`), which must -/// abort the whole turn rather than degrade into a tool result. -const HardFailTool = struct { - name_owned: []u8, - - fn create(allocator: Allocator, name: []const u8) !Tool { - const self = try allocator.create(HardFailTool); - errdefer allocator.destroy(self); - self.name_owned = try allocator.dupe(u8, name); - return .{ - .decl = .{ .name = self.name_owned, .description = "hard fail", .schema_json = "{}" }, - .ctx = self, - .vtable = &vt, - }; - } - - const vt: Tool.VTable = .{ .invoke = invoke, .deinit = deinit }; - - fn invoke(_: *anyopaque, _: []const u8, _: Allocator) anyerror!tool_mod.ResultParts { - return error.Canceled; - } - - fn deinit(ctx: *anyopaque, allocator: Allocator) void { - const self: *HardFailTool = @ptrCast(@alignCast(ctx)); - allocator.free(self.name_owned); - allocator.destroy(self); - } -}; - -/// An in-memory `SessionStore` test double: records every appended -/// `StoredMessage` (role + provider/model stamp) so tests can assert the -/// agent persisted the right turn without touching disk. Honors the store -/// ownership contract by freeing each consumed message after recording its -/// salient fields. -const CapturingStore = struct { - allocator: Allocator, - roles: std.ArrayList(conversation.MessageRole) = .empty, - base_urls: std.ArrayList([]const u8) = .empty, - - fn init(allocator: Allocator) CapturingStore { - return .{ .allocator = allocator }; - } - - fn deinit(self: *CapturingStore) void { - for (self.base_urls.items) |s| self.allocator.free(s); - self.base_urls.deinit(self.allocator); - self.roles.deinit(self.allocator); - } - - fn createVT(ctx: *anyopaque) session_store_mod.Session { - const self: *CapturingStore = @ptrCast(@alignCast(ctx)); - const a = self.allocator; - const info: session_store_mod.SessionInfo = .{ - .id = a.dupe(u8, "cap") catch "cap", - .created = a.dupe(u8, "") catch "", - .modified = a.dupe(u8, "") catch "", - .message_count = 0, - .last_user_message = a.dupe(u8, "") catch "", - .api_style = .openai_chat, - .base_url = a.dupe(u8, "") catch "", - .model = a.dupe(u8, "") catch "", - .reasoning = .default, - }; - return .{ .info = info, .store = self.store() }; - } - - fn listVT(ctx: *anyopaque) anyerror![]session_store_mod.SessionInfo { - const self: *CapturingStore = @ptrCast(@alignCast(ctx)); - return self.allocator.alloc(session_store_mod.SessionInfo, 0); - } - fn freeSessionInfosVT(ctx: *anyopaque, infos: []session_store_mod.SessionInfo) void { - const self: *CapturingStore = @ptrCast(@alignCast(ctx)); - for (infos) |i| i.deinit(self.allocator); - self.allocator.free(infos); - } - fn resolveVT(_: *anyopaque, _: []const u8) anyerror!?session_store_mod.Session { - return null; - } - fn latestVT(_: *anyopaque) anyerror!?session_store_mod.Session { - return null; - } - fn loadVT(_: *anyopaque, _: []const u8) anyerror!?conversation.Conversation { - return null; - } - - fn appendMessagesVT( - ctx: *anyopaque, - _: []const u8, - messages: []session_store_mod.PersistentMessage, - ) anyerror!void { - const self: *CapturingStore = @ptrCast(@alignCast(ctx)); - for (messages) |m| { - try self.roles.append(self.allocator, m.message.role); - try self.base_urls.append(self.allocator, try self.allocator.dupe(u8, m.identity.base_url)); - } - } - - const vtable: session_store_mod.SessionStore.VTable = .{ - .create = createVT, - .list = listVT, - .freeSessionInfos = freeSessionInfosVT, - .resolve = resolveVT, - .latest = latestVT, - .load = loadVT, - .appendMessages = appendMessagesVT, - }; - - fn store(self: *CapturingStore) session_store_mod.SessionStore { - return .{ .ptr = self, .vtable = &vtable }; - } -}; - -test "agent persists user, assistant, and tool-result messages of a turn" { - const allocator = testing.allocator; - - const scripted = [_]StubProvider.ScriptedTurn{ - .{ .blocks = &.{ - .{ .ToolUse = .{ .id = "tc_1", .name = "echo", .input = "hello" } }, - } }, - .{ .blocks = &.{ - .{ .Text = "ok" }, - } }, - }; - var stub = StubProvider{ .allocator = allocator, .scripted = &scripted }; - var threaded: std.Io.Threaded = .init(allocator, .{}); - defer threaded.deinit(); - const io = threaded.io(); - var h = TestHarness.init(allocator); - defer h.deinit(); - try h.registry.register(try EchoTool.create(allocator, "echo", "ECHO:")); - h.activate(); - - var cap = CapturingStore.init(allocator); - defer cap.deinit(); - const agent = try Agent.init(allocator, io, &h.config, cap.store().create(), null); - defer agent.deinit(); - h.seedInto(agent); - agent._open_stream_fn = stub.install(); - - try drainTurn(agent, "call a tool"); - - // Persisted, in order: user prompt, assistant(ToolUse), user(ToolResult), - // assistant(text). - try testing.expectEqual(@as(usize, 4), cap.roles.items.len); - try testing.expectEqual(conversation.MessageRole.user, cap.roles.items[0]); - try testing.expectEqual(conversation.MessageRole.assistant, cap.roles.items[1]); - try testing.expectEqual(conversation.MessageRole.user, cap.roles.items[2]); - try testing.expectEqual(conversation.MessageRole.assistant, cap.roles.items[3]); - - // The wire identity (base_url from the active config) rode through on - // every entry. - for (cap.base_urls.items) |b| { - try testing.expectEqualStrings("u", b); - } -} - -test "interruption: persistence stays current and excludes a dangling tool call" { - const allocator = testing.allocator; - - const scripted = [_]StubProvider.ScriptedTurn{ - .{ .blocks = &.{ - .{ .ToolUse = .{ .id = "tc_1", .name = "echo", .input = "hello" } }, - } }, - .{ .blocks = &.{ - .{ .Text = "ok" }, - } }, - }; - var stub = StubProvider{ .allocator = allocator, .scripted = &scripted }; - var threaded: std.Io.Threaded = .init(allocator, .{}); - defer threaded.deinit(); - const io = threaded.io(); - var h = TestHarness.init(allocator); - defer h.deinit(); - try h.registry.register(try EchoTool.create(allocator, "echo", "ECHO:")); - h.activate(); - - var cap = CapturingStore.init(allocator); - defer cap.deinit(); - const agent = try Agent.init(allocator, io, &h.config, cap.store().create(), null); - defer agent.deinit(); - h.seedInto(agent); - agent._open_stream_fn = stub.install(); - - var s = try runUserText(agent, "call a tool"); - - // `run` persisted the user prompt up front, and nothing else yet. - try testing.expectEqual(@as(usize, 1), cap.roles.items.len); - try testing.expectEqual(conversation.MessageRole.user, cap.roles.items[0]); - - // Pull until the assistant's tool-call message completes, then break — the - // embedder's interruption point (e.g. to handle the tool calls itself). - while (try s.next()) |ev| { - if (ev == .message_complete) break; - } - - // The tool-call message is live in the conversation... - const msgs = agent.conversation.messages.items; - try testing.expectEqual(conversation.MessageRole.assistant, msgs[msgs.len - 1].role); - try testing.expect(Agent.hasToolUseBlock(msgs[msgs.len - 1])); - // ...but it is a dangling tool call (no results), so the store still holds - // only the coherent prefix: the interruption guarantee. - try testing.expectEqual(@as(usize, 1), cap.roles.items.len); - - // Dropping the stream mid-turn must not persist the dangling call either. - s.deinit(); - try testing.expectEqual(@as(usize, 1), cap.roles.items.len); - - // The conversation is coherent enough to resume from: feeding the tool - // result as a fresh user turn resolves the dangling call, and now both the - // assistant tool-call and the user result become persistable. - var parts: std.ArrayList(conversation.ResultPartStored) = .empty; - try parts.append(allocator, .{ .text = try conversation.textualBlockFromSlice(allocator, "42") }); - var s2 = try agent.run(.{ .blocks = &.{ - .{ .ToolResult = .{ - .tool_use_id = try allocator.dupe(u8, "tc_1"), - .parts = parts, - .is_error = false, - } }, - } }); - defer s2.deinit(); - // user prompt, assistant(ToolUse), user(ToolResult) are all persisted once - // the dangling call is resolved by the new user turn. - try testing.expectEqual(@as(usize, 3), cap.roles.items.len); - try testing.expectEqual(conversation.MessageRole.user, cap.roles.items[0]); - try testing.expectEqual(conversation.MessageRole.assistant, cap.roles.items[1]); - try testing.expectEqual(conversation.MessageRole.user, cap.roles.items[2]); -} - -test "agent runs a turn against NullStore without persisting or erroring" { - const allocator = testing.allocator; - - const scripted = [_]StubProvider.ScriptedTurn{ - .{ .blocks = &.{.{ .Text = "hi" }} }, - }; - var stub = StubProvider{ .allocator = allocator, .scripted = &scripted }; - var threaded: std.Io.Threaded = .init(allocator, .{}); - defer threaded.deinit(); - const io = threaded.io(); - var h = TestHarness.init(allocator); - defer h.deinit(); - h.activate(); - - var ns = null_store_mod.NullStore.init(allocator); - const agent = try Agent.init(allocator, io, &h.config, ns.store().create(), null); - defer agent.deinit(); - h.seedInto(agent); - agent._open_stream_fn = stub.install(); - - try drainTurn(agent, "hello"); - - // Nothing crashed; the conversation has the user + assistant messages. - try testing.expectEqual(@as(usize, 2), agent.conversation.messages.items.len); -} - -test "override mutators rewrite the conversation tail's tool blocks" { - const allocator = testing.allocator; - var threaded: std.Io.Threaded = .init(allocator, .{}); - defer threaded.deinit(); - const io = threaded.io(); - var h = TestHarness.init(allocator); - defer h.deinit(); - h.activate(); - var ns = null_store_mod.NullStore.init(allocator); - const agent = try Agent.init(allocator, io, &h.config, ns.store().create(), null); - defer agent.deinit(); - - const conv = &agent.conversation; - // Committed assistant message carrying one ToolUse. - var a_content: std.ArrayList(conversation.ContentBlock) = .empty; - try a_content.append(allocator, .{ .ToolUse = .{ - .id = try allocator.dupe(u8, "tc_1"), - .name = try allocator.dupe(u8, "read"), - .input = try conversation.textualBlockFromSlice(allocator, "{\"path\":\"orig\"}"), - } }); - try conv.messages.append(allocator, .{ .role = .assistant, .content = a_content }); - - // Input override rewrites the tail's matching ToolUse; unknown ids miss. - try testing.expect(try agent.overrideToolUseInput("tc_1", "{\"path\":\"new\"}")); - try testing.expect(!try agent.overrideToolUseInput("tc_missing", "{}")); - const tu = conv.messages.items[0].content.items[0].ToolUse; - try testing.expectEqualStrings("{\"path\":\"new\"}", tu.input.items); - - // Tool-result user message: a text part followed by a media part. - var parts: std.ArrayList(conversation.ResultPartStored) = .empty; - try parts.append(allocator, .{ .text = try conversation.textualBlockFromSlice(allocator, "original output") }); - try parts.append(allocator, .{ .media = .{ - .media_type = try allocator.dupe(u8, "image/png"), - .data = try conversation.textualBlockFromSlice(allocator, "aW1n"), - } }); - var u_content: std.ArrayList(conversation.ContentBlock) = .empty; - try u_content.append(allocator, .{ .ToolResult = .{ - .tool_use_id = try allocator.dupe(u8, "tc_1"), - .parts = parts, - } }); - try conv.messages.append(allocator, .{ .role = .user, .content = u_content }); - - // The ToolUse override only targets the tail — now a user message. - try testing.expect(!try agent.overrideToolUseInput("tc_1", "{}")); - - // Output override replaces the text, keeps the media part. - try testing.expect(try agent.overrideToolResultOutput("tc_1", "replaced output")); - try testing.expect(!try agent.overrideToolResultOutput("tc_other", "x")); - const tr = conv.messages.items[1].content.items[0].ToolResult; - try testing.expectEqual(@as(usize, 2), tr.parts.items.len); - try testing.expectEqualStrings("replaced output", tr.parts.items[0].text.items); - try testing.expectEqualStrings("image/png", tr.parts.items[1].media.media_type); -} - -/// A configurable ToolSource for testing the grouped-dispatch path. -/// Stores every batch it receives so tests can assert "calls X and Y -/// arrived in the same batch on the same thread". -const TestSource = struct { - name_owned: []u8, - decls: []tool_source_mod.ToolDecl, - decl_strings: std.array_list.Managed([]u8), - /// Sequence of (thread_id, [tool_name; n]) per batch received. - /// Only mutated inside `invoke_batch`. Because libpanto guarantees - /// at most one outstanding `invoke_batch` per source at any time - /// (one batch per turn per source), no synchronization is needed. - batches: std.array_list.Managed(Batch), - allocator: Allocator, - - const Batch = struct { - thread_id: u64, - names: std.array_list.Managed([]u8), - }; - - fn create( - allocator: Allocator, - source_name: []const u8, - tool_names: []const []const u8, - ) !ToolSource { - const self = try allocator.create(TestSource); - errdefer allocator.destroy(self); - - var strings = std.array_list.Managed([]u8).init(allocator); - errdefer { - for (strings.items) |s| allocator.free(s); - strings.deinit(); - } - - const name_owned = try allocator.dupe(u8, source_name); - try strings.append(name_owned); - - const decls = try allocator.alloc(tool_source_mod.ToolDecl, tool_names.len); - errdefer allocator.free(decls); - for (tool_names, 0..) |tn, i| { - const n = try allocator.dupe(u8, tn); - try strings.append(n); - const d = try allocator.dupe(u8, "test src tool"); - try strings.append(d); - const s = try allocator.dupe(u8, "{}"); - try strings.append(s); - decls[i] = .{ .name = n, .description = d, .schema_json = s }; - } - - self.* = .{ - .name_owned = name_owned, - .decls = decls, - .decl_strings = strings, - .batches = std.array_list.Managed(Batch).init(allocator), - .allocator = allocator, - }; - - return ToolSource{ - .name = self.name_owned, - .tools = self.decls, - .ctx = self, - .vtable = &vt, - }; - } - - const vt: ToolSource.VTable = .{ - .invoke_batch = invokeBatch, - .deinit = deinitSrc, - }; - - fn invokeBatch( - ctx: *anyopaque, - calls: []const tool_source_mod.Call, - results: []tool_source_mod.CallResult, - allocator: Allocator, - ) anyerror!void { - const self: *TestSource = @ptrCast(@alignCast(ctx)); - var batch: Batch = .{ - .thread_id = std.Thread.getCurrentId(), - .names = std.array_list.Managed([]u8).init(self.allocator), - }; - for (calls) |c| { - const copy = try self.allocator.dupe(u8, c.tool_name); - try batch.names.append(copy); - } - try self.batches.append(batch); - - for (calls, 0..) |c, i| { - const msg = std.fmt.allocPrint( - allocator, - "{s}->{s}", - .{ c.tool_name, c.input }, - ) catch |e| { - results[i] = .{ .err = e }; - continue; - }; - results[i] = .{ - .ok = tool_mod.ResultParts.fromTextOwned(allocator, msg) catch |e| { - results[i] = .{ .err = e }; - continue; - }, - }; - } - } - - fn deinitSrc(ctx: *anyopaque, _: Allocator) void { - const self: *TestSource = @ptrCast(@alignCast(ctx)); - for (self.decl_strings.items) |s| self.allocator.free(s); - self.decl_strings.deinit(); - for (self.batches.items) |*b| { - for (b.names.items) |n| self.allocator.free(n); - b.names.deinit(); - } - self.batches.deinit(); - self.allocator.free(self.decls); - self.allocator.destroy(self); - } -}; - -/// A source that always fails the whole batch by returning an error -/// from invoke_batch (rather than recording per-call errors). Used to -/// verify libpanto's whole-batch-failure path. -const FailingSource = struct { - name_owned: []u8, - decls: []tool_source_mod.ToolDecl, - decl_strings: std.array_list.Managed([]u8), - allocator: Allocator, - - fn create(allocator: Allocator, source_name: []const u8, tool_names: []const []const u8) !ToolSource { - const self = try allocator.create(FailingSource); - errdefer allocator.destroy(self); - - var strings = std.array_list.Managed([]u8).init(allocator); - errdefer { - for (strings.items) |s| allocator.free(s); - strings.deinit(); - } - - const name_owned = try allocator.dupe(u8, source_name); - try strings.append(name_owned); - - const decls = try allocator.alloc(tool_source_mod.ToolDecl, tool_names.len); - errdefer allocator.free(decls); - for (tool_names, 0..) |tn, i| { - const n = try allocator.dupe(u8, tn); - try strings.append(n); - const d = try allocator.dupe(u8, "fails"); - try strings.append(d); - const s = try allocator.dupe(u8, "{}"); - try strings.append(s); - decls[i] = .{ .name = n, .description = d, .schema_json = s }; - } - - self.* = .{ - .name_owned = name_owned, - .decls = decls, - .decl_strings = strings, - .allocator = allocator, - }; - return ToolSource{ .name = self.name_owned, .tools = self.decls, .ctx = self, .vtable = &vt }; - } - - const vt: ToolSource.VTable = .{ .invoke_batch = invokeBatch, .deinit = deinitSrc }; - - fn invokeBatch( - _: *anyopaque, - _: []const tool_source_mod.Call, - _: []tool_source_mod.CallResult, - _: Allocator, - ) anyerror!void { - return error.SourceExploded; - } - - fn deinitSrc(ctx: *anyopaque, _: Allocator) void { - const self: *FailingSource = @ptrCast(@alignCast(ctx)); - for (self.decl_strings.items) |s| self.allocator.free(s); - self.decl_strings.deinit(); - self.allocator.free(self.decls); - self.allocator.destroy(self); - } -}; - -/// A source that succeeds the first member call and fails the rest with a -/// per-call error (returning void from `invoke_batch`). Exercises the -/// per-call error path distinct from a whole-batch failure. -const PartialSource = struct { - name_owned: []u8, - decls: []tool_source_mod.ToolDecl, - decl_strings: std.array_list.Managed([]u8), - allocator: Allocator, - - fn create(allocator: Allocator, source_name: []const u8, tool_names: []const []const u8) !ToolSource { - const self = try allocator.create(PartialSource); - errdefer allocator.destroy(self); - var strings = std.array_list.Managed([]u8).init(allocator); - errdefer { - for (strings.items) |s| allocator.free(s); - strings.deinit(); - } - const name_owned = try allocator.dupe(u8, source_name); - try strings.append(name_owned); - const decls = try allocator.alloc(tool_source_mod.ToolDecl, tool_names.len); - errdefer allocator.free(decls); - for (tool_names, 0..) |tn, i| { - const n = try allocator.dupe(u8, tn); - try strings.append(n); - const d = try allocator.dupe(u8, "partial"); - try strings.append(d); - const s = try allocator.dupe(u8, "{}"); - try strings.append(s); - decls[i] = .{ .name = n, .description = d, .schema_json = s }; - } - self.* = .{ .name_owned = name_owned, .decls = decls, .decl_strings = strings, .allocator = allocator }; - return ToolSource{ .name = self.name_owned, .tools = self.decls, .ctx = self, .vtable = &vt }; - } - - const vt: ToolSource.VTable = .{ .invoke_batch = invokeBatch, .deinit = deinitSrc }; - - fn invokeBatch( - _: *anyopaque, - calls: []const tool_source_mod.Call, - results: []tool_source_mod.CallResult, - allocator: Allocator, - ) anyerror!void { - for (calls, 0..) |_, j| { - if (j == 0) { - results[j] = .{ .ok = try tool_mod.ResultParts.fromText(allocator, "ok") }; - } else { - results[j] = .{ .err = error.PerCallBoom }; - } - } - } - - fn deinitSrc(ctx: *anyopaque, _: Allocator) void { - const self: *PartialSource = @ptrCast(@alignCast(ctx)); - for (self.decl_strings.items) |s| self.allocator.free(s); - self.decl_strings.deinit(); - self.allocator.free(self.decls); - self.allocator.destroy(self); - } -}; - -test "registry register and lookup" { - var h = TestHarness.init(testing.allocator); - defer h.deinit(); - try h.registry.register(try EchoTool.create(testing.allocator, "echo", "ECHO:")); - try testing.expectEqual(@as(usize, 1), h.registry.count()); - try testing.expect(h.registry.lookup("echo") != null); -} - -test "duplicate register returns error" { - var h = TestHarness.init(testing.allocator); - defer h.deinit(); - try h.registry.register(try EchoTool.create(testing.allocator, "echo", "A:")); - - var dup = try EchoTool.create(testing.allocator, "echo", "B:"); - try testing.expectError(error.DuplicateTool, h.registry.register(dup)); - dup.vtable.deinit(dup.ctx, testing.allocator); -} - -test "runStep dispatches a tool call and loops to a final text turn" { - const allocator = testing.allocator; - - const scripted = [_]StubProvider.ScriptedTurn{ - .{ .blocks = &.{ - .{ .ToolUse = .{ .id = "tc_1", .name = "echo", .input = "hello" } }, - } }, - .{ .blocks = &.{ - .{ .Text = "ok" }, - } }, - }; - var stub = StubProvider{ .allocator = allocator, .scripted = &scripted }; - var threaded: std.Io.Threaded = .init(allocator, .{}); - defer threaded.deinit(); - const io = threaded.io(); - var h = TestHarness.init(allocator); - defer h.deinit(); - try h.registry.register(try EchoTool.create(allocator, "echo", "ECHO:")); - h.activate(); - var ns = null_store_mod.NullStore.init(allocator); - const agent = try Agent.init(allocator, io, &h.config, ns.store().create(), null); - defer agent.deinit(); - h.seedInto(agent); - agent._open_stream_fn = stub.install(); - - const conv = &agent.conversation; - - try drainTurn(agent, "call a tool"); - - try testing.expectEqual(@as(usize, 4), conv.messages.items.len); - - try testing.expectEqual(conversation.MessageRole.assistant, conv.messages.items[1].role); - try testing.expectEqual(@as(usize, 1), conv.messages.items[1].content.items.len); - try testing.expectEqualStrings("tc_1", conv.messages.items[1].content.items[0].ToolUse.id); - - try testing.expectEqual(conversation.MessageRole.user, conv.messages.items[2].role); - try testing.expectEqual(@as(usize, 1), conv.messages.items[2].content.items.len); - const tr = conv.messages.items[2].content.items[0].ToolResult; - try testing.expectEqualStrings("tc_1", tr.tool_use_id); - try testing.expectEqualStrings("ECHO:hello", trText(tr)); - - try testing.expectEqual(conversation.MessageRole.assistant, conv.messages.items[3].role); - try testing.expectEqualStrings("ok", conv.messages.items[3].content.items[0].Text.items); -} - -test "Stream emits tool_dispatch_start before running tools" { - const allocator = testing.allocator; - - const scripted = [_]StubProvider.ScriptedTurn{ - .{ .blocks = &.{ - .{ .ToolUse = .{ .id = "tc_1", .name = "echo", .input = "hello" } }, - } }, - .{ .blocks = &.{.{ .Text = "ok" }} }, - }; - var stub = StubProvider{ .allocator = allocator, .scripted = &scripted }; - var threaded: std.Io.Threaded = .init(allocator, .{}); - defer threaded.deinit(); - const io = threaded.io(); - var h = TestHarness.init(allocator); - defer h.deinit(); - try h.registry.register(try EchoTool.create(allocator, "echo", "ECHO:")); - h.activate(); - var ns = null_store_mod.NullStore.init(allocator); - const agent = try Agent.init(allocator, io, &h.config, ns.store().create(), null); - defer agent.deinit(); - h.seedInto(agent); - agent._open_stream_fn = stub.install(); - - var s = try runUserText(agent, "call a tool"); - defer s.deinit(); - - const first = (try s.next()).?; - try testing.expect(first == .message_complete); - try testing.expectEqual(@as(usize, 2), agent.conversation.messages.items.len); - - const start = (try s.next()).?; - try testing.expect(start == .tool_dispatch_start); - try testing.expectEqual(@as(usize, 1), start.tool_dispatch_start.count); - // The ToolResult user message must not exist yet; otherwise callers do - // not get a chance to render the tool as running before dispatch blocks. - try testing.expectEqual(@as(usize, 2), agent.conversation.messages.items.len); - - const complete = (try s.next()).?; - try testing.expect(complete == .tool_dispatch_complete); - try testing.expectEqual(@as(usize, 3), agent.conversation.messages.items.len); -} - -test "runStep dispatches multiple tool calls in parallel" { - const allocator = testing.allocator; - - var barrier: BarrierTool.Barrier = .{ .target = 3 }; - - const scripted = [_]StubProvider.ScriptedTurn{ - .{ .blocks = &.{ - .{ .ToolUse = .{ .id = "a", .name = "barrierA", .input = "" } }, - .{ .ToolUse = .{ .id = "b", .name = "barrierB", .input = "" } }, - .{ .ToolUse = .{ .id = "c", .name = "barrierC", .input = "" } }, - } }, - .{ .blocks = &.{ - .{ .Text = "done" }, - } }, - }; - var stub = StubProvider{ .allocator = allocator, .scripted = &scripted }; - var threaded: std.Io.Threaded = .init(allocator, .{}); - defer threaded.deinit(); - const io = threaded.io(); - var h = TestHarness.init(allocator); - defer h.deinit(); - try h.registry.register(try BarrierTool.create(allocator, "barrierA", &barrier)); - try h.registry.register(try BarrierTool.create(allocator, "barrierB", &barrier)); - try h.registry.register(try BarrierTool.create(allocator, "barrierC", &barrier)); - h.activate(); - var ns = null_store_mod.NullStore.init(allocator); - const agent = try Agent.init(allocator, io, &h.config, ns.store().create(), null); - defer agent.deinit(); - h.seedInto(agent); - agent._open_stream_fn = stub.install(); - - const conv = &agent.conversation; - - try drainTurn(agent, "go"); - - const tr_msg = conv.messages.items[2]; - try testing.expectEqual(@as(usize, 3), tr_msg.content.items.len); - try testing.expectEqualStrings("a", tr_msg.content.items[0].ToolResult.tool_use_id); - try testing.expectEqualStrings("b", tr_msg.content.items[1].ToolResult.tool_use_id); - try testing.expectEqualStrings("c", tr_msg.content.items[2].ToolResult.tool_use_id); - - const t0 = barrier.thread_ids[0].load(.acquire); - const t1 = barrier.thread_ids[1].load(.acquire); - const t2 = barrier.thread_ids[2].load(.acquire); - try testing.expect(t0 != 0 and t1 != 0 and t2 != 0); - try testing.expect(t0 != t1 and t1 != t2 and t0 != t2); -} - -test "runStep: native tool handler error becomes an error result and the model gets another turn" { - const allocator = testing.allocator; - - const scripted = [_]StubProvider.ScriptedTurn{ - .{ .blocks = &.{ - .{ .ToolUse = .{ .id = "x", .name = "boom", .input = "" } }, - } }, - .{ .blocks = &.{.{ .Text = "i will recover" }} }, - }; - var stub = StubProvider{ .allocator = allocator, .scripted = &scripted }; - var threaded: std.Io.Threaded = .init(allocator, .{}); - defer threaded.deinit(); - const io = threaded.io(); - var h = TestHarness.init(allocator); - defer h.deinit(); - try h.registry.register(try FailingTool.create(allocator, "boom")); - h.activate(); - var ns = null_store_mod.NullStore.init(allocator); - const agent = try Agent.init(allocator, io, &h.config, ns.store().create(), null); - defer agent.deinit(); - h.seedInto(agent); - agent._open_stream_fn = stub.install(); - - const conv = &agent.conversation; - - try drainTurn(agent, "break it"); - - // user, assistant(tool_use), user(tool_result), assistant(text) - try testing.expectEqual(@as(usize, 4), conv.messages.items.len); - const tr = conv.messages.items[2].content.items[0].ToolResult; - try testing.expectEqualStrings("x", tr.tool_use_id); - try testing.expect(tr.is_error); - try testing.expect(std.mem.indexOf(u8, trText(tr), "ToolExploded") != null); -} - -test "runStep: unknown tool becomes an error tool result and the loop continues" { - const allocator = testing.allocator; - - const scripted = [_]StubProvider.ScriptedTurn{ - .{ .blocks = &.{ - .{ .ToolUse = .{ .id = "z", .name = "ghost", .input = "" } }, - } }, - .{ .blocks = &.{.{ .Text = "ok, that tool does not exist" }} }, - }; - var stub = StubProvider{ .allocator = allocator, .scripted = &scripted }; - var threaded: std.Io.Threaded = .init(allocator, .{}); - defer threaded.deinit(); - const io = threaded.io(); - var h = TestHarness.init(allocator); - defer h.deinit(); - h.activate(); - var ns = null_store_mod.NullStore.init(allocator); - const agent = try Agent.init(allocator, io, &h.config, ns.store().create(), null); - defer agent.deinit(); - h.seedInto(agent); - agent._open_stream_fn = stub.install(); - - const conv = &agent.conversation; - - 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); - const tr_msg = conv.messages.items[2]; - try testing.expectEqual(conversation.MessageRole.user, tr_msg.role); - const tr = tr_msg.content.items[0].ToolResult; - try testing.expectEqualStrings("z", tr.tool_use_id); - try testing.expect(tr.is_error); - try testing.expect(std.mem.indexOf(u8, trText(tr), "UnknownTool") != null); -} - -test "runStep with no tool calls returns after one provider step" { - const allocator = testing.allocator; - - const scripted = [_]StubProvider.ScriptedTurn{ - .{ .blocks = &.{.{ .Text = "hi" }} }, - }; - var stub = StubProvider{ .allocator = allocator, .scripted = &scripted }; - var threaded: std.Io.Threaded = .init(allocator, .{}); - defer threaded.deinit(); - const io = threaded.io(); - var h = TestHarness.init(allocator); - defer h.deinit(); - h.activate(); - var ns = null_store_mod.NullStore.init(allocator); - const agent = try Agent.init(allocator, io, &h.config, ns.store().create(), null); - defer agent.deinit(); - h.seedInto(agent); - agent._open_stream_fn = stub.install(); - - const conv = &agent.conversation; - - 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); -} - -test "runStep surfaces EmptyAssistantResponse when provider commits an empty message" { - const allocator = testing.allocator; - - const scripted = [_]StubProvider.ScriptedTurn{ - .{ .blocks = &.{} }, - }; - var stub = StubProvider{ .allocator = allocator, .scripted = &scripted }; - var threaded: std.Io.Threaded = .init(allocator, .{}); - defer threaded.deinit(); - const io = threaded.io(); - var h = TestHarness.init(allocator); - defer h.deinit(); - h.activate(); - var ns = null_store_mod.NullStore.init(allocator); - const agent = try Agent.init(allocator, io, &h.config, ns.store().create(), null); - defer agent.deinit(); - h.seedInto(agent); - agent._open_stream_fn = stub.install(); - - try testing.expectError(error.EmptyAssistantResponse, drainTurn(agent, "hi")); -} - -// ------------ ToolSource tests ------------ - -test "runStep delivers all source-backed calls in one batch on one thread" { - const allocator = testing.allocator; - - const scripted = [_]StubProvider.ScriptedTurn{ - .{ .blocks = &.{ - .{ .ToolUse = .{ .id = "a", .name = "lua_x", .input = "1" } }, - .{ .ToolUse = .{ .id = "b", .name = "lua_y", .input = "2" } }, - .{ .ToolUse = .{ .id = "c", .name = "lua_x", .input = "3" } }, - } }, - .{ .blocks = &.{.{ .Text = "done" }} }, - }; - var stub = StubProvider{ .allocator = allocator, .scripted = &scripted }; - var threaded: std.Io.Threaded = .init(allocator, .{}); - defer threaded.deinit(); - const io = threaded.io(); - var h = TestHarness.init(allocator); - defer h.deinit(); - try h.registry.registerSource(try TestSource.create(allocator, "panto-lua", &.{ "lua_x", "lua_y" })); - h.activate(); - var ns = null_store_mod.NullStore.init(allocator); - const agent = try Agent.init(allocator, io, &h.config, ns.store().create(), null); - defer agent.deinit(); - h.seedInto(agent); - agent._open_stream_fn = stub.install(); - - const conv = &agent.conversation; - - try drainTurn(agent, "go"); - - // Locate the source and inspect its observed batches. - const view = agent._registry.lookup("lua_x") orelse return error.NotFound; - const src_ptr = view.entry.source.source; - const test_src: *TestSource = @ptrCast(@alignCast(src_ptr.ctx)); - - try testing.expectEqual(@as(usize, 1), test_src.batches.items.len); - const b = test_src.batches.items[0]; - try testing.expectEqual(@as(usize, 3), b.names.items.len); - try testing.expectEqualStrings("lua_x", b.names.items[0]); - try testing.expectEqualStrings("lua_y", b.names.items[1]); - try testing.expectEqualStrings("lua_x", b.names.items[2]); - - // ToolResults arrived in the original call order. - const tr_msg = conv.messages.items[2]; - try testing.expectEqual(@as(usize, 3), tr_msg.content.items.len); - try testing.expectEqualStrings("a", tr_msg.content.items[0].ToolResult.tool_use_id); - try testing.expectEqualStrings("lua_x->1", trText(tr_msg.content.items[0].ToolResult)); - try testing.expectEqualStrings("b", tr_msg.content.items[1].ToolResult.tool_use_id); - try testing.expectEqualStrings("lua_y->2", trText(tr_msg.content.items[1].ToolResult)); - try testing.expectEqualStrings("c", tr_msg.content.items[2].ToolResult.tool_use_id); - try testing.expectEqualStrings("lua_x->3", trText(tr_msg.content.items[2].ToolResult)); -} - -test "runStep: distinct sources run on distinct threads in parallel" { - const allocator = testing.allocator; - - const scripted = [_]StubProvider.ScriptedTurn{ - .{ .blocks = &.{ - .{ .ToolUse = .{ .id = "a", .name = "src_a_t", .input = "" } }, - .{ .ToolUse = .{ .id = "b", .name = "src_b_t", .input = "" } }, - } }, - .{ .blocks = &.{.{ .Text = "done" }} }, - }; - var stub = StubProvider{ .allocator = allocator, .scripted = &scripted }; - var threaded: std.Io.Threaded = .init(allocator, .{}); - defer threaded.deinit(); - const io = threaded.io(); - var h = TestHarness.init(allocator); - defer h.deinit(); - try h.registry.registerSource(try TestSource.create(allocator, "src_a", &.{"src_a_t"})); - try h.registry.registerSource(try TestSource.create(allocator, "src_b", &.{"src_b_t"})); - h.activate(); - var ns = null_store_mod.NullStore.init(allocator); - const agent = try Agent.init(allocator, io, &h.config, ns.store().create(), null); - defer agent.deinit(); - h.seedInto(agent); - agent._open_stream_fn = stub.install(); - - try drainTurn(agent, "go"); - - const view_a = agent._registry.lookup("src_a_t") orelse return error.NotFound; - const view_b = agent._registry.lookup("src_b_t") orelse return error.NotFound; - const sa: *TestSource = @ptrCast(@alignCast(view_a.entry.source.source.ctx)); - const sb: *TestSource = @ptrCast(@alignCast(view_b.entry.source.source.ctx)); - - try testing.expectEqual(@as(usize, 1), sa.batches.items.len); - try testing.expectEqual(@as(usize, 1), sb.batches.items.len); - // The two sources ran on distinct OS threads. - try testing.expect(sa.batches.items[0].thread_id != sb.batches.items[0].thread_id); -} - -test "runStep: source whole-batch error becomes per-call error results and continues" { - const allocator = testing.allocator; - - const scripted = [_]StubProvider.ScriptedTurn{ - .{ .blocks = &.{ - .{ .ToolUse = .{ .id = "a", .name = "fa", .input = "" } }, - .{ .ToolUse = .{ .id = "b", .name = "fb", .input = "" } }, - } }, - .{ .blocks = &.{.{ .Text = "recovered" }} }, - }; - var stub = StubProvider{ .allocator = allocator, .scripted = &scripted }; - var threaded: std.Io.Threaded = .init(allocator, .{}); - defer threaded.deinit(); - const io = threaded.io(); - var h = TestHarness.init(allocator); - defer h.deinit(); - try h.registry.registerSource(try FailingSource.create(allocator, "fs", &.{ "fa", "fb" })); - h.activate(); - var ns = null_store_mod.NullStore.init(allocator); - const agent = try Agent.init(allocator, io, &h.config, ns.store().create(), null); - defer agent.deinit(); - h.seedInto(agent); - agent._open_stream_fn = stub.install(); - - const conv = &agent.conversation; - - 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); - const tr_msg = conv.messages.items[2]; - try testing.expectEqual(@as(usize, 2), tr_msg.content.items.len); - // Every member of the failed batch produced an error result, in order. - const tr_a = tr_msg.content.items[0].ToolResult; - const tr_b = tr_msg.content.items[1].ToolResult; - try testing.expectEqualStrings("a", tr_a.tool_use_id); - try testing.expectEqualStrings("b", tr_b.tool_use_id); - try testing.expect(tr_a.is_error); - try testing.expect(tr_b.is_error); - try testing.expect(std.mem.indexOf(u8, trText(tr_a), "SourceExploded") != null); -} - -test "runStep: mixed single Tools and source-backed tools coexist in one turn" { - const allocator = testing.allocator; - - const scripted = [_]StubProvider.ScriptedTurn{ - .{ .blocks = &.{ - .{ .ToolUse = .{ .id = "a", .name = "single", .input = "X" } }, - .{ .ToolUse = .{ .id = "b", .name = "src_t1", .input = "Y" } }, - .{ .ToolUse = .{ .id = "c", .name = "src_t2", .input = "Z" } }, - } }, - .{ .blocks = &.{.{ .Text = "done" }} }, - }; - var stub = StubProvider{ .allocator = allocator, .scripted = &scripted }; - var threaded: std.Io.Threaded = .init(allocator, .{}); - defer threaded.deinit(); - const io = threaded.io(); - var h = TestHarness.init(allocator); - defer h.deinit(); - try h.registry.register(try EchoTool.create(allocator, "single", "S:")); - try h.registry.registerSource(try TestSource.create(allocator, "src", &.{ "src_t1", "src_t2" })); - h.activate(); - var ns = null_store_mod.NullStore.init(allocator); - const agent = try Agent.init(allocator, io, &h.config, ns.store().create(), null); - defer agent.deinit(); - h.seedInto(agent); - agent._open_stream_fn = stub.install(); - - const conv = &agent.conversation; - - try drainTurn(agent, "go"); - - const tr_msg = conv.messages.items[2]; - try testing.expectEqual(@as(usize, 3), tr_msg.content.items.len); - try testing.expectEqualStrings("S:X", trText(tr_msg.content.items[0].ToolResult)); - try testing.expectEqualStrings("src_t1->Y", trText(tr_msg.content.items[1].ToolResult)); - try testing.expectEqualStrings("src_t2->Z", trText(tr_msg.content.items[2].ToolResult)); -} - -test "setConfig swaps provider between turns; agent tool set persists" { - // Post-R1 the tool set lives on the `Agent`, not on `Config`. Swapping - // the config pointer (`setConfig`) changes provider/model at the next - // turn boundary but leaves the agent's registered tools intact: a turn - // after the swap still resolves a tool registered before it. - const allocator = testing.allocator; - - const scripted = [_]StubProvider.ScriptedTurn{ - .{ .blocks = &.{.{ .ToolUse = .{ .id = "2", .name = "late", .input = "B" } }} }, - .{ .blocks = &.{.{ .Text = "done" }} }, - }; - var stub = StubProvider{ .allocator = allocator, .scripted = &scripted }; - var threaded: std.Io.Threaded = .init(allocator, .{}); - defer threaded.deinit(); - const io = threaded.io(); - - const cfg_a: config_mod.Config = .{ - .provider = .{ .openai_chat = .{ .api_key = "k", .base_url = "a", .model = "m" } }, - }; - const cfg_b: config_mod.Config = .{ - .provider = .{ .openai_chat = .{ .api_key = "k", .base_url = "b", .model = "m" } }, - }; - - var ns = null_store_mod.NullStore.init(allocator); - const agent = try Agent.init(allocator, io, &cfg_a, ns.store().create(), null); - defer agent.deinit(); - try agent.registerTool(try EchoTool.create(allocator, "late", "B:")); - agent._open_stream_fn = stub.install(); - - // The tool is visible regardless of which config is active. - try testing.expect(agent._registry.lookup("late") != null); - agent.setConfig(&cfg_b); - try testing.expect(agent._registry.lookup("late") != null); - - // A real turn after the swap still resolves `late`, then loops to the - // final text turn. - const conv = &agent.conversation; - try drainTurn(agent, "go"); - - const tr = conv.messages.items[2].content.items[0].ToolResult; - try testing.expectEqualStrings("2", tr.tool_use_id); - try testing.expectEqualStrings("B:B", trText(tr)); -} - -test "compact: summarizes prefix, keeps suffix, system survives" { - const allocator = testing.allocator; - - // The stub returns a single text turn — used as the summary text. - const scripted = [_]StubProvider.ScriptedTurn{ - .{ .blocks = &.{.{ .Text = "SUMMARY OF EARLIER" }} }, - }; - var stub = StubProvider{ .allocator = allocator, .scripted = &scripted }; - var threaded: std.Io.Threaded = .init(allocator, .{}); - defer threaded.deinit(); - const io = threaded.io(); - var h = TestHarness.init(allocator); - defer h.deinit(); - h.activate(); - // keep_verbatim sized so only the last (short) turn fits: q2+a2 are - // 3 words each => ceil(3*1.3)=4 tokens each => 8 total <= 10, while - // adding the longer first turn exceeds it. - h.config.compaction = .{ .keep_verbatim = 10 }; - var ns = null_store_mod.NullStore.init(allocator); - const agent = try Agent.init(allocator, io, &h.config, ns.store().create(), null); - defer agent.deinit(); - h.seedInto(agent); - agent._open_stream_fn = stub.install(); - - const conv = &agent.conversation; - try conv.addSystemMessage("you are helpful"); - try addUserText(conv, "first question here with several words"); - try conv.addAssistantMessage(&.{ - .{ .Text = try conversation.textualBlockFromSlice(allocator, "first answer with several words") }, - }, null); - try addUserText(conv, "second recent question"); - try conv.addAssistantMessage(&.{ - .{ .Text = try conversation.textualBlockFromSlice(allocator, "second recent answer") }, - }, null); - - const res = try agent._compactInPlace("Summarize the conversation.", null); - try testing.expect(res.compacted); - - // Expected rebuilt: [system, compaction summary(user), user q2, asst a2] - try testing.expectEqual(@as(usize, 4), conv.messages.items.len); - try testing.expectEqual(conversation.MessageRole.system, conv.messages.items[0].role); - try testing.expectEqualStrings( - "you are helpful", - conv.messages.items[0].content.items[0].System.text.items, - ); - try testing.expectEqual(conversation.MessageRole.user, conv.messages.items[1].role); - try testing.expectEqualStrings( - "SUMMARY OF EARLIER", - conv.messages.items[1].content.items[0].CompactionSummary.text.items, - ); - try testing.expectEqualStrings( - "second recent question", - conv.messages.items[2].content.items[0].Text.items, - ); - try testing.expectEqualStrings( - "second recent answer", - conv.messages.items[3].content.items[0].Text.items, - ); -} - -test "interruption: resuming next() after a break dispatches the pending tools" { - const allocator = testing.allocator; - - const scripted = [_]StubProvider.ScriptedTurn{ - .{ .blocks = &.{ - .{ .ToolUse = .{ .id = "tc_1", .name = "echo", .input = "hello" } }, - } }, - .{ .blocks = &.{ - .{ .Text = "ok" }, - } }, - }; - var stub = StubProvider{ .allocator = allocator, .scripted = &scripted }; - var threaded: std.Io.Threaded = .init(allocator, .{}); - defer threaded.deinit(); - const io = threaded.io(); - var h = TestHarness.init(allocator); - defer h.deinit(); - try h.registry.register(try EchoTool.create(allocator, "echo", "ECHO:")); - h.activate(); - - var cap = CapturingStore.init(allocator); - defer cap.deinit(); - const agent = try Agent.init(allocator, io, &h.config, cap.store().create(), null); - defer agent.deinit(); - h.seedInto(agent); - agent._open_stream_fn = stub.install(); - - var s = try runUserText(agent, "call a tool"); - defer s.deinit(); - - // Break right after the assistant's tool-call message completes — the - // stream is paused, NOT closed. The tool has not been dispatched. - while (try s.next()) |ev| { - if (ev == .message_complete) break; - } - try testing.expectEqual(@as(usize, 1), cap.roles.items.len); // only the prompt - - // Resuming `next()` on the same stream picks up where it left off and runs - // the agent's registered-tool machinery: no new user turn required. - var saw_dispatch = false; - while (try s.next()) |ev| { - if (ev == .tool_dispatch_complete) saw_dispatch = true; - } - try testing.expect(saw_dispatch); - - // The full turn is now persisted: prompt, assistant(ToolUse), - // user(ToolResult), assistant(text). - try testing.expectEqual(@as(usize, 4), cap.roles.items.len); - try testing.expectEqual(conversation.MessageRole.user, cap.roles.items[2]); -} - -test "compact: tolerates a trailing dangling tool call (interrupted turn)" { - const allocator = testing.allocator; - - const scripted = [_]StubProvider.ScriptedTurn{ - .{ .blocks = &.{.{ .Text = "SUMMARY OF EARLIER" }} }, - }; - var stub = StubProvider{ .allocator = allocator, .scripted = &scripted }; - var threaded: std.Io.Threaded = .init(allocator, .{}); - defer threaded.deinit(); - const io = threaded.io(); - var h = TestHarness.init(allocator); - defer h.deinit(); - h.activate(); - h.config.compaction = .{ .keep_verbatim = 10 }; - var ns = null_store_mod.NullStore.init(allocator); - const agent = try Agent.init(allocator, io, &h.config, ns.store().create(), null); - defer agent.deinit(); - h.seedInto(agent); - agent._open_stream_fn = stub.install(); - - const conv = &agent.conversation; - try conv.addSystemMessage("you are helpful"); - try addUserText(conv, "first question here with several words"); - try conv.addAssistantMessage(&.{ - .{ .Text = try conversation.textualBlockFromSlice(allocator, "first answer with several words") }, - }, null); - // A fresh turn whose assistant message ends in a tool call with no results - // yet — exactly the state a break right after `message_complete` leaves. - try addUserText(conv, "second recent question"); - try conv.addAssistantMessage(&.{ - .{ .ToolUse = .{ - .id = try allocator.dupe(u8, "tc_1"), - .name = try allocator.dupe(u8, "echo"), - .input = try conversation.textualBlockFromSlice(allocator, "hello"), - } }, - }, null); - - // Compaction must not choke on the dangling tail: it summarizes the older - // turn and keeps the interrupted turn (incl. the dangling call) verbatim. - const res = try agent._compactInPlace("Summarize the conversation.", null); - try testing.expect(res.compacted); - - // The dangling tool call survives as the tail, intact and resumable. - const last = conv.messages.items[conv.messages.items.len - 1]; - try testing.expectEqual(conversation.MessageRole.assistant, last.role); - try testing.expect(Agent.hasToolUseBlock(last)); - try testing.expectEqualStrings("tc_1", last.content.items[0].ToolUse.id); -} - -test "compact: restated suffix usage reconstructs a fresh cumulative chain" { - const allocator = testing.allocator; - - // The compaction turn reports a known output token count (the summary - // size used as the new anchor). - const scripted = [_]StubProvider.ScriptedTurn{ - .{ - .blocks = &.{.{ .Text = "SUMMARY" }}, - .usage = .{ .input = 9999, .output = 100 }, // input is ignored; only output anchors - }, - }; - var stub = StubProvider{ .allocator = allocator, .scripted = &scripted }; - var threaded: std.Io.Threaded = .init(allocator, .{}); - defer threaded.deinit(); - const io = threaded.io(); - var h = TestHarness.init(allocator); - defer h.deinit(); - h.activate(); - // Budget keeps the last two turns verbatim (deltas 8 + 6 = 14) but - // summarizes the prefix. - h.config.compaction = .{ .keep_verbatim = 20 }; - var ns = null_store_mod.NullStore.init(allocator); - const agent = try Agent.init(allocator, io, &h.config, ns.store().create(), null); - defer agent.deinit(); - h.seedInto(agent); - agent._open_stream_fn = stub.install(); - - const conv = &agent.conversation; - // Prefix turn (will be summarized). Cumulative footprint = 500+40+10+50 - // = 600. Its real usage (with cache buckets) is irrelevant - // post-compaction. - try addUserText(conv, "first question here with several words"); - try conv.addAssistantMessage( - &.{.{ .Text = try conversation.textualBlockFromSlice(allocator, "first answer with words") }}, - .{ .input = 500, .output = 50, .cache_read = 40, .cache_write = 10 }, - ); - // Kept turn 1: cumulative 608 (delta 8 over prefix). Reasoning is a - // subset of output. - try addUserText(conv, "kept question"); // 2 words => ceil(2*1.3)=3 tokens - try conv.addAssistantMessage( - &.{.{ .Text = try conversation.textualBlockFromSlice(allocator, "kept answer") }}, - .{ .input = 600, .output = 8, .reasoning = 5 }, - ); - // Kept turn 2: cumulative 614 (delta 6). Cache buckets present to prove - // they collapse into `input` on restatement. - try addUserText(conv, "two more words here"); // 4 words => ceil(4*1.3)=6 - try conv.addAssistantMessage( - &.{.{ .Text = try conversation.textualBlockFromSlice(allocator, "final answer") }}, - .{ .input = 600, .output = 4, .cache_read = 8, .cache_write = 2 }, - ); - - const res = try agent._compactInPlace("Summarize.", null); - try testing.expect(res.compacted); - - // Rebuilt: [summary(user), u1, a1, u2, a2]. - try testing.expectEqual(@as(usize, 5), conv.messages.items.len); - const asst1 = conv.messages.items[2]; - const asst2 = conv.messages.items[4]; - try testing.expect(conv.messages.items[1].usage == null); // kept user - - // First restated assistant: input = summary_size(100) + - // user_size("kept question" => 3) = 103. Cache zeroed; output/reasoning - // verbatim. - const us1 = asst1.usage.?; - try testing.expectEqual(@as(u64, 103), us1.input); - try testing.expectEqual(@as(u64, 0), us1.cache_read); - try testing.expectEqual(@as(u64, 0), us1.cache_write); - try testing.expectEqual(@as(u64, 8), us1.output); - try testing.expectEqual(@as(u64, 5), us1.reasoning); - - // Second restated assistant: input = prev.input(103) + prev.output(8) + - // user_size("two more words here" => 6) = 117. Original cache buckets - // (8+2) collapse away; only the synthetic cumulative total survives in - // `input`. - const us2 = asst2.usage.?; - try testing.expectEqual(@as(u64, 117), us2.input); - try testing.expectEqual(@as(u64, 0), us2.cache_read); - try testing.expectEqual(@as(u64, 0), us2.cache_write); - try testing.expectEqual(@as(u64, 4), us2.output); -} - -test "compact: no-op when conversation already fits the budget" { - const allocator = testing.allocator; - - const scripted = [_]StubProvider.ScriptedTurn{ - .{ .blocks = &.{.{ .Text = "should not be used" }} }, - }; - var stub = StubProvider{ .allocator = allocator, .scripted = &scripted }; - var threaded: std.Io.Threaded = .init(allocator, .{}); - defer threaded.deinit(); - const io = threaded.io(); - var h = TestHarness.init(allocator); - defer h.deinit(); - h.activate(); - h.config.compaction = .{ .keep_verbatim = 1_000_000 }; - var ns = null_store_mod.NullStore.init(allocator); - const agent = try Agent.init(allocator, io, &h.config, ns.store().create(), null); - defer agent.deinit(); - h.seedInto(agent); - agent._open_stream_fn = stub.install(); - - const conv = &agent.conversation; - try conv.addSystemMessage("sys"); - try addUserText(conv, "hi"); - try conv.addAssistantMessage(&.{ - .{ .Text = try conversation.textualBlockFromSlice(allocator, "hello") }, - }, null); - - const res = try agent._compactInPlace("Summarize.", null); - try testing.expect(!res.compacted); - try testing.expectEqual(@as(usize, 3), conv.messages.items.len); - // Stub was never consumed. - try testing.expectEqual(@as(usize, 0), stub.next); -} - -test "compact: extra instructions are appended to the system prompt" { - const allocator = testing.allocator; - - // Capture the system prompt the stub sees by scripting a turn and - // inspecting the throwaway conversation isn't directly possible via the - // current stub; instead we just assert compaction succeeds with extra - // instructions present (smoke test of the append path). - const scripted = [_]StubProvider.ScriptedTurn{ - .{ .blocks = &.{.{ .Text = "S" }} }, - }; - var stub = StubProvider{ .allocator = allocator, .scripted = &scripted }; - var threaded: std.Io.Threaded = .init(allocator, .{}); - defer threaded.deinit(); - const io = threaded.io(); - var h = TestHarness.init(allocator); - defer h.deinit(); - h.activate(); - h.config.compaction = .{ .keep_verbatim = 1 }; - var ns = null_store_mod.NullStore.init(allocator); - const agent = try Agent.init(allocator, io, &h.config, ns.store().create(), null); - defer agent.deinit(); - h.seedInto(agent); - agent._open_stream_fn = stub.install(); - - const conv = &agent.conversation; - try addUserText(conv, "question one two three"); - try conv.addAssistantMessage(&.{ - .{ .Text = try conversation.textualBlockFromSlice(allocator, "answer one two three") }, - }, null); - try addUserText(conv, "question two"); - try conv.addAssistantMessage(&.{ - .{ .Text = try conversation.textualBlockFromSlice(allocator, "answer two") }, - }, null); - - const res = try agent._compactInPlace("Base prompt.", "keep bug #3 details"); - try testing.expect(res.compacted); -} - -test "runStep: auto-compacts on context overflow and retries once" { - const allocator = testing.allocator; - - // First stream call overflows; then the compaction request returns a - // summary; then the retried main request returns a final text turn. - const scripted = [_]StubProvider.ScriptedTurn{ - .{ .blocks = &.{.{ .Text = "COMPACTED SUMMARY" }} }, // compaction call - .{ .blocks = &.{.{ .Text = "final answer" }} }, // retried main call - }; - var stub = StubProvider{ - .allocator = allocator, - .scripted = &scripted, - .overflow_calls = 1, - }; - var threaded: std.Io.Threaded = .init(allocator, .{}); - defer threaded.deinit(); - const io = threaded.io(); - var h = TestHarness.init(allocator); - defer h.deinit(); - h.activate(); - h.config.compaction = .{ .keep_verbatim = 10, .compaction_prompt = "Summarize the conversation." }; - var ns = null_store_mod.NullStore.init(allocator); - const agent = try Agent.init(allocator, io, &h.config, ns.store().create(), null); - defer agent.deinit(); - h.seedInto(agent); - agent._open_stream_fn = stub.install(); - - const conv = &agent.conversation; - try conv.addSystemMessage("you are helpful"); - try addUserText(conv, "first question with several words here"); - try conv.addAssistantMessage(&.{ - .{ .Text = try conversation.textualBlockFromSlice(allocator, "first answer with several words") }, - }, null); - - try drainTurn(agent, "second recent question"); - - try testing.expect(agent._auto_compacted); - // After compaction + retry: [system, summary, user q2, assistant final]. - const msgs = conv.messages.items; - try testing.expectEqual(conversation.MessageRole.system, msgs[0].role); - try testing.expectEqualStrings( - "COMPACTED SUMMARY", - msgs[1].content.items[0].CompactionSummary.text.items, - ); - try testing.expectEqualStrings("second recent question", msgs[2].content.items[0].Text.items); - try testing.expectEqualStrings("final answer", msgs[msgs.len - 1].content.items[0].Text.items); -} - -test "runStep: context overflow without compaction prompt propagates" { - const allocator = testing.allocator; - - const scripted = [_]StubProvider.ScriptedTurn{ - .{ .blocks = &.{.{ .Text = "unused" }} }, - }; - var stub = StubProvider{ - .allocator = allocator, - .scripted = &scripted, - .overflow_calls = 1, - }; - var threaded: std.Io.Threaded = .init(allocator, .{}); - defer threaded.deinit(); - const io = threaded.io(); - var h = TestHarness.init(allocator); - defer h.deinit(); - h.activate(); - var ns = null_store_mod.NullStore.init(allocator); - const agent = try Agent.init(allocator, io, &h.config, ns.store().create(), null); - defer agent.deinit(); - h.seedInto(agent); - agent._open_stream_fn = stub.install(); - // No compaction_system_prompt set -> overflow propagates. - - try testing.expectError(error.ContextOverflow, drainTurn(agent, "hi")); -} - -// ----------------------------------------------------------------------------- -// Phase 6: provider retry + tool-error holistic tests -// ----------------------------------------------------------------------------- - -/// 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, - - /// Append a copy of `info`, dup'ing its borrowed `message` so assertions - /// stay valid after the source `Stream` is deinit'd. - fn append(self: *RetryRecorder, info: provider_mod.ProviderRetryInfo) !void { - var owned = info; - if (info.message) |m| owned.message = try self.allocator.dupe(u8, m); - try self.infos.append(self.allocator, owned); - } - - fn deinit(self: *RetryRecorder) void { - for (self.infos.items) |info| { - if (info.message) |m| self.allocator.free(m); - } - self.infos.deinit(self.allocator); - } -}; - -/// 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 runUserText(agent, text); - defer s.deinit(); - while (try s.next()) |ev| { - if (ev == .provider_retry) try rr.append(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 { - h.activate(); - // Make sleeps negligible and deterministic (no jitter). - h.config.retry = .{ - .max_attempts = 4, - .initial_delay_ms = 0, - .max_delay_ms = 0, - .multiplier = 2.0, - .jitter = false, - }; -} - -test "runStep: provider 429 retries then succeeds without duplicate messages" { - const allocator = testing.allocator; - - const errs = [_]StubProvider.ScriptedError{ - .{ .err = error.ProviderRateLimited, .status_code = 429 }, - .{ .err = error.ProviderRateLimited, .status_code = 429 }, - }; - const scripted = [_]StubProvider.ScriptedTurn{ - .{ .blocks = &.{.{ .Text = "finally" }} }, - }; - var stub = StubProvider{ - .allocator = allocator, - .scripted = &scripted, - .scripted_errors = &errs, - }; - var threaded: std.Io.Threaded = .init(allocator, .{}); - defer threaded.deinit(); - const io = threaded.io(); - var h = TestHarness.init(allocator); - defer h.deinit(); - fastRetryHarness(&h); - var ns = null_store_mod.NullStore.init(allocator); - const agent = try Agent.init(allocator, io, &h.config, ns.store().create(), null); - defer agent.deinit(); - h.seedInto(agent); - agent._open_stream_fn = stub.install(); - - const conv = &agent.conversation; - - var rr = RetryRecorder{ .allocator = allocator }; - defer rr.deinit(); - try drainTurnRecording(agent, "hi", &rr); - - // Two failures + one success. - try testing.expectEqual(@as(usize, 3), stub.calls_made); - // No duplicate assistant messages: user + single assistant. - try testing.expectEqual(@as(usize, 2), conv.messages.items.len); - try testing.expectEqual(conversation.MessageRole.assistant, conv.messages.items[1].role); - // Two retry notifications, delivered before each delayed retry. - try testing.expectEqual(@as(usize, 2), rr.infos.items.len); - try testing.expectEqual(@as(?u16, 429), rr.infos.items[0].status_code); - try testing.expectEqual(@as(usize, 1), rr.infos.items[0].attempt); - try testing.expectEqual(@as(usize, 2), rr.infos.items[1].attempt); -} - -test "runStep: provider 500 retries with backoff notification" { - const allocator = testing.allocator; - - const errs = [_]StubProvider.ScriptedError{ - .{ .err = error.ProviderServerError, .status_code = 500 }, - }; - const scripted = [_]StubProvider.ScriptedTurn{ - .{ .blocks = &.{.{ .Text = "ok" }} }, - }; - var stub = StubProvider{ - .allocator = allocator, - .scripted = &scripted, - .scripted_errors = &errs, - }; - var threaded: std.Io.Threaded = .init(allocator, .{}); - defer threaded.deinit(); - const io = threaded.io(); - var h = TestHarness.init(allocator); - defer h.deinit(); - fastRetryHarness(&h); - var ns = null_store_mod.NullStore.init(allocator); - const agent = try Agent.init(allocator, io, &h.config, ns.store().create(), null); - defer agent.deinit(); - h.seedInto(agent); - agent._open_stream_fn = stub.install(); - - var rr = RetryRecorder{ .allocator = allocator }; - defer rr.deinit(); - try drainTurnRecording(agent, "hi", &rr); - - try testing.expectEqual(@as(usize, 2), stub.calls_made); - try testing.expectEqual(@as(usize, 1), rr.infos.items.len); - try testing.expectEqual(error.ProviderServerError, rr.infos.items[0].err); - try testing.expectEqual(@as(usize, 4), rr.infos.items[0].max_attempts); - try testing.expect(!rr.infos.items[0].compaction); -} - -test "runStep: provider auth failure does not retry" { - const allocator = testing.allocator; - - const errs = [_]StubProvider.ScriptedError{ - .{ .err = error.ProviderAuthFailed, .status_code = 401 }, - }; - const scripted = [_]StubProvider.ScriptedTurn{ - .{ .blocks = &.{.{ .Text = "unreachable" }} }, - }; - var stub = StubProvider{ - .allocator = allocator, - .scripted = &scripted, - .scripted_errors = &errs, - }; - var threaded: std.Io.Threaded = .init(allocator, .{}); - defer threaded.deinit(); - const io = threaded.io(); - var h = TestHarness.init(allocator); - defer h.deinit(); - fastRetryHarness(&h); - var ns = null_store_mod.NullStore.init(allocator); - const agent = try Agent.init(allocator, io, &h.config, ns.store().create(), null); - defer agent.deinit(); - h.seedInto(agent); - agent._open_stream_fn = stub.install(); - - var rr = RetryRecorder{ .allocator = allocator }; - defer rr.deinit(); - try testing.expectError(error.ProviderAuthFailed, drainTurnRecording(agent, "hi", &rr)); - - // Exactly one attempt, no retry notification. - try testing.expectEqual(@as(usize, 1), stub.calls_made); - try testing.expectEqual(@as(usize, 0), rr.infos.items.len); -} - -test "runStep: retries exhaust and hard-fail after max_attempts" { - const allocator = testing.allocator; - - const errs = [_]StubProvider.ScriptedError{ - .{ .err = error.ProviderUnavailable, .status_code = 503 }, - .{ .err = error.ProviderUnavailable, .status_code = 503 }, - .{ .err = error.ProviderUnavailable, .status_code = 503 }, - .{ .err = error.ProviderUnavailable, .status_code = 503 }, - }; - const scripted = [_]StubProvider.ScriptedTurn{ - .{ .blocks = &.{.{ .Text = "unreachable" }} }, - }; - var stub = StubProvider{ - .allocator = allocator, - .scripted = &scripted, - .scripted_errors = &errs, - }; - var threaded: std.Io.Threaded = .init(allocator, .{}); - defer threaded.deinit(); - const io = threaded.io(); - var h = TestHarness.init(allocator); - defer h.deinit(); - fastRetryHarness(&h); - var ns = null_store_mod.NullStore.init(allocator); - const agent = try Agent.init(allocator, io, &h.config, ns.store().create(), null); - defer agent.deinit(); - h.seedInto(agent); - agent._open_stream_fn = stub.install(); - - var rr = RetryRecorder{ .allocator = allocator }; - defer rr.deinit(); - 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); - try testing.expectEqual(@as(usize, 3), rr.infos.items.len); -} - -test "runStep: mid-stream failure retries then succeeds" { - const allocator = testing.allocator; - - // The open succeeds every time; the FIRST two responses fail during - // `produce` (mid-stream), the third is a normal scripted turn. - const stream_errs = [_]StubProvider.StreamFailure{ - .{ .err = error.ProviderOverloaded, .message = "overloaded_error: Overloaded" }, - .{ .err = error.ProviderOverloaded, .message = "overloaded_error: Overloaded" }, - }; - const scripted = [_]StubProvider.ScriptedTurn{ - .{ .blocks = &.{.{ .Text = "finally" }} }, - }; - var stub = StubProvider{ - .allocator = allocator, - .scripted = &scripted, - .stream_errors = &stream_errs, - }; - var threaded: std.Io.Threaded = .init(allocator, .{}); - defer threaded.deinit(); - const io = threaded.io(); - var h = TestHarness.init(allocator); - defer h.deinit(); - fastRetryHarness(&h); - var ns = null_store_mod.NullStore.init(allocator); - const agent = try Agent.init(allocator, io, &h.config, ns.store().create(), null); - defer agent.deinit(); - h.seedInto(agent); - agent._open_stream_fn = stub.install(); - - const conv = &agent.conversation; - - var rr = RetryRecorder{ .allocator = allocator }; - defer rr.deinit(); - try drainTurnRecording(agent, "hi", &rr); - - // Three opens: two that fail mid-stream + one that succeeds. - try testing.expectEqual(@as(usize, 3), stub.calls_made); - // No duplicate assistant messages: the mid-stream failures committed - // nothing, so we end with user + a single assistant. - try testing.expectEqual(@as(usize, 2), conv.messages.items.len); - try testing.expectEqual(conversation.MessageRole.assistant, conv.messages.items[1].role); - // Two retry notifications, with a monotonically increasing attempt count - // (the bug was that this stayed pinned at 1 forever). - try testing.expectEqual(@as(usize, 2), rr.infos.items.len); - try testing.expectEqual(@as(usize, 1), rr.infos.items[0].attempt); - try testing.expectEqual(@as(usize, 2), rr.infos.items[1].attempt); - // The provider's diagnostic is surfaced on the retry notice (was the - // whole point: the user should see "overloaded", not a bare error name). - try testing.expectEqualStrings("overloaded_error: Overloaded", rr.infos.items[0].message.?); -} - -test "runStep: mid-stream failures exhaust and hard-fail after max_attempts" { - const allocator = testing.allocator; - - // Every open succeeds but every response fails mid-stream. Without the - // attempt counter this would re-open forever; it must hard-fail after - // `max_attempts` opens. - const stream_errs = [_]StubProvider.StreamFailure{ - .{ .err = error.ProviderStreamMalformed }, - .{ .err = error.ProviderStreamMalformed }, - .{ .err = error.ProviderStreamMalformed }, - .{ .err = error.ProviderStreamMalformed }, - }; - const scripted = [_]StubProvider.ScriptedTurn{ - .{ .blocks = &.{.{ .Text = "unreachable" }} }, - }; - var stub = StubProvider{ - .allocator = allocator, - .scripted = &scripted, - .stream_errors = &stream_errs, - }; - var threaded: std.Io.Threaded = .init(allocator, .{}); - defer threaded.deinit(); - const io = threaded.io(); - var h = TestHarness.init(allocator); - defer h.deinit(); - fastRetryHarness(&h); - var ns = null_store_mod.NullStore.init(allocator); - const agent = try Agent.init(allocator, io, &h.config, ns.store().create(), null); - defer agent.deinit(); - h.seedInto(agent); - agent._open_stream_fn = stub.install(); - - var rr = RetryRecorder{ .allocator = allocator }; - defer rr.deinit(); - try testing.expectError(error.ProviderStreamMalformed, drainTurnRecording(agent, "hi", &rr)); - - // 4 opens total (max_attempts), 3 retry notifications — then hard-fail - // instead of looping endlessly. - try testing.expectEqual(@as(usize, 4), stub.calls_made); - try testing.expectEqual(@as(usize, 3), rr.infos.items.len); -} - -test "runStep: Retry-After is honored and reported" { - const allocator = testing.allocator; - - const errs = [_]StubProvider.ScriptedError{ - .{ .err = error.ProviderRateLimited, .status_code = 429, .retry_after_ms = 7000 }, - }; - const scripted = [_]StubProvider.ScriptedTurn{ - .{ .blocks = &.{.{ .Text = "ok" }} }, - }; - var stub = StubProvider{ - .allocator = allocator, - .scripted = &scripted, - .scripted_errors = &errs, - }; - var threaded: std.Io.Threaded = .init(allocator, .{}); - defer threaded.deinit(); - const io = threaded.io(); - var h = TestHarness.init(allocator); - defer h.deinit(); - h.activate(); - // Cap below the Retry-After to verify the policy cap applies. - h.config.retry = .{ .initial_delay_ms = 0, .max_delay_ms = 1, .jitter = false }; - var ns = null_store_mod.NullStore.init(allocator); - const agent = try Agent.init(allocator, io, &h.config, ns.store().create(), null); - defer agent.deinit(); - h.seedInto(agent); - agent._open_stream_fn = stub.install(); - - var rr = RetryRecorder{ .allocator = allocator }; - defer rr.deinit(); - try drainTurnRecording(agent, "hi", &rr); - - try testing.expectEqual(@as(usize, 1), rr.infos.items.len); - // Reported Retry-After is the raw provider value... - try testing.expectEqual(@as(?u64, 7000), rr.infos.items[0].retry_after_ms); - // ...but the actual delay is capped by policy.max_delay_ms. - try testing.expectEqual(@as(u64, 1), rr.infos.items[0].delay_ms); -} - -test "runStep: cancellation from a tool still hard-fails" { - const allocator = testing.allocator; - - const scripted = [_]StubProvider.ScriptedTurn{ - .{ .blocks = &.{ - .{ .ToolUse = .{ .id = "x", .name = "hard", .input = "" } }, - } }, - .{ .blocks = &.{.{ .Text = "unreachable" }} }, - }; - var stub = StubProvider{ .allocator = allocator, .scripted = &scripted }; - var threaded: std.Io.Threaded = .init(allocator, .{}); - defer threaded.deinit(); - const io = threaded.io(); - var h = TestHarness.init(allocator); - defer h.deinit(); - try h.registry.register(try HardFailTool.create(allocator, "hard")); - h.activate(); - var ns = null_store_mod.NullStore.init(allocator); - const agent = try Agent.init(allocator, io, &h.config, ns.store().create(), null); - defer agent.deinit(); - h.seedInto(agent); - agent._open_stream_fn = stub.install(); - - const conv = &agent.conversation; - - 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); -} - -test "runStep: source per-call error produces a per-call error result and continues" { - const allocator = testing.allocator; - - const scripted = [_]StubProvider.ScriptedTurn{ - .{ .blocks = &.{ - .{ .ToolUse = .{ .id = "a", .name = "pa", .input = "" } }, - .{ .ToolUse = .{ .id = "b", .name = "pb", .input = "" } }, - } }, - .{ .blocks = &.{.{ .Text = "moving on" }} }, - }; - var stub = StubProvider{ .allocator = allocator, .scripted = &scripted }; - var threaded: std.Io.Threaded = .init(allocator, .{}); - defer threaded.deinit(); - const io = threaded.io(); - var h = TestHarness.init(allocator); - defer h.deinit(); - try h.registry.registerSource(try PartialSource.create(allocator, "ps", &.{ "pa", "pb" })); - h.activate(); - var ns = null_store_mod.NullStore.init(allocator); - const agent = try Agent.init(allocator, io, &h.config, ns.store().create(), null); - defer agent.deinit(); - h.seedInto(agent); - agent._open_stream_fn = stub.install(); - - const conv = &agent.conversation; - - try drainTurn(agent, "go"); - - try testing.expectEqual(@as(usize, 4), conv.messages.items.len); - const tr_msg = conv.messages.items[2]; - const tr_a = tr_msg.content.items[0].ToolResult; // first call succeeded - const tr_b = tr_msg.content.items[1].ToolResult; // second failed - try testing.expect(!tr_a.is_error); - try testing.expectEqualStrings("ok", trText(tr_a)); - try testing.expect(tr_b.is_error); - try testing.expect(std.mem.indexOf(u8, trText(tr_b), "PerCallBoom") != null); -} - -test "runStep: context-overflow compaction fires a compaction retry notification" { - const allocator = testing.allocator; - - const scripted = [_]StubProvider.ScriptedTurn{ - .{ .blocks = &.{.{ .Text = "COMPACTED SUMMARY" }} }, // compaction call - .{ .blocks = &.{.{ .Text = "final answer" }} }, // retried main call - }; - var stub = StubProvider{ - .allocator = allocator, - .scripted = &scripted, - .overflow_calls = 1, - }; - var threaded: std.Io.Threaded = .init(allocator, .{}); - defer threaded.deinit(); - const io = threaded.io(); - var h = TestHarness.init(allocator); - defer h.deinit(); - h.activate(); - h.config.compaction = .{ .keep_verbatim = 10, .compaction_prompt = "Summarize the conversation." }; - var ns = null_store_mod.NullStore.init(allocator); - const agent = try Agent.init(allocator, io, &h.config, ns.store().create(), null); - defer agent.deinit(); - h.seedInto(agent); - agent._open_stream_fn = stub.install(); - - const conv = &agent.conversation; - try conv.addSystemMessage("you are helpful"); - try addUserText(conv, "first question with several words here"); - try conv.addAssistantMessage(&.{ - .{ .Text = try conversation.textualBlockFromSlice(allocator, "first answer with several words") }, - }, null); - - var rr = RetryRecorder{ .allocator = allocator }; - defer rr.deinit(); - try drainTurnRecording(agent, "second recent question", &rr); - - try testing.expect(agent._auto_compacted); - // Exactly one notification, flagged as a compaction retry with no delay. - try testing.expectEqual(@as(usize, 1), rr.infos.items.len); - try testing.expect(rr.infos.items[0].compaction); - try testing.expectEqual(@as(u64, 0), rr.infos.items[0].delay_ms); - try testing.expectEqual(error.ContextOverflow, rr.infos.items[0].err); -} |