Move libpanto projects to libpantograph dependencyHEADmain

Remove the in-repo libpanto sources and binding projects from pantograph.
Consume libpantograph through the Zig package URL at
code.tjp.lol/libpantograph.git, including the Lua module artifact used by
CLI extensions.

1 files changed, 0 insertions, 3925 deletions

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