yoagent 0.23.0

//! The core agent loop: prompt → LLM stream → tool execution → repeat.
//!
//! This is the heart of yoagent:
//!
//! - `agent_loop()` starts with new prompt messages
//! - `agent_loop_continue()` resumes from existing context
//!
//! Both return a stream of `AgentEvent`s.

use crate::context::{
    self, CompactionStrategy, ContextConfig, ContextTracker, DefaultCompaction, ExecutionLimits,
    ExecutionTracker,
};
use crate::provider::{
    ModelConfig, ProviderError, StreamConfig, StreamEvent, StreamProvider, ToolDefinition,
};
use crate::types::*;
use std::sync::Arc;

/// Type alias for convert_to_llm callback.
pub type ConvertToLlmFn = Box<dyn Fn(&[AgentMessage]) -> Vec<Message> + Send + Sync>;
/// Type alias for transform_context callback.
pub type TransformContextFn = Box<dyn Fn(Vec<AgentMessage>) -> Vec<AgentMessage> + Send + Sync>;
/// Type alias for steering/follow-up message callbacks.
pub type GetMessagesFn = Box<dyn Fn() -> Vec<AgentMessage> + Send + Sync>;
/// Called before each LLM turn. Return `false` to abort the loop.
pub type BeforeTurnFn = Arc<dyn Fn(&[AgentMessage], usize) -> bool + Send + Sync>;
/// Called after each LLM turn with the current messages and the turn's usage.
pub type AfterTurnFn = Arc<dyn Fn(&[AgentMessage], &Usage) + Send + Sync>;
/// Called when the LLM returns a `StopReason::Error`.
pub type OnErrorFn = Arc<dyn Fn(&str) + Send + Sync>;
use tokio::sync::mpsc;
use tracing::warn;

/// Configuration for the agent loop
pub struct AgentLoopConfig {
    pub provider: Arc<dyn StreamProvider>,
    pub model: String,
    pub api_key: String,
    pub thinking_level: ThinkingLevel,
    pub max_tokens: Option<u32>,
    pub temperature: Option<f32>,

    /// Optional model configuration for multi-provider support.
    /// When set, passed through to StreamConfig so providers can use
    /// base_url, headers, compat flags, etc.
    pub model_config: Option<ModelConfig>,

    /// Convert AgentMessage[] → Message[] before each LLM call.
    /// Default: keep only LLM-compatible messages.
    pub convert_to_llm: Option<ConvertToLlmFn>,

    /// Transform context before convert_to_llm (for pruning, compaction).
    pub transform_context: Option<TransformContextFn>,

    /// Get steering messages (user interruptions mid-run).
    pub get_steering_messages: Option<GetMessagesFn>,

    /// Get follow-up messages (queued work after agent finishes).
    pub get_follow_up_messages: Option<GetMessagesFn>,

    /// Context window configuration (auto-compaction).
    pub context_config: Option<ContextConfig>,

    /// Custom compaction strategy. When set, replaces the default
    /// `compact_messages()` call. Invoked when `context_config` is `Some`.
    pub compaction_strategy: Option<Arc<dyn CompactionStrategy>>,

    /// Execution limits (max turns, tokens, duration).
    pub execution_limits: Option<ExecutionLimits>,

    /// Prompt caching configuration.
    pub cache_config: CacheConfig,
    /// Where to stash the full text of a truncated tool result, so the model
    /// can retrieve what head-tail truncation elided.
    ///
    /// `None` (default) means truncation behaves exactly as before: the middle
    /// is gone and only the event stream, which the agent cannot read, still
    /// has it. Set it and the marker names a `shared_state` key instead.
    ///
    /// The pointer lives in the transcript, so it is lost when lossy compaction
    /// drops that turn — the stash entry outlives it and keeps consuming the
    /// backend's cap. Retrieval is best-effort by construction.
    pub tool_output_sink: Option<crate::shared_state::SharedState>,

    /// Tool execution strategy (sequential, parallel, or batched).
    pub tool_execution: ToolExecutionStrategy,

    /// Tool middleware chain — approve/deny/modify every tool call before it
    /// executes (see [`ToolMiddleware`]). Empty = allow all.
    pub tool_middleware: Vec<Arc<dyn ToolMiddleware>>,

    /// Structured-output constraint, passed through to the provider (see
    /// [`OutputSchema`](crate::provider::OutputSchema)). Usually set via
    /// [`Agent::prompt_structured`](crate::Agent::prompt_structured).
    pub output_schema: Option<crate::provider::OutputSchema>,

    /// Retry configuration for transient provider errors.
    pub retry_config: crate::retry::RetryConfig,

    /// Called before each LLM turn. Return `false` to abort the loop.
    pub before_turn: Option<BeforeTurnFn>,
    /// Called after each LLM turn with the current messages and the turn's usage.
    pub after_turn: Option<AfterTurnFn>,
    /// Called when the LLM returns a `StopReason::Error`.
    pub on_error: Option<OnErrorFn>,

    /// Input filters applied to user messages before the LLM call.
    /// Filters run in order; first `Reject` wins and discards any accumulated
    /// warnings. `Warn` messages accumulate and are appended to the user message.
    /// An [`AsyncFilter`] in the list is awaited (see [`InputFilter::as_async`]).
    pub input_filters: Vec<Arc<dyn InputFilter>>,

    /// Optional delay between turns. Useful for rate-limit-sensitive scenarios
    /// (e.g., OAuth tokens with low request-per-minute caps). Skipped on the
    /// first turn so the agent starts immediately.
    pub turn_delay: Option<std::time::Duration>,
}

/// Default convert_to_llm: keep only user/assistant/toolResult messages.
fn default_convert_to_llm(messages: &[AgentMessage]) -> Vec<Message> {
    messages
        .iter()
        .filter_map(|m| m.as_llm().cloned())
        .collect()
}

/// Start an agent loop with new prompt messages.
/// Prefix of the message the loop appends when it stops a run itself.
///
/// A run stopped by a limit or by loop detection ends with a `Message::User`
/// carrying this prefix. The last *assistant* message still reports whatever
/// stop reason it had — `ToolUse` for a loop abort — so a consumer inspecting
/// only assistant messages cannot tell a stopped run from a finished one.
/// `SubAgentTool` uses this to avoid reporting an aborted delegation as a
/// bland success.
pub const AGENT_STOPPED_PREFIX: &str = "[Agent stopped:";

/// The stop marker for a run halted by loop detection specifically.
///
/// Distinct from the limit stops because the two mean opposite things to a
/// caller. Hitting `max_turns` is a bound: the work was cut short but what it
/// produced is real, and `SubAgentTool` returns it. A loop abort is a failure:
/// the model was emitting the same call forever and there is nothing to keep.
pub const LOOP_ABORT_PREFIX: &str = "[Agent stopped: repeated tool call —";

/// Prefix of the user-role nudge the loop injects when loop detection steers
/// a model that keeps repeating one call.
pub(crate) const LOOP_NUDGE_PREFIX: &str = "[You have called ";

/// Per-run state the loop shares with the hooks that run inside its task.
#[derive(Default)]
struct LoopScope {
    /// Decision-model spend, recorded by the decision integrations and
    /// folded into the run's `SessionStats`.
    decision: DecisionStats,
    /// The user messages this run was given — its prompts, steering and
    /// follow-ups — kept apart from the context so compaction cannot remove
    /// them. Exposed as `ToolCallRequest::run_prompts` /
    /// `TurnContext::run_prompts`.
    prompts: Vec<Message>,
}

tokio::task_local! {
    /// The running loop's [`LoopScope`]. Hooks run inside the loop's task; a
    /// nested sub-agent loop opens its own scope.
    static LOOP_SCOPE: std::cell::RefCell<LoopScope>;
}

/// Record decision-model spend into the enclosing loop's stats. A no-op
/// outside a loop.
#[cfg_attr(not(feature = "decision"), allow(dead_code))]
pub(crate) fn record_decision(f: impl FnOnce(&mut DecisionStats)) {
    let _ = LOOP_SCOPE.try_with(|cell| f(&mut cell.borrow_mut().decision));
}

/// The user messages the enclosing loop's run was given so far (empty
/// outside a loop).
pub(crate) fn run_prompts() -> Vec<Message> {
    LOOP_SCOPE
        .try_with(|cell| cell.borrow().prompts.clone())
        .unwrap_or_default()
}

/// The user-role LLM messages among `messages`.
fn user_messages(messages: &[AgentMessage]) -> Vec<Message> {
    messages
        .iter()
        .filter_map(|m| match m {
            AgentMessage::Llm(msg @ Message::User { .. }) => Some(msg.clone()),
            _ => None,
        })
        .collect()
}

