//! 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 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");

pub const Tool = tool_mod.Tool;
pub const ToolSource = tool_source_mod.ToolSource;
pub const ToolRegistry = tool_registry_mod.ToolRegistry;

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;

/// 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 };
}

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;
            break :blk .{ .Thinking = .{ .text = tb, .signature = sig } };
        },
        .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),
        } },
    };
}

/// A minimal receiver that captures the assistant's streamed message for
/// compaction. We don't need incremental events — the assembled message is
/// read off the conversation after the turn — so all callbacks are no-ops.
const CompactionCapture = struct {
    allocator: Allocator,

    fn receiver(self: *CompactionCapture) provider_mod.Receiver {
        return .{ .ptr = self, .vtable = &vt };
    }
    fn deinit(self: *CompactionCapture) void {
        _ = self;
    }
    const vt: provider_mod.ReceiverVTable = .{
        .onMessageStart = onMessageStart,
        .onBlockStart = onBlockStart,
        .onToolDetails = onToolDetails,
        .onContentDelta = onContentDelta,
        .onBlockComplete = onBlockComplete,
        .onMessageComplete = onMessageComplete,
        .onError = onError,
        .onProviderRetry = onProviderRetry,
    };
    fn onMessageStart(_: *anyopaque, _: conversation.MessageRole) anyerror!void {}
    fn onBlockStart(_: *anyopaque, _: provider_mod.ContentBlockType, _: usize) anyerror!void {}
    fn onToolDetails(_: *anyopaque, _: usize, _: []const u8, _: []const u8) anyerror!void {}
    fn onContentDelta(_: *anyopaque, _: usize, _: []const u8) anyerror!void {}
    fn onBlockComplete(_: *anyopaque, _: usize, _: conversation.ContentBlock) anyerror!void {}
    fn onMessageComplete(_: *anyopaque, _: conversation.Message, _: ?provider_mod.Usage) anyerror!void {}
    fn onError(_: *anyopaque, _: anyerror) void {}
    fn onProviderRetry(_: *anyopaque, _: provider_mod.ProviderRetryInfo) void {}
};

