zeph-core 0.22.3

// SPDX-FileCopyrightText: 2026 Andrei G <bug-ops>
// SPDX-License-Identifier: MIT OR Apache-2.0

use futures::FutureExt as _;
use tracing::Instrument;
use zeph_durable::{EffectIntentSubClass, OnAmbiguous, StepDescriptor};
use zeph_llm::provider::{Message, MessagePart, Role};
use zeph_tools::ExecutionContext;
use zeph_tools::executor::ToolCall;

use zeph_llm::provider::ToolDefinition;

use super::{
    CacheCheckResult, TierLoopData, TierLoopOutput, ToolDispatchContext, ToolExecFut,
    strip_tafc_fields, tool_args_hash,
};
use crate::agent::Agent;
use crate::channel::{Channel, StopHint, ToolStartEvent};

/// Build MAGE hard-block tier results when `is_blocked()` fired at dispatch time.
///
/// Returns `Some(TierLoopData)` synthesizing `ToolError::TrajectoryRiskExceeded` for every
/// call in the batch — bypassing `run_tier_execution_loop` entirely — when the hard-block
/// tier (`mage_blocked`, spec 004-16 FR-005) fired. Returns `None` when it did not, so the
/// caller runs the normal tier execution loop (this also covers the soft-escalation tier,
/// spec 004-16 FR-006, which gates on a single batch-level confirmation but then falls
/// through to the normal tier loop — see `Agent::confirm_mage_escalation` — so that
/// `check_trust`/`PermissionPolicy`/shadow-probe still apply per call; critic finding F1
/// caught an earlier version of this function that bypassed those gates for escalation too).
/// Extracted from `handle_native_tool_calls` to stay under the clippy line limit.
fn build_mage_bypass_tier_data(
    mage_blocked: Option<(f64, Vec<String>)>,
    calls: &[ToolCall],
) -> Option<TierLoopData> {
    let (score, top_signals) = mage_blocked?;
    Some(TierLoopData {
        tool_results: calls
            .iter()
            .map(|_| {
                Err(zeph_tools::ToolError::TrajectoryRiskExceeded {
                    score,
                    top_signals: top_signals.clone(),
                })
            })
            .collect(),
        pending_focus_checkpoint: None,
        pending_system_hints: Vec::new(),
    })
}

fn make_tool_hook_env(
    tool_name: &str,
    tool_input: &serde_json::Value,
    session_id: Option<&str>,
) -> std::collections::HashMap<String, String> {
    let mut env = zeph_subagent::make_base_hook_env(tool_name, tool_input);
    if let Some(sid) = session_id {
        env.insert("ZEPH_SESSION_ID".to_owned(), sid.to_owned());
    }
    crate::agent::hooks_dispatch::insert_main_agent_ctx(&mut env, session_id);
    env
}