/// Note messages handed to the run (steering, follow-ups) as run prompts.
fn note_run_prompts(messages: &[AgentMessage]) {
    let users = user_messages(messages);
    if !users.is_empty() {
        let _ = LOOP_SCOPE.try_with(|cell| cell.borrow_mut().prompts.extend(users));
    }
}

/// Run `fut` in a fresh [`LoopScope`] seeded with the run's prompts, and
/// return its output plus the decision spend recorded.
async fn with_loop_scope<T>(
    prompts: Vec<Message>,
    fut: impl std::future::Future<Output = T>,
) -> (T, DecisionStats) {
    let scope = LoopScope {
        prompts,
        ..Default::default()
    };
    LOOP_SCOPE
        .scope(std::cell::RefCell::new(scope), async {
            let out = fut.await;
            let recorded = LOOP_SCOPE.with(|cell| std::mem::take(&mut cell.borrow_mut().decision));
            (out, recorded)
        })
        .await
}

/// The error tool result given to each tool call in a response that ended as
/// [`StopReason::Refusal`] — the model declined, or a content filter stopped
/// the response. The call is never executed.
const REFUSAL_TOOL_RESULT_TEXT: &str = "Tool call not run: the response was stopped as a refusal \
     (declined by the model or stopped by the content filter).";

pub async fn agent_loop(
    prompts: Vec<AgentMessage>,
    context: &mut AgentContext,
    config: &AgentLoopConfig,
    tx: mpsc::UnboundedSender<AgentEvent>,
    cancel: tokio_util::sync::CancellationToken,
) -> Vec<AgentMessage> {
    agent_loop_with_stats(prompts, context, config, tx, cancel)
        .await
        .0
}

/// [`agent_loop`], also returning the [`SessionStats`] it sent on
/// `AgentEnd` — for callers inside the crate that must not depend on someone
/// draining the event channel (`Agent`'s sub-agent bucket, `SubAgentTool`'s
/// spend report).
pub(crate) async fn agent_loop_with_stats(
    prompts: Vec<AgentMessage>,
    context: &mut AgentContext,
    config: &AgentLoopConfig,
    tx: mpsc::UnboundedSender<AgentEvent>,
    cancel: tokio_util::sync::CancellationToken,
) -> (Vec<AgentMessage>, SessionStats) {
    tx.send(AgentEvent::AgentStart).ok();

    // One scope for the whole run, input filters included, so decision
    // spend in an async filter is counted — also when it rejects.
    let (outcome, decision) = with_loop_scope(Vec::new(), async {
        let prompts = apply_input_filters(prompts, config).await?;
        note_run_prompts(&prompts);

        let mut new_messages: Vec<AgentMessage> = prompts.clone();

        // Add prompts to context
        for prompt in &prompts {
            context.messages.push(prompt.clone());
        }

        tx.send(AgentEvent::TurnStart).ok();

        // Emit events for each prompt message
        for prompt in &prompts {
            tx.send(AgentEvent::MessageStart {
                message: prompt.clone(),
            })
            .ok();
            tx.send(AgentEvent::MessageEnd {
                message: prompt.clone(),
            })
            .ok();
        }

        let stats = {
            use tracing::Instrument;
            run_loop(context, &mut new_messages, config, &tx, &cancel)
                .instrument(tracing::info_span!("agent_loop", model = %config.model))
                .await
        };
        Ok::<_, String>((new_messages, stats))
    })
    .await;

    let (new_messages, mut stats) = match outcome {
        Ok(done) => done,
        Err(reason) => {
            tx.send(AgentEvent::InputRejected { reason }).ok();
            let stats = SessionStats {
                decision,
                ..Default::default()
            };
            tx.send(AgentEvent::AgentEnd {
                messages: vec![],
                stats: stats.clone(),
            })
            .ok();
            return (vec![], stats);
        }
    };
    stats.decision.merge(&decision);

    tx.send(AgentEvent::AgentEnd {
        messages: new_messages.clone(),
        stats: stats.clone(),
    })
    .ok();
    (new_messages, stats)
}

/// Run the input filters over the prompts: `Err(reason)` on the first
/// reject, otherwise the prompts with any warnings appended to the last
/// user message.
async fn apply_input_filters(
    prompts: Vec<AgentMessage>,
    config: &AgentLoopConfig,
) -> Result<Vec<AgentMessage>, String> {
    let prompts = if !config.input_filters.is_empty() {
        let user_text: String = prompts
            .iter()
            .filter_map(|m| {
                if let AgentMessage::Llm(Message::User { content, .. }) = m {
                    Some(
                        content
                            .iter()
                            .filter_map(|c| {
                                if let Content::Text { text } = c {
                                    Some(text.as_str())
                                } else {
                                    None
                                }
                            })
                            .collect::<Vec<_>>()
                            .join("\n"),
                    )
                } else {
                    None
                }
            })
            .collect::<Vec<_>>()
            .join("\n");

        let mut warnings: Vec<String> = Vec::new();
        for filter in &config.input_filters {
            let verdict = match filter.as_async() {
                // A panicking async filter must not take the loop task (and
                // with it the agent's tools and history) down: contain it and
                // fail closed, as for middleware.
                Some(async_filter) => {
                    use futures::FutureExt;
                    std::panic::AssertUnwindSafe(async_filter.filter(&user_text))
                        .catch_unwind()
                        .await
                        .unwrap_or_else(|_| {
                            warn!("async input filter panicked; rejecting the input");
                            FilterResult::Reject("input filter panicked".into())
                        })
                }
                None => filter.filter(&user_text),
            };
            match verdict {
                FilterResult::Pass => {}
                FilterResult::Warn(w) => warnings.push(w),
                FilterResult::Reject(reason) => return Err(reason),
            }
        }

        // Append warnings to the last user message's content (avoids consecutive user messages)
        if !warnings.is_empty() {
            let warning_text = warnings
                .iter()
                .map(|w| format!("[Warning: {}]", w))
                .collect::<Vec<_>>()
                .join("\n");

            let mut modified = prompts;
            // Find and extend the last user message
            for msg in modified.iter_mut().rev() {
                if let AgentMessage::Llm(Message::User { content, .. }) = msg {
                    content.push(Content::Text { text: warning_text });
                    break;
                }
            }
            modified
        } else {
            prompts
        }
    } else {
        prompts
    };
    Ok(prompts)
}

/// Continue an agent loop from existing context (for retries).
pub async fn agent_loop_continue(
    context: &mut AgentContext,
    config: &AgentLoopConfig,
    tx: mpsc::UnboundedSender<AgentEvent>,
    cancel: tokio_util::sync::CancellationToken,
) -> Vec<AgentMessage> {
    agent_loop_continue_with_stats(context, config, tx, cancel)
        .await
        .0
}

/// [`agent_loop_continue`], also returning its [`SessionStats`].
pub(crate) async fn agent_loop_continue_with_stats(
    context: &mut AgentContext,
    config: &AgentLoopConfig,
    tx: mpsc::UnboundedSender<AgentEvent>,
    cancel: tokio_util::sync::CancellationToken,
) -> (Vec<AgentMessage>, SessionStats) {
    assert!(
        !context.messages.is_empty(),
        "Cannot continue: no messages in context"
    );

    if let Some(last) = context.messages.last() {
        assert!(
            last.role() != "assistant",
            "Cannot continue from assistant message"
        );
    }

    let mut new_messages: Vec<AgentMessage> = Vec::new();

    tx.send(AgentEvent::AgentStart).ok();
    tx.send(AgentEvent::TurnStart).ok();

    let stats = {
        use tracing::Instrument;
        let (mut stats, decision) = with_loop_scope(
            Vec::new(),
            run_loop(context, &mut new_messages, config, &tx, &cancel)
                .instrument(tracing::info_span!("agent_loop", model = %config.model)),
        )
        .await;
        stats.decision.merge(&decision);
        stats
    };

    tx.send(AgentEvent::AgentEnd {
        messages: new_messages.clone(),
        stats: stats.clone(),
    })
    .ok();
    (new_messages, stats)
}