fn isValidToolInput(input: []const u8) bool {
    if (input.len == 0) return true;
    if (input[0] != '{') return true; // legacy tests/tools may use opaque bytes
    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.ResultPart {
    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.ownedTextResult(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.ResultPart {
    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.ownedTextResult(allocator, msg);
}


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 and/or
    /// the visible tool set atomically. The pointee and its registry are
    /// owned by the embedder, not the agent.
    config: *const Config,
    /// Injectable streaming seam. Defaults to the real provider dispatch
    /// (`provider_mod.streamStep`); tests override it with a stub.
    stream_fn: provider_mod.StreamFn = provider_mod.streamStep,
    /// Compaction system prompt used for automatic compaction on context
    /// overflow. Borrowed; set by the embedder (resolved from its
    /// `COMPACTION.md` layers). When null, auto-compaction is disabled and
    /// a context-overflow error propagates unchanged.
    compaction_system_prompt: ?[]const u8 = null,
    /// 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,

    pub fn init(allocator: Allocator, io: Io, config: *const Config) Agent {
        return .{
            .allocator = allocator,
            .io = io,
            .config = config,
        };
    }

    pub fn deinit(self: *Agent) void {
        // The agent owns neither the config snapshot nor the registry it
        // borrows; the embedder tears those down.
        _ = self;
    }

    /// 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;
    }

    /// The registry exposed by the active snapshot.
    pub fn registry(self: *const Agent) *const ToolRegistry {
        return self.config.registry;
    }

    /// Drive the conversation forward until the model stops calling tools.
    pub fn runStep(
        self: *Agent,
        conv: *conversation.Conversation,
        receiver: *provider_mod.Receiver,
    ) !void {
        self.auto_compacted = false;
        while (true) {
            // Re-read the config snapshot at the top of each turn so a
            // mid-conversation swap takes effect here, never mid-stream.
            const cfg = self.config;
            try self.streamWithRetries(cfg, conv, receiver);

            const last = conv.messages.items[conv.messages.items.len - 1];
            std.debug.assert(last.role == .assistant);

            // Defense-in-depth: a provider that silently committed an
            // empty assistant message means the turn made no observable
            // progress. Surface it instead of looping back to the prompt.
            if (last.content.items.len == 0) return error.EmptyAssistantResponse;

            if (!hasToolUseBlock(last)) return;

            try self.dispatchToolCalls(conv, last);
        }
    }

    fn hasToolUseBlock(msg: conversation.Message) bool {
        for (msg.content.items) |block| {
            if (block == .ToolUse) return true;
        }
        return false;
    }

    /// Drive one provider turn with the configured retry policy.
    ///
    /// Decision path for a failed attempt:
    ///   - `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 attempt never mutates the conversation (providers commit the
    /// assistant message only on success), so each retry runs against the
    /// same snapshot.
    fn streamWithRetries(
        self: *Agent,
        cfg: *const Config,
        conv: *conversation.Conversation,
        receiver: *provider_mod.Receiver,
    ) !void {
        const policy = cfg.retry;
        var attempt: usize = 1;
        while (true) {
            var diag: provider_mod.ProviderDiagnostic = .{};
            self.stream_fn(self.allocator, self.io, cfg, conv, receiver, &diag) catch |err| {
                if (err == error.ContextOverflow) {
                    try self.handleContextOverflow(cfg, conv, receiver, err);
                    return;
                }
                if (!provider_mod.isRetryableProviderError(err)) return err;
                // Out of attempts: hard-fail with the last error.
                if (attempt >= policy.max_attempts) return err;

                const delay_ms = self.backoffDelayMs(policy, attempt, diag.retry_after_ms);
                receiver.onProviderRetry(.{
                    .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;
        }
    }

    /// One-shot context-overflow recovery: compact once, retry once. Mirrors
    /// the prior inline behavior, now fired from `streamWithRetries`. The
    /// retry is announced through `onProviderRetry` with `compaction = true`
    /// and `delay_ms = 0`.
    fn handleContextOverflow(
        self: *Agent,
        cfg: *const Config,
        conv: *conversation.Conversation,
        receiver: *provider_mod.Receiver,
        err: anyerror,
    ) !void {
        if (self.auto_compacted) return err; // already retried once this turn
        const sys = self.compaction_system_prompt orelse return err;
        const res = try self.compact(conv, sys, null);
        if (!res.compacted) return err; // nothing to shed; give up
        self.auto_compacted = true;
        receiver.onProviderRetry(.{
            .attempt = 1,
            .max_attempts = 2,
            .delay_ms = 0,
            .err = err,
            .compaction = true,
        });
        // Retry the same request against the compacted context. A second
        // overflow (or any other error) propagates.
        var diag: provider_mod.ProviderDiagnostic = .{};
        try self.stream_fn(self.allocator, self.io, cfg, conv, receiver, &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;
    }

    /// 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,
    };

    /// Compact the conversation: summarize an older prefix into a single
    /// `.CompactionSummary` block and keep a recent suffix of whole turns
    /// verbatim. Mutates `conv` in place. The embedder is responsible for
    /// persisting the resulting new messages (the agent never touches the
    /// session log).
    ///
    /// 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).
    pub fn compact(
        self: *Agent,
        conv: *conversation.Conversation,
        system_prompt: []const u8,
        extra_instructions: ?[]const u8,
    ) !CompactionResult {
        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);

        try self.rewriteWithSummary(conv, split.prefix_end, summary);

        return .{
            .compacted = true,
            .kept_turns = split.kept_turns,
            .summarized_messages = summarized,
        };
    }

    /// 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,
    ) !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.
        for (old[prefix_end..]) |*m| {
            try rebuilt.append(alloc, try cloneMessage(alloc, m.*));
        }

        // 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;
    }

    /// 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,
    ) ![]u8 {
        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,
                .registry = &empty_registry,
                .compaction = self.config.compaction,
            };
            if (self.runSingleCompactionTurn(&cfg, 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,
            .registry = &empty_registry,
            .compaction = self.config.compaction,
        };
        return self.runSingleCompactionTurn(&cfg, sys_text, body);
    }

    /// 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.
    fn runSingleCompactionTurn(
        self: *Agent,
        cfg: *const config_mod.Config,
        system_prompt: []const u8,
        body: []const u8,
    ) ![]u8 {
        const alloc = self.allocator;
        var conv = conversation.Conversation.init(alloc);
        defer conv.deinit();
        try conv.addSystemMessage(system_prompt);
        try conv.addUserMessage(body);

        var capture = CompactionCapture{ .allocator = alloc };
        defer capture.deinit();
        var recv = capture.receiver();

        try self.stream_fn(alloc, self.io, cfg, &conv, &recv, null);

        // 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;
        return out.toOwnedSlice(alloc);
    }

    /// 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,
        conv: *conversation.Conversation,
        assistant_msg: conversation.Message,
    ) !void {
        // 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.config.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| tool_mod.freeResultParts(self.allocator, r);
            }
        }

        // 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| tool_mod.freeResultParts(self.allocator, r);
                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| {
                tool_mod.freeResultParts(self.allocator, r);
                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 tool_mod.freeResultParts(self.allocator, result_parts);

            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.len);
            for (result_parts) |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,
        });
    }
};

