From f83578fdc9264019a1a1cef8c5484a161167d3dd Mon Sep 17 00:00:00 2001 From: t Date: Tue, 7 Jul 2026 11:26:32 -0600 Subject: initial commit, moved libpanto over from the pantograph repo --- src/agent.zig | 3925 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 3925 insertions(+) create mode 100644 src/agent.zig (limited to 'src/agent.zig') diff --git a/src/agent.zig b/src/agent.zig new file mode 100644 index 0000000..09b0994 --- /dev/null +++ b/src/agent.zig @@ -0,0 +1,3925 @@ +//! 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); +} -- cgit v1.3