/// Main loop logic shared by agent_loop and agent_loop_continue.
///
/// Outer loop: continues when follow-up messages arrive after agent would stop.
/// Inner loop: process tool calls and steering messages.
async fn run_loop(
    context: &mut AgentContext,
    new_messages: &mut Vec<AgentMessage>,
    config: &AgentLoopConfig,
    tx: &mpsc::UnboundedSender<AgentEvent>,
    cancel: &tokio_util::sync::CancellationToken,
) -> SessionStats {
    let mut stats = SessionStats::default();
    let mut first_turn = true;
    let mut turn_number: usize = 0;
    // Rolling growth measurement feeding `compact_headroom_turns`.
    let mut last_context_tokens: Option<usize> = None;
    let mut growth_total: usize = 0;
    let mut growth_samples: usize = 0;
    // Blends real provider usage with estimation for compaction sizing.
    let mut context_tracker = ContextTracker::new();
    let mut tracker = config
        .execution_limits
        .as_ref()
        .map(|limits| ExecutionTracker::new(limits.clone()));

    // Check for steering messages at start
    let mut pending: Vec<AgentMessage> = config
        .get_steering_messages
        .as_ref()
        .map(|f| f())
        .unwrap_or_default();

    // Outer loop: follow-ups after agent would stop
    loop {
        if cancel.is_cancelled() {
            return stats;
        }

        let mut steering_after_tools: Option<Vec<AgentMessage>> = None;

        // Inner loop: runs at least once, then continues if tool calls or pending messages
        loop {
            if cancel.is_cancelled() {
                return stats;
            }

            if !first_turn {
                tx.send(AgentEvent::TurnStart).ok();
            } else {
                first_turn = false;
            }

            // Inject pending messages
            if !pending.is_empty() {
                note_run_prompts(&pending);
                for msg in pending.drain(..) {
                    tx.send(AgentEvent::MessageStart {
                        message: msg.clone(),
                    })
                    .ok();
                    tx.send(AgentEvent::MessageEnd {
                        message: msg.clone(),
                    })
                    .ok();
                    context.messages.push(msg.clone());
                    new_messages.push(msg);
                }
            }

            // Check execution limits
            if let Some(ref tracker) = tracker {
                if let Some(reason) = tracker.check_limits() {
                    warn!("Execution limit reached: {}", reason);
                    let limit_msg = AgentMessage::Llm(Message::User {
                        content: vec![Content::Text {
                            text: format!("{AGENT_STOPPED_PREFIX} {}]", reason),
                        }],
                        timestamp: now_ms(),
                    });
                    tx.send(AgentEvent::MessageStart {
                        message: limit_msg.clone(),
                    })
                    .ok();
                    tx.send(AgentEvent::MessageEnd {
                        message: limit_msg.clone(),
                    })
                    .ok();
                    context.messages.push(limit_msg.clone());
                    new_messages.push(limit_msg);
                    return stats;
                }
            }

            // before_turn callback — abort if it returns false
            if let Some(ref before_turn) = config.before_turn {
                if !before_turn(&context.messages, turn_number) {
                    return stats;
                }
            }

            // Inter-turn delay — throttle API calls to stay under rate limits.
            // Skipped on the first turn so the agent starts immediately.
            if turn_number > 0 {
                if let Some(delay) = config.turn_delay {
                    crate::rt::sleep(delay).await;
                }
            }

            turn_number += 1;

            // Compact context if configured (tiered: tool outputs → summarize → drop).
            //
            // Calibration: the tracker's hybrid figure is anchored on provider
            // usage, which counts the FULL request (system prompt + tool
            // schemas + any estimation shortfall on messages), while
            // `total_tokens` counts messages only. The difference is a
            // measured overhead; subtracting it from the budget makes the
            // strategy's message-token checks equivalent to "true window
            // occupancy <= max_context_tokens". The static
            // `system_prompt_tokens` reserve is zeroed in the calibrated
            // config because the measured overhead already includes the real
            // system prompt. A floor of 10% of the configured budget
            // guarantees a mis-measured overhead can never wipe the history.
            if let Some(ref ctx_config) = config.context_config {
                let estimated = context::total_tokens(&context.messages);
                let hybrid = context_tracker.estimate_context_tokens(&context.messages);
                let overhead = hybrid.saturating_sub(estimated);

                // Growth since the last turn's compaction step, averaged. This
                // is what lets the headroom policy target an interval between
                // compactions instead of a fixed fraction of the budget: a
                // ratio cannot know how fast the session is growing, so the
                // room it leaves shrinks as history accumulates.
                if let Some(previous) = last_context_tokens {
                    growth_samples += 1;
                    growth_total += estimated.saturating_sub(previous);
                }
                let growth_per_turn = if growth_samples > 0 {
                    growth_total as f64 / growth_samples as f64
                } else {
                    0.0
                };

                let calibrated;
                let effective_config = if overhead > 0 {
                    let floor = ctx_config.max_context_tokens / 10;
                    calibrated = ContextConfig {
                        max_context_tokens: ctx_config
                            .max_context_tokens
                            .saturating_sub(overhead)
                            .max(floor),
                        system_prompt_tokens: 0,
                        ..ctx_config.clone()
                    };
                    tracing::debug!(
                        "compaction budget calibrated: {} -> {} (measured overhead: {} tokens)",
                        ctx_config.max_context_tokens,
                        calibrated.max_context_tokens,
                        overhead
                    );
                    &calibrated
                } else {
                    ctx_config
                };

                // Resolve the headroom policy against the calibrated budget.
                let with_headroom;
                let effective_config = {
                    let ratio = effective_config.effective_target_ratio(growth_per_turn);
                    if ratio != effective_config.compact_target_ratio {
                        tracing::debug!(
                            "compaction target adapted: ratio {} -> {:.3} \
                             ({} tokens/turn growth, {:?} turns of headroom)",
                            effective_config.compact_target_ratio,
                            ratio,
                            growth_per_turn as usize,
                            effective_config.compact_headroom_turns,
                        );
                        with_headroom = ContextConfig {
                            compact_target_ratio: ratio,
                            ..effective_config.clone()
                        };
                        &with_headroom
                    } else {
                        effective_config
                    }
                };

                let strategy: &dyn CompactionStrategy = config
                    .compaction_strategy
                    .as_deref()
                    .unwrap_or(&DefaultCompaction);
                let before_len = context.messages.len();
                // Already computed above over the same unmutated history.
                let before_tokens = estimated;
                context.messages =
                    strategy.compact(std::mem::take(&mut context.messages), effective_config);
                let after_tokens = context::total_tokens(&context.messages);
                // Length alone would miss level-1 truncation, which rewrites
                // tool outputs in place and leaves the count unchanged — and
                // that is exactly the case where the tracker's baseline goes
                // stale, so both the reset and the counter key off the same
                // test rather than the reset keying off length alone.
                if context.messages.len() != before_len || after_tokens != before_tokens {
                    // History moved; re-baseline from the next real usage.
                    context_tracker.reset();
                    stats.compactions += 1;
                }
                last_context_tokens = Some(after_tokens);
            }

            // Stream assistant response, under an llm_stream span that
            // records tokens and (when rates are configured) dollar cost.
            let llm_span = tracing::info_span!(
                "llm_stream",
                turn = turn_number,
                model = %config.model,
                tokens_in = tracing::field::Empty,
                tokens_out = tracing::field::Empty,
                tokens_cached = tracing::field::Empty,
                cost_usd = tracing::field::Empty,
                error = tracing::field::Empty,
            );
            let message = {
                use tracing::Instrument;
                stream_assistant_response(context, config, tx, cancel)
                    .instrument(llm_span.clone())
                    .await
            };
            if let Message::Assistant {
                usage, stop_reason, ..
            } = &message
            {
                llm_span.record("error", *stop_reason == StopReason::Error);
                llm_span.record("tokens_in", usage.input);
                llm_span.record("tokens_out", usage.output);
                llm_span.record("tokens_cached", usage.cache_read);
                // Unpriced models (`cost: None`) leave the field empty: unknown,
                // never $0. A free model (`Some`, all-zero) records 0.0.
                if let Some(cost) = config.model_config.as_ref().and_then(|mc| mc.cost.as_ref()) {
                    llm_span.record("cost_usd", cost.cost_usd(usage));
                }
            }
            // Tool-forcing providers (Anthropic) deliver structured output as
            // a forced tool call — unwrap it into plain text BEFORE tool-call
            // extraction, so the loop never tries to execute the synthetic tool.
            // Skipped on Anthropic's native path (`native_structured_output`):
            // no synthetic tool was offered there, so a call named after the
            // schema is a real user tool and must execute. A no-op for the
            // other natively-constraining providers (OpenAI-compat, Gemini)
            // unless a user tool shares the schema's name.
            let message = if structured_output_is_tool_forced(config.model_config.as_ref()) {
                unwrap_structured_tool_call(message, config.output_schema.as_ref())
            } else {
                message
            };

            let agent_msg: AgentMessage = message.clone().into();
            context.messages.push(agent_msg.clone());
            new_messages.push(agent_msg.clone());
            if let Message::Assistant { usage, .. } = &message {
                context_tracker.record_usage(usage, context.messages.len() - 1);
                stats.record_turn(
                    usage,
                    config.model_config.as_ref().and_then(|mc| mc.cost.as_ref()),
                );
            }

            // Check for error/abort
            if let Message::Assistant {
                ref stop_reason,
                ref error_message,
                ref usage,
                ..
            } = message
            {
                if *stop_reason == StopReason::Error || *stop_reason == StopReason::Aborted {
                    if *stop_reason == StopReason::Error {
                        if let Some(ref on_error) = config.on_error {
                            let err_str = error_message.as_deref().unwrap_or("Unknown error");
                            on_error(err_str);
                        }
                    }
                    // Call after_turn even on error/abort so callers tracking usage don't miss this turn
                    if let Some(ref after_turn) = config.after_turn {
                        after_turn(&context.messages, usage);
                    }
                    tx.send(AgentEvent::TurnEnd {
                        message: agent_msg,
                        tool_results: vec![],
                    })
                    .ok();
                    return stats;
                }
            }

            // A refusal (the model declined, or a content filter cut the
            // response) is terminal: nothing in it runs, and there is no
            // further LLM turn — the run ends as Error/Aborted end it, leaving
            // any queued steering/follow-up messages queued. A tool call in the
            // refused message may be complete and well-formed (a content filter
            // can land after it), but executing it would act on a response the
            // provider withdrew. Every call is still answered with an error
            // result so the transcript stays one the provider accepts on the
            // next prompt.
            if let Message::Assistant {
                stop_reason: StopReason::Refusal,
                ref content,
                ref usage,
                ..
            } = message
            {
                let mut tool_results: Vec<Message> = Vec::new();
                for c in content {
                    if let Content::ToolCall {
                        id,
                        name,
                        arguments,
                        ..
                    } = c
                    {
                        tracing::warn!(
                            tool = name.as_str(),
                            tool_call_id = id.as_str(),
                            "tool call not executed: the response was a refusal"
                        );
                        let (result, _) = unexecuted_tool_call(
                            id,
                            name,
                            arguments,
                            REFUSAL_TOOL_RESULT_TEXT.to_string(),
                            tx,
                        );
                        let am: AgentMessage = result.clone().into();
                        context.messages.push(am.clone());
                        new_messages.push(am);
                        tool_results.push(result);
                    }
                }
                if let Some(ref after_turn) = config.after_turn {
                    after_turn(&context.messages, usage);
                }
                tx.send(AgentEvent::TurnEnd {
                    message: agent_msg,
                    tool_results,
                })
                .ok();
                return stats;
            }

            // Extract tool calls
            let tool_calls: Vec<_> = match &message {
                Message::Assistant { content, .. } => content
                    .iter()
                    .filter_map(|c| match c {
                        Content::ToolCall {
                            id,
                            name,
                            arguments,
                            ..
                        } => Some((id.clone(), name.clone(), arguments.clone())),
                        _ => None,
                    })
                    .collect(),
                _ => vec![],
            };

            // A loop-detection nudge, held until the assistant's tool calls
            // have been answered. Injecting it before the `tool_result`s would
            // orphan them, which every provider rejects.
            let mut loop_nudge: Option<AgentMessage> = None;

            // Repetition check runs *before* execution, so a stuck model does
            // not also pay for the tool run it was never going to learn from.
            if !tool_calls.is_empty() {
                if let Some(ref mut tracker) = tracker {
                    let sigs: Vec<(String, serde_json::Value)> = tool_calls
                        .iter()
                        .map(|(_, name, args)| (name.clone(), args.clone()))
                        .collect();
                    match tracker.record_tool_calls(&sigs) {
                        context::LoopVerdict::Continue => {}
                        context::LoopVerdict::Steer {
                            tool_name,
                            repetitions,
                        } => {
                            warn!(
                                "loop detection: {tool_name} called {repetitions}x with identical arguments; steering"
                            );
                            tx.send(AgentEvent::LoopDetected {
                                tool_name: tool_name.clone(),
                                repetitions,
                                aborted: false,
                            })
                            .ok();
                            // A nudge, not a stop: a model repeating a call is
                            // often retrying something transient, and it
                            // usually recovers once told the result will not
                            // change.
                            let nudge = AgentMessage::Llm(Message::User {
                                content: vec![Content::Text {
                                    text: format!(
                                        "{LOOP_NUDGE_PREFIX}{tool_name} {repetitions} times with identical arguments. The result will not change — change approach, or say why the repetition is needed.]"
                                    ),
                                }],
                                timestamp: now_ms(),
                            });
                            // Deferred, not pushed here. The assistant's
                            // `tool_use` blocks are still unanswered at this
                            // point, and every provider rejects a transcript
                            // where anything but their `tool_result`s comes
                            // next. Appended after the results, below.
                            loop_nudge = Some(nudge);
                        }
                        context::LoopVerdict::Abort {
                            tool_name,
                            repetitions,
                        } => {
                            warn!("loop detection: {tool_name} repeated after steering; stopping");
                            tx.send(AgentEvent::LoopDetected {
                                tool_name: tool_name.clone(),
                                repetitions,
                                aborted: true,
                            })
                            .ok();
                            let stop = AgentMessage::Llm(Message::User {
                                content: vec![Content::Text {
                                    text: format!(
                                        "{LOOP_ABORT_PREFIX} {tool_name} was called repeatedly with identical \
                                         arguments after being asked to change approach.]"
                                    ),
                                }],
                                timestamp: now_ms(),
                            });
                            // The tools are deliberately not run — a stuck
                            // model should not also pay for a call it was
                            // never going to learn from. But the assistant's
                            // `tool_use` blocks must still be answered, or the
                            // transcript is one every provider rejects and the
                            // agent is unusable for any later prompt. Synthesize
                            // one error result per outstanding call.
                            for (id, name, _) in &tool_calls {
                                let denied = Message::ToolResult {
                                    tool_call_id: id.clone(),
                                    tool_name: name.clone(),
                                    content: vec![Content::Text {
                                        text: "Not executed: the run was stopped for repeating \
                                               this call with identical arguments."
                                            .into(),
                                    }],
                                    is_error: true,
                                    timestamp: now_ms(),
                                };
                                let am: AgentMessage = denied.into();
                                context.messages.push(am.clone());
                                new_messages.push(am);
                            }
                            tx.send(AgentEvent::MessageStart {
                                message: stop.clone(),
                            })
                            .ok();
                            tx.send(AgentEvent::MessageEnd {
                                message: stop.clone(),
                            })
                            .ok();
                            context.messages.push(stop.clone());
                            new_messages.push(stop);
                            // Match every other abort path in this function:
                            // callers tracking usage must not miss this turn,
                            // and a UI pairing TurnStart/TurnEnd must not be
                            // left with an unmatched TurnStart.
                            if let Some(after_turn) = config.after_turn.as_ref() {
                                let usage = match &message {
                                    Message::Assistant { usage, .. } => usage.clone(),
                                    _ => Usage::default(),
                                };
                                after_turn(&context.messages, &usage);
                            }
                            tx.send(AgentEvent::TurnEnd {
                                message: agent_msg,
                                tool_results: Vec::new(),
                            })
                            .ok();
                            return stats;
                        }
                    }
                }
            }

            let has_tool_calls = !tool_calls.is_empty();
            let mut tool_results: Vec<Message> = Vec::new();

            if has_tool_calls {
                let execution = execute_tool_calls(
                    &context.tools,
                    &tool_calls,
                    tx,
                    cancel,
                    config.get_steering_messages.as_ref(),
                    &config.tool_execution,
                    Gate {
                        middleware: &config.tool_middleware,
                        history: &context.messages,
                    },
                )
                .await;

                tool_results = execution.tool_results;
                steering_after_tools = execution.steering_messages;
                // Separate bucket: `usage`/`cost_usd` stay this agent's own.
                for child in &execution.sub_agent_stats {
                    stats.sub_agents.record_run(child);
                    // Decision spend has one bucket for the whole tree.
                    stats.decision.merge(&child.decision);
                }

                // Cap oversized output on the way in when configured, so
                // compaction never has to rewrite a tool result the provider
                // has already cached. Per-tool budgets apply, so a tool that
                // tolerates head+tail badly can opt out. The events emitted
                // above still carry the untruncated output — this is a context
                // concern only.
                let append_cap = config
                    .context_config
                    .as_ref()
                    .filter(|c| c.truncate_tool_output_on_append);

                for result in &tool_results {
                    let original: AgentMessage = result.clone().into();
                    let mut am = original.clone();
                    if let Some(ctx_config) = append_cap {
                        // First pass, unkeyed: tells us what the context will
                        // hold and which blocks carry a marker to name a key.
                        let (plain, marked) =
                            context::truncate_tool_output_keyed(original.clone(), ctx_config, None);
                        am = plain;

                        // Stash here and nowhere else. This is the one place
                        // the full text still exists — by the time compaction
                        // runs, the append-path truncation has already
                        // discarded the middle, so there would be nothing left
                        // to store.
                        //
                        // `marked` is the gate, not "did the text change": a
                        // budget too small for a marker still truncates, and
                        // stashing then would leave an entry no marker names
                        // and nothing can reach.
                        if let (Some(sink), false) = (&config.tool_output_sink, marked.is_empty()) {
                            if let Message::ToolResult { tool_call_id, .. } = result {
                                let blocks = context::block_texts(&original);
                                // One key per marked block. A single shared key
                                // made every marker resolve to all blocks
                                // concatenated, silently dropping any image
                                // between them, so what the model fetched was
                                // never the block whose marker it followed.
                                // Via the shared helper, so the hash input
                                // cannot drift from what `message_text`
                                // documents and silently move every key.
                                let base = context::tool_output_key(
                                    tool_call_id,
                                    &context::message_text(&original),
                                );
                                let mut all_stored = true;
                                let mut written: Vec<String> = Vec::new();
                                for idx in &marked {
                                    let Some((_, text)) = blocks.iter().find(|(i, _)| i == idx)
                                    else {
                                        // Unreachable — `marked` and `blocks`
                                        // both enumerate the same `original`.
                                        // Fail closed regardless: continuing
                                        // leaves `all_stored` true and emits a
                                        // marker for a block never stored,
                                        // which is the defect this feature
                                        // exists to prevent.
                                        debug_assert!(
                                            false,
                                            "marked block {idx} absent from block_texts"
                                        );
                                        all_stored = false;
                                        break;
                                    };
                                    let key = context::block_key(&base, *idx);
                                    match sink.set(&key, text.clone()).await {
                                        Ok(()) => written.push(key),
                                        Err(e) => {
                                            all_stored = false;
                                            warn!(
                                                tool_call_id = %tool_call_id,
                                                block = idx,
                                                bytes = text.len(),
                                                "could not stash tool output: {e}; no marker in \
                                                 this result will offer retrieval"
                                            );
                                            break;
                                        }
                                    }
                                }

                                // `Ok(())` per write is not proof the set
                                // survived. `FileBackend` evicts to fit and
                                // exempts only the key it just wrote, so a
                                // later sibling can evict an earlier one —
                                // and its (mtime, filename) ordering makes
                                // `-b0` lose to `-b2` deterministically. Naming
                                // a key that was deleted microseconds ago is
                                // exactly the marker-points-at-nothing failure
                                // the backend's own guard was added to stop.
                                if all_stored {
                                    for key in &written {
                                        if sink.get(key).await.is_none() {
                                            all_stored = false;
                                            warn!(
                                                tool_call_id = %tool_call_id,
                                                key = %key,
                                                stored = written.len(),
                                                "a stashed block did not survive storing its \
                                                 siblings — the backend cap cannot hold this \
                                                 result; no marker will offer retrieval"
                                            );
                                            break;
                                        }
                                    }
                                }

                                // Roll back whatever is left, keyed or not.
                                // Half a result in the store is unreachable
                                // bytes that still count against the cap, and
                                // on `FileBackend` they evict the caller's own
                                // artifacts to make room for debris.
                                if !all_stored {
                                    for key in &written {
                                        if !sink.remove(key).await {
                                            // `remove` already swallowed any backend error into
                                            // `false` + a warn. Surfacing it here too matters
                                            // because a failed rollback leaves a stashed value
                                            // no marker names, consuming cap quota for the rest
                                            // of the run.
                                            warn!("shared state: rollback could not remove {key}");
                                        }
                                    }
                                }
                                if all_stored {
                                    // Re-truncate from the original with the
                                    // base key, so each marker names its own
                                    // block only once every block is stored.
                                    am = context::truncate_tool_output_keyed(
                                        original.clone(),
                                        ctx_config,
                                        Some(&base),
                                    )
                                    .0;
                                }
                            }
                        }
                    }
                    context.messages.push(am.clone());
                    new_messages.push(am);
                }
            }

            // Track turn for execution limits
            if let Some(ref mut tracker) = tracker {
                let turn_tokens = match &message {
                    Message::Assistant { usage, .. } => {
                        (usage.input + usage.output + usage.cache_read + usage.cache_write) as usize
                    }
                    _ => context::message_tokens(&agent_msg),
                };
                tracker.record_turn(turn_tokens);
            }

            // after_turn callback
            if let Some(ref after_turn) = config.after_turn {
                let usage = match &message {
                    Message::Assistant { usage, .. } => usage.clone(),
                    _ => Usage::default(),
                };
                after_turn(&context.messages, &usage);
            }

            // Now that every `tool_result` is appended, the transcript is
            // well-formed again and the nudge can land.
            if let Some(nudge) = loop_nudge.take() {
                tx.send(AgentEvent::MessageStart {
                    message: nudge.clone(),
                })
                .ok();
                tx.send(AgentEvent::MessageEnd {
                    message: nudge.clone(),
                })
                .ok();
                context.messages.push(nudge.clone());
                new_messages.push(nudge);
            }

            tx.send(AgentEvent::TurnEnd {
                message: agent_msg,
                tool_results,
            })
            .ok();

            // Check steering after turn
            if let Some(steering) = steering_after_tools.take() {
                if !steering.is_empty() {
                    pending = steering;
                    continue;
                }
            }

            pending = config
                .get_steering_messages
                .as_ref()
                .map(|f| f())
                .unwrap_or_default();

            // Exit inner loop if no more tool calls and no pending messages
            if !has_tool_calls && pending.is_empty() {
                break;
            }
        }

        // Agent would stop. Check for follow-ups.
        let follow_ups = config
            .get_follow_up_messages
            .as_ref()
            .map(|f| f())
            .unwrap_or_default();

        if !follow_ups.is_empty() {
            pending = follow_ups;
            continue;
        }

        break;
    }

    stats
}