/// 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.ResultPart,

    /// 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| tool_mod.freeResultParts(agent.allocator, b),
            .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: 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 `stream_fn` seam.
///
/// `provider_mod.StreamFn` carries no user context (it mirrors the real
/// free function exactly), so the stub parks its state in a module-level
/// pointer that `stubStreamStep` reads. The Zig test runner executes tests
/// serially in one process, so a single global slot is safe; each test
/// sets it via `install` before driving the agent.
var stub_active: ?*StubProvider = null;

const StubProvider = struct {
    allocator: Allocator,
    scripted: []const ScriptedTurn,
    next: usize = 0,
    /// Number of leading stream calls that should fail with
    /// `error.ContextOverflow` before any scripted turn is served. Used to
    /// drive the auto-compaction path. Decremented on each overflow.
    overflow_calls: usize = 0,
    /// A queue of provider errors to return, in order, before any scripted
    /// turn is served. Each entry is consumed on one stream call. Used to
    /// drive the transient-retry path. `diag_retry_after_ms`, when set on an
    /// entry, is stashed into the caller's `ProviderDiagnostic`.
    scripted_errors: []const ScriptedError = &.{},
    error_idx: usize = 0,
    /// Count of stream calls 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 ScriptedTurn = struct {
        blocks: []const TestBlock,
    };

    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.stream_fn`. Call once per test, after constructing the
    /// stub on the stack.
    fn install(self: *StubProvider) provider_mod.StreamFn {
        stub_active = self;
        return stubStreamStep;
    }
};

fn stubStreamStep(
    allocator: Allocator,
    _: Io,
    _: *const config_mod.Config,
    conv: *conversation.Conversation,
    _: *provider_mod.Receiver,
    diag: ?*provider_mod.ProviderDiagnostic,
) anyerror!void {
    const self = stub_active orelse return error.NoStubInstalled;
    _ = allocator;
    self.calls_made += 1;
    if (self.error_idx < self.scripted_errors.len) {
        const e = self.scripted_errors[self.error_idx];
        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.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;

    var blocks: std.ArrayList(conversation.ContentBlock) = .empty;
    errdefer {
        for (blocks.items) |*b| b.deinit(self.allocator);
        blocks.deinit(self.allocator);
    }
    for (turn.blocks) |tb| {
        switch (tb) {
            .Text => |s| {
                try blocks.append(self.allocator, .{
                    .Text = try conversation.textualBlockFromSlice(self.allocator, s),
                });
            },
            .ToolUse => |tu| {
                const id = try self.allocator.dupe(u8, tu.id);
                errdefer self.allocator.free(id);
                const name = try self.allocator.dupe(u8, tu.name);
                errdefer self.allocator.free(name);
                var input_buf: conversation.TextualBlock = .empty;
                errdefer input_buf.deinit(self.allocator);
                try input_buf.appendSlice(self.allocator, tu.input);
                try blocks.append(self.allocator, .{ .ToolUse = .{
                    .id = id,
                    .name = name,
                    .input = input_buf,
                } });
            },
        }
    }
    const moved = try blocks.toOwnedSlice(self.allocator);
    defer self.allocator.free(moved);
    try conv.addAssistantMessage(moved);
}

/// Build a stack registry + active `Config` snapshot wired together, for
/// tests that drive the agent. 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 to point at this harness's registry.
    /// 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" } },
            .registry = &self.registry,
        };
    }

    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.ResultPart {
        const self: *EchoTool = @ptrCast(@alignCast(ctx));
        const msg = try std.fmt.allocPrint(allocator, "{s}{s}", .{ self.prefix_owned, input });
        return tool_mod.ownedTextResult(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.ResultPart {
        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.textResult(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.ResultPart {
        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.ResultPart {
        return error.Canceled;
    }

    fn deinit(ctx: *anyopaque, allocator: Allocator) void {
        const self: *HardFailTool = @ptrCast(@alignCast(ctx));
        allocator.free(self.name_owned);
        allocator.destroy(self);
    }
};

const NoopReceiver = struct {
    fn make() provider_mod.Receiver {
        return .{ .ptr = @ptrCast(@constCast(&dummy)), .vtable = &vt };
    }
    var dummy: u8 = 0;
    const vt: provider_mod.ReceiverVTable = .{
        .onMessageStart = noop1,
        .onBlockStart = noop2,
        .onToolDetails = noopToolDetails,
        .onContentDelta = noop3,
        .onBlockComplete = noop4,
        .onMessageComplete = noop5,
        .onError = noop6,
        .onProviderRetry = noop7,
    };
    fn noop1(_: *anyopaque, _: conversation.MessageRole) anyerror!void {}
    fn noop2(_: *anyopaque, _: provider_mod.ContentBlockType, _: usize) anyerror!void {}
    fn noopToolDetails(_: *anyopaque, _: usize, _: []const u8, _: []const u8) anyerror!void {}
    fn noop3(_: *anyopaque, _: usize, _: []const u8) anyerror!void {}
    fn noop4(_: *anyopaque, _: usize, _: conversation.ContentBlock) anyerror!void {}
    fn noop5(_: *anyopaque, _: conversation.Message, _: ?provider_mod.Usage) anyerror!void {}
    fn noop6(_: *anyopaque, _: anyerror) void {}
    fn noop7(_: *anyopaque, _: provider_mod.ProviderRetryInfo) void {}
};

/// 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.ownedTextResult(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.textResult(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 agent = Agent.init(allocator, io, &h.config);
    agent.stream_fn = stub.install();

    var conv = conversation.Conversation.init(allocator);
    defer conv.deinit();
    try conv.addUserMessage("call a tool");

    var recv = NoopReceiver.make();
    try agent.runStep(&conv, &recv);

    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 "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 agent = Agent.init(allocator, io, &h.config);
    agent.stream_fn = stub.install();

    var conv = conversation.Conversation.init(allocator);
    defer conv.deinit();
    try conv.addUserMessage("go");

    var recv = NoopReceiver.make();
    try agent.runStep(&conv, &recv);

    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 agent = Agent.init(allocator, io, &h.config);
    agent.stream_fn = stub.install();

    var conv = conversation.Conversation.init(allocator);
    defer conv.deinit();
    try conv.addUserMessage("break it");

    var recv = NoopReceiver.make();
    try agent.runStep(&conv, &recv);

    // 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 agent = Agent.init(allocator, io, &h.config);
    agent.stream_fn = stub.install();

    var conv = conversation.Conversation.init(allocator);
    defer conv.deinit();
    try conv.addUserMessage("call a ghost");

    var recv = NoopReceiver.make();
    try agent.runStep(&conv, &recv);

    // 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 agent = Agent.init(allocator, io, &h.config);
    agent.stream_fn = stub.install();

    var conv = conversation.Conversation.init(allocator);
    defer conv.deinit();
    try conv.addUserMessage("hello");

    var recv = NoopReceiver.make();
    try agent.runStep(&conv, &recv);

    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 agent = Agent.init(allocator, io, &h.config);
    agent.stream_fn = stub.install();

    var conv = conversation.Conversation.init(allocator);
    defer conv.deinit();
    try conv.addUserMessage("hi");

    var recv = NoopReceiver.make();
    try testing.expectError(error.EmptyAssistantResponse, agent.runStep(&conv, &recv));
}

// ------------ 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 agent = Agent.init(allocator, io, &h.config);
    agent.stream_fn = stub.install();

    var conv = conversation.Conversation.init(allocator);
    defer conv.deinit();
    try conv.addUserMessage("go");

    var recv = NoopReceiver.make();
    try agent.runStep(&conv, &recv);

    // Locate the source and inspect its observed batches.
    const view = h.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 agent = Agent.init(allocator, io, &h.config);
    agent.stream_fn = stub.install();

    var conv = conversation.Conversation.init(allocator);
    defer conv.deinit();
    try conv.addUserMessage("go");

    var recv = NoopReceiver.make();
    try agent.runStep(&conv, &recv);

    const view_a = h.registry.lookup("src_a_t") orelse return error.NotFound;
    const view_b = h.registry.lookup("src_b_t") orelse return error.NotFound;
    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 agent = Agent.init(allocator, io, &h.config);
    agent.stream_fn = stub.install();

    var conv = conversation.Conversation.init(allocator);
    defer conv.deinit();
    try conv.addUserMessage("kaboom");

    var recv = NoopReceiver.make();
    try agent.runStep(&conv, &recv);

    // 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 agent = Agent.init(allocator, io, &h.config);
    agent.stream_fn = stub.install();

    var conv = conversation.Conversation.init(allocator);
    defer conv.deinit();
    try conv.addUserMessage("go");

    var recv = NoopReceiver.make();
    try agent.runStep(&conv, &recv);

    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 the visible tool set between turns" {
    // The core RCU promise: the agent reads `*const Config` fresh each
    // turn, so swapping the pointer mid-conversation changes the tool set
    // the next turn sees. Config A exposes only `echo`; config B only
    // `late`. After `setConfig(&cfg_b)`, a turn that calls `late` resolves
    // — proving both the swap and per-turn re-consultation.
    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();

    // Config A: only `echo`.
    var reg_a = ToolRegistry.init(allocator);
    defer reg_a.deinit();
    try reg_a.register(try EchoTool.create(allocator, "echo", "A:"));
    const cfg_a: config_mod.Config = .{
        .provider = .{ .openai_chat = .{ .api_key = "k", .base_url = "u", .model = "m" } },
        .registry = &reg_a,
    };

    // Config B: only `late`.
    var reg_b = ToolRegistry.init(allocator);
    defer reg_b.deinit();
    try reg_b.register(try EchoTool.create(allocator, "late", "B:"));
    const cfg_b: config_mod.Config = .{
        .provider = .{ .openai_chat = .{ .api_key = "k", .base_url = "u", .model = "m" } },
        .registry = &reg_b,
    };

    var agent = Agent.init(allocator, io, &cfg_a);
    agent.stream_fn = stub.install();

    // Under A: `echo` visible, `late` not.
    try testing.expect(agent.config.registry.lookup("echo") != null);
    try testing.expect(agent.config.registry.lookup("late") == null);

    // Swap. Under B: the visibility inverts.
    agent.setConfig(&cfg_b);
    try testing.expect(agent.config.registry.lookup("echo") == null);
    try testing.expect(agent.config.registry.lookup("late") != null);

    // A real turn under B resolves `late` (which would have been
    // UnknownTool under A), then loops to the final text turn.
    var conv = conversation.Conversation.init(allocator);
    defer conv.deinit();
    try conv.addUserMessage("go");
    var recv = NoopReceiver.make();
    try agent.runStep(&conv, &recv);

    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 agent = Agent.init(allocator, io, &h.config);
    agent.stream_fn = stub.install();

    var conv = conversation.Conversation.init(allocator);
    defer conv.deinit();
    try conv.addSystemMessage("you are helpful");
    try conv.addUserMessage("first question here with several words");
    try conv.addAssistantMessage(&.{
        .{ .Text = try conversation.textualBlockFromSlice(allocator, "first answer with several words") },
    });
    try conv.addUserMessage("second recent question");
    try conv.addAssistantMessage(&.{
        .{ .Text = try conversation.textualBlockFromSlice(allocator, "second recent answer") },
    });

    const res = try agent.compact(&conv, "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 "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 agent = Agent.init(allocator, io, &h.config);
    agent.stream_fn = stub.install();

    var conv = conversation.Conversation.init(allocator);
    defer conv.deinit();
    try conv.addSystemMessage("sys");
    try conv.addUserMessage("hi");
    try conv.addAssistantMessage(&.{
        .{ .Text = try conversation.textualBlockFromSlice(allocator, "hello") },
    });

    const res = try agent.compact(&conv, "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 agent = Agent.init(allocator, io, &h.config);
    agent.stream_fn = stub.install();

    var conv = conversation.Conversation.init(allocator);
    defer conv.deinit();
    try conv.addUserMessage("question one two three");
    try conv.addAssistantMessage(&.{
        .{ .Text = try conversation.textualBlockFromSlice(allocator, "answer one two three") },
    });
    try conv.addUserMessage("question two");
    try conv.addAssistantMessage(&.{
        .{ .Text = try conversation.textualBlockFromSlice(allocator, "answer two") },
    });

    const res = try agent.compact(&conv, "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 };
    var agent = Agent.init(allocator, io, &h.config);
    agent.stream_fn = stub.install();
    agent.compaction_system_prompt = "Summarize the conversation.";

    var conv = conversation.Conversation.init(allocator);
    defer conv.deinit();
    try conv.addSystemMessage("you are helpful");
    try conv.addUserMessage("first question with several words here");
    try conv.addAssistantMessage(&.{
        .{ .Text = try conversation.textualBlockFromSlice(allocator, "first answer with several words") },
    });
    try conv.addUserMessage("second recent question");

    var recv = NoopReceiver.make();
    try agent.runStep(&conv, &recv);

    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 agent = Agent.init(allocator, io, &h.config);
    agent.stream_fn = stub.install();
    // No compaction_system_prompt set -> overflow propagates.

    var conv = conversation.Conversation.init(allocator);
    defer conv.deinit();
    try conv.addUserMessage("hi");

    var recv = NoopReceiver.make();
    try testing.expectError(error.ContextOverflow, agent.runStep(&conv, &recv));
}

// -----------------------------------------------------------------------------
// Phase 6: provider retry + tool-error holistic tests
// -----------------------------------------------------------------------------

/// Receiver that records `onProviderRetry` notifications (and nothing else),
/// so tests can assert retry scheduling without a live provider.
const RetryRecordingReceiver = struct {
    infos: std.ArrayList(provider_mod.ProviderRetryInfo) = .empty,
    allocator: Allocator,

    fn make(self: *RetryRecordingReceiver) provider_mod.Receiver {
        return .{ .ptr = self, .vtable = &vt };
    }
    fn deinit(self: *RetryRecordingReceiver) void {
        self.infos.deinit(self.allocator);
    }
    const vt: provider_mod.ReceiverVTable = .{
        .onMessageStart = onMessageStart,
        .onBlockStart = onBlockStart,
        .onToolDetails = onToolDetails,
        .onContentDelta = onContentDelta,
        .onBlockComplete = onBlockComplete,
        .onMessageComplete = onMessageComplete,
        .onError = onError,
        .onProviderRetry = onProviderRetry,
    };
    fn onMessageStart(_: *anyopaque, _: conversation.MessageRole) anyerror!void {}
    fn onBlockStart(_: *anyopaque, _: provider_mod.ContentBlockType, _: usize) anyerror!void {}
    fn onToolDetails(_: *anyopaque, _: usize, _: []const u8, _: []const u8) anyerror!void {}
    fn onContentDelta(_: *anyopaque, _: usize, _: []const u8) anyerror!void {}
    fn onBlockComplete(_: *anyopaque, _: usize, _: conversation.ContentBlock) anyerror!void {}
    fn onMessageComplete(_: *anyopaque, _: conversation.Message, _: ?provider_mod.Usage) anyerror!void {}
    fn onError(_: *anyopaque, _: anyerror) void {}
    fn onProviderRetry(ptr: *anyopaque, info: provider_mod.ProviderRetryInfo) void {
        const self: *RetryRecordingReceiver = @ptrCast(@alignCast(ptr));
        self.infos.append(self.allocator, info) catch {};
    }
};

/// 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 agent = Agent.init(allocator, io, &h.config);
    agent.stream_fn = stub.install();

    var conv = conversation.Conversation.init(allocator);
    defer conv.deinit();
    try conv.addUserMessage("hi");

    var rr = RetryRecordingReceiver{ .allocator = allocator };
    defer rr.deinit();
    var recv = rr.make();
    try agent.runStep(&conv, &recv);

    // 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 agent = Agent.init(allocator, io, &h.config);
    agent.stream_fn = stub.install();

    var conv = conversation.Conversation.init(allocator);
    defer conv.deinit();
    try conv.addUserMessage("hi");

    var rr = RetryRecordingReceiver{ .allocator = allocator };
    defer rr.deinit();
    var recv = rr.make();
    try agent.runStep(&conv, &recv);

    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 agent = Agent.init(allocator, io, &h.config);
    agent.stream_fn = stub.install();

    var conv = conversation.Conversation.init(allocator);
    defer conv.deinit();
    try conv.addUserMessage("hi");

    var rr = RetryRecordingReceiver{ .allocator = allocator };
    defer rr.deinit();
    var recv = rr.make();
    try testing.expectError(error.ProviderAuthFailed, agent.runStep(&conv, &recv));

    // 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 agent = Agent.init(allocator, io, &h.config);
    agent.stream_fn = stub.install();

    var conv = conversation.Conversation.init(allocator);
    defer conv.deinit();
    try conv.addUserMessage("hi");

    var rr = RetryRecordingReceiver{ .allocator = allocator };
    defer rr.deinit();
    var recv = rr.make();
    try testing.expectError(error.ProviderUnavailable, agent.runStep(&conv, &recv));

    // 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: 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 agent = Agent.init(allocator, io, &h.config);
    agent.stream_fn = stub.install();

    var conv = conversation.Conversation.init(allocator);
    defer conv.deinit();
    try conv.addUserMessage("hi");

    var rr = RetryRecordingReceiver{ .allocator = allocator };
    defer rr.deinit();
    var recv = rr.make();
    try agent.runStep(&conv, &recv);

    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 agent = Agent.init(allocator, io, &h.config);
    agent.stream_fn = stub.install();

    var conv = conversation.Conversation.init(allocator);
    defer conv.deinit();
    try conv.addUserMessage("go");

    var recv = NoopReceiver.make();
    try testing.expectError(error.Canceled, agent.runStep(&conv, &recv));
    // 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 agent = Agent.init(allocator, io, &h.config);
    agent.stream_fn = stub.install();

    var conv = conversation.Conversation.init(allocator);
    defer conv.deinit();
    try conv.addUserMessage("go");

    var recv = NoopReceiver.make();
    try agent.runStep(&conv, &recv);

    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 };
    var agent = Agent.init(allocator, io, &h.config);
    agent.stream_fn = stub.install();
    agent.compaction_system_prompt = "Summarize the conversation.";

    var conv = conversation.Conversation.init(allocator);
    defer conv.deinit();
    try conv.addSystemMessage("you are helpful");
    try conv.addUserMessage("first question with several words here");
    try conv.addAssistantMessage(&.{
        .{ .Text = try conversation.textualBlockFromSlice(allocator, "first answer with several words") },
    });
    try conv.addUserMessage("second recent question");

    var rr = RetryRecordingReceiver{ .allocator = allocator };
    defer rr.deinit();
    var recv = rr.make();
    try agent.runStep(&conv, &recv);

    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);
}