impl<C: Channel> Agent<C> {
    #[tracing::instrument(
        name = "core.tool.run_post_dispatch_phases",
        skip_all,
        level = "debug",
        err
    )]
    /// Runs the confirmation, retry, and parameter-reformat phases in sequence.
    ///
    /// Returns `Ok(true)` as soon as any phase reports that the user cancelled the turn, skipping
    /// the remaining phases — each phase already persists its own `[Cancelled]` tombstone, so
    /// running further phases after one reports cancellation would duplicate that write (#5513).
    async fn run_post_dispatch_phases(
        &mut self,
        tool_calls: &[zeph_llm::provider::ToolUseRequest],
        calls: &[ToolCall],
        tool_results: &mut [Result<Option<zeph_tools::ToolOutput>, zeph_tools::ToolError>],
        max_retries: usize,
        cancel: &tokio_util::sync::CancellationToken,
    ) -> Result<bool, crate::agent::error::AgentError> {
        if self
            .handle_confirmation_phase(tool_calls, calls, tool_results, cancel)
            .await?
        {
            return Ok(true);
        }
        if self
            .handle_retry_phase(tool_calls, calls, tool_results, max_retries, cancel)
            .await?
        {
            return Ok(true);
        }
        if self
            .handle_reformat_phase(tool_calls, calls, tool_results, cancel)
            .await?
        {
            return Ok(true);
        }
        Ok(false)
    }

    /// Resets skill env, records the cancellation, persists the tombstone tool results, and
    /// sends the `[Cancelled]` notice — the fixed sequence every cancellation checkpoint in
    /// this file must follow (#5654). Skipping or reordering a step here has previously left
    /// orphaned `ToolUse` messages with no matching `ToolResult` (#5464, #5513, #5646) — see
    /// the recurring defect-class notes in issue history.
    ///
    /// The tombstone persist runs before the notification send and its failure is logged
    /// rather than propagated: a closed/dropped channel receiver or disconnected adapter must
    /// never skip the tombstone and reintroduce the orphaned-`tool_calls` defect (#5717).
    ///
    /// `pub(super)`: called from the confirmation and retry/reformat phase modules after the
    /// `tool_execution` split.
    pub(super) async fn cancel_tool_batch(
        &mut self,
        tool_calls: &[zeph_llm::provider::ToolUseRequest],
        log_msg: &str,
    ) -> Result<(), crate::agent::error::AgentError> {
        self.tool_executor.set_skill_env(None);
        tracing::info!("{log_msg}");
        self.update_metrics(|m| m.cancellations += 1);
        self.persist_cancelled_tool_results(tool_calls, None).await;
        if let Err(e) = self.channel.send("[Cancelled]").await {
            tracing::warn!(
                error = %e,
                "cancel_tool_batch: failed to notify channel of cancellation"
            );
        }
        Ok(())
    }

    fn run_pre_execution_verifiers(&mut self, calls: &[ToolCall]) -> Vec<bool> {
        let mut pre_exec_blocked = vec![false; calls.len()];
        if self.tool_orchestrator.pre_execution_verifiers.is_empty() {
            return pre_exec_blocked;
        }
        for (idx, call) in calls.iter().enumerate() {
            let args_value = serde_json::Value::Object(call.params.clone());
            for verifier in &self.tool_orchestrator.pre_execution_verifiers {
                match verifier.verify(call.tool_id.as_str(), &args_value) {
                    zeph_tools::VerificationResult::Block { reason } => {
                        tracing::warn!(
                            tool = %call.tool_id,
                            verifier = verifier.name(),
                            %reason,
                            "pre-execution verifier blocked tool call"
                        );
                        self.update_metrics(|m| m.pre_execution_blocks += 1);
                        self.push_security_event(
                            zeph_common::SecurityEventCategory::PreExecutionBlock,
                            call.tool_id.as_str(),
                            format!("{}: {}", verifier.name(), reason),
                        );
                        if let Some(ref logger) = self.tool_orchestrator.audit_logger {
                            let args_json = serde_json::to_string(&args_value).unwrap_or_default();
                            let entry = zeph_tools::AuditEntry {
                                source_kind: None,
                                trust_level: None,
                                timestamp: zeph_tools::chrono_now(),
                                tool: call.tool_id.clone(),
                                command: args_json,
                                result: zeph_tools::AuditResult::Blocked {
                                    reason: format!("{}: {}", verifier.name(), reason),
                                },
                                duration_ms: 0,
                                error_category: Some("pre_execution_block".to_owned()),
                                error_domain: Some("security".to_owned()),
                                error_phase: Some(
                                    zeph_tools::error_taxonomy::ToolInvocationPhase::Setup
                                        .label()
                                        .to_owned(),
                                ),
                                claim_source: None,
                                mcp_server_id: None,
                                injection_flagged: false,
                                embedding_anomalous: false,
                                cross_boundary_mcp_to_acp: false,
                                adversarial_policy_decision: None,
                                exit_code: None,
                                truncated: false,
                                caller_id: call.caller_id.clone(),
                                skill_name: call.skill_name.clone(),
                                policy_match: None,
                                correlation_id: None,
                                vigil_risk: None,
                                execution_env: None,
                                resolved_cwd: None,
                                scope_at_definition: None,
                                scope_at_dispatch: None,
                            };
                            let logger = std::sync::Arc::clone(logger);
                            self.runtime.lifecycle.supervisor.spawn(
                                crate::agent::agent_supervisor::TaskClass::Telemetry,
                                "audit-log",
                                async move { logger.log(&entry).await },
                            );
                        }
                        pre_exec_blocked[idx] = true;
                        break;
                    }
                    zeph_tools::VerificationResult::Warn { message } => {
                        tracing::warn!(
                            tool = %call.tool_id,
                            verifier = verifier.name(),
                            %message,
                            "pre-execution verifier warning (not blocked)"
                        );
                        self.update_metrics(|m| m.pre_execution_warnings += 1);
                        self.push_security_event(
                            zeph_common::SecurityEventCategory::PreExecutionWarn,
                            call.tool_id.as_str(),
                            format!("{}: {}", verifier.name(), message),
                        );
                    }
                    _ => {}
                }
            }
        }
        pre_exec_blocked
    }

    /// Block tool calls whose names are absent from the channel-level allowlist (#3879).
    ///
    /// No-op when no allowlist is configured (`None`). Skips already-blocked indices.
    /// Comparison is case-sensitive; channel configs must use canonical lowercase names.
    fn apply_channel_tool_allowlist(&mut self, calls: &[ToolCall], pre_exec_blocked: &mut [bool]) {
        let Some(ref allowlist) = self.runtime.config.channel_tool_allowlist else {
            return;
        };
        for (idx, call) in calls.iter().enumerate() {
            if pre_exec_blocked[idx] {
                continue;
            }
            if !allowlist.iter().any(|t| t == call.tool_id.as_str()) {
                tracing::warn!(tool = %call.tool_id, "tool blocked by channel allowlist");
                self.update_metrics(|m| m.pre_execution_blocks += 1);
                self.push_security_event(
                    zeph_common::SecurityEventCategory::PreExecutionBlock,
                    call.tool_id.as_str(),
                    format!(
                        "channel allowlist: '{}' is not permitted on this channel",
                        call.tool_id
                    ),
                );
                pre_exec_blocked[idx] = true;
            }
        }
    }

    fn compute_utility_actions(
        &mut self,
        calls: &[ToolCall],
        pre_exec_blocked: &[bool],
        pending_system_hints: &mut Vec<String>,
    ) -> (Vec<zeph_tools::UtilityAction>, bool) {
        #[allow(clippy::cast_possible_truncation)]
        let tokens_consumed =
            usize::try_from(self.runtime.providers.cached_prompt_tokens).unwrap_or(usize::MAX);
        // token_budget = 0 signals "unknown" to UtilityContext — cost component is zeroed.
        let token_budget: usize = 0;
        let tool_calls_this_turn = self.tool_orchestrator.recent_tool_calls.len();
        // Detect explicit tool request from the last user message text only.
        // We only read MessagePart::Text parts so tool outputs/thinking blocks are excluded.
        let explicit_request = self
            .msg
            .messages
            .iter()
            .rfind(|m| m.role == zeph_llm::provider::Role::User)
            .is_some_and(|m| {
                let text = if m.parts.is_empty() {
                    m.content.clone()
                } else {
                    m.parts
                        .iter()
                        .filter_map(|p| {
                            if let zeph_llm::provider::MessagePart::Text { text } = p {
                                Some(text.as_str())
                            } else {
                                None
                            }
                        })
                        .collect::<Vec<_>>()
                        .join(" ")
                };
                zeph_tools::has_explicit_tool_request(&text)
            });
        let mut actions = Vec::with_capacity(calls.len());
        // Once window exhaustion fires, all remaining calls in the batch are downgraded to Stop.
        let mut window_exhausted = false;
        for (idx, call) in calls.iter().enumerate() {
            if window_exhausted {
                actions.push(zeph_tools::UtilityAction::Stop);
                continue;
            }
            if pre_exec_blocked[idx] {
                actions.push(zeph_tools::UtilityAction::ToolCall);
                continue;
            }
            if self
                .tool_orchestrator
                .utility_scorer
                .is_exempt(call.tool_id.as_str())
            {
                actions.push(zeph_tools::UtilityAction::ToolCall);
                continue;
            }
            // Consume any pending mandated-retry marker before scoring: a call that was vetoed
            // by the Retrieve rule earlier this turn, and that the injected hint told the LLM
            // to retry with the same arguments, must not be re-vetoed as a redundant duplicate
            // when the LLM complies (#5719).
            let mandated_retry = self
                .tool_orchestrator
                .utility_scorer
                .take_mandated_retry(call);
            let ctx = zeph_tools::UtilityContext {
                tool_calls_this_turn: tool_calls_this_turn + idx,
                tokens_consumed,
                token_budget,
                user_requested: explicit_request,
                mandated_retry,
            };
            let score = self.tool_orchestrator.utility_scorer.score(call, &ctx);
            let action = self
                .tool_orchestrator
                .utility_scorer
                .recommend_action(score.as_ref(), &ctx);
            tracing::debug!(
                tool = %call.tool_id,
                score = ?score.as_ref().map(|s| s.total),
                threshold = self.tool_orchestrator.utility_scorer.threshold(),
                action = ?action,
                "utility gate: action recommended"
            );
            if action != zeph_tools::UtilityAction::ToolCall {
                tracing::info!(
                    tool = %call.tool_id,
                    action = ?action,
                    "utility gate: non-execute action"
                );
            }
            // Record call regardless so subsequent calls in this batch see it as prior.
            self.tool_orchestrator.utility_scorer.record_call(call);
            // note_action increments the consecutive-low counter for scored calls only.
            // Exempt and pre-exec-blocked calls above bypass scoring and are not tracked.
            if self.tool_orchestrator.utility_scorer.note_action(&action) {
                let n = self.tool_orchestrator.utility_scorer.utility_window();
                tracing::info!(
                    window = n,
                    "utility gate: consecutive-low window exhausted, early-stopping loop"
                );
                pending_system_hints.push(format!(
                    "Tool loop stopped early: utility below threshold for {n} consecutive calls."
                ));
                window_exhausted = true;
            }
            actions.push(action);
        }
        (actions, window_exhausted)
    }

    #[tracing::instrument(
        name = "core.tool.handle_native_tool_calls",
        skip_all,
        level = "debug",
        fields(tool_count = tool_calls.len()),
        err
    )]
    /// Returns `true` when the utility-window was exhausted and the outer iteration loop
    /// should break immediately after this batch.
    pub(super) async fn handle_native_tool_calls(
        &mut self,
        text: Option<&str>,
        tool_calls: &[zeph_llm::provider::ToolUseRequest],
    ) -> Result<bool, crate::agent::error::AgentError> {
        let t_tool_exec = std::time::Instant::now();
        tracing::debug!("turn timing: tool_exec start");

        // Issues #6558/#6569 (write-time memory-consent gate TOCTOU): ratchet the shared
        // trust slot to this batch's worst case BEFORE any tool call below — including
        // `memory_save` itself — starts executing. Must run ahead of tier dispatch, not
        // merely ahead of `memory_save`'s own future, since tier calls execute concurrently
        // via `join_all` and `MemoryToolExecutor` cannot observe `self.msg` to check for
        // itself. See `ratchet_memory_consent_trust_for_dispatch`'s doc comment.
        self.ratchet_memory_consent_trust_for_dispatch(tool_calls);

        // Scan for image-exfiltration in accompanying text, send to channel, persist
        // the assistant ToolUse message.
        self.push_assistant_tool_use_message(text, tool_calls)
            .await?;

        // Build calls, assign IDs, run exfiltration guard, gate checks (pre-exec/utility/
        // quota/repeat/cache), and inject skill env. Extracted to keep this function under
        // the clippy line limit.
        let ToolDispatchContext {
            calls,
            tool_call_ids,
            mut tool_started_ats,
            pre_exec_blocked,
            utility_actions,
            quota_blocked,
            args_hashes,
            repeat_blocked,
            cache_hits,
            mcp_tool_ids,
            mage_blocked,
            mage_escalate,
            mut early_stop_hints,
            window_exhausted,
        } = self.prepare_tool_dispatch(tool_calls);

        let max_retries = self.tool_orchestrator.max_tool_retries;
        // Clamp to 1 to prevent Semaphore(0) deadlock when config is set to 0.
        let max_parallel = self.runtime.config.timeouts.max_parallel_tools.max(1);
        let cancel = self.runtime.lifecycle.cancel_token.clone();

        // Causal IPI pre-probe: record behavioral baseline before tool batch dispatch.
        let causal_pre_response = self.run_causal_pre_probe().await;

        // Phase 1: Tiered parallel execution bounded by a shared semaphore.
        // Extracted to run_tier_execution_loop to satisfy the line-count limit.
        // Returns None when the user cancelled (caller must return Ok(())).
        // MAGE hard block: bypass the tier loop and build TrajectoryRiskExceeded results
        // directly so process_tool_result_batch renders them normally.
        // MAGE soft escalation: gate on a single batch-level confirmation, then fall through
        // to the normal tier loop below — check_trust/PermissionPolicy/shadow-probe must still
        // run per call (critic finding F1: an earlier version bypassed those gates here).
        let tier_data: TierLoopOutput =
            if let Some(bypass) = build_mage_bypass_tier_data(mage_blocked, &calls) {
                Some(bypass)
            } else {
                if mage_escalate && self.confirm_mage_escalation(tool_calls).await? {
                    return Ok(false);
                }
                self.run_tier_execution_loop(
                    tool_calls,
                    &calls,
                    &pre_exec_blocked,
                    &utility_actions,
                    quota_blocked,
                    &args_hashes,
                    &repeat_blocked,
                    &cache_hits,
                    &mcp_tool_ids,
                    max_parallel,
                    &cancel,
                    &tool_call_ids,
                    &mut tool_started_ats,
                )
                .await?
            };

        // Unpack tier execution output. None means the user cancelled — return early.
        let Some(TierLoopData {
            mut tool_results,
            pending_focus_checkpoint,
            mut pending_system_hints,
        }) = tier_data
        else {
            return Ok(false);
        };
        // Prepend window-exhaustion hints so the LLM sees them before per-call skipped results.
        if !early_stop_hints.is_empty() {
            early_stop_hints.extend(pending_system_hints);
            pending_system_hints = early_stop_hints;
        }

        // Phases 2a / 2 / 3: confirmation, transient retry, parameter reformat.
        // Each phase may signal cancellation (Ok(true)), which already persisted its own
        // tombstone — skip process_tool_result_batch below to avoid a duplicate batch write (#5513).
        let post_dispatch_cancelled = self
            .run_post_dispatch_phases(tool_calls, &calls, &mut tool_results, max_retries, &cancel)
            .await?;
        if post_dispatch_cancelled {
            return Ok(false);
        }

        // Process results, persist messages, run LSP hooks, fire deferred reflection.
        // Also clears skill env and syncs cache counters after execution.
        // Extracted to process_tool_result_batch to satisfy the line-count limit.
        self.process_tool_result_batch(
            tool_calls,
            &tool_call_ids,
            &tool_started_ats,
            tool_results,
            causal_pre_response,
            pending_focus_checkpoint,
            pending_system_hints,
        )
        .await?;

        let tool_exec_ms = u64::try_from(t_tool_exec.elapsed().as_millis()).unwrap_or(u64::MAX);
        tracing::debug!(ms = tool_exec_ms, "turn timing: tool_exec done");
        self.runtime.metrics.pending_timings.tool_exec_ms = self
            .runtime
            .metrics
            .pending_timings
            .tool_exec_ms
            .saturating_add(tool_exec_ms);

        Ok(window_exhausted)
    }

    async fn push_assistant_tool_use_message(
        &mut self,
        text: Option<&str>,
        tool_calls: &[zeph_llm::provider::ToolUseRequest],
    ) -> Result<(), crate::agent::error::AgentError> {
        // S4: scan text accompanying ToolUse responses for markdown image exfiltration.
        let cleaned_text: Option<String> = if let Some(t) = text
            && !t.is_empty()
        {
            Some(self.scan_output_and_warn(t))
        } else {
            None
        };

        if let Some(ref t) = cleaned_text
            && !t.is_empty()
        {
            let display = self.maybe_redact(t);
            self.channel.send(&display).await?;
        }

        let mut parts: Vec<MessagePart> = Vec::new();
        if let Some(ref t) = cleaned_text
            && !t.is_empty()
        {
            parts.push(MessagePart::Text { text: t.clone() });
        }
        for tc in tool_calls {
            parts.push(MessagePart::ToolUse {
                id: tc.id.clone(),
                name: tc.name.to_string(),
                input: tc.input.clone(),
            });
        }
        let assistant_msg = Message::from_parts(Role::Assistant, parts);
        self.persist_message(
            Role::Assistant,
            &assistant_msg.content,
            &assistant_msg.parts,
            false,
        )
        .await;
        self.push_message(assistant_msg);
        if let (Some(id), Some(last)) = (
            self.msg.last_persisted_message_id,
            self.msg.messages.last_mut(),
        ) {
            last.metadata.db_id = Some(id);
        }
        Ok(())
    }

    fn skill_attribution(&self) -> Option<Vec<String>> {
        (!self.services.skill.active_skill_names.is_empty())
            .then(|| self.services.skill.active_skill_names.clone())
    }

    #[allow(clippy::too_many_lines)] // unmask-miss telemetry aggregation (#5437 S1) crossed the 100-line limit
    fn prepare_tool_dispatch(
        &mut self,
        tool_calls: &[zeph_llm::provider::ToolUseRequest],
    ) -> ToolDispatchContext {
        let tafc_enabled = self.tool_orchestrator.tafc.enabled;
        // PAAC secret unmasking (#5437): resolved once per dispatch batch (Arc clone, cheap)
        // so the closure below doesn't need to re-borrow `self` for every tool call.
        let secret_registry = self.services.security.secret_registry.clone();
        // When the orchestration scheduler has set a named execution environment for the
        // current task, inject it into every ToolCall so ShellExecutor::resolve_context
        // uses the right env/cwd without the LLM having to supply it.
        let task_ctx = self
            .services
            .orchestration
            .task_execution_env
            .as_deref()
            .map(|name| ExecutionContext::default().with_name(name));
        let tool_call_ids: Vec<String> = tool_calls
            .iter()
            .map(|_| uuid::Uuid::new_v4().to_string())
            .collect();

        // S1: unmask-miss telemetry (#5437) — counts string leaves that still carry a
        // `<SECRET:` prefix after `unmask_json_value`, meaning the model failed to reproduce a
        // placeholder byte-for-byte. Aggregated across the whole batch and reported once below.
        let mut unmask_misses = 0usize;
        let mut unmask_miss_tools: Vec<String> = Vec::new();

        // INVARIANT: `calls` must stay the same length as `tool_calls`, with `calls[i]`
        // always corresponding to `tool_calls[i]`. The tool-call DAG (built from `tool_calls`)
        // and every per-call vector below (tool_started_ats, pre_exec_blocked, args_hashes,
        // repeat_blocked, cache_hits, utility_actions) are all indexed by that same original
        // position downstream — dropping an entry here would desynchronize every one of them
        // against `tool_calls`, silently misattributing timing/state or executing the wrong
        // tool call under another call's identity (#5731).
        let mut tafc_stripped = vec![false; tool_calls.len()];
        let calls: Vec<ToolCall> = tool_calls
            .iter()
            .enumerate()
            .map(|(idx, tc)| {
                let mut params: serde_json::Map<String, serde_json::Value> =
                    if let serde_json::Value::Object(map) = &tc.input {
                        map.clone()
                    } else {
                        serde_json::Map::new()
                    };
                // Unmask secret placeholders in tool arguments before dispatch (e.g. the model
                // echoing a masked value it saw in a prior tool result back into a shell `code`
                // or HTTP header argument). No-op when secret masking is disabled/empty.
                if let Some(registry) = secret_registry.as_deref() {
                    let mut tc_misses = 0usize;
                    for value in params.values_mut() {
                        tc_misses += unmask_json_value(value, registry);
                    }
                    if tc_misses > 0 {
                        unmask_misses += tc_misses;
                        unmask_miss_tools.push(tc.name.to_string());
                    }
                }
                if tafc_enabled && strip_tafc_fields(&mut params, tc.name.as_str()).is_err() {
                    // Model produced only think fields — keep the slot (see invariant above)
                    // but mark it so the existing pre-exec gate skips it without dispatch.
                    tafc_stripped[idx] = true;
                }
                ToolCall {
                    tool_id: tc.name.clone(),
                    params,
                    caller_id: None,
                    context: task_ctx.clone(),
                    tool_call_id: tool_call_ids[idx].clone(),
                    skill_name: self.skill_attribution(),
                }
            })
            .collect();

        if unmask_misses > 0 {
            tracing::warn!(
                misses = unmask_misses,
                tools = ?unmask_miss_tools,
                "secret placeholder(s) in tool arguments did not resolve — the model likely \
                 mangled a <SECRET:...> token (whitespace/truncation); the affected tool call(s) \
                 will run with the literal placeholder text, not the real secret"
            );
            self.update_metrics(|m| m.secret_unmask_misses += unmask_misses as u64);
        }
        // Timestamps filled just before each tier's join_all so audit reflects actual start.
        let tool_started_ats = vec![std::time::Instant::now(); tool_calls.len()];

        self.check_exfiltration_urls(tool_calls);

        // Pre-execution verification (TrustBench, #1630): runs before repeat-detection.
        let mut pre_exec_blocked = self.run_pre_execution_verifiers(&calls);

        self.apply_channel_tool_allowlist(&calls, &mut pre_exec_blocked);

        // TAFC-stripped calls (kept as placeholders above to preserve index alignment)
        // are routed through the existing pre-exec gate so they skip dispatch with a
        // synthetic result instead of being silently dropped from `calls`.
        for (idx, stripped) in tafc_stripped.into_iter().enumerate() {
            if stripped {
                pre_exec_blocked[idx] = true;
            }
        }

        // Utility gate: score each call and recommend an action (#2477).
        // user_requested is from the last user message only (prompt-injection guard).
        let mut early_stop_hints: Vec<String> = Vec::new();
        let (utility_actions, window_exhausted) =
            self.compute_utility_actions(&calls, &pre_exec_blocked, &mut early_stop_hints);

        // M3: quota counted once per batch; retries do not consume additional slots.
        let quota_blocked = self.check_and_update_quota(calls.len());

        // Build args hashes and check for repeats. Blocked calls get a pre-built error result.
        let args_hashes: Vec<u64> = calls.iter().map(|c| tool_args_hash(&c.params)).collect();
        let repeat_blocked: Vec<bool> = calls
            .iter()
            .zip(args_hashes.iter())
            .map(|(call, &hash)| {
                let blocked = self
                    .tool_orchestrator
                    .is_repeat(call.tool_id.as_str(), hash);
                if blocked {
                    tracing::warn!(
                        tool = %call.tool_id,
                        "[repeat-detect] identical tool call detected, skipping execution"
                    );
                }
                blocked
            })
            .collect();
        // CRIT-3: push calls before execution; cache hits included (P1 invariant).
        for (call, &hash) in calls.iter().zip(args_hashes.iter()) {
            self.tool_orchestrator
                .push_tool_call(call.tool_id.as_str(), hash);
        }

        // Resolved once per dispatch batch and threaded through `ToolDispatchContext` so both
        // the cache-lookup gate below and the cache-store gate later in `apply_tier_results`
        // share one registry scan instead of each re-scanning `tool_definitions_erased()`
        // per call/per tier (#5733 follow-up, M1). Also sidesteps a borrow-checker conflict:
        // capturing a whole-`self` method inside the closure below would conflict with that
        // closure's existing `&mut self.tool_orchestrator` borrow.
        let mcp_tool_ids: std::collections::HashSet<String> = self
            .tool_executor
            .tool_definitions_erased()
            .into_iter()
            .filter(zeph_tools::registry::ToolDef::is_mcp_tool)
            .map(|d| d.id.into_owned())
            .collect();

        // Cache lookup: hits pre-built before dispatch; cache store happens after join_all.
        let cache_hits: Vec<Option<zeph_tools::ToolOutput>> = calls
            .iter()
            .zip(args_hashes.iter())
            .zip(repeat_blocked.iter())
            .map(|((call, &hash), &blocked)| {
                if blocked
                    || !zeph_tools::is_cacheable(
                        call.tool_id.as_str(),
                        mcp_tool_ids.contains(call.tool_id.as_str()),
                    )
                {
                    return None;
                }
                let key = zeph_tools::CacheKey::new(call.tool_id.as_str(), hash);
                self.tool_orchestrator.result_cache.get(&key)
            })
            .collect();

        // Inject active skill secrets before tool execution.
        self.inject_active_skill_env();

        // MAGE trajectory risk gate (spec 004-16 FR-004, FR-005).
        // Extracted to keep prepare_tool_dispatch under the line limit.
        let mage_blocked = self.check_mage_block();
        // Soft-escalation tier (spec 004-16 FR-006): only meaningful when the hard block
        // above did not already fire — the two threshold ranges never overlap, but the
        // guard keeps this call site independent of that invariant.
        let mage_escalate = mage_blocked.is_none() && self.check_mage_escalation();

        ToolDispatchContext {
            calls,
            tool_call_ids,
            tool_started_ats,
            pre_exec_blocked,
            utility_actions,
            quota_blocked,
            args_hashes,
            repeat_blocked,
            cache_hits,
            mcp_tool_ids,
            mage_blocked,
            mage_escalate,
            early_stop_hints,
            window_exhausted,
        }
    }

    #[tracing::instrument(
        name = "core.tool.run_tier_execution_loop",
        skip_all,
        level = "debug",
        fields(tool_count = tool_calls.len()),
        err
    )]
    #[allow(clippy::too_many_arguments, clippy::too_many_lines)]
    async fn run_tier_execution_loop(
        &mut self,
        tool_calls: &[zeph_llm::provider::ToolUseRequest],
        calls: &[ToolCall],
        pre_exec_blocked: &[bool],
        utility_actions: &[zeph_tools::UtilityAction],
        quota_blocked: bool,
        args_hashes: &[u64],
        repeat_blocked: &[bool],
        cache_hits: &[Option<zeph_tools::ToolOutput>],
        mcp_tool_ids: &std::collections::HashSet<String>,
        max_parallel: usize,
        cancel: &tokio_util::sync::CancellationToken,
        tool_call_ids: &[String],
        tool_started_ats: &mut [std::time::Instant],
    ) -> Result<TierLoopOutput, crate::agent::error::AgentError> {
        // Build a dependency DAG over tool_use_id references in call arguments. When the
        // DAG is trivial (no dependencies — the common case), we execute all calls in a
        // single tier with zero overhead. When dependencies exist, we partition calls into
        // topological tiers and execute each tier in parallel, awaiting the previous tier
        // before starting the next.
        //
        // ToolStartEvent is sent at the beginning of each tier so the UI reflects actual
        // execution start time rather than pre-build time.
        let semaphore = std::sync::Arc::new(tokio::sync::Semaphore::new(max_parallel));
        let dag = super::tool_call_dag::ToolCallDag::build(tool_calls);
        let trivial = dag.is_trivial();
        let tiers = dag.tiers();
        let tier_count = tiers.len();
        // Clone the Arc before the mutable borrow loop so try_commit can be called without
        // holding a borrow on self across await points.
        let speculation_engine = self.services.speculation_engine.clone();
        tracing::debug!(
            trivial,
            tier_count,
            tool_count = tool_calls.len(),
            "tool dispatch: partitioned into tiers"
        );

        // Pre-allocate result vector; slots are filled as tiers complete.
        let mut tool_results: Vec<Result<Option<zeph_tools::ToolOutput>, zeph_tools::ToolError>> =
            (0..tool_calls.len()).map(|_| Ok(None)).collect();

        // Pre-process focus tool calls (#1850) and compress_context (#2218).
        // These need &mut self and cannot run inside the parallel tier futures.
        // Pre-populate their results so the tier loop skips them.
        let pending_focus_checkpoint = self
            .preprocess_focus_compress_calls(tool_calls, &mut tool_results)
            .await;

        // Track which indices have a failed/ConfirmationRequired prerequisite so that
        // dependent calls in later tiers receive a synthetic error instead of executing.
        // IMP-02: ConfirmationRequired is treated as a failure for dependency propagation —
        // a dependent tool must not proceed when its prerequisite is awaiting user approval.
        let mut failed_ids: std::collections::HashSet<String> = std::collections::HashSet::new();
        // Utility gate hints (Retrieve/Verify) are deferred so they are pushed after
        // User(tool_results), maintaining valid OpenAI message ordering (#2615).
        let mut pending_system_hints: Vec<String> = Vec::new();

        for (tier_idx, tier) in tiers.into_iter().enumerate() {
            if cancel.is_cancelled() {
                self.cancel_tool_batch(tool_calls, "tool execution cancelled by user")
                    .await?;
                return Ok(None);
            }

            if tier_count > 1 {
                self.channel
                    .send_status_best_effort(&format!(
                        "Executing tools (tier {}/{})\u{2026}",
                        tier_idx + 1,
                        tier_count
                    ))
                    .await;
            }

            // Pre-scan: commit speculative handles and emit speculative ToolStartEvents.
            let speculative_commits = self
                .commit_speculative_tier(
                    &tier.indices,
                    calls,
                    tool_calls,
                    tool_call_ids,
                    tool_started_ats,
                    speculation_engine.as_ref(),
                    mcp_tool_ids,
                )
                .await?;

            // Stamp execution start time and send ToolStartEvent for non-committed calls (§3.7).
            let non_committed_indices: Vec<usize> = tier
                .indices
                .iter()
                .copied()
                .filter(|idx| !speculative_commits.contains_key(idx))
                .collect();
            self.stamp_and_send_tier_start(
                &non_committed_indices,
                tool_calls,
                tool_call_ids,
                tool_started_ats,
                mcp_tool_ids,
            )
            .await?;

            // Build futures for non-committed calls in this tier.
            let mut tier_futs = self
                .build_tier_call_futures(
                    tool_calls,
                    calls,
                    &non_committed_indices,
                    &dag,
                    &failed_ids,
                    quota_blocked,
                    pre_exec_blocked,
                    utility_actions,
                    repeat_blocked,
                    cache_hits,
                    &semaphore,
                    &mut pending_system_hints,
                )
                .await?;

            // Inject committed speculative results as ready futures.
            for (idx, result) in speculative_commits {
                tier_futs.push((idx, Box::pin(std::future::ready(result))));
            }

            // Execute futures concurrently with cancellation and MCP elicitation drain.
            let (indices, futs): (Vec<usize>, Vec<ToolExecFut>) = tier_futs.into_iter().unzip();
            let Some(tier_results) = self.execute_tier_join(futs, cancel, tool_calls).await? else {
                return Ok(None);
            };

            // Store results, update dependency graph, and run after_tool hooks.
            self.apply_tier_results(
                indices,
                tier_results,
                tool_calls,
                calls,
                cache_hits,
                mcp_tool_ids,
                args_hashes,
                tool_started_ats,
                max_parallel,
                &mut failed_ids,
                &mut tool_results,
            )
            .await;

            if tier_count > 1 {
                self.channel.send_status_best_effort("").await;
            }

            // Check hook block cap after each tier (RF-1: counter is per-turn, not per-tier).
            // hook_block_cap = 0 means no cap.
            let cap = self.tool_orchestrator.hook_block_cap;
            if cap > 0 && self.tool_orchestrator.hook_block_count >= cap {
                tracing::warn!(
                    hook_block_count = self.tool_orchestrator.hook_block_count,
                    hook_block_cap = cap,
                    "hook block cap reached — ending turn"
                );
                let _ = self
                    .channel
                    .send(&format!(
                        "Stopping: PreToolUse hook blocked {}/{} tool calls this turn.",
                        self.tool_orchestrator.hook_block_count, cap
                    ))
                    .await;
                break;
            }
        }

        // Pad with empty results if needed (defensive; should not happen).
        while tool_results.len() < tool_calls.len() {
            tool_results.push(Ok(None));
        }

        Ok(Some(TierLoopData {
            tool_results,
            pending_focus_checkpoint,
            pending_system_hints,
        }))
    }

    pub(super) async fn stamp_and_send_tier_start(
        &mut self,
        tier_indices: &[usize],
        tool_calls: &[zeph_llm::provider::ToolUseRequest],
        tool_call_ids: &[String],
        tool_started_ats: &mut [std::time::Instant],
        mcp_tool_ids: &std::collections::HashSet<String>,
    ) -> Result<(), crate::agent::error::AgentError> {
        let tier_start = std::time::Instant::now();
        for &idx in tier_indices {
            tool_started_ats[idx] = tier_start;
        }
        for &idx in tier_indices {
            let tc = &tool_calls[idx];
            self.channel
                .send_tool_start(ToolStartEvent {
                    tool_name: tc.name.clone(),
                    tool_call_id: tool_call_ids[idx].clone(),
                    params: Some(tc.input.clone()),
                    parent_tool_use_id: self.services.session.parent_tool_use_id.clone(),
                    started_at: std::time::Instant::now(),
                    speculative: false,
                    sandbox_profile: None,
                    is_mcp: mcp_tool_ids.contains(tc.name.as_str()),
                })
                .await?;
        }
        Ok(())
    }

    #[tracing::instrument(
        name = "core.tool.commit_speculative_tier",
        skip_all,
        level = "debug",
        fields(tier_size = tier_indices.len()),
        err
    )]
    #[allow(clippy::too_many_arguments)]
    pub(super) async fn commit_speculative_tier(
        &mut self,
        tier_indices: &[usize],
        calls: &[ToolCall],
        tool_calls: &[zeph_llm::provider::ToolUseRequest],
        tool_call_ids: &[String],
        tool_started_ats: &mut [std::time::Instant],
        engine: Option<&std::sync::Arc<crate::agent::speculative::SpeculationEngine>>,
        mcp_tool_ids: &std::collections::HashSet<String>,
    ) -> Result<
        std::collections::HashMap<
            usize,
            Result<Option<zeph_tools::ToolOutput>, zeph_tools::ToolError>,
        >,
        crate::agent::error::AgentError,
    > {
        let mut commits: std::collections::HashMap<
            usize,
            Result<Option<zeph_tools::ToolOutput>, zeph_tools::ToolError>,
        > = std::collections::HashMap::new();

        let Some(engine) = engine else {
            return Ok(commits);
        };

        for &idx in tier_indices {
            let Some(result) = engine.try_commit(&calls[idx]).await else {
                continue;
            };
            if let Err(ref e) = result {
                tracing::warn!(
                    tool = %calls[idx].tool_id,
                    error = %e,
                    "speculative commit returned Err — result will be used as-is"
                );
                // Invariant: ConfirmationRequired must never reach the commit boundary —
                // try_dispatch guards against it at dispatch time via requires_confirmation_erased.
                #[cfg(debug_assertions)]
                if matches!(e, zeph_tools::ToolError::ConfirmationRequired { .. }) {
                    tracing::error!(
                        tool = %calls[idx].tool_id,
                        "invariant violated: committed speculative result is ConfirmationRequired"
                    );
                }
            }
            // M1: stamp actual dispatch time so build_tool_output_messages computes correct elapsed.
            tool_started_ats[idx] = std::time::Instant::now();
            commits.insert(idx, result);
        }

        // Emit ToolStartEvent with speculative: true for all committed calls.
        for &idx in tier_indices {
            if commits.contains_key(&idx) {
                let tc = &tool_calls[idx];
                self.channel
                    .send_tool_start(ToolStartEvent {
                        tool_name: tc.name.clone(),
                        tool_call_id: tool_call_ids[idx].clone(),
                        params: Some(tc.input.clone()),
                        parent_tool_use_id: self.services.session.parent_tool_use_id.clone(),
                        started_at: tool_started_ats[idx],
                        speculative: true,
                        sandbox_profile: None,
                        is_mcp: mcp_tool_ids.contains(tc.name.as_str()),
                    })
                    .await?;
            }
        }

        Ok(commits)
    }

    /// Handles `UtilityAction::Retrieve` — either mandates a context-retrieval-then-retry
    /// round, or lets the originally requested tool proceed directly instead of demanding
    /// another doomed retry, when either:
    /// - a prior retrieval attempt this turn already failed in a way that makes
    ///   `memory_search` unusable — a retryable/network-class error (e.g. Qdrant
    ///   unreachable, #5584) or `ConfirmationRequired` (e.g. the query content itself trips
    ///   a sanitizer check, #5774); or
    /// - the gate has already issued `MAX_RETRIEVE_MANDATES_PER_TURN` mandates this turn,
    ///   as a defense-in-depth circuit breaker against any other variant of the same stall.
    async fn handle_retrieve_action(
        &mut self,
        idx: usize,
        tc: &zeph_llm::provider::ToolUseRequest,
        call: &ToolCall,
        pending_system_hints: &mut Vec<String>,
    ) -> Result<Option<(usize, ToolExecFut)>, crate::agent::error::AgentError> {
        if self.tool_orchestrator.has_blocked_retrieval_this_turn() {
            self.channel
                .send_status_best_effort(&format!(
                    "Utility action: Retrieve skipped, context unavailable ({})",
                    tc.name
                ))
                .await;
            pending_system_hints.push(format!(
                "[utility:retrieve] Context retrieval failed and appears unavailable. \
                 Proceed with the '{}' tool call using the best information already \
                 available rather than retrying the failed retrieval.",
                tc.name
            ));
            return Ok(None);
        }
        if self.tool_orchestrator.retrieve_mandate_limit_reached() {
            tracing::warn!(
                tool = %tc.name,
                mandates = self.tool_orchestrator.retrieve_mandate_count,
                "utility gate: Retrieve mandate limit reached this turn, proceeding directly"
            );
            self.channel
                .send_status_best_effort(&format!(
                    "Utility action: Retrieve loop detected, proceeding directly ({})",
                    tc.name
                ))
                .await;
            pending_system_hints.push(format!(
                "[utility:retrieve] Repeated context-retrieval cycle detected this turn. \
                 Proceed with the '{}' tool call directly without further retrieval to \
                 avoid a stalled loop.",
                tc.name
            ));
            return Ok(None);
        }
        self.channel
            .send_status_best_effort(&format!("Utility action: Retrieve ({})", tc.name))
            .await;
        // Inject a system message directing the LLM to retrieve context first (#2620).
        pending_system_hints.push(format!(
            "[utility:retrieve] Before executing the '{}' tool, retrieve \
             relevant context via memory_search or a related lookup to ensure \
             the call is well-targeted. After retrieving context, you MUST call \
             the '{}' tool again with the same arguments.",
            tc.name, tc.name
        ));
        // The hint above mandates a retry of this exact call — mark it so the utility gate
        // does not re-veto the compliant retry as a redundant duplicate (#5719).
        self.tool_orchestrator
            .utility_scorer
            .mark_mandated_retry(call);
        self.tool_orchestrator.record_retrieve_mandate();
        Ok(Some(ready_fut(
            idx,
            skipped_output(
                tc.name.clone(),
                format!(
                    "[skipped] Tool call to {} skipped — utility policy recommends \
                     retrieving additional context first.",
                    tc.name
                ),
            ),
        )))
    }

    async fn handle_utility_gate(
        &mut self,
        idx: usize,
        tc: &zeph_llm::provider::ToolUseRequest,
        call: &ToolCall,
        utility_actions: &[zeph_tools::UtilityAction],
        pending_system_hints: &mut Vec<String>,
    ) -> Result<Option<(usize, ToolExecFut)>, crate::agent::error::AgentError> {
        match utility_actions[idx] {
            zeph_tools::UtilityAction::Respond => {
                self.channel
                    .send_status_best_effort(&format!("Utility action: Respond ({})", tc.name))
                    .await;
                Ok(Some(ready_fut(
                    idx,
                    skipped_output(
                        tc.name.clone(),
                        format!(
                            "[skipped] Tool call to {} skipped — utility policy recommends a \
                             direct response without further tool use.",
                            tc.name
                        ),
                    ),
                )))
            }
            zeph_tools::UtilityAction::Retrieve => {
                self.handle_retrieve_action(idx, tc, call, pending_system_hints)
                    .await
            }
            zeph_tools::UtilityAction::Verify => {
                self.channel
                    .send_status_best_effort(&format!("Utility action: Verify ({})", tc.name))
                    .await;
                pending_system_hints.push(format!(
                    "[utility:verify] Before executing the '{}' tool again, verify \
                     the result of the previous tool call to confirm it is correct \
                     and that further tool use is necessary.",
                    tc.name
                ));
                Ok(Some(ready_fut(
                    idx,
                    skipped_output(
                        tc.name.clone(),
                        format!(
                            "[skipped] Tool call to {} skipped — utility policy recommends \
                             verifying the previous result first.",
                            tc.name
                        ),
                    ),
                )))
            }
            zeph_tools::UtilityAction::Stop => {
                self.channel
                    .send_status_best_effort(&format!("Utility action: Stop ({})", tc.name))
                    .await;
                let threshold = self.tool_orchestrator.utility_scorer.threshold();
                Ok(Some(ready_fut(
                    idx,
                    skipped_output(
                        tc.name.clone(),
                        format!(
                            "[stopped] Tool call to {} halted by the utility gate — \
                             budget exhausted or score below threshold {threshold:.2}.",
                            tc.name
                        ),
                    ),
                )))
            }
            _ => {
                // Reached only when this call is not being intercepted by Respond/Retrieve/
                // Verify/Stop — i.e. it proceeds to real dispatch. Critic-flagged S3
                // regression: `memory_search` itself always reaches this arm (gain 0.8 takes
                // the direct ToolCall branch), so resetting unconditionally here made the
                // breaker reset every single cycle of the very "retrieve, dispatch, decline"
                // loop it exists to catch — the repeated-decline case never trips it. Only a
                // *different* tool reaching real dispatch is genuine forward progress on the
                // user's actual request; the retrieval detour's own dispatch is part of the
                // stall pattern, not progress past it (#5774 S3).
                if tc.name.as_str() != "memory_search" {
                    self.tool_orchestrator.reset_retrieve_mandate_count();
                }
                Ok(None)
            }
        }
    }

    async fn run_before_tool_hooks(
        &mut self,
        idx: usize,
        tc: &zeph_llm::provider::ToolUseRequest,
        call: &ToolCall,
    ) -> Option<(usize, ToolExecFut)> {
        if self.runtime.config.layers.is_empty() {
            return None;
        }
        let conv_id_str = self
            .services
            .memory
            .persistence
            .conversation_id
            .map(|id| id.0.to_string());
        let ctx = crate::runtime_layer::LayerContext {
            conversation_id: conv_id_str.as_deref(),
            turn_number: u32::try_from(self.services.sidequest.turn_counter).unwrap_or(u32::MAX),
        };
        let mut sc_result: crate::runtime_layer::BeforeToolResult = None;
        for layer in &self.runtime.config.layers {
            let hook_result = std::panic::AssertUnwindSafe(layer.before_tool(&ctx, call))
                .catch_unwind()
                .await;
            match hook_result {
                Ok(Some(r)) => {
                    sc_result = Some(r);
                    break;
                }
                Ok(None) => {}
                Err(_) => {
                    tracing::warn!("RuntimeLayer::before_tool panicked, continuing");
                }
            }
        }
        let r = sc_result?;
        // TODO: implement retry-on-{"retry":true} stdout signal (#3292)
        self.fire_permission_denied_hooks(tc, &r.reason).await;
        Some((idx, Box::pin(std::future::ready(r.result))))
    }

    #[allow(clippy::too_many_arguments)]
    async fn check_call_gates(
        &mut self,
        idx: usize,
        tc: &zeph_llm::provider::ToolUseRequest,
        call: &ToolCall,
        has_failed_dep: bool,
        quota_blocked: bool,
        pre_exec_blocked: &[bool],
        utility_actions: &[zeph_tools::UtilityAction],
        repeat_blocked: &[bool],
        pending_system_hints: &mut Vec<String>,
    ) -> Result<Option<((usize, ToolExecFut), String)>, crate::agent::error::AgentError> {
        if has_failed_dep {
            let reason = "prerequisite tool failed or requires confirmation".to_owned();
            return Ok(Some((
                ready_fut(
                    idx,
                    skipped_output(
                        tc.name.clone(),
                        "[error] Skipped: a prerequisite tool failed or requires confirmation",
                    ),
                ),
                reason,
            )));
        }
        if quota_blocked {
            let max = self
                .tool_orchestrator
                .max_tool_calls_per_session
                .unwrap_or(0);
            let reason = format!("session tool call quota exceeded (limit: {max} calls)");
            return Ok(Some((
                ready_fut(
                    idx,
                    skipped_output(
                        tc.name.clone(),
                        format!(
                            "[error] Tool call quota exceeded (session limit: {max} calls). \
                             No further tool calls are allowed this session."
                        ),
                    ),
                ),
                reason,
            )));
        }
        if pre_exec_blocked[idx] {
            let reason = format!(
                "blocked by pre-execution verifier: {} is not permitted",
                tc.name
            );
            return Ok(Some((
                ready_fut(
                    idx,
                    skipped_output(
                        tc.name.clone(),
                        format!(
                            "[error] Tool call to {} was blocked by pre-execution verifier. \
                             The requested operation is not permitted.",
                            tc.name
                        ),
                    ),
                ),
                reason,
            )));
        }
        if let Some(fut) = self
            .handle_utility_gate(idx, tc, call, utility_actions, pending_system_hints)
            .await?
        {
            let reason = format!(
                "utility gate ({:?}) intercepted {}",
                utility_actions[idx], tc.name
            );
            return Ok(Some((fut, reason)));
        }
        if repeat_blocked[idx] {
            let reason = format!("repeated identical call to {} detected", tc.name);
            return Ok(Some((
                ready_fut(
                    idx,
                    skipped_output(
                        tc.name.clone(),
                        format!(
                            "[error] Repeated identical call to {} detected. \
                             Use different arguments or a different approach.",
                            tc.name
                        ),
                    ),
                ),
                reason,
            )));
        }
        Ok(None)
    }

    #[tracing::instrument(
        name = "core.tool.build_tier_call_futures",
        skip_all,
        level = "debug",
        fields(tier_size = tier_indices.len()),
        err
    )]
    #[allow(clippy::too_many_arguments, clippy::ptr_arg, clippy::too_many_lines)]
    async fn build_tier_call_futures(
        &mut self,
        tool_calls: &[zeph_llm::provider::ToolUseRequest],
        calls: &[ToolCall],
        tier_indices: &[usize],
        dag: &super::tool_call_dag::ToolCallDag,
        failed_ids: &std::collections::HashSet<String>,
        quota_blocked: bool,
        pre_exec_blocked: &[bool],
        utility_actions: &[zeph_tools::UtilityAction],
        repeat_blocked: &[bool],
        cache_hits: &[Option<zeph_tools::ToolOutput>],
        semaphore: &std::sync::Arc<tokio::sync::Semaphore>,
        pending_system_hints: &mut Vec<String>,
    ) -> Result<Vec<(usize, ToolExecFut)>, crate::agent::error::AgentError> {
        let tier_tool_names: Vec<&str> = tier_indices
            .iter()
            .map(|&i| tool_calls[i].name.as_str())
            .collect();
        let rate_results = self
            .runtime
            .config
            .rate_limiter
            .check_batch(&tier_tool_names);

        // Phase 1: fire PreToolUse hooks for every call in the tier concurrently, bounded by the
        // same tier semaphore used for tool execution below — mirrors apply_tier_results Phase 2
        // (#6128), which already parallelized the PostToolUse/RuntimeLayer side of this same
        // per-index hook-firing pattern. There is no ordering constraint between different
        // tool-call indices' hook firing; the only required invariant — a call's own gate checks
        // must observe that same call's own hook result — holds because this whole phase
        // completes before Phase 2's per-index gate checks below begin.
        //
        // Focus/compress tools are synthetic internal calls that never reach hooks or gates (see
        // the matching `continue` in Phase 2 below), so they are excluded here too.
        let pre_hooks = self.services.session.hooks_config.pre_tool_use.clone();
        let mut pre_hook_blocked: std::collections::HashMap<usize, String> =
            std::collections::HashMap::new();
        if !pre_hooks.is_empty() {
            let conv_id_str = self
                .services
                .memory
                .persistence
                .conversation_id
                .map(|id| id.0.to_string());
            let dispatch = self.mcp_dispatch();
            let mcp: Option<&dyn zeph_subagent::McpDispatch> = dispatch
                .as_ref()
                .map(|d| d as &dyn zeph_subagent::McpDispatch);

            let futs = tier_indices.iter().filter_map(|&idx| {
                let tc = &tool_calls[idx];
                if tc.name == "compress_context"
                    || tc.name == "request_compaction"
                    || (self.services.focus.config.enabled
                        && (tc.name == "start_focus" || tc.name == "complete_focus"))
                {
                    return None;
                }
                let matched: Vec<&zeph_config::HookDef> =
                    zeph_subagent::matching_hooks(&pre_hooks, tc.name.as_str());
                if matched.is_empty() {
                    return None;
                }
                let has_fail_closed = matched.iter().any(|h| h.fail_closed);
                let owned: Vec<zeph_config::HookDef> = matched.into_iter().cloned().collect();
                let env = make_tool_hook_env(tc.name.as_str(), &tc.input, conv_id_str.as_deref());
                let sem = std::sync::Arc::clone(semaphore);
                let tool_name = tc.name.clone();
                Some(async move {
                    let Ok(_permit) = sem.acquire().await else {
                        tracing::warn!(
                            tool = %tool_name,
                            "semaphore closed during pre-tool hook firing, skipping \
                             PreToolUse hook for this call"
                        );
                        return (idx, None);
                    };
                    let result = zeph_subagent::hooks::fire_hooks(&owned, &env, mcp, None)
                        .instrument(tracing::info_span!(
                            "core.hooks.pre_tool_use",
                            tool = %tool_name
                        ))
                        .await;
                    (idx, Some(result.map_err(|e| (e, has_fail_closed))))
                })
            });

            for (idx, outcome) in futures::future::join_all(futs).await {
                let Some(Err((e, has_fail_closed))) = outcome else {
                    continue;
                };
                let tool_name = tool_calls[idx].name.as_str();
                if has_fail_closed {
                    self.tool_orchestrator.hook_block_count += 1;
                    tracing::warn!(
                        error = %e,
                        tool = %tool_name,
                        hook_block_count = self.tool_orchestrator.hook_block_count,
                        hook_block_cap = self.tool_orchestrator.hook_block_cap,
                        "PreToolUse hook blocked tool (fail_closed)"
                    );
                    pre_hook_blocked.insert(
                        idx,
                        format!("[blocked] PreToolUse hook blocked tool `{tool_name}`: {e}"),
                    );
                } else {
                    tracing::warn!(error = %e, tool = %tool_name, "PreToolUse hook failed");
                }
            }
        }

        // Phase 2: per-index gate checks, cache lookups, and execution-future construction.
        // Stays sequential — it needs `&mut self` throughout, and each idx's control flow
        // depends on that same idx's own PreToolUse hook outcome computed in Phase 1 above.
        let mut tier_futs: Vec<(usize, ToolExecFut)> = Vec::with_capacity(tier_indices.len());
        for (tier_local_idx, &idx) in tier_indices.iter().enumerate() {
            let tc = &tool_calls[idx];
            let call = &calls[idx];

            // Skip focus tools, compress_context, and request_compaction — pre-handled before the tier loop.
            if tc.name == "compress_context"
                || tc.name == "request_compaction"
                || (self.services.focus.config.enabled
                    && (tc.name == "start_focus" || tc.name == "complete_focus"))
            {
                continue;
            }

            if let Some(msg) = pre_hook_blocked.remove(&idx) {
                tier_futs.push((
                    idx,
                    Box::pin(std::future::ready(Ok(Some(skipped_output(
                        tc.name.clone(),
                        msg,
                    ))))),
                ));
                continue;
            }

            // Check static gates: dep failure, quota, pre-exec block, utility gate, repeat.
            let has_failed_dep = dag
                .string_values_for(idx)
                .iter()
                .any(|v| failed_ids.contains(v));
            if let Some((fut, reason)) = self
                .check_call_gates(
                    idx,
                    tc,
                    call,
                    has_failed_dep,
                    quota_blocked,
                    pre_exec_blocked,
                    utility_actions,
                    repeat_blocked,
                    pending_system_hints,
                )
                .await?
            {
                self.fire_permission_denied_hooks(tc, &reason).await;
                tier_futs.push(fut);
                continue;
            }

            // Cache hit: return pre-computed result without executing the tool.
            if let Some(cached_output) = cache_hits[idx].clone() {
                tracing::debug!(
                    tool = %tc.name,
                    "[tool-cache] returning cached result, skipping execution"
                );
                tier_futs.push((idx, Box::pin(std::future::ready(Ok(Some(cached_output))))));
                continue;
            }

            // Rate limiter: check the pre-computed batch result for this call.
            if let Some(ref exceeded) = rate_results[tier_local_idx] {
                tracing::warn!(
                    tool = %tc.name,
                    category = exceeded.category.as_str(),
                    limit = exceeded.limit,
                    "tool rate limiter: blocking call"
                );
                self.update_metrics(|m| m.rate_limit_trips += 1);
                self.push_security_event(
                    zeph_common::SecurityEventCategory::RateLimit,
                    tc.name.as_str(),
                    format!(
                        "{} calls exceeded {}/min",
                        exceeded.category.as_str(),
                        exceeded.limit
                    ),
                );
                self.fire_permission_denied_hooks(tc, &exceeded.to_error_message())
                    .await;
                tier_futs.push(ready_fut(
                    idx,
                    skipped_output(tc.name.clone(), exceeded.to_error_message()),
                ));
                continue;
            }

            if let Some(fut) = self.run_before_tool_hooks(idx, tc, call).await {
                tier_futs.push(fut);
                continue;
            }

            // Speculative try_commit (#3641): reuse a pre-executed result when available.
            // Uses the LLM-assigned `tool_use_id` (tc.id) for result routing (critic H3).
            // TODO(#3645): add circuit-breaker check when implemented.
            if let Some(engine) = self.services.speculation_engine.as_ref()
                && let Some(result) =
                    crate::agent::speculative::stream_drainer::try_commit_with_timeout(engine, call)
                        .await
            {
                tracing::debug!(tool = %tc.name, llm_id = %tc.id, "speculative try_commit hit");
                tier_futs.push((idx, Box::pin(std::future::ready(result))));
                continue;
            }

            tier_futs.push(self.make_exec_future(idx, tc, call, semaphore));
        }
        Ok(tier_futs)
    }

    fn make_exec_future(
        &self,
        idx: usize,
        tc: &zeph_llm::provider::ToolUseRequest,
        call: &ToolCall,
        semaphore: &std::sync::Arc<tokio::sync::Semaphore>,
    ) -> (usize, ToolExecFut) {
        let sem = std::sync::Arc::clone(semaphore);
        let executor = std::sync::Arc::clone(&self.tool_executor);
        let call = call.clone();
        let tool_name = tc.name.clone();
        let tool_id = tc.id.clone();
        let fut = async move {
            let _permit = sem.acquire().await.map_err(|_| {
                zeph_tools::ToolError::Execution(std::io::Error::other(
                    "semaphore closed during tool execution",
                ))
            })?;
            executor
                .execute_tool_call_erased(&call)
                .instrument(tracing::info_span!(
                    "tool_exec",
                    tool_name = %tool_name,
                    idx = %tool_id
                ))
                .await
        };
        (idx, Box::pin(fut))
    }

    #[tracing::instrument(name = "core.tool.execute_tier_join", skip_all, level = "debug", err)]
    #[allow(clippy::type_complexity)]
    async fn execute_tier_join(
        &mut self,
        futs: Vec<ToolExecFut>,
        cancel: &tokio_util::sync::CancellationToken,
        tool_calls: &[zeph_llm::provider::ToolUseRequest],
    ) -> Result<
        Option<Vec<Result<Option<zeph_tools::ToolOutput>, zeph_tools::ToolError>>>,
        crate::agent::error::AgentError,
    > {
        let mut join_fut = std::pin::pin!(futures::future::join_all(futs));
        // Take elicitation_rx out of self so we can hold &mut self for handling.
        let mut elicitation_rx = self.services.mcp.elicitation_rx.take();
        let result = loop {
            tokio::select! {
                results = &mut join_fut => break results,
                () = cancel.cancelled() => {
                    self.services.mcp.elicitation_rx = elicitation_rx;
                    self.cancel_tool_batch(tool_calls, "tool execution cancelled by user")
                        .await?;
                    return Ok(None);
                }
                event = recv_elicitation(&mut elicitation_rx) => {
                    if let Some(ev) = event {
                        self.handle_elicitation_event(ev).await;
                    } else {
                        tracing::debug!("elicitation channel closed during tier exec");
                        elicitation_rx = None;
                    }
                }
            }
        };
        self.services.mcp.elicitation_rx = elicitation_rx;
        Ok(Some(result))
    }

    #[allow(clippy::too_many_arguments, clippy::too_many_lines)]
    async fn apply_tier_results(
        &mut self,
        indices: Vec<usize>,
        tier_results: Vec<Result<Option<zeph_tools::ToolOutput>, zeph_tools::ToolError>>,
        tool_calls: &[zeph_llm::provider::ToolUseRequest],
        calls: &[ToolCall],
        cache_hits: &[Option<zeph_tools::ToolOutput>],
        mcp_tool_ids: &std::collections::HashSet<String>,
        args_hashes: &[u64],
        tool_started_ats: &[std::time::Instant],
        max_parallel: usize,
        failed_ids: &mut std::collections::HashSet<String>,
        tool_results: &mut [Result<Option<zeph_tools::ToolOutput>, zeph_tools::ToolError>],
    ) {
        // Phase 1: sequential bookkeeping (dependency-graph, result cache, failed_ids). Cheap,
        // synchronous, and needs `&mut self` — kept sequential rather than folded into phase 2.
        let mut pending: Vec<(
            usize,
            Result<Option<zeph_tools::ToolOutput>, zeph_tools::ToolError>,
        )> = Vec::with_capacity(indices.len());
        for (idx, result) in indices.into_iter().zip(tier_results) {
            // IMP-02: Err(_) covers all error variants including ConfirmationRequired.
            // Ok(Some(out)) with "[error]" prefix covers synthetic/blocked results.
            let is_failed = match &result {
                Err(_) => true,
                Ok(Some(out)) => out.summary.starts_with("[error]"),
                Ok(None) => false,
            };
            if is_failed {
                failed_ids.insert(tool_calls[idx].id.clone());
            }

            // Store successful, non-cached results in the tool result cache. Utility-gate
            // synthetic outputs ("[skipped]"/"[stopped]") are never real execution results and
            // must not be cached: caching one would make a later mandated retry (#5719) replay
            // the stale skip text under the same args hash instead of actually executing the
            // tool, silently reintroducing the exact stall the mandated-retry bypass fixes.
            if !is_failed
                && cache_hits[idx].is_none()
                && zeph_tools::is_cacheable(
                    tool_calls[idx].name.as_str(),
                    mcp_tool_ids.contains(tool_calls[idx].name.as_str()),
                )
                && let Ok(Some(ref out)) = result
                && !out.summary.starts_with("[skipped]")
                && !out.summary.starts_with("[stopped]")
            {
                let key =
                    zeph_tools::CacheKey::new(tool_calls[idx].name.to_string(), args_hashes[idx]);
                self.tool_orchestrator.result_cache.put(key, out.clone());
            }

            // Record successful tool completions for the dependency graph (#2024).
            if !is_failed && self.services.tool_state.dependency_graph.is_some() {
                self.services
                    .tool_state
                    .completed_tool_ids
                    .insert(tool_calls[idx].name.to_string());
            }

            pending.push((idx, result));
        }

        let layers_empty = self.runtime.config.layers.is_empty();
        let post_hooks = self.services.session.hooks_config.post_tool_use.clone();
        if layers_empty && post_hooks.is_empty() {
            for (idx, result) in pending {
                tool_results[idx] = result;
            }
            return;
        }

        // Phase 2: RuntimeLayer::after_tool and PostToolUse hook firing per index. Unlike
        // process_one_tool_result (which must stay strictly sequential to preserve
        // OpenAI/Claude message-ordering — see the comment above its call site), there is no
        // ordering constraint between different tool-result indices here: each index's layer
        // chain and hook firing only observes/mutates that index's own result. Run them
        // concurrently, bounded by the same `max_parallel` semaphore the tier's tool execution
        // itself already uses, so a tier with many hook-matching calls doesn't serialize N
        // subprocess spawns after already paying for bounded-parallel tool execution (#6128).
        let conv_id_str = self
            .services
            .memory
            .persistence
            .conversation_id
            .map(|id| id.0.to_string());
        let ctx = crate::runtime_layer::LayerContext {
            conversation_id: conv_id_str.as_deref(),
            turn_number: u32::try_from(self.services.sidequest.turn_counter).unwrap_or(u32::MAX),
        };
        let dispatch = self.mcp_dispatch();
        let mcp: Option<&dyn zeph_subagent::McpDispatch> = dispatch
            .as_ref()
            .map(|d| d as &dyn zeph_subagent::McpDispatch);
        let semaphore = std::sync::Arc::new(tokio::sync::Semaphore::new(max_parallel.max(1)));
        let layers = &self.runtime.config.layers;
        // Reborrow as plain references (`&T` is `Copy`) so the `FnMut` closure below can
        // capture them by value on every invocation instead of moving the owned originals.
        let ctx_ref = &ctx;
        let conv_id = conv_id_str.as_deref();

        let futs = pending.into_iter().map(|(idx, result)| {
            let sem = std::sync::Arc::clone(&semaphore);
            let tc = &tool_calls[idx];
            let call = &calls[idx];
            let matched: Vec<&zeph_config::HookDef> =
                zeph_subagent::matching_hooks(&post_hooks, tc.name.as_str());
            async move {
                let Ok(_permit) = sem.acquire().await else {
                    tracing::warn!(
                        tool = %tc.name,
                        "semaphore closed during post-tool hook firing, skipping \
                         RuntimeLayer::after_tool/PostToolUse hooks for this result"
                    );
                    return (idx, result);
                };
                let mut result = result;

                // RuntimeLayer after_tool hooks.
                for layer in layers {
                    let hook_result =
                        std::panic::AssertUnwindSafe(layer.after_tool(ctx_ref, call, &result))
                            .catch_unwind()
                            .await;
                    if hook_result.is_err() {
                        tracing::warn!("RuntimeLayer::after_tool panicked, continuing");
                    }
                }

                // Fire PostToolUse hooks after the tool result is available (fail-open).
                if !matched.is_empty() {
                    let mut env = make_tool_hook_env(tc.name.as_str(), &tc.input, conv_id);
                    let duration_ms = u64::try_from(tool_started_ats[idx].elapsed().as_millis())
                        .unwrap_or(u64::MAX);
                    env.insert("ZEPH_TOOL_DURATION_MS".to_owned(), duration_ms.to_string());
                    let owned: Vec<zeph_config::HookDef> = matched.into_iter().cloned().collect();

                    // Build stdin JSON context for the hook process.
                    let tool_output_text = match &result {
                        Ok(Some(out)) => Some(out.summary.as_str()),
                        _ => None,
                    };
                    let tool_error_text = match &result {
                        Err(e) => Some(e.to_string()),
                        _ => None,
                    };
                    let hook_input = zeph_subagent::PostToolUseHookInput {
                        tool_name: tc.name.as_str(),
                        tool_args: &tc.input,
                        session_id: conv_id,
                        duration_ms,
                        tool_output: tool_output_text,
                        tool_error: tool_error_text.as_deref(),
                        agent_id: conv_id,
                        agent_type: "main",
                    };
                    let stdin_bytes = serde_json::to_vec(&hook_input).ok();

                    match zeph_subagent::hooks::fire_hooks(
                        &owned,
                        &env,
                        mcp,
                        stdin_bytes.as_deref(),
                    )
                    .instrument(tracing::info_span!(
                        "core.hooks.post_tool_use",
                        tool = %tc.name
                    ))
                    .await
                    {
                        Ok(run_result) => {
                            if let Some(replacement) = run_result.output.updated_tool_output {
                                // Apply hook-requested output substitution.
                                if let Ok(Some(ref mut out)) = result {
                                    tracing::debug!(
                                        tool = %tc.name,
                                        original_len = out.summary.len(),
                                        replacement_len = replacement.len(),
                                        "PostToolUse hook replaced tool output"
                                    );
                                    out.summary = replacement;
                                }
                            }
                        }
                        Err(e) => {
                            tracing::warn!(
                                error = %e,
                                tool = %tc.name,
                                "PostToolUse hook failed"
                            );
                        }
                    }
                }

                (idx, result)
            }
        });

        for (idx, result) in futures::future::join_all(futs).await {
            tool_results[idx] = result;
        }
    }

    #[tracing::instrument(
        name = "core.tool.process_tool_result_batch",
        skip_all,
        level = "debug",
        fields(batch_size = tool_calls.len()),
        err
    )]
    #[allow(clippy::too_many_arguments, clippy::too_many_lines)]
    async fn process_tool_result_batch(
        &mut self,
        tool_calls: &[zeph_llm::provider::ToolUseRequest],
        tool_call_ids: &[String],
        tool_started_ats: &[std::time::Instant],
        mut tool_results: Vec<Result<Option<zeph_tools::ToolOutput>, zeph_tools::ToolError>>,
        causal_pre_response: Option<(String, String)>,
        pending_focus_checkpoint: Option<zeph_llm::provider::Message>,
        pending_system_hints: Vec<String>,
    ) -> Result<(), crate::agent::error::AgentError> {
        self.tool_executor.set_skill_env(None);

        // Sync cache counters to metrics after all tool execution is complete.
        {
            let hits = self.tool_orchestrator.result_cache.hits();
            let misses = self.tool_orchestrator.result_cache.misses();
            let entries = self.tool_orchestrator.result_cache.len();
            self.update_metrics(|m| {
                m.tool_cache_hits = hits;
                m.tool_cache_misses = misses;
                m.tool_cache_entries = entries;
            });
        }

        // Collect (name, params, output) for LSP hooks. Built during the results loop below.
        let mut lsp_tool_calls: Vec<(String, serde_json::Value, String)> = Vec::new();

        // Process results sequentially (metrics, channel sends, message parts).
        // self_reflection is deferred until after all result_parts are assembled and user_msg
        // is pushed to history. Calling it inside the loop would insert a reflection dialogue
        // (User{prompt} + Assistant{response}) between Assistant{ToolUse} and User{ToolResults},
        // violating the OpenAI/Claude API message ordering protocol → HTTP 400.
        let mut result_parts: Vec<MessagePart> = Vec::new();
        // Accumulates injection flags across all tools in the batch (Bug #1490 fix).
        let mut has_any_injection_flags = false;
        // Deferred self-reflection: set to the sanitized error output of the first failing tool
        // that is eligible for reflection. Consumed after user_msg is pushed to history.
        let mut pending_reflection: Option<String> = None;
        // Accumulate skill outcomes during the tool loop; flushed once after the loop via
        // flush_skill_outcomes to avoid N×M×13 sequential SQLite awaits (#2770).
        let mut pending_outcomes: Vec<crate::agent::learning::PendingSkillOutcome> = Vec::new();
        // Running per-turn counter for attached MCP-sourced images (spec-072 §3.4
        // max_images_per_turn), aggregated across every tool call in this batch.
        let mut images_attached_this_turn: usize = 0;
        // Batch-level worst-case content-trust tier (issue #6490, MemGhost): the whole batch
        // is persisted as one message, so its write-time provenance tag reflects the
        // least-trusted tool call in the batch — same OR-aggregation shape as
        // has_any_injection_flags/flagged_urls below. `batch_source_kind` is paired with it
        // (issue #6556): it tracks the actual `ContentSourceKind` of the last tool call
        // observed at or above the running trust maximum (`>=` tie-break, tool_result.rs),
        // instead of being re-derived from the trust tier alone.
        let mut batch_trust_level = zeph_sanitizer::ContentTrustLevel::Trusted;
        let mut batch_source_kind = zeph_sanitizer::ContentSourceKind::ToolResult;
        for idx in 0..tool_calls.len() {
            let tc = &tool_calls[idx];
            let tool_call_id = &tool_call_ids[idx];
            let started_at = &tool_started_ats[idx];
            let tool_result = std::mem::replace(&mut tool_results[idx], Ok(None));
            self.process_one_tool_result(
                tc,
                tool_call_id,
                started_at,
                tool_result,
                &mut result_parts,
                &mut lsp_tool_calls,
                &mut has_any_injection_flags,
                &mut pending_reflection,
                &mut pending_outcomes,
                &mut images_attached_this_turn,
                &mut batch_trust_level,
                &mut batch_source_kind,
            )
            .await?;
        }

        // Flush all accumulated skill outcomes from the tool batch in a single pass.
        // This replaces the per-tool record_skill_outcomes calls that caused N×M sequential
        // SQLite awaits (#2770).
        self.flush_skill_outcomes(pending_outcomes).await;

        // Extract goal_summary from causal pre-probe response before it is consumed.
        // Used by shadow memory below; empty when causal IPI is disabled (deviation_score = 0.0).
        let goal_summary_for_shadow: String = causal_pre_response
            .as_ref()
            .map(|(pre, _)| {
                let end = pre.floor_char_boundary(100);
                pre[..end].to_owned()
            })
            .unwrap_or_default();

        self.run_causal_ipi_post_probe(causal_pre_response, &result_parts)
            .await;

        // Spec 010-7 FR-001–FR-004: record shadow event and check goal drift.
        self.record_shadow_event(tool_calls, goal_summary_for_shadow);

        // Acon: compress tool results before they enter message history (#4021).
        self.apply_acon_compression(tool_calls, &mut result_parts);

        let mut user_msg = Message::from_parts(Role::User, result_parts);
        // Issue #6558: tag the in-memory message with its write-time provenance so
        // `Agent::context_max_trust_level` can still recognize this batch's untrusted content
        // as long as the message remains in `self.msg.messages` — including across a user-turn
        // boundary, unlike the turn-scoped slot reset alone. Mirrors the SQLite-side provenance
        // tagging below, at the same per-batch granularity.
        user_msg.metadata.trust_level = Some(batch_trust_level as u8);
        // flagged_urls accumulates across ALL tools in this batch (cross-tool trust boundary).
        // A URL from tool N's output can flag tool M's arguments even if tool M returned clean
        // output. has_any_injection_flags covers pure text injections (no URL); flagged_urls
        // covers URL-based exfiltration. Both are OR-combined for conservative guarding.
        // Individual per-tool granularity would require separate persist_message calls per
        // result, which would change message history structure.
        let tool_results_have_flags =
            has_any_injection_flags || !self.services.security.flagged_urls.is_empty();
        tracing::debug!("tool_batch: calling persist_message for tool results");
        // Issue #6490 (MemGhost): tag the batch with its write-time provenance. `source_kind`
        // is a coarse per-batch label (multiple tool calls with different origins can share one
        // persisted message) — `batch_source_kind` (issue #6556) tracks the actual
        // `ContentSourceKind` of the last tool call observed at or above the batch's running
        // trust maximum, rather than being re-derived from `batch_trust_level` alone (which
        // previously collapsed every `ExternalUntrusted` batch — MCP responses, memory-recall
        // replays, and web-scrapes alike — into `WebScrape`). Trusted batches use the default
        // trusted `persist_message`.
        if batch_trust_level == zeph_sanitizer::ContentTrustLevel::Trusted {
            self.persist_message(
                Role::User,
                &user_msg.content,
                &user_msg.parts,
                tool_results_have_flags,
            )
            .await;
        } else {
            self.persist_message_with_provenance(
                Role::User,
                &user_msg.content,
                &user_msg.parts,
                tool_results_have_flags,
                batch_source_kind,
                batch_trust_level,
            )
            .await;
        }
        tracing::debug!("tool_batch: persist_message done, pushing message");
        self.push_message(user_msg);
        tracing::debug!("tool_batch: message pushed, starting LSP hooks");
        if let (Some(id), Some(last)) = (
            self.msg.last_persisted_message_id,
            self.msg.messages.last_mut(),
        ) {
            last.metadata.db_id = Some(id);
        }

        // Flush deferred start_focus checkpoint AFTER User(tool_results) so the ordering
        // Assistant→User→System is valid for OpenAI (#3262).
        if let Some(checkpoint) = pending_focus_checkpoint {
            self.push_message(checkpoint);
        }

        // Flush deferred utility gate hints (Retrieve/Verify). Pushed after User(tool_results)
        // so the ordering Assistant→User→System is valid for OpenAI (#2615).
        for hint in pending_system_hints {
            self.push_message(zeph_llm::provider::Message::from_legacy(
                zeph_llm::provider::Role::System,
                &hint,
            ));
        }

        // Deferred self-reflection: user_msg is now in history so the reflection dialogue
        // (User{prompt} + Assistant{response}) appends after User{ToolResults}, preserving
        // API message ordering. Only the first eligible error per batch triggers reflection.
        if let Some(sanitized_out) = pending_reflection {
            match self
                .attempt_self_reflection(&sanitized_out, &sanitized_out)
                .await
            {
                Ok(_) | Err(_) => {
                    // Whether reflection succeeded, declined, or errored: the ToolResults are
                    // already committed to history. Return Ok regardless so the caller continues
                    // the tool loop normally (#2197).
                }
            }
        }

        // Fire LSP hooks for each completed tool call (non-blocking: diagnostics fetch
        // is spawned in background; hover calls are awaited but short-lived).
        // `lsp_tool_calls` collects (name, params, output) tuples built during the
        // results loop above. They are captured into a separate Vec so we can call
        // `&mut self.services.session.lsp_hooks` without conflicting borrows.
        //
        // The entire batch is capped at 30s to prevent stalls when many files are
        // modified in one tool batch (#2750). Per the critic review, a single outer
        // timeout is more effective than per-call timeouts because it bounds total
        // blocking time regardless of N.
        if self.services.session.lsp_hooks.is_some() {
            let tc_arc = std::sync::Arc::clone(&self.runtime.metrics.token_counter);
            let sanitizer = self.services.security.sanitizer.clone();
            self.channel
                .send_status_best_effort("Analyzing changes...")
                .await;
            // TODO: cooperative MCP cancellation — dropped futures here may leave
            // in-flight MCP JSON-RPC requests pending until the server-side timeout.
            let lsp_result = tokio::time::timeout(std::time::Duration::from_secs(30), async {
                for (name, input, output) in lsp_tool_calls {
                    if let Some(ref mut lsp) = self.services.session.lsp_hooks {
                        lsp.after_tool(
                            &name,
                            &input,
                            &output,
                            &tc_arc,
                            &sanitizer,
                            &mut self.runtime.lifecycle.supervisor,
                        )
                        .await;
                    }
                }
            })
            .await;
            self.channel.send_status_best_effort("").await;
            if lsp_result.is_err() {
                tracing::warn!("LSP after_tool batch timed out (30s)");
            }
            tracing::debug!("tool_batch: LSP hooks done");
        }

        // Defense-in-depth: check if process cwd changed during this tool batch.
        // Normally only changes via set_working_directory; this also catches any
        // future code path that calls set_current_dir.
        self.check_cwd_changed().await;

        Ok(())
    }
}