/// Stream an assistant response from the LLM.
async fn stream_assistant_response(
    context: &AgentContext,
    config: &AgentLoopConfig,
    tx: &mpsc::UnboundedSender<AgentEvent>,
    cancel: &tokio_util::sync::CancellationToken,
) -> Message {
    // Apply context transform
    let messages = if let Some(transform) = &config.transform_context {
        transform(context.messages.clone())
    } else {
        context.messages.clone()
    };

    // Convert to LLM messages
    let convert = config.convert_to_llm.as_ref();
    let llm_messages = match convert {
        Some(f) => f(&messages),
        None => default_convert_to_llm(&messages),
    };

    // Build tool definitions
    let tool_defs: Vec<ToolDefinition> = context
        .tools
        .iter()
        .map(|t| ToolDefinition {
            name: t.name().to_string(),
            description: t.description().to_string(),
            parameters: t.parameters_schema(),
        })
        .collect();

    // Retry loop for transient provider errors
    let retry = &config.retry_config;
    let mut attempt = 0;
    let result = loop {
        let stream_config = StreamConfig {
            model: config.model.clone(),
            system_prompt: context.system_prompt.clone(),
            messages: llm_messages.clone(),
            tools: tool_defs.clone(),
            thinking_level: config.thinking_level,
            api_key: config.api_key.clone(),
            max_tokens: config.max_tokens,
            temperature: config.temperature,
            model_config: config.model_config.clone(),
            cache_config: config.cache_config.clone(),
            output_schema: config.output_schema.clone(),
        };

        let (stream_tx, mut stream_rx) = mpsc::unbounded_channel();
        let provider_cancel = cancel.clone();

        // Spawn a task to forward events in real-time as the provider streams
        let event_tx = tx.clone();
        let model_for_events = config.model.clone();
        let forward_handle = crate::rt::spawn(async move {
            let mut partial_message: Option<AgentMessage> = None;
            let mut ended = false;
            while let Some(event) = stream_rx.recv().await {
                match &event {
                    StreamEvent::Start => {
                        let placeholder = AgentMessage::Llm(Message::Assistant {
                            content: Vec::new(),
                            stop_reason: StopReason::Stop,
                            model: model_for_events.clone(),
                            provider: String::new(),
                            usage: Usage::default(),
                            timestamp: now_ms(),
                            error_message: None,
                        });
                        partial_message = Some(placeholder.clone());
                        event_tx
                            .send(AgentEvent::MessageStart {
                                message: placeholder,
                            })
                            .ok();
                    }
                    StreamEvent::TextDelta { delta, .. } => {
                        if let Some(ref msg) = partial_message {
                            event_tx
                                .send(AgentEvent::MessageUpdate {
                                    message: msg.clone(),
                                    delta: StreamDelta::Text {
                                        delta: delta.clone(),
                                    },
                                })
                                .ok();
                        }
                    }
                    StreamEvent::ThinkingDelta { delta, .. } => {
                        if let Some(ref msg) = partial_message {
                            event_tx
                                .send(AgentEvent::MessageUpdate {
                                    message: msg.clone(),
                                    delta: StreamDelta::Thinking {
                                        delta: delta.clone(),
                                    },
                                })
                                .ok();
                        }
                    }
                    StreamEvent::ToolCallDelta { delta, .. } => {
                        if let Some(ref msg) = partial_message {
                            event_tx
                                .send(AgentEvent::MessageUpdate {
                                    message: msg.clone(),
                                    delta: StreamDelta::ToolCallDelta {
                                        delta: delta.clone(),
                                    },
                                })
                                .ok();
                        }
                    }
                    StreamEvent::Done { message } => {
                        let am: AgentMessage = message.clone().into();
                        partial_message = Some(am.clone());
                        ended = true;
                        event_tx.send(AgentEvent::MessageEnd { message: am }).ok();
                    }
                    StreamEvent::Error { message } => {
                        let am: AgentMessage = message.clone().into();
                        if partial_message.is_none() {
                            event_tx
                                .send(AgentEvent::MessageStart {
                                    message: am.clone(),
                                })
                                .ok();
                        }
                        partial_message = Some(am.clone());
                        ended = true;
                        event_tx.send(AgentEvent::MessageEnd { message: am }).ok();
                    }
                    _ => {}
                }
            }
            AttemptEvents {
                started: partial_message.is_some(),
                ended,
            }
        });

        // Provider streams concurrently — events are forwarded in real-time
        // When provider returns, stream_tx is dropped, ending the forwarder
        let result = config
            .provider
            .stream(stream_config, stream_tx, provider_cancel)
            .await;

        // The provider has returned, so its sender is gone and the forwarder
        // ends once it has forwarded what the attempt produced. Draining it
        // (rather than aborting it) makes what consumers see deterministic:
        // every event of every attempt, never a timing-dependent subset.
        let attempt_events = forward_handle.await.unwrap_or_default();

        match &result {
            Err(e) if e.is_retryable() && attempt < retry.max_retries && !cancel.is_cancelled() => {
                // Close the failed attempt's message so consumers can discard
                // whatever partial content it streamed before the retry. The
                // `ProviderRetry` that follows marks it as retried, not final.
                if attempt_events.needs_end() {
                    let failed = error_message(
                        &config.model,
                        format!(
                            "attempt {} of {} failed and will be retried: {e}",
                            attempt + 1,
                            retry.max_retries + 1
                        ),
                    );
                    tx.send(AgentEvent::MessageEnd {
                        message: failed.into(),
                    })
                    .ok();
                }
                attempt += 1;
                // Server-provided Retry-After wins over backoff, but is
                // clamped to max_delay_ms so a bad header can't stall the loop.
                let delay = e
                    .retry_after()
                    .map(|d| d.min(std::time::Duration::from_millis(retry.max_delay_ms)))
                    .unwrap_or_else(|| retry.delay_for_attempt(attempt));
                crate::retry::log_retry(attempt, retry.max_retries, &delay, e);
                tx.send(AgentEvent::provider_retry(
                    attempt,
                    retry.max_retries + 1,
                    e.to_string(),
                    delay,
                ))
                .ok();
                // Cancelling during the backoff ends the turn now rather than
                // after the delay (and a wasted request).
                let backoff = std::pin::pin!(crate::rt::sleep(delay));
                let cancelled = std::pin::pin!(cancel.cancelled());
                if let futures::future::Either::Right(_) =
                    futures::future::select(backoff, cancelled).await
                {
                    break (Err(ProviderError::Cancelled), AttemptEvents::default());
                }
                continue;
            }
            _ => break (result, attempt_events),
        }
    };

    let (result, attempt_events) = result;
    match result {
        Ok(msg) => msg,
        Err(e) => {
            warn!("Provider error: {}", e);
            let mut failed = error_message(&config.model, e.to_string());
            // A cancelled run is aborted, not failed: it does not reach
            // `on_error`, and consumers can tell the two apart.
            if matches!(e, ProviderError::Cancelled) || cancel.is_cancelled() {
                if let Message::Assistant { stop_reason, .. } = &mut failed {
                    *stop_reason = StopReason::Aborted;
                }
            }
            // Close the message the final attempt opened, with the same
            // error message this turn returns.
            if attempt_events.needs_end() {
                tx.send(AgentEvent::MessageEnd {
                    message: failed.clone().into(),
                })
                .ok();
            }
            failed
        }
    }
}