/// Recursively unmask PAAC secret placeholders in a JSON value's string leaves, in place.
///
/// Applied to tool-call parameters at the dispatch boundary (`prepare_tool_dispatch`) so a
/// model-emitted placeholder (e.g. one copied verbatim from a prior masked tool result into a
/// shell `code` or HTTP header argument) resolves to the real secret only at execution time —
/// never during LLM-facing context assembly. `unmask` is nonce-scoped and passthrough-on-miss,
/// so a placeholder the model could not have legitimately seen is left untouched (#5437).
///
/// Returns the number of string leaves that still contain a `<SECRET:` prefix after unmasking
/// (S1: unmask-miss telemetry). A non-zero count means the model failed to reproduce a
/// placeholder byte-for-byte (LLMs routinely normalize/space-break long opaque tokens) — the
/// tool call proceeds with the literal placeholder text (fail-safe: no leak, but the flow that
/// depended on the real secret will likely fail). Callers should log a `tracing::warn!` and
/// increment a metric so operators can detect this class of silent breakage.
fn unmask_json_value(
    value: &mut serde_json::Value,
    registry: &zeph_sanitizer::secret_mask::SecretMaskRegistry,
) -> usize {
    match value {
        serde_json::Value::String(s) => {
            let unmasked = registry.unmask(s);
            let still_masked = usize::from(unmasked.contains("<SECRET:"));
            if unmasked != *s {
                *s = unmasked;
            }
            still_masked
        }
        serde_json::Value::Array(arr) => {
            arr.iter_mut().map(|v| unmask_json_value(v, registry)).sum()
        }
        serde_json::Value::Object(map) => map
            .values_mut()
            .map(|v| unmask_json_value(v, registry))
            .sum(),
        serde_json::Value::Null | serde_json::Value::Bool(_) | serde_json::Value::Number(_) => 0,
    }
}

async fn recv_elicitation(
    rx: &mut Option<tokio::sync::mpsc::Receiver<zeph_mcp::ElicitationEvent>>,
) -> Option<zeph_mcp::ElicitationEvent> {
    match rx {
        Some(r) => r.recv().await,
        None => std::future::pending().await,
    }
}

// `pub(super)`: called from `focus_compression` after the `tool_execution` split.
pub(super) fn skipped_output(
    tool_name: impl Into<zeph_common::ToolName>,
    summary: impl Into<String>,
) -> zeph_tools::ToolOutput {
    zeph_tools::ToolOutput {
        tool_name: tool_name.into(),
        summary: summary.into(),
        blocks_executed: 0,
        filter_stats: None,
        diff: None,
        streamed: false,
        terminal_id: None,
        locations: None,
        raw_response: None,
        claim_source: None,
        ..Default::default()
    }
}

fn ready_fut(idx: usize, out: zeph_tools::ToolOutput) -> (usize, ToolExecFut) {
    (idx, Box::pin(std::future::ready(Ok(Some(out)))))
}

impl<C: Channel> Agent<C> {
    #[tracing::instrument(name = "core.tool.native_loop", skip_all, level = "debug", err)]
    pub(super) async fn process_response_native_tools(
        &mut self,
    ) -> Result<(), crate::agent::error::AgentError> {
        self.tool_orchestrator.clear_doom_history();
        self.tool_orchestrator.clear_recent_tool_calls();
        self.tool_orchestrator.clear_utility_state();
        self.tool_orchestrator.reset_hook_block_count();

        // `mut` required when context-compression is enabled to inject focus tool definitions.
        let tafc = &self.tool_orchestrator.tafc;
        let mut tool_defs: Vec<ToolDefinition> = Vec::new();

        // `[tools] enabled = false` (#6386): omit every tool definition so the LLM request
        // carries no tools at all and the model cannot attempt a tool call.
        if self.services.tool_state.tools_enabled {
            tool_defs = self
                .tool_executor
                .tool_definitions_erased()
                .iter()
                .map(|def| super::tool_def_to_definition_with_tafc(def, tafc))
                .collect();

            // Inject focus tool definitions when the feature is enabled and configured (#1850).
            if self.services.focus.config.enabled {
                tool_defs.extend(crate::agent::focus::focus_tool_definitions());
            }

            // Inject compress_context tool — always available when context-compression is enabled (#2218).
            tool_defs.push(crate::agent::focus::compress_context_tool_definition());

            // Inject request_compaction tool when ARC agent-initiated compaction is enabled (#4020).
            if self
                .services
                .memory
                .subsystems
                .arc_config
                .allow_agent_compaction
            {
                tool_defs.push(crate::agent::focus::request_compaction_tool_definition());
            }
        }

        // Pre-compute the full tool set for iterations 1+ before filtering.
        let all_tool_defs = tool_defs.clone();

        // Iteration 0: apply dynamic tool schema filter (#2020) if cached IDs are available.
        if let Some(ref filtered_ids) = self.services.tool_state.cached_filtered_tool_ids {
            tool_defs.retain(|d| filtered_ids.contains(d.name.as_str()));
            tracing::debug!(
                filtered = tool_defs.len(),
                total = all_tool_defs.len(),
                "tool schema filter: iteration 0 using filtered tool set"
            );
        }

        tracing::debug!(
            tool_count = tool_defs.len(),
            tools = ?tool_defs.iter().map(|t| &t.name).collect::<Vec<_>>(),
            "native tool_use: collected tool definitions"
        );

        let query_embedding = match self.check_response_cache().await? {
            CacheCheckResult::Hit(cached) => {
                self.persist_message(Role::Assistant, &cached, &[], false)
                    .await;
                self.push_message(Message::from_legacy(Role::Assistant, cached.as_str()));
                if cached.contains(zeph_llm::provider::MAX_TOKENS_TRUNCATION_MARKER) {
                    let _ = self.channel.send_stop_hint(StopHint::MaxTokens).await;
                }
                self.channel.flush_chunks().await?;
                return Ok(());
            }
            CacheCheckResult::Miss { query_embedding } => query_embedding,
        };

        for iteration in 0..self.tool_orchestrator.max_iterations {
            if *self.runtime.lifecycle.shutdown.borrow() {
                tracing::info!("native tool loop interrupted by shutdown");
                break;
            }
            if self.runtime.lifecycle.cancel_token.is_cancelled() {
                tracing::info!("native tool loop cancelled by user");
                break;
            }
            // Iteration 0 uses filtered tool_defs (schema filter + dependency gates).
            // Iterations 1+ expand to the full set but still apply hard dependency gates
            // so tools with unmet `requires` cannot re-enter through the expansion path (#2024).
            let defs_for_iter: Vec<ToolDefinition>;
            let defs_for_turn: &[ToolDefinition] = if iteration == 0 {
                &tool_defs
            } else {
                defs_for_iter = build_gated_defs_for_iteration(
                    iteration,
                    &all_tool_defs,
                    &self.services.tool_state,
                );
                &defs_for_iter
            };
            // None = continue loop, Some(()) = return Ok, Err = propagate
            if self
                .process_single_native_turn(defs_for_turn, iteration, query_embedding.clone())
                .await?
                .is_some()
            {
                return Ok(());
            }
            if self.check_doom_loop(iteration).await? {
                break;
            }
        }

        let _ = self.channel.send_stop_hint(StopHint::MaxTurnRequests).await;
        self.channel.flush_chunks().await?;
        Ok(())
    }