/// What one provider attempt's forwarder emitted.
#[derive(Debug, Default, Clone, Copy)]
struct AttemptEvents {
    /// A `MessageStart` was sent for this attempt.
    started: bool,
    /// Its `MessageEnd` was sent too.
    ended: bool,
}

impl AttemptEvents {
    /// The attempt opened a message that nothing closed.
    fn needs_end(self) -> bool {
        self.started && !self.ended
    }
}

/// The assistant message for a provider failure.
fn error_message(model: &str, error: String) -> Message {
    Message::Assistant {
        content: vec![Content::Text {
            text: String::new(),
        }],
        stop_reason: StopReason::Error,
        model: model.to_string(),
        provider: "unknown".into(),
        usage: Usage::default(),
        timestamp: now_ms(),
        error_message: Some(error),
    }
}

// ---------------------------------------------------------------------------
// Tool execution
// ---------------------------------------------------------------------------

struct ToolExecutionResult {
    tool_results: Vec<Message>,
    steering_messages: Option<Vec<AgentMessage>>,
    /// Stats of every sub-agent run these tool calls delegated to, failed
    /// ones included, for the run's `SessionStats::sub_agents`.
    sub_agent_stats: Vec<SessionStats>,
}

/// Whether a structured-output request may have been enforced by forcing a
/// synthetic tool call, which the loop must unwrap. False only on Anthropic's
/// native path (`AnthropicCompat::native_structured_output`), where the API
/// constrains the reply text and offers no synthetic tool. Without a model
/// config the provider falls back to its defaults (tool-forcing on Anthropic),
/// so unwrapping stays on.
fn structured_output_is_tool_forced(model_config: Option<&ModelConfig>) -> bool {
    !model_config.is_some_and(|mc| {
        mc.api == crate::provider::ApiProtocol::AnthropicMessages
            && mc
                .anthropic
                .as_ref()
                .is_some_and(|c| c.native_structured_output)
    })
}