    /// Drive an LLM chat call through the durable step journal when a [`DurableContext`] is
    /// attached to the session.
    ///
    /// The step commits an `ExactlyOnceGuarded` / `CostBearingOrBoundaryIdempotent` intent
    /// before the real call runs, so a crash between the API response and journal acknowledgement
    /// is handled safely (`OnAmbiguous::Skip` — the cost is already incurred). The closure payload
    /// is `None` — the actual call always executes in every branch because `&mut self` cannot be
    /// captured inside the durable closure. The [`DurableStep::was_replayed`] flag is used to
    /// suppress double-printing in the caller.
    ///
    /// When `durable_ctx` is `None` the call is forwarded directly without any journaling.
    async fn call_llm_durable(
        &mut self,
        tool_defs: &[ToolDefinition],
        iteration: usize,
    ) -> Result<Option<zeph_llm::provider::ChatResponse>, crate::agent::error::AgentError> {
        self.ensure_session_durable_ctx().await;
        let Some(ctx) = self.services.session.durable_ctx.clone() else {
            return self.call_chat_with_tools_retry(tool_defs, 2).await;
        };

        let turn_span = tracing::info_span!(
            "core.durable.turn",
            iteration,
            execution_id = %ctx.execution_id().as_uuid(),
        );
        let cached_tokens = self.runtime.providers.cached_prompt_tokens;
        let fp_input = format!("llm_call:iter={iteration}:tokens={cached_tokens}").into_bytes();
        let desc = StepDescriptor::exactly_once_guarded(
            "llm_call",
            EffectIntentSubClass::CostBearingOrBoundaryIdempotent,
            Some(OnAmbiguous::Skip),
            fp_input,
        )
        .expect("CostBearingOrBoundaryIdempotent never requires explicit policy");

        let step = ctx
            .step_recorded::<Option<i64>, _, _>(desc, |_handle| async move { Ok(None::<i64>) })
            .instrument(turn_span)
            .await;

        match step {
            Ok(record) if record.was_replayed() => {
                // Already delivered to the user in a prior run; suppress re-printing.
                self.services.session.durable_turn_replayed = true;
                self.call_chat_with_tools_retry(tool_defs, 2).await
            }
            Ok(_) => self.call_chat_with_tools_retry(tool_defs, 2).await,
            Err(e) => {
                tracing::warn!(error = %e, "durable LLM step error; degrading to non-durable");
                self.call_chat_with_tools_retry(tool_defs, 2).await
            }
        }
    }

    /// Execute one turn of the native tool loop. Returns `Ok(Some(()))` when the LLM produced
    /// a terminal text response (caller should return `Ok(())`), `Ok(None)` to continue the
    /// loop, or `Err` on a hard error.
    #[tracing::instrument(
        name = "core.tool.single_turn",
        skip_all,
        level = "debug",
        fields(iteration),
        err
    )]
    async fn process_single_native_turn(
        &mut self,
        tool_defs: &[ToolDefinition],
        iteration: usize,
        query_embedding: Option<Vec<f32>>,
    ) -> Result<Option<()>, crate::agent::error::AgentError> {
        // Clear the per-turn replay flag; it is set below when the LLM step is replayed.
        self.services.session.durable_turn_replayed = false;

        // Track iteration for BudgetHint injection (#2267).
        self.services.tool_state.current_tool_iteration = iteration;
        self.channel.send_typing().await?;

        if let Some(ref budget) = self.context_manager.budget {
            let used =
                usize::try_from(self.runtime.providers.cached_prompt_tokens).unwrap_or(usize::MAX);
            let threshold = budget.max_tokens() * 4 / 5;
            if used >= threshold {
                tracing::warn!(
                    iteration,
                    used,
                    threshold,
                    "stopping tool loop: context budget nearing limit"
                );
                self.channel
                    .send("Stopping: context window is nearly full.")
                    .await?;
                return Ok(Some(()));
            }
        }

        // Show triage status indicator before inference when triage routing is active.
        if matches!(self.provider, zeph_llm::any::AnyProvider::Triage(_)) {
            self.channel
                .send_status_best_effort("Evaluating complexity...")
                .await;
        } else {
            self.channel.send_status_best_effort("thinking...").await;
        }

        let chat_result = self.call_llm_durable(tool_defs, iteration).await?;

        self.channel.send_status_best_effort("").await;

        let Some(chat_result) = chat_result else {
            tracing::debug!("chat_with_tools returned None (timeout)");
            return Ok(Some(()));
        };

        tracing::debug!(iteration, ?chat_result, "native tool loop iteration");

        if let zeph_llm::provider::ChatResponse::Text(text) = &chat_result {
            // RV-1: response verification before delivery.
            if self.run_response_verification(text) {
                let _ = self
                    .channel
                    .send("[security] Response blocked by injection detection.")
                    .await;
                self.channel.flush_chunks().await?;
                return Ok(Some(()));
            }
            let cleaned = self.scan_output_and_warn(text);
            // Double-print suppression: when the LLM step was replayed from the journal, the
            // assistant text was already delivered in a previous run; skip re-sending it to the
            // channel. Persistence and in-memory push still run so the context is consistent
            // (spec-064 §001 §15 RuntimeLayer observe-only; replay control is NOT in RuntimeLayer).
            if !self.services.session.durable_turn_replayed {
                if !cleaned.is_empty() {
                    let display = self.maybe_redact(&cleaned);
                    self.channel.send(&display).await?;
                }
                self.store_response_in_cache(&cleaned, query_embedding)
                    .await;
            }
            self.persist_message(Role::Assistant, &cleaned, &[], false)
                .await;
            self.push_message(Message::from_legacy(Role::Assistant, cleaned.as_str()));
            // Detect context loss after compaction and log failure pair if found.
            self.maybe_log_compression_failure(&cleaned).await;
            if cleaned.contains(zeph_llm::provider::MAX_TOKENS_TRUNCATION_MARKER) {
                let _ = self.channel.send_stop_hint(StopHint::MaxTokens).await;
            }
            return Ok(Some(()));
        }

        let zeph_llm::provider::ChatResponse::ToolUse {
            text,
            tool_calls,
            thinking_blocks,
        } = chat_result
        else {
            tracing::warn!(
                ?chat_result,
                "unexpected ChatResponse variant in native tool loop"
            );
            return Ok(Some(()));
        };
        self.preserve_thinking_blocks(thinking_blocks);
        let window_exhausted = self
            .handle_native_tool_calls(text.as_deref(), &tool_calls)
            .await?;

        // Summarize before pruning; apply deferred summaries after pruning.
        self.maybe_summarize_tool_pair().await;
        let keep_recent = 2 * self.services.memory.persistence.tool_call_cutoff + 2;
        self.prune_stale_tool_outputs(keep_recent);
        self.maybe_apply_deferred_summaries();
        self.flush_deferred_summaries().await;
        // Mid-iteration soft compaction: fires after summarization so fresh results are
        // either summarized or protected before pruning. Does not touch turn counters,
        // cooldown, or trigger Hard tier (no LLM call during tool loop).
        self.maybe_soft_compact_mid_iteration();
        self.flush_deferred_summaries().await;

        if window_exhausted {
            return Ok(Some(()));
        }

        Ok(None)
    }
}

/// Build the tool definition slice for iterations 1+ of the native tool loop.
///
/// Applies hard dependency-gate filtering when a dependency graph is configured, ensuring tools
/// with unmet `requires` cannot re-enter through the expansion path after iteration 0 (#2024).
///
/// Returns the allowed set as an owned `Vec`; the caller holds a reference into it.
/// When no dependency graph is present the full `all_tool_defs` slice is returned as-is (cloned).
fn build_gated_defs_for_iteration(
    iteration: usize,
    all_tool_defs: &[ToolDefinition],
    tool_state: &crate::agent::state::ToolState,
) -> Vec<ToolDefinition> {
    let Some(ref dep_graph) = tool_state.dependency_graph else {
        return all_tool_defs.to_vec();
    };
    if dep_graph.is_empty() {
        return all_tool_defs.to_vec();
    }

    let names: Vec<&str> = all_tool_defs.iter().map(|d| d.name.as_str()).collect();
    let allowed = dep_graph.filter_tool_names(
        &names,
        &tool_state.completed_tool_ids,
        &tool_state.dependency_always_on,
    );
    let allowed_set: std::collections::HashSet<&str> = allowed.into_iter().collect();

    // Deadlock fallback: if all non-always-on tools would be blocked, use the full set.
    let non_ao_allowed = allowed_set
        .iter()
        .filter(|n| !tool_state.dependency_always_on.contains(**n))
        .count();
    let non_ao_total = all_tool_defs
        .iter()
        .filter(|d| !tool_state.dependency_always_on.contains(d.name.as_str()))
        .count();
    if non_ao_allowed == 0 && non_ao_total > 0 {
        tracing::warn!(
            iteration,
            "tool dependency graph: all non-always-on tools gated on iter 1+; \
             disabling hard gates for this iteration"
        );
        return all_tool_defs.to_vec();
    }

    all_tool_defs
        .iter()
        .filter(|d| allowed_set.contains(d.name.as_str()))
        .cloned()
        .collect()
}

#[cfg(test)]
mod tests {
    use super::*;
    use zeph_subagent::TOOL_ARGS_JSON_LIMIT;

    fn json_val(s: &str) -> serde_json::Value {
        serde_json::from_str(s).unwrap()
    }

    fn make_tool_req(name: &str) -> zeph_llm::provider::ToolUseRequest {
        zeph_llm::provider::ToolUseRequest {
            id: format!("id_{name}"),
            name: name.into(),
            input: serde_json::Value::Null,
        }
    }

    fn make_tool_call(name: &str) -> ToolCall {
        ToolCall {
            tool_id: zeph_common::ToolName::new(name),
            ..Default::default()
        }
    }

    #[test]
    fn make_tool_hook_env_sets_tool_name() {
        let env = make_tool_hook_env("Edit", &serde_json::Value::Null, None);
        assert_eq!(env.get("ZEPH_TOOL_NAME").map(String::as_str), Some("Edit"));
    }