/// Convert a forced structured-output tool call back into a plain-text
/// assistant message with `StopReason::Stop`. No-op unless a schema is set
/// and the message carries a tool call named after it.
fn unwrap_structured_tool_call(
    message: Message,
    schema: Option<&crate::provider::OutputSchema>,
) -> Message {
    let Some(schema) = schema else {
        return message;
    };
    let Message::Assistant {
        content,
        model,
        provider,
        usage,
        stop_reason,
        error_message,
        ..
    } = &message
    else {
        return message;
    };
    let Some(payload) = content.iter().find_map(|c| match c {
        Content::ToolCall {
            name, arguments, ..
        } if *name == schema.name => Some(arguments.clone()),
        _ => None,
    }) else {
        return message;
    };

    // Remove ONLY the synthetic call; any real tool calls (shouldn't occur
    // under forced tool_choice, but defensively) stay and execute normally.
    // The payload is appended AFTER any preamble text — consumers take the
    // last text block.
    let mut new_content: Vec<Content> = content
        .iter()
        .filter(|c| !matches!(c, Content::ToolCall { name, .. } if *name == schema.name))
        .cloned()
        .collect();
    new_content.push(Content::Text {
        text: payload.to_string(),
    });
    // ToolUse becomes Stop (the forced call was the "answer"); every other
    // stop reason (Length = truncated payload, Error, ...) is preserved so
    // truncation isn't laundered into success.
    let new_stop = if *stop_reason == StopReason::ToolUse {
        StopReason::Stop
    } else {
        stop_reason.clone()
    };
    let rebuilt = Message::assistant(
        new_content,
        new_stop,
        model.clone(),
        provider.clone(),
        usage.clone(),
    );
    // `Message::assistant` nulls `error_message`, so carry it across explicitly.
    // The stop reason is preserved a few lines up for the same reason; dropping
    // the explanation with it leaves the caller a bare `Error` and
    // `StructuredPromptError::Provider { message: "provider error (no detail)" }`.
    match error_message {
        Some(msg) => rebuilt.with_error_message(msg.clone()),
        None => rebuilt,
    }
}