    #[test]
    fn make_tool_hook_env_sets_args_json_for_small_payload() {
        let input = json_val(r#"{"path": "/tmp/foo.txt"}"#);
        let env = make_tool_hook_env("Write", &input, None);
        let args = env
            .get("ZEPH_TOOL_ARGS_JSON")
            .expect("ZEPH_TOOL_ARGS_JSON missing");
        let parsed: serde_json::Value = serde_json::from_str(args).unwrap();
        assert_eq!(parsed["path"], "/tmp/foo.txt");
    }

    #[test]
    fn make_tool_hook_env_truncates_large_payload_safely() {
        // Build a JSON string > 64 KiB with a multi-byte char near the boundary.
        let mut big = String::from(r#"{"data":""#);
        // Fill mostly with ASCII, then add a 3-byte char (€ = 0xE2 0x82 0xAC) right at boundary.
        // We want the char boundary to fall inside the limit so truncation must round down.
        while big.len() < TOOL_ARGS_JSON_LIMIT - 3 {
            big.push('a');
        }
        big.push('€'); // 3 bytes — may straddle the limit
        while big.len() < TOOL_ARGS_JSON_LIMIT + 100 {
            big.push('b');
        }
        big.push_str(r#""}"#);
        let input: serde_json::Value = serde_json::from_str(&big).unwrap_or_default();
        // Must not panic and must end with the ellipsis character.
        let env = make_tool_hook_env("Shell", &input, None);
        let args = env
            .get("ZEPH_TOOL_ARGS_JSON")
            .expect("ZEPH_TOOL_ARGS_JSON missing");
        assert!(
            args.ends_with('…'),
            "truncated value should end with ellipsis"
        );
        assert!(
            args.is_char_boundary(args.len()),
            "truncation must land on char boundary"
        );
    }

    #[test]
    fn make_tool_hook_env_sets_session_id_when_present() {
        let env = make_tool_hook_env("Read", &serde_json::Value::Null, Some("sess-42"));
        assert_eq!(
            env.get("ZEPH_SESSION_ID").map(String::as_str),
            Some("sess-42")
        );
    }

    #[test]
    fn make_tool_hook_env_omits_session_id_when_none() {
        let env = make_tool_hook_env("Read", &serde_json::Value::Null, None);
        assert!(!env.contains_key("ZEPH_SESSION_ID"));
    }

    // Regression guard for issue #3738: pre_tool_use hooks must fire for tools that are
    // intercepted by the utility gate (Retrieve / Verify / Stop / Respond). The fix moves
    // pre-hook dispatch before check_call_gates. This test verifies that matching_hooks
    // correctly matches gate-intercepted tools so the hook system would observe them, and
    // that internal focus/compress tools are excluded when the caller skips them explicitly.
    #[test]
    fn pre_tool_use_hook_matches_gate_intercepted_tools_but_not_internal() {
        use zeph_config::{HookAction, HookDef, HookMatcher};
        use zeph_subagent::matching_hooks;

        let hook = HookDef {
            action: HookAction::Command {
                command: "true".to_owned(),
            },
            timeout_secs: 5,
            fail_closed: false,
            r#if: None,
        };
        // A wildcard-style matcher that matches any tool name token.
        let matchers = vec![HookMatcher {
            matcher: "shell|read|write|retrieve_memory".to_owned(),
            hooks: vec![hook],
        }];

        // Tools that a utility gate may intercept — pre-hook MUST fire for these.
        assert!(!matching_hooks(&matchers, "retrieve_memory").is_empty());
        assert!(!matching_hooks(&matchers, "shell").is_empty());

        // Internal tools — they are skipped before the hook dispatch block, so
        // matching_hooks is never called for them. Confirm they do NOT match the
        // hook matchers in the first place (extra guard).
        assert!(matching_hooks(&matchers, "compress_context").is_empty());
        assert!(matching_hooks(&matchers, "request_compaction").is_empty());
        assert!(matching_hooks(&matchers, "start_focus").is_empty());
        assert!(matching_hooks(&matchers, "complete_focus").is_empty());
    }

    // Gap 3: handle_request_compaction must return the rate-limit error when
    // CompactionState is already CompactedThisTurn.
    #[test]
    fn request_compaction_rate_limit_fires_when_compacted_this_turn() {
        use crate::agent::context_manager::CompactionState;
        use crate::testing::{MockChannel, MockToolExecutor, mock_provider};
        use zeph_skills::registry::SkillRegistry;
        let rt = tokio::runtime::Builder::new_current_thread()
            .build()
            .unwrap();
        rt.block_on(async {
            let mut agent = Agent::new(
                mock_provider(vec![]),
                MockChannel::new(vec![] as Vec<String>),
                SkillRegistry::empty(),
                None,
                5,
                MockToolExecutor::no_tools(),
            );
            // Simulate a compaction that already happened this turn.
            agent
                .context_manager
                .set_compaction_state(CompactionState::CompactedThisTurn { cooldown: 0 });
            let input = serde_json::json!({"reason": "context is growing"});
            let result = agent.handle_request_compaction(&input).await;
            assert!(
                result.contains("already performed this turn"),
                "rate-limit guard must fire: got {result:?}"
            );
        });
    }

    // Regression guard for #5584: when a retryable failure (e.g. Qdrant unreachable) was
    // already recorded this turn, handle_retrieve_action must not mandate another doomed
    // retry — it should let the originally-requested tool call proceed (Ok(None)) and inject
    // a graceful-degradation hint instead of the "you MUST call again" hint.
    #[tokio::test]
    async fn handle_retrieve_action_skips_mandatory_retry_after_retryable_failure() {
        use crate::testing::{MockChannel, MockToolExecutor, mock_provider};
        use zeph_skills::registry::SkillRegistry;
        use zeph_tools::error_taxonomy::ToolErrorCategory;

        let mut agent = Agent::new(
            mock_provider(vec![]),
            MockChannel::new(vec![] as Vec<String>),
            SkillRegistry::empty(),
            None,
            5,
            MockToolExecutor::no_tools(),
        );
        agent
            .tool_orchestrator
            .last_tool_error
            .insert("memory_search".to_owned(), ToolErrorCategory::NetworkError);

        let tc = make_tool_req("bash");
        let mut hints = Vec::new();
        let result = agent
            .handle_retrieve_action(0, &tc, &make_tool_call("bash"), &mut hints)
            .await
            .unwrap();

        assert!(
            result.is_none(),
            "must return None so the originally-requested tool proceeds to dispatch"
        );
        assert_eq!(hints.len(), 1);
        assert!(
            hints[0].contains("Proceed with the 'bash' tool call"),
            "hint must direct graceful degradation, got: {}",
            hints[0]
        );
        assert!(
            !hints[0].contains("you MUST call"),
            "must not mandate another retry, got: {}",
            hints[0]
        );
    }

    #[tokio::test]
    async fn handle_retrieve_action_mandates_retry_without_prior_failure() {
        use crate::testing::{MockChannel, MockToolExecutor, mock_provider};
        use zeph_skills::registry::SkillRegistry;

        let mut agent = Agent::new(
            mock_provider(vec![]),
            MockChannel::new(vec![] as Vec<String>),
            SkillRegistry::empty(),
            None,
            5,
            MockToolExecutor::no_tools(),
        );

        let tc = make_tool_req("bash");
        let mut hints = Vec::new();
        let result = agent
            .handle_retrieve_action(0, &tc, &make_tool_call("bash"), &mut hints)
            .await
            .unwrap();

        assert!(
            result.is_some(),
            "without a prior failure, the tool call is skipped pending retrieval"
        );
        assert_eq!(hints.len(), 1);
        assert!(
            hints[0].contains("you MUST call the 'bash' tool again"),
            "hint must mandate a retry, got: {}",
            hints[0]
        );
    }

    // Regression guard for critic-flagged S3 (#5584 follow-up): a retryable failure of an
    // UNRELATED tool (e.g. web_fetch) must not suppress the Retrieve mandatory-retry hint
    // for the rest of the turn — only a failure of memory_search, the retrieval tool the
    // Retrieve branch's hint actually recommends, should trigger graceful degradation.
    #[tokio::test]
    async fn handle_retrieve_action_mandates_retry_when_only_unrelated_tool_failed() {
        use crate::testing::{MockChannel, MockToolExecutor, mock_provider};
        use zeph_skills::registry::SkillRegistry;
        use zeph_tools::error_taxonomy::ToolErrorCategory;

        let mut agent = Agent::new(
            mock_provider(vec![]),
            MockChannel::new(vec![] as Vec<String>),
            SkillRegistry::empty(),
            None,
            5,
            MockToolExecutor::no_tools(),
        );
        agent
            .tool_orchestrator
            .last_tool_error
            .insert("web_fetch".to_owned(), ToolErrorCategory::NetworkError);

        let tc = make_tool_req("bash");
        let mut hints = Vec::new();
        let result = agent
            .handle_retrieve_action(0, &tc, &make_tool_call("bash"), &mut hints)
            .await
            .unwrap();

        assert!(
            result.is_some(),
            "an unrelated tool's stale retryable failure must not suppress Retrieve"
        );
        assert_eq!(hints.len(), 1);
        assert!(
            hints[0].contains("you MUST call the 'bash' tool again"),
            "hint must still mandate a retry, got: {}",
            hints[0]
        );
    }

    // Regression guard for #5774: memory_search failing with ConfirmationRequired (e.g. the
    // query content itself trips a sanitizer/exfiltration-guard check) must be treated the
    // same as a retryable/network failure — otherwise the Retrieve gate keeps mandating a
    // fresh memory_search detour every iteration, which fails identically every time, forming
    // an unbreakable loop.
    #[tokio::test]
    async fn handle_retrieve_action_skips_mandatory_retry_after_confirmation_required() {
        use crate::testing::{MockChannel, MockToolExecutor, mock_provider};
        use zeph_skills::registry::SkillRegistry;
        use zeph_tools::error_taxonomy::ToolErrorCategory;

        let mut agent = Agent::new(
            mock_provider(vec![]),
            MockChannel::new(vec![] as Vec<String>),
            SkillRegistry::empty(),
            None,
            5,
            MockToolExecutor::no_tools(),
        );
        agent.tool_orchestrator.last_tool_error.insert(
            "memory_search".to_owned(),
            ToolErrorCategory::ConfirmationRequired,
        );

        let tc = make_tool_req("bash");
        let mut hints = Vec::new();
        let result = agent
            .handle_retrieve_action(0, &tc, &make_tool_call("bash"), &mut hints)
            .await
            .unwrap();

        assert!(
            result.is_none(),
            "must return None so the originally-requested tool proceeds to dispatch"
        );
        assert_eq!(hints.len(), 1);
        assert!(
            !hints[0].contains("you MUST call"),
            "must not mandate another memory_search detour, got: {}",
            hints[0]
        );
    }

    // Regression guard for #5774: even when memory_search itself never records a failure (or
    // the failing tool varies from iteration to iteration, e.g. the LLM retries a different
    // but similar native tool each time), the Retrieve gate must stop mandating fresh detours
    // after MAX_RETRIEVE_MANDATES_PER_TURN cycles in the same turn — the defense-in-depth
    // circuit breaker for any undiscovered variant of the stall.
    #[tokio::test]
    async fn handle_retrieve_action_circuit_breaker_after_mandate_limit() {
        use crate::testing::{MockChannel, MockToolExecutor, mock_provider};
        use zeph_skills::registry::SkillRegistry;

        let mut agent = Agent::new(
            mock_provider(vec![]),
            MockChannel::new(vec![] as Vec<String>),
            SkillRegistry::empty(),
            None,
            5,
            MockToolExecutor::no_tools(),
        );

        // Drive MAX_RETRIEVE_MANDATES_PER_TURN mandates, each for a distinct tool call (as
        // happens when the model retries a different, similarly-gated tool each time).
        for i in 0..crate::agent::MAX_RETRIEVE_MANDATES_PER_TURN {
            let name = format!("tool_{i}");
            let tc = make_tool_req(&name);
            let mut hints = Vec::new();
            let result = agent
                .handle_retrieve_action(0, &tc, &make_tool_call(&name), &mut hints)
                .await
                .unwrap();
            assert!(result.is_some(), "mandate #{i} should still be issued");
        }

        // The next Retrieve-gated call, for yet another distinct tool, must bypass the gate.
        let tc = make_tool_req("tool_over_limit");
        let mut hints = Vec::new();
        let result = agent
            .handle_retrieve_action(0, &tc, &make_tool_call("tool_over_limit"), &mut hints)
            .await
            .unwrap();

        assert!(
            result.is_none(),
            "must proceed directly once the mandate limit is reached"
        );
        assert_eq!(hints.len(), 1);
        assert!(
            !hints[0].contains("you MUST call"),
            "must not mandate yet another detour, got: {}",
            hints[0]
        );
    }

    // --- PAAC tool-dispatch unmasking: unmask_json_value (#5437) ---

    mod unmask_json_value_tests {
        use super::unmask_json_value;
        use zeph_sanitizer::secret_mask::{SecretCategory, SecretMaskRegistry};

        #[test]
        fn unmasks_top_level_string() {
            let registry = SecretMaskRegistry::new();
            registry.register("KEY", "hunter2password", SecretCategory::Password);
            let masked = registry.mask("hunter2password");
            let mut value = serde_json::Value::String(masked);
            unmask_json_value(&mut value, &registry);
            assert_eq!(value, serde_json::json!("hunter2password"));
        }

        #[test]
        fn unmasks_string_nested_in_object_and_array() {
            let registry = SecretMaskRegistry::new();
            registry.register("KEY", "supersecretvalue1", SecretCategory::ApiKey);
            let masked_placeholder = registry.mask("supersecretvalue1");
            let mut value = serde_json::json!({
                "code": format!("curl -H 'Authorization: Bearer {masked_placeholder}'"),
                "headers": [masked_placeholder.clone(), "plain-value"],
                "nested": {"token": masked_placeholder},
            });
            unmask_json_value(&mut value, &registry);
            assert!(
                value["code"]
                    .as_str()
                    .unwrap()
                    .contains("supersecretvalue1")
            );
            assert_eq!(value["headers"][0], serde_json::json!("supersecretvalue1"));
            assert_eq!(value["headers"][1], serde_json::json!("plain-value"));
            assert_eq!(
                value["nested"]["token"],
                serde_json::json!("supersecretvalue1")
            );
        }

        /// Placeholder-injection safety (#5437): a model-crafted placeholder that was never
        /// issued by this registry (wrong/foreign nonce) must be left verbatim — `unmask` is
        /// nonce-scoped and passthrough-on-miss, so a model cannot forge a placeholder to
        /// exfiltrate a secret it never legitimately saw.
        #[test]
        fn foreign_placeholder_is_left_untouched() {
            let registry = SecretMaskRegistry::new();
            registry.register("KEY", "realsecretvalue1", SecretCategory::ApiKey);
            let forged = "<SECRET:api_key:0000000000000000:0>".to_owned();
            let mut value = serde_json::Value::String(forged.clone());
            unmask_json_value(&mut value, &registry);
            assert_eq!(
                value,
                serde_json::Value::String(forged),
                "a placeholder this registry never issued must pass through unchanged"
            );
        }

        #[test]
        fn non_string_values_are_untouched() {
            let registry = SecretMaskRegistry::new();
            registry.register("KEY", "somesecretvalue1", SecretCategory::Generic);
            let mut value =
                serde_json::json!({"count": 3, "enabled": true, "ratio": 1.5, "n": null});
            let before = value.clone();
            unmask_json_value(&mut value, &registry);
            assert_eq!(value, before);
        }

        // --- S1: unmask-miss telemetry (#5437 critique) ---

        #[test]
        fn mangled_placeholder_returns_a_miss_and_is_left_verbatim() {
            let registry = SecretMaskRegistry::new();
            registry.register("KEY", "realsecretvalue1", SecretCategory::ApiKey);
            let masked = registry.mask("realsecretvalue1");
            // Simulate an LLM inserting a space into the opaque token — a real, observed
            // failure mode for long tokens (S1).
            let mangled = masked.replacen(':', ": ", 1);
            let mut value = serde_json::Value::String(mangled.clone());
            let misses = unmask_json_value(&mut value, &registry);
            assert_eq!(
                misses, 1,
                "a mangled placeholder must be reported as a miss"
            );
            assert_eq!(
                value,
                serde_json::Value::String(mangled),
                "mangled placeholder is left verbatim (fail-safe, no leak)"
            );
        }

        #[test]
        fn successful_unmask_reports_zero_misses() {
            let registry = SecretMaskRegistry::new();
            registry.register("KEY", "realsecretvalue1", SecretCategory::ApiKey);
            let masked = registry.mask("realsecretvalue1");
            let mut value = serde_json::Value::String(masked);
            let misses = unmask_json_value(&mut value, &registry);
            assert_eq!(misses, 0);
            assert_eq!(value, serde_json::json!("realsecretvalue1"));
        }

        #[test]
        fn miss_count_aggregates_across_nested_structure() {
            let registry = SecretMaskRegistry::new();
            registry.register("KEY", "realsecretvalue1", SecretCategory::ApiKey);
            let masked = registry.mask("realsecretvalue1");
            let mangled = masked.replacen(':', ": ", 1);
            let mut value = serde_json::json!({
                "ok": masked,
                "nested": {"a": mangled.clone(), "b": mangled},
                "plain": "no placeholder here",
            });
            let misses = unmask_json_value(&mut value, &registry);
            assert_eq!(
                misses, 2,
                "both mangled leaves must be counted, the valid one must not"
            );
        }

        // --- secret values containing JSON/regex-special characters ---

        #[test]
        fn secret_with_json_and_regex_special_chars_roundtrips() {
            let registry = SecretMaskRegistry::new();
            let tricky_secret = r#"p@ss"w0rd\n{[(.*+?)]}$^|\}"#;
            registry.register("KEY", tricky_secret, SecretCategory::Password);
            let masked = registry.mask(tricky_secret);
            assert!(!masked.contains(tricky_secret));

            let mut value = serde_json::json!({
                "code": format!("echo '{masked}'"),
                "list": [masked.clone()],
            });
            let misses = unmask_json_value(&mut value, &registry);
            assert_eq!(misses, 0);
            assert!(value["code"].as_str().unwrap().contains(tricky_secret));
            assert_eq!(value["list"][0], serde_json::json!(tricky_secret));
        }
    }

    /// Regression tests for #5513: cancellation during the post-dispatch phases
    /// (confirmation / retry / reformat) must write exactly one `[Cancelled]` tombstone
    /// `ToolResult` per `tool_use_id` and must never leave a `ToolUse` orphaned or let the
    /// batch persist run again afterward.
    mod cancellation_regression_tests {
        use super::*;
        use crate::agent::agent_tests::*;

        /// Always fails with a `Transient` error and is marked retryable, so
        /// `handle_retry_phase` enters its backoff-sleep branch on every attempt.
        struct AlwaysTransientExecutor;

        impl zeph_tools::executor::ToolExecutor for AlwaysTransientExecutor {
            fn execute(
                &self,
                _response: &str,
            ) -> impl std::future::Future<
                Output = Result<Option<zeph_tools::ToolOutput>, zeph_tools::ToolError>,
            > + Send {
                std::future::ready(Ok(None))
            }

            fn execute_tool_call(
                &self,
                _call: &ToolCall,
            ) -> impl std::future::Future<
                Output = Result<Option<zeph_tools::ToolOutput>, zeph_tools::ToolError>,
            > + Send {
                std::future::ready(Err(zeph_tools::ToolError::Execution(std::io::Error::new(
                    std::io::ErrorKind::TimedOut,
                    "always transient",
                ))))
            }

            fn is_tool_retryable(&self, _tool_id: &str) -> bool {
                true
            }
            zeph_tools::tool_executor_no_inner_defaults!();
        }

        fn tool_result_ids(agent: &Agent<MockChannel>, id: &str) -> Vec<&'static str> {
            agent
                .msg
                .messages
                .iter()
                .flat_map(|m| m.parts.iter())
                .filter(|p| {
                    matches!(p, MessagePart::ToolResult { tool_use_id, .. } if tool_use_id == id)
                })
                .map(|_| "match")
                .collect()
        }

        /// Bug A: a token already cancelled before `handle_confirmation_phase` runs must
        /// short-circuit `run_post_dispatch_phases` after the first phase reports
        /// cancellation, instead of cascading into `handle_retry_phase` and
        /// `handle_reformat_phase` as well. Before the fix, each of the three phases
        /// independently detected the same cancellation and wrote its own tombstone batch,
        /// producing up to 3 duplicate `[Cancelled]` `ToolResult`s per `tool_use_id`.
        #[tokio::test]
        async fn cancelled_before_confirmation_phase_writes_one_tombstone_and_skips_later_phases() {
            let provider = mock_provider(vec![]);
            let channel = MockChannel::new(vec![]);
            let registry = create_test_registry();
            let executor = MockToolExecutor::no_tools();
            let mut agent = Agent::new(provider, channel, registry, None, 5, executor);
            agent.tool_orchestrator.max_tool_retries = 2;
            agent.tool_orchestrator.parameter_reformat_provider = "fast".to_owned();

            let tool_calls = vec![
                zeph_llm::provider::ToolUseRequest {
                    id: "id-1".to_owned(),
                    name: "bash".to_owned().into(),
                    input: serde_json::json!({}),
                },
                zeph_llm::provider::ToolUseRequest {
                    id: "id-2".to_owned(),
                    name: "bash".to_owned().into(),
                    input: serde_json::json!({}),
                },
            ];
            let calls = vec![
                ToolCall {
                    tool_id: zeph_common::ToolName::new("bash"),
                    ..Default::default()
                },
                ToolCall {
                    tool_id: zeph_common::ToolName::new("bash"),
                    ..Default::default()
                },
            ];
            let mut tool_results: Vec<
                Result<Option<zeph_tools::ToolOutput>, zeph_tools::ToolError>,
            > = vec![Ok(None), Ok(None)];

            let cancel = tokio_util::sync::CancellationToken::new();
            cancel.cancel();

            let cancelled = agent
                .run_post_dispatch_phases(&tool_calls, &calls, &mut tool_results, 2, &cancel)
                .await
                .unwrap();

            assert!(
                cancelled,
                "run_post_dispatch_phases must report cancellation"
            );

            for id in ["id-1", "id-2"] {
                let matches = tool_result_ids(&agent, id);
                assert_eq!(
                    matches.len(),
                    1,
                    "tool_use_id {id} must have exactly one [Cancelled] tombstone, got {}",
                    matches.len()
                );
            }
        }

        /// Bug C: cancellation landing specifically inside the retry-phase backoff-sleep
        /// `tokio::select!` must still persist a tombstone `ToolResult` for the pending
        /// `ToolUse`. Before the fix this was the only cancellation checkpoint in the file
        /// that returned without calling `persist_cancelled_tool_results`, leaving the
        /// `ToolUse` message genuinely orphaned (zero `ToolResult`s) for the rest of the
        /// live session.
        #[tokio::test]
        async fn handle_retry_phase_cancelled_during_backoff_sleep_persists_tombstone() {
            let provider = mock_provider(vec![]);
            let channel = MockChannel::new(vec![]);
            let registry = create_test_registry();
            let executor = AlwaysTransientExecutor;
            let mut agent = Agent::new(provider, channel, registry, None, 5, executor);
            agent.tool_orchestrator.max_tool_retries = 3;
            // Large, deterministic-enough backoff window so the spawned cancellation below
            // (fired after a short real-time delay) lands during the sleep rather than after
            // it — full-jitter backoff makes an exact guarantee impossible, but at this
            // magnitude the chance of picking a delay under 200ms is negligible.
            agent.tool_orchestrator.retry_base_ms = 600_000;
            agent.tool_orchestrator.retry_max_ms = 600_000;
            agent.tool_orchestrator.max_retry_duration_secs = 0;

            let tool_calls = vec![zeph_llm::provider::ToolUseRequest {
                id: "id-retry".to_owned(),
                name: "bash".to_owned().into(),
                input: serde_json::json!({}),
            }];
            let calls = vec![ToolCall {
                tool_id: zeph_common::ToolName::new("bash"),
                ..Default::default()
            }];
            let mut tool_results: Vec<
                Result<Option<zeph_tools::ToolOutput>, zeph_tools::ToolError>,
            > = vec![Err(zeph_tools::ToolError::Execution(std::io::Error::new(
                std::io::ErrorKind::TimedOut,
                "initial transient failure",
            )))];

            let cancel = tokio_util::sync::CancellationToken::new();
            let cancel_trigger = cancel.clone();
            tokio::spawn(async move {
                tokio::time::sleep(std::time::Duration::from_millis(200)).await;
                cancel_trigger.cancel();
            });

            let cancelled = agent
                .handle_retry_phase(&tool_calls, &calls, &mut tool_results, 3, &cancel)
                .await
                .unwrap();

            assert!(cancelled, "handle_retry_phase must report cancellation");

            let matches = tool_result_ids(&agent, "id-retry");
            assert_eq!(
                matches.len(),
                1,
                "cancellation during backoff sleep must still write exactly one tombstone \
                 ToolResult, got {}",
                matches.len()
            );
        }

        /// Bug B + Bug C, exercised through the full `handle_native_tool_calls` entry point:
        /// once `run_post_dispatch_phases` reports cancellation (here triggered during the
        /// retry-phase backoff sleep), `handle_native_tool_calls` must return immediately and
        /// must NOT call `process_tool_result_batch` afterward. Before the fix, the batch
        /// persist ran unconditionally, appending a second (contradicting, non-cancelled)
        /// `ToolResult` message right after the phases' own tombstone write.
        #[tokio::test]
        async fn handle_native_tool_calls_cancelled_during_retry_backoff_skips_batch_persist() {
            let provider = mock_provider(vec![]);
            let channel = MockChannel::new(vec![]);
            let registry = create_test_registry();
            let executor = AlwaysTransientExecutor;
            let mut agent = Agent::new(provider, channel, registry, None, 5, executor);
            agent.tool_orchestrator.max_tool_retries = 3;
            agent.tool_orchestrator.retry_base_ms = 600_000;
            agent.tool_orchestrator.retry_max_ms = 600_000;
            agent.tool_orchestrator.max_retry_duration_secs = 0;

            let tool_calls = vec![zeph_llm::provider::ToolUseRequest {
                id: "id-e2e".to_owned(),
                name: "bash".to_owned().into(),
                input: serde_json::json!({"command": "echo hi"}),
            }];

            let cancel_trigger = agent.runtime.lifecycle.cancel_token.clone();
            tokio::spawn(async move {
                tokio::time::sleep(std::time::Duration::from_millis(200)).await;
                cancel_trigger.cancel();
            });

            let window_exhausted = agent
                .handle_native_tool_calls(None, &tool_calls)
                .await
                .unwrap();
            assert!(
                !window_exhausted,
                "a cancelled turn must not report utility-window exhaustion"
            );

            let matches = tool_result_ids(&agent, "id-e2e");
            assert_eq!(
                matches.len(),
                1,
                "exactly one ToolResult must exist for id-e2e after cancellation, got {}",
                matches.len()
            );

            // The tombstone must be the very last message — proving process_tool_result_batch
            // did not run afterward and append a second, contradicting result.
            let last = agent.msg.messages.last().expect("at least one message");
            let has_tombstone = last.parts.iter().any(|p| {
                matches!(
                    p,
                    MessagePart::ToolResult { tool_use_id, content, is_error }
                        if tool_use_id == "id-e2e" && content == "[Cancelled]" && *is_error
                )
            });
            assert!(
                has_tombstone,
                "last message must be the [Cancelled] tombstone for id-e2e, got: {last:?}"
            );
        }

        /// #5717 regression: `cancel_tool_batch` previously ran
        /// `self.channel.send("[Cancelled]").await?` before `persist_cancelled_tool_results`,
        /// using `?` to propagate any send failure immediately. A channel-send failure (closed
        /// mpsc receiver, dropped Telegram/Discord connection) therefore short-circuited the
        /// function and skipped the tombstone persist entirely, leaving the in-flight `ToolUse`
        /// batch without a matching `ToolResult` — the same orphaned-`tool_calls` defect class
        /// already fixed 3 times over (#5464, #5513, #5646). The fix persists the tombstone
        /// first and only logs (never propagates) a subsequent send failure.
        #[tokio::test]
        async fn cancel_tool_batch_persists_tombstone_even_when_channel_send_fails() {
            let provider = mock_provider(vec![]);
            let channel = MockChannel::new(vec![]).with_failing_send();
            let registry = create_test_registry();
            let executor = MockToolExecutor::no_tools();
            let mut agent = Agent::new(provider, channel, registry, None, 5, executor);

            let tool_calls = vec![zeph_llm::provider::ToolUseRequest {
                id: "id-cancel".to_owned(),
                name: "bash".to_owned().into(),
                input: serde_json::json!({}),
            }];

            let result = agent
                .cancel_tool_batch(&tool_calls, "test cancellation with failing channel")
                .await;
            assert!(
                result.is_ok(),
                "cancel_tool_batch must not fail even when channel.send fails, got {result:?}"
            );

            let matches = tool_result_ids(&agent, "id-cancel");
            assert_eq!(
                matches.len(),
                1,
                "tombstone must be persisted even when the notification send fails, got {}",
                matches.len()
            );
        }
    }

    mod mandated_retry_regression_tests {
        use super::*;
        use crate::agent::agent_tests::*;

        fn tool_result_content(agent: &Agent<MockChannel>, id: &str) -> Option<(String, bool)> {
            agent.msg.messages.iter().find_map(|m| {
                m.parts.iter().find_map(|p| match p {
                    MessagePart::ToolResult {
                        tool_use_id,
                        content,
                        is_error,
                    } if tool_use_id == id => Some((content.clone(), *is_error)),
                    _ => None,
                })
            })
        }

        /// #5719 end-to-end regression: `find_path` (`default_gain` 0.65) triggers the
        /// `Retrieve` rule on a fresh first call and is skipped with a hint instructing the LLM
        /// to call it again with the same arguments. Before the fix, that mandated retry was
        /// re-vetoed by the redundancy (`Respond`) rule on the very next round — the tool never
        /// executed and the turn ended with a fabricated "restriction" apology. The retry must
        /// now execute for real.
        #[tokio::test]
        async fn find_path_mandated_retry_executes_instead_of_redundant_respond() {
            let provider = mock_provider(vec![]);
            let channel = MockChannel::new(vec![]);
            let registry = create_test_registry();
            let executor = MockToolExecutor::with_output("find_path", "src/main.rs\nsrc/lib.rs");
            let mut agent = Agent::new(provider, channel, registry, None, 5, executor);
            agent
                .tool_orchestrator
                .set_utility_config(zeph_tools::UtilityScoringConfig {
                    enabled: true,
                    ..zeph_tools::UtilityScoringConfig::default()
                });

            let args = serde_json::json!({"pattern": "*.rs"});
            let round1 = vec![zeph_llm::provider::ToolUseRequest {
                id: "call-1".to_owned(),
                name: "find_path".to_owned().into(),
                input: args.clone(),
            }];
            agent.handle_native_tool_calls(None, &round1).await.unwrap();

            let (content1, _) =
                tool_result_content(&agent, "call-1").expect("call-1 must have a ToolResult");
            assert!(
                content1.contains("[skipped]"),
                "first call must be skipped by the Retrieve rule, got: {content1}"
            );

            // Second round: identical tool name + args under a fresh tool_use_id, as a real LLM
            // would assign, mimicking compliance with the injected "you MUST call it again" hint.
            let round2 = vec![zeph_llm::provider::ToolUseRequest {
                id: "call-2".to_owned(),
                name: "find_path".to_owned().into(),
                input: args,
            }];
            agent.handle_native_tool_calls(None, &round2).await.unwrap();

            let (content2, is_error2) =
                tool_result_content(&agent, "call-2").expect("call-2 must have a ToolResult");
            assert!(
                content2.contains("src/main.rs") && !is_error2,
                "mandated retry must execute for real instead of being re-vetoed as a \
                 redundant duplicate, got: {content2}"
            );
        }

        /// #5719 companion regression, found while writing the test above: the utility gate's
        /// synthetic `[skipped]` output must never be written into the tool result cache.
        /// `find_path`/`list_directory`/`grep`/`glob`/`search_code` are all cacheable (none are
        /// in the cache's deny-list), so a Retrieve-skipped call poisoned the cache under the
        /// real args hash — even after the mandated-retry bypass correctly recommended
        /// `ToolCall`, the tier loop served the stale `[skipped]` text back from cache instead
        /// of actually executing the tool, silently reintroducing the stall the bypass fixes.
        #[tokio::test]
        async fn retrieve_skip_output_is_never_cached() {
            let provider = mock_provider(vec![]);
            let channel = MockChannel::new(vec![]);
            let registry = create_test_registry();
            let executor = MockToolExecutor::with_output("find_path", "src/main.rs");
            let mut agent = Agent::new(provider, channel, registry, None, 5, executor);
            agent
                .tool_orchestrator
                .set_utility_config(zeph_tools::UtilityScoringConfig {
                    enabled: true,
                    ..zeph_tools::UtilityScoringConfig::default()
                });

            let args = serde_json::json!({"pattern": "*.rs"});
            let round1 = vec![zeph_llm::provider::ToolUseRequest {
                id: "call-1".to_owned(),
                name: "find_path".to_owned().into(),
                input: args.clone(),
            }];
            agent.handle_native_tool_calls(None, &round1).await.unwrap();

            let (content1, _) =
                tool_result_content(&agent, "call-1").expect("call-1 must have a ToolResult");
            assert!(content1.contains("[skipped]"));

            let params = match &args {
                serde_json::Value::Object(m) => m.clone(),
                _ => serde_json::Map::new(),
            };
            let hash = tool_args_hash(&params);
            let cached = agent
                .tool_orchestrator
                .result_cache
                .get(&zeph_tools::CacheKey::new("find_path", hash));
            assert!(
                cached.is_none(),
                "utility-gate skip output must never be cached, got: {cached:?}"
            );
        }

        /// #5774 end-to-end regression, driving the REAL `handle_confirmation_phase` ->
        /// classification -> `record_gate_feedback` pipeline (not a hand-inserted
        /// `last_tool_error` entry — see critic finding S1). Reproduces the pathological
        /// sub-case the code's own `execute_tool_call_confirmed_erased` comment documents:
        /// `memory_search` fails with `ConfirmationRequired`, the user confirms, and the
        /// confirmed re-execution *itself* fails with `ConfirmationRequired` again (e.g. a
        /// misconfigured executor stack re-runs the same sanitizer/exfiltration-guard check).
        /// Proves `has_blocked_retrieval_this_turn` actually engages from that real dispatch
        /// outcome, and that the bypass applies to a *different* tool than the one that
        /// originally triggered the `Retrieve` mandate — matching the issue's own observation
        /// that the LLM sometimes retries a different, similarly-gated tool each cycle.
        #[tokio::test]
        async fn memory_search_confirmation_required_breaks_retrieve_loop_via_real_pipeline() {
            let provider = mock_provider(vec![]);
            let channel = MockChannel::new(vec![]);
            let registry = create_test_registry();
            let executor = MockToolExecutor::new(vec![
                // memory_search's initial dispatch attempt.
                Err(ToolError::ConfirmationRequired {
                    command: "search query embedding /tmp/somefile.txt".to_owned(),
                }),
                // memory_search's confirmed re-execution (MockChannel auto-confirms) — still
                // fails, simulating the misconfigured-executor-stack sub-case.
                Err(ToolError::ConfirmationRequired {
                    command: "search query embedding /tmp/somefile.txt".to_owned(),
                }),
                // grep's dispatch, reached only if the Retrieve gate correctly bypasses.
                Ok(Some(ToolOutput {
                    tool_name: "grep".to_owned().into(),
                    summary: "match: foo.rs:10".to_owned(),
                    blocks_executed: 1,
                    filter_stats: None,
                    diff: None,
                    streamed: false,
                    terminal_id: None,
                    locations: None,
                    raw_response: None,
                    claim_source: None,
                    ..Default::default()
                })),
            ]);
            let mut agent = Agent::new(provider, channel, registry, None, 5, executor);
            agent
                .tool_orchestrator
                .set_utility_config(zeph_tools::UtilityScoringConfig {
                    enabled: true,
                    ..zeph_tools::UtilityScoringConfig::default()
                });

            // Round 1: LLM calls `list_directory` (gain 0.65, uncertainty ~1.0) -> Retrieve.
            let round1 = vec![zeph_llm::provider::ToolUseRequest {
                id: "call-1".to_owned(),
                name: "list_directory".to_owned().into(),
                input: serde_json::json!({"path": "."}),
            }];
            agent.handle_native_tool_calls(None, &round1).await.unwrap();
            let (content1, _) =
                tool_result_content(&agent, "call-1").expect("call-1 must have a ToolResult");
            assert!(
                content1.contains("[skipped]"),
                "first call must be skipped by the Retrieve rule, got: {content1}"
            );

            // Round 2: LLM complies with the hint and calls memory_search (gain 0.8 -> dispatches
            // for real), fails with ConfirmationRequired, MockChannel auto-confirms, and the
            // confirmed re-execution also fails with ConfirmationRequired.
            let round2 = vec![zeph_llm::provider::ToolUseRequest {
                id: "call-2".to_owned(),
                name: "memory_search".to_owned().into(),
                input: serde_json::json!({"query": "somefile.txt"}),
            }];
            agent.handle_native_tool_calls(None, &round2).await.unwrap();
            let (content2, is_error2) =
                tool_result_content(&agent, "call-2").expect("call-2 must have a ToolResult");
            assert!(
                is_error2 && content2.contains("confirmation_required"),
                "memory_search must surface the real ConfirmationRequired failure, got: {content2}"
            );
            assert_eq!(
                agent.tool_orchestrator.last_tool_error.get("memory_search"),
                Some(&zeph_tools::error_taxonomy::ToolErrorCategory::ConfirmationRequired),
                "record_gate_feedback must record the real dispatch outcome via the actual \
                 confirmation-phase pipeline, not a hand-inserted entry"
            );

            // Round 3: LLM tries a DIFFERENT tool than round 1 (as the issue observed happening
            // in practice) that would normally also trigger Retrieve. It must dispatch directly
            // instead of mandating yet another memory_search detour.
            let round3 = vec![zeph_llm::provider::ToolUseRequest {
                id: "call-3".to_owned(),
                name: "grep".to_owned().into(),
                input: serde_json::json!({"pattern": "foo"}),
            }];
            agent.handle_native_tool_calls(None, &round3).await.unwrap();
            let (content3, is_error3) =
                tool_result_content(&agent, "call-3").expect("call-3 must have a ToolResult");
            assert!(
                content3.contains("match: foo.rs:10") && !is_error3,
                "grep must dispatch directly once memory_search is known blocked this turn, \
                 not be redirected into another Retrieve mandate, got: {content3}"
            );
        }

        /// #5774 S3 regression (critic-flagged, post-S2): reproduces the issue's own
        /// "answering the dialog does not break the cycle" observation — the user repeatedly
        /// **declines** the `memory_search` confirmation prompt. A decline is a *successful*
        /// `[cancelled by user]` result, so `has_blocked_retrieval_this_turn` self-heals
        /// `last_tool_error` back to `None` after every cycle and never bypasses the `Retrieve`
        /// gate. Before this fix, `reset_retrieve_mandate_count` fired unconditionally whenever
        /// ANY call reached real dispatch — including `memory_search`'s own dispatch (gain 0.8
        /// always takes the direct `ToolCall` branch) — so the consecutive-mandate counter was
        /// reset to 0 on every single one of these cycles and the circuit breaker never tripped
        /// either. Neither of the two loop-prevention bounds fired: the exact scenario the issue
        /// reported as unbreakable. Proves the fix: the counter now only resets on a
        /// *non-`memory_search`* tool reaching real dispatch, so 3 consecutive
        /// Retrieve-mandate-then-decline cycles still trip the breaker on the 4th distinct tool.
        #[tokio::test]
        async fn repeated_decline_still_trips_circuit_breaker() {
            let provider = mock_provider(vec![]);
            let channel = MockChannel::new(vec![]).with_confirmations(vec![false, false, false]);
            let registry = create_test_registry();
            let executor = MockToolExecutor::new(vec![
                // memory_search dispatch attempts across 3 decline cycles.
                Err(ToolError::ConfirmationRequired {
                    command: "search query A".to_owned(),
                }),
                Err(ToolError::ConfirmationRequired {
                    command: "search query B".to_owned(),
                }),
                Err(ToolError::ConfirmationRequired {
                    command: "search query C".to_owned(),
                }),
                // find_path's dispatch, reached only once the circuit breaker trips.
                Ok(Some(ToolOutput {
                    tool_name: "find_path".to_owned().into(),
                    summary: "src/main.rs".to_owned(),
                    blocks_executed: 1,
                    filter_stats: None,
                    diff: None,
                    streamed: false,
                    terminal_id: None,
                    locations: None,
                    raw_response: None,
                    claim_source: None,
                    ..Default::default()
                })),
            ]);
            let mut agent = Agent::new(provider, channel, registry, None, 5, executor);
            agent
                .tool_orchestrator
                .set_utility_config(zeph_tools::UtilityScoringConfig {
                    enabled: true,
                    ..zeph_tools::UtilityScoringConfig::default()
                });

            // Each Retrieve-mandate round uses a distinct tool: reusing the same tool+args would
            // let the #5719 mandated-retry bypass dispatch it directly on the next round instead
            // of re-triggering Retrieve, which is not what this scenario is testing (and matches
            // the issue's own observation that the LLM tries a different tool each cycle).
            let mandate_tools = ["list_directory", "grep", "glob"];
            let mandate_queries = ["query A", "query B", "query C"];
            for (i, tool) in mandate_tools.iter().enumerate() {
                let call_id = format!("call-mandate-{i}");
                let round = vec![zeph_llm::provider::ToolUseRequest {
                    id: call_id.clone(),
                    name: (*tool).to_owned().into(),
                    input: serde_json::json!({"path": "."}),
                }];
                agent.handle_native_tool_calls(None, &round).await.unwrap();
                let (content, _) = tool_result_content(&agent, &call_id)
                    .unwrap_or_else(|| panic!("{call_id} must have a ToolResult"));
                assert!(
                    content.contains("[skipped]"),
                    "mandate round {i} ({tool}) must be skipped by the Retrieve rule, got: {content}"
                );

                let search_id = format!("call-search-{i}");
                let search_round = vec![zeph_llm::provider::ToolUseRequest {
                    id: search_id.clone(),
                    name: "memory_search".to_owned().into(),
                    input: serde_json::json!({"query": mandate_queries[i]}),
                }];
                agent
                    .handle_native_tool_calls(None, &search_round)
                    .await
                    .unwrap();
                let (search_content, search_is_error) = tool_result_content(&agent, &search_id)
                    .unwrap_or_else(|| panic!("{search_id} must have a ToolResult"));
                assert!(
                    !search_is_error && search_content.contains("[cancelled by user]"),
                    "declined memory_search must be a successful cancellation, got: \
                     {search_content}"
                );
                assert_eq!(
                    agent.tool_orchestrator.last_tool_error.get("memory_search"),
                    None,
                    "a declined confirmation must self-heal last_tool_error back to None \
                     (round {i})"
                );
            }

            // A 4th distinct tool: the circuit breaker (not the confirmation-outcome bypass,
            // which self-healed away 3 rounds ago) must now let it dispatch directly.
            let round_final = vec![zeph_llm::provider::ToolUseRequest {
                id: "call-final".to_owned(),
                name: "find_path".to_owned().into(),
                input: serde_json::json!({"pattern": "*.rs"}),
            }];
            agent
                .handle_native_tool_calls(None, &round_final)
                .await
                .unwrap();
            let (content_final, is_error_final) = tool_result_content(&agent, "call-final")
                .expect("call-final must have a ToolResult");
            assert!(
                content_final.contains("src/main.rs") && !is_error_final,
                "after 3 consecutive mandate+decline cycles, the 4th distinct tool must dispatch \
                 directly via the circuit breaker, got: {content_final}"
            );
        }
    }
}