/// What gates a tool call: the middleware chain, and the conversation it may
/// consult ([`ToolCallRequest::messages`]).
#[derive(Clone, Copy)]
struct Gate<'a> {
    middleware: &'a [Arc<dyn ToolMiddleware>],
    history: &'a [AgentMessage],
}

async fn execute_tool_calls(
    tools: &[Box<dyn AgentTool>],
    tool_calls: &[(String, String, serde_json::Value)],
    tx: &mpsc::UnboundedSender<AgentEvent>,
    cancel: &tokio_util::sync::CancellationToken,
    get_steering: Option<&GetMessagesFn>,
    strategy: &ToolExecutionStrategy,
    gate: Gate<'_>,
) -> ToolExecutionResult {
    match strategy {
        ToolExecutionStrategy::Sequential => {
            execute_sequential(tools, tool_calls, tx, cancel, get_steering, gate).await
        }
        ToolExecutionStrategy::Parallel => {
            execute_batch(tools, tool_calls, tx, cancel, get_steering, gate).await
        }
        ToolExecutionStrategy::Batched { size } => {
            let mut results: Vec<Message> = Vec::new();
            let mut steering_messages: Option<Vec<AgentMessage>> = None;
            let mut sub_agent_stats: Vec<SessionStats> = Vec::new();

            for (batch_idx, batch) in tool_calls.chunks(*size).enumerate() {
                let batch_result = execute_batch(tools, batch, tx, cancel, None, gate).await;
                results.extend(batch_result.tool_results);
                sub_agent_stats.extend(batch_result.sub_agent_stats);

                // Check steering between batches
                if let Some(get_steering_fn) = get_steering {
                    let steering = get_steering_fn();
                    if !steering.is_empty() {
                        steering_messages = Some(steering);
                        // Skip remaining batches
                        let executed = (batch_idx + 1) * *size;
                        if executed < tool_calls.len() {
                            for (skip_id, skip_name, _) in &tool_calls[executed..] {
                                results.push(skip_tool_call(skip_id, skip_name, tx));
                            }
                        }
                        break;
                    }
                }
            }

            ToolExecutionResult {
                tool_results: results,
                steering_messages,
                sub_agent_stats,
            }
        }
    }
}

/// Execute tool calls one at a time, checking steering between each.
async fn execute_sequential(
    tools: &[Box<dyn AgentTool>],
    tool_calls: &[(String, String, serde_json::Value)],
    tx: &mpsc::UnboundedSender<AgentEvent>,
    cancel: &tokio_util::sync::CancellationToken,
    get_steering: Option<&GetMessagesFn>,
    gate: Gate<'_>,
) -> ToolExecutionResult {
    let mut results: Vec<Message> = Vec::new();
    let mut steering_messages: Option<Vec<AgentMessage>> = None;
    let mut sub_agent_stats: Vec<SessionStats> = Vec::new();

    for (index, (id, name, args)) in tool_calls.iter().enumerate() {
        let (result_msg, delegated) =
            execute_single_tool(tools, id, name, args, tx, cancel, gate).await;
        results.push(result_msg);
        sub_agent_stats.extend(delegated);

        // Check for steering — skip remaining tools if user interrupted
        if let Some(get_steering_fn) = get_steering {
            let steering = get_steering_fn();
            if !steering.is_empty() {
                steering_messages = Some(steering);
                for (skip_id, skip_name, _) in &tool_calls[index + 1..] {
                    results.push(skip_tool_call(skip_id, skip_name, tx));
                }
                break;
            }
        }
    }

    ToolExecutionResult {
        tool_results: results,
        steering_messages,
        sub_agent_stats,
    }
}

/// Execute a batch of tool calls concurrently using futures::join_all.
async fn execute_batch(
    tools: &[Box<dyn AgentTool>],
    tool_calls: &[(String, String, serde_json::Value)],
    tx: &mpsc::UnboundedSender<AgentEvent>,
    cancel: &tokio_util::sync::CancellationToken,
    get_steering: Option<&GetMessagesFn>,
    gate: Gate<'_>,
) -> ToolExecutionResult {
    use futures::future::join_all;

    let futures: Vec<_> = tool_calls
        .iter()
        .map(|(id, name, args)| execute_single_tool(tools, id, name, args, tx, cancel, gate))
        .collect();

    let batch_results = join_all(futures).await;

    let mut sub_agent_stats: Vec<SessionStats> = Vec::new();
    let results: Vec<Message> = batch_results
        .into_iter()
        .map(|(msg, delegated)| {
            sub_agent_stats.extend(delegated);
            msg
        })
        .collect();

    // Check steering after batch completes
    let steering_messages = if let Some(get_steering_fn) = get_steering {
        let steering = get_steering_fn();
        if steering.is_empty() {
            None
        } else {
            Some(steering)
        }
    } else {
        None
    };

    ToolExecutionResult {
        tool_results: results,
        steering_messages,
        sub_agent_stats,
    }
}

/// Execute a single tool call and emit events.
///
/// Also returns the stats of any sub-agent runs the tool reported, so the
/// caller can fold delegated spend into the run's rollup.
async fn execute_single_tool(
    tools: &[Box<dyn AgentTool>],
    id: &str,
    name: &str,
    args: &serde_json::Value,
    tx: &mpsc::UnboundedSender<AgentEvent>,
    cancel: &tokio_util::sync::CancellationToken,
    gate: Gate<'_>,
) -> (Message, Vec<SessionStats>) {
    // A call whose streamed arguments did not resolve to a JSON object (cut
    // off at the output token limit, in practice, or double-encoded as a
    // string) is answered with an error, never run: the provider kept the
    // raw text instead of substituting `{}`, and running the tool on its
    // defaults would silently replace what the model asked for. This sits
    // ahead of middleware — there is no real call to approve or rewrite.
    if let Some(raw) = crate::provider::unparsed_tool_arguments(args) {
        let (msg, _) = unparsed_arguments_tool_call(id, name, args, raw, tx);
        return (msg, Vec::new());
    }

    // Middleware chain runs next: each hook may rewrite the args seen by
    // later hooks; the first Deny short-circuits into an error tool result
    // (the LLM sees the reason and can adapt — the loop continues).
    let mut effective_args = args.clone();
    let prompts = if gate.middleware.is_empty() {
        Vec::new()
    } else {
        run_prompts()
    };
    for mw in gate.middleware {
        let call = ToolCallRequest {
            tool_call_id: id,
            tool_name: name,
            args: &effective_args,
            messages: gate.history,
            run_prompts: &prompts,
        };
        // A panicking middleware must not kill the loop task (which would
        // strip the agent of its tools) — contain it and fail closed.
        let decision = {
            use futures::FutureExt;
            std::panic::AssertUnwindSafe(mw.before_tool(&call))
                .catch_unwind()
                .await
                .unwrap_or_else(|_| {
                    tracing::warn!(tool = name, "tool middleware panicked; denying the call");
                    ToolDecision::Deny("tool middleware panicked".into())
                })
        };
        match decision {
            ToolDecision::Allow => {}
            ToolDecision::Modify(new_args) => effective_args = new_args,
            ToolDecision::Deny(reason) => {
                let (msg, _) = denied_tool_call(id, name, &effective_args, &reason, tx);
                return (msg, Vec::new());
            }
        }
    }
    let args = &effective_args;

    let tool = tools.iter().find(|t| t.name() == name);

    // The Start event carries the effective (post-middleware) args — what
    // actually runs.
    tx.send(AgentEvent::ToolExecutionStart {
        tool_call_id: id.to_string(),
        tool_name: name.to_string(),
        args: args.clone(),
    })
    .ok();

    let on_update: Option<ToolUpdateFn> = {
        let tx = tx.clone();
        let id = id.to_string();
        let name = name.to_string();
        Some(Arc::new(move |partial: ToolResult| {
            tx.send(AgentEvent::ToolExecutionUpdate {
                tool_call_id: id.clone(),
                tool_name: name.clone(),
                partial_result: partial,
            })
            .ok();
        }))
    };

    let on_progress: Option<ProgressFn> = {
        let tx = tx.clone();
        let id = id.to_string();
        let name = name.to_string();
        Some(Arc::new(move |text: String| {
            tx.send(AgentEvent::ProgressMessage {
                tool_call_id: id.clone(),
                tool_name: name.clone(),
                text,
            })
            .ok();
        }))
    };

    let sub_agent_report: SubAgentReport = Arc::default();
    let ctx = ToolContext {
        tool_call_id: id.to_string(),
        tool_name: name.to_string(),
        cancel: cancel.child_token(),
        on_update,
        on_progress,
        sub_agent_report: Some(sub_agent_report.clone()),
    };

    let tool_span = tracing::info_span!(
        "tool",
        tool = %name,
        tool_call_id = %id,
        is_error = tracing::field::Empty,
    );
    use tracing::Instrument;
    let (result, is_error) = match tool {
        Some(tool) => {
            let execution = tool
                .execute(args.clone(), ctx)
                .instrument(tool_span.clone())
                .await;
            match execution {
                Ok(r) => (r, false),
                Err(e) => (
                    ToolResult {
                        content: vec![Content::Text {
                            text: e.to_string(),
                        }],
                        details: serde_json::Value::Null,
                    },
                    true,
                ),
            }
        }
        None => (
            ToolResult {
                content: vec![Content::Text {
                    text: format!("Tool {} not found", name),
                }],
                details: serde_json::Value::Null,
            },
            true,
        ),
    };

    tool_span.record("is_error", is_error);

    // Delegated spend travels out of band, so it is here even when the
    // sub-agent failed and the tool returned `Err`. Attach it to the result
    // too — a streaming consumer reads it off `ToolExecutionEnd` — but only
    // where the tool did not already: `SubAgentTool` sets it on success, and
    // a custom tool's own details are not ours to overwrite. A call that
    // reported several runs gets their combination, so the details never
    // show less than the rollup counted (see `from_sub_agent_result`).
    let delegated =
        std::mem::take(&mut *sub_agent_report.lock().unwrap_or_else(|e| e.into_inner()));
    let mut result = result;
    if let Some((first, rest)) = delegated.split_first() {
        let mut combined = first.clone();
        for stats in rest {
            combined.merge(stats);
        }
        attach_sub_agent_stats(&mut result.details, &combined);
    }

    tx.send(AgentEvent::ToolExecutionEnd {
        tool_call_id: id.to_string(),
        tool_name: name.to_string(),
        result: result.clone(),
        is_error,
    })
    .ok();

    let tool_result_msg = Message::ToolResult {
        tool_call_id: id.to_string(),
        tool_name: name.to_string(),
        content: result.content,
        is_error,
        timestamp: now_ms(),
    };

    tx.send(AgentEvent::MessageStart {
        message: tool_result_msg.clone().into(),
    })
    .ok();
    tx.send(AgentEvent::MessageEnd {
        message: tool_result_msg.clone().into(),
    })
    .ok();

    (tool_result_msg, delegated)
}

/// Put a sub-agent's stats under [`SUB_AGENT_STATS_KEY`] unless the details
/// already carry them. `Null` (every error result) becomes an object; any
/// other non-object is left alone rather than clobbered.
fn attach_sub_agent_stats(details: &mut serde_json::Value, stats: &SessionStats) {
    if details.is_null() {
        *details = serde_json::json!({});
    }
    if let Some(obj) = details.as_object_mut() {
        if !obj.contains_key(SUB_AGENT_STATS_KEY) {
            if let Ok(v) = serde_json::to_value(stats) {
                obj.insert(SUB_AGENT_STATS_KEY.to_string(), v);
            }
        }
    }
}

/// Emit events and build the error tool result for a middleware-denied call.
fn denied_tool_call(
    id: &str,
    name: &str,
    args: &serde_json::Value,
    reason: &str,
    tx: &mpsc::UnboundedSender<AgentEvent>,
) -> (Message, bool) {
    // Operator-visible signal: without this, a denial exists only in the
    // event stream / message history, invisible to telemetry.
    tracing::warn!(
        tool = name,
        tool_call_id = id,
        reason,
        "tool call denied by middleware"
    );
    unexecuted_tool_call(id, name, args, format!("Tool call denied: {}", reason), tx)
}

/// Emit events and build the error tool result for a call whose arguments
/// did not resolve to a JSON object: either they did not parse (cut off —
/// the response likely hit its output token limit) or they parsed to some
/// other JSON value (e.g. double-encoded as a string).
///
/// The text deliberately does not say "retry": the same call resent unchanged
/// would be cut off at the same point again.
fn unparsed_arguments_tool_call(
    id: &str,
    name: &str,
    args: &serde_json::Value,
    raw: &str,
    tx: &mpsc::UnboundedSender<AgentEvent>,
) -> (Message, bool) {
    let text = match serde_json::from_str::<serde_json::Value>(raw) {
        Err(e) => {
            tracing::warn!(
                tool = name,
                tool_call_id = id,
                len = raw.len(),
                parse_error = %e,
                "tool call not executed: arguments did not parse as JSON"
            );
            // Only an early end of input means the text was cut off. Complete
            // but malformed JSON (a trailing comma, two objects run together)
            // must not be answered with "make it smaller": the model would
            // resend the same syntax error in smaller pieces.
            if e.is_eof() {
                format!(
                    "The arguments for tool `{name}` were cut off before they were complete \
                     (the response likely hit the output token limit): {e}. The tool was not \
                     run. Do not resend the same call unchanged — make the arguments smaller \
                     (for example, split large content across several calls)."
                )
            } else {
                format!(
                    "The arguments for tool `{name}` are not valid JSON: {e}. The tool was not \
                     run. Send the arguments as a single valid JSON object."
                )
            }
        }
        Ok(v) => {
            let kind = match v {
                serde_json::Value::String(_) => "a string",
                serde_json::Value::Number(_) => "a number",
                serde_json::Value::Array(_) => "an array",
                serde_json::Value::Bool(_) => "a boolean",
                // Unreachable for a provider-built marker (`null` resolves to
                // `{}`, objects pass through), but a hand-built one could hold
                // either; say only what is known rather than invent a cause.
                serde_json::Value::Null | serde_json::Value::Object(_) => "",
            };
            tracing::warn!(
                tool = name,
                tool_call_id = id,
                kind,
                "tool call not executed: arguments were not delivered as a JSON object"
            );
            if kind.is_empty() {
                format!(
                    "The arguments for tool `{name}` were not delivered as a parsed JSON \
                     object. The tool was not run."
                )
            } else {
                format!(
                    "The arguments for tool `{name}` were not a JSON object (got {kind}). \
                     The tool was not run. Send the arguments as a single JSON object — \
                     not encoded as a string."
                )
            }
        }
    };
    unexecuted_tool_call(id, name, args, text, tx)
}

/// Emit events and build an error tool result for a call that was not run.
/// Start/End are both emitted so UI event pairing stays intact.
fn unexecuted_tool_call(
    id: &str,
    name: &str,
    args: &serde_json::Value,
    text: String,
    tx: &mpsc::UnboundedSender<AgentEvent>,
) -> (Message, bool) {
    tx.send(AgentEvent::ToolExecutionStart {
        tool_call_id: id.to_string(),
        tool_name: name.to_string(),
        args: args.clone(),
    })
    .ok();

    let result = ToolResult {
        content: vec![Content::Text { text }],
        details: serde_json::Value::Null,
    };

    tx.send(AgentEvent::ToolExecutionEnd {
        tool_call_id: id.to_string(),
        tool_name: name.to_string(),
        result: result.clone(),
        is_error: true,
    })
    .ok();

    let msg = Message::ToolResult {
        tool_call_id: id.to_string(),
        tool_name: name.to_string(),
        content: result.content,
        is_error: true,
        timestamp: now_ms(),
    };

    tx.send(AgentEvent::MessageStart {
        message: msg.clone().into(),
    })
    .ok();
    tx.send(AgentEvent::MessageEnd {
        message: msg.clone().into(),
    })
    .ok();

    (msg, true)
}

fn skip_tool_call(
    tool_call_id: &str,
    tool_name: &str,
    tx: &mpsc::UnboundedSender<AgentEvent>,
) -> Message {
    let result = ToolResult {
        content: vec![Content::Text {
            text: "Skipped due to queued user message.".into(),
        }],
        details: serde_json::Value::Null,
    };

    tx.send(AgentEvent::ToolExecutionStart {
        tool_call_id: tool_call_id.into(),
        tool_name: tool_name.into(),
        args: serde_json::Value::Null,
    })
    .ok();

    tx.send(AgentEvent::ToolExecutionEnd {
        tool_call_id: tool_call_id.into(),
        tool_name: tool_name.into(),
        result: result.clone(),
        is_error: true,
    })
    .ok();

    let msg = Message::ToolResult {
        tool_call_id: tool_call_id.into(),
        tool_name: tool_name.into(),
        content: result.content,
        is_error: true,
        timestamp: now_ms(),
    };

    tx.send(AgentEvent::MessageStart {
        message: msg.clone().into(),
    })
    .ok();
    tx.send(AgentEvent::MessageEnd {
        message: msg.clone().into(),
    })
    .ok();

    msg
}