supercode/

agent.rs

//! The agent loop.

use std::collections::HashSet;

use crate::config::{CachePlan, Config, SteeringMode, ToolAdvertising};
use crate::error::{Error, Result};
use crate::event::AgentEvent;
use crate::message::{ChatMessage, Role};
use crate::provider::{self, ChatRequest, OpenAiProvider, Provider, ToolSchema};
use crate::reduce::{self, ReductionLog, ReductionPolicy};
use crate::session::Session;
use crate::sidecar::SidecarWriter;
use crate::tools::{ToolContext, ToolRegistry};

/// Tool name of the `tool_search` agent intrinsic (B6). Never a registered
/// [`crate::tools::Tool`] — intercepted in [`Agent::run_tool`] before registry
/// lookup, so it works under any [`ToolAdvertising`] mode.
const TOOL_SEARCH: &str = "tool_search";

/// Tool name of the `expand_reduction` agent intrinsic (T12/TR-1) — the
/// model-invocable rehydration counterpart to `tool_search`, same
/// interception pattern. Advertised whenever a [`ReductionPolicy`] is
/// installed, regardless of [`ToolAdvertising`] mode (see [`Self::tool_schemas`]).
const EXPAND_REDUCTION: &str = "expand_reduction";

/// Tool name of the `sidecar_search` agent intrinsic (T12/TR-1).
const SIDECAR_SEARCH: &str = "sidecar_search";

/// Tool name of the `spawn_subagent` agent intrinsic (P5-3, §2 module 9 D1
/// "spawn tool"). Same interception pattern as [`TOOL_SEARCH`] — never a
/// registered [`crate::tools::Tool`], intercepted in [`Agent::run_tool`]
/// before registry lookup — but ALSO needs full `&mut self` async access
/// (running a whole child agent loop, or `tokio::spawn`-ing one), which
/// [`Agent::prepare_tool_call`]'s purely-synchronous intrinsics don't, so
/// the interception point is `Self::run_tool`'s top, not
/// `prepare_tool_call`.
const SPAWN_SUBAGENT: &str = "spawn_subagent";

/// Claude Code's native name for [`SPAWN_SUBAGENT`]. It is exposed only when
/// `Config::subagents_claude_agent_alias` is enabled for a Claude import.
const CLAUDE_AGENT: &str = "Agent";

/// Claude Code spellings for core filesystem/shell tools. Imported Claude
/// context frequently continues to call these names even when another model
/// is driving the turn, so emulation must translate execution as well as
/// preserve the original call/result names in the transcript.
const CLAUDE_BASH: &str = "Bash";
const CLAUDE_READ: &str = "Read";
const CLAUDE_WRITE: &str = "Write";
const CLAUDE_EDIT: &str = "Edit";
const CLAUDE_GLOB: &str = "Glob";
const CLAUDE_GREP: &str = "Grep";

/// Claude Code scheduler compatibility intrinsics. They edit an imported
/// [`crate::ClaudeRuntimeManifest`]; actual timer execution belongs to an
/// embedding scheduler driver, never this agent loop.
const CLAUDE_CRON_CREATE: &str = "CronCreate";
const CLAUDE_CRON_DELETE: &str = "CronDelete";
const CLAUDE_CRON_LIST: &str = "CronList";
const CLAUDE_SCHEDULE_WAKEUP: &str = "ScheduleWakeup";

/// Shared SDK steering mailbox. `accepting` and `queue` share one lock so a
/// turn's final boundary can close acceptance atomically with its last drain;
/// a steer can therefore never be acknowledged into the following turn.
#[derive(Default)]
pub(crate) struct SteerInbox {
    queue: std::collections::VecDeque<QueuedSteer>,
    accepting: bool,
}

struct QueuedSteer {
    message: String,
    sdk_bound: bool,
}

impl SteerInbox {
    pub(crate) fn open(&mut self) {
        self.queue.clear();
        self.accepting = true;
    }

    pub(crate) fn enqueue(&mut self, message: String) -> bool {
        if !self.accepting {
            return false;
        }
        self.queue.push_back(QueuedSteer {
            message,
            sdk_bound: true,
        });
        true
    }

    pub(crate) fn close(&mut self) {
        self.accepting = false;
        self.queue.retain(|queued| !queued.sdk_bound);
    }

    fn drain(&mut self, mode: SteeringMode) -> Option<String> {
        if self.queue.is_empty() {
            return None;
        }
        match mode {
            SteeringMode::All => Some(
                self.queue
                    .drain(..)
                    .map(|queued| queued.message)
                    .collect::<Vec<_>>()
                    .join("\n\n"),
            ),
            SteeringMode::OneAtATime => self.queue.pop_front().map(|queued| queued.message),
        }
    }

    fn drain_or_close(&mut self, mode: SteeringMode) -> Option<String> {
        if !self.queue.iter().any(|queued| queued.sdk_bound) {
            self.accepting = false;
            return None;
        }
        self.drain(mode)
    }

    fn queue_unchecked(&mut self, message: String) {
        self.queue.push_back(QueuedSteer {
            message,
            sdk_bound: false,
        });
    }

    fn len(&self) -> usize {
        self.queue.len()
    }
}

struct SteerTurnGuard {
    inbox: std::sync::Arc<std::sync::Mutex<SteerInbox>>,
}

impl SteerTurnGuard {
    fn new(inbox: std::sync::Arc<std::sync::Mutex<SteerInbox>>) -> Self {
        inbox
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .accepting = true;
        Self { inbox }
    }
}

impl Drop for SteerTurnGuard {
    fn drop(&mut self) {
        self.inbox
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .close();
    }
}

/// Tool name of the `subagent_status` agent intrinsic (P5-3, D3
/// "background+resume"): poll (and reap, once finished) a background child
/// spawned via [`SPAWN_SUBAGENT`]. Only advertised when
/// `Config::subagents_background` is on (see [`Agent::tool_schemas`]).
const SUBAGENT_STATUS: &str = "subagent_status";

/// Tool name of the `background_exec` agent intrinsic (P5-6, §2 module 4
/// `tools.background` D1 "background exec"). Unlike [`SPAWN_SUBAGENT`], this
/// needs no async child-agent loop — spawning a process
/// (`tokio::process::Command::spawn`) is itself synchronous — so, like
/// [`TOOL_SEARCH`], it is intercepted in [`Agent::prepare_tool_call`], not
/// [`Agent::run_tool`].
const BACKGROUND_EXEC: &str = "background_exec";

/// Tool name of the `background_status` agent intrinsic (P5-6, D1 "monitor/
/// event feed"): poll a background job's run status, drain its newly
/// captured output as an [`AgentEvent::BackgroundOutput`] event, and reap it
/// (remove it from [`Agent::background_jobs`]) once it has exited or been
/// killed.
const BACKGROUND_STATUS: &str = "background_status";

/// Tool name of the `background_list` agent intrinsic (P5-6, D10
/// "bg-manager"): list every background job this agent is currently
/// tracking (running or finished-but-unreaped), without draining output or
/// reaping anything.
const BACKGROUND_LIST: &str = "background_list";

/// Tool name of the `background_kill` agent intrinsic (P5-6, D10
/// "bg-manager"): kill a background job's real OS process
/// (`tokio::process::Child::start_kill`) and reap it immediately.
const BACKGROUND_KILL: &str = "background_kill";

/// P4e (§3.1 `core.parallel_tool_calls`): the synchronous outcome of
/// [`Agent::prepare_tool_call`] — either a result already in hand (an
/// intrinsic, or a call refused before it ever reached `Tool::execute`), or
/// a plain registry-tool call ready for the (possibly concurrent) async
/// `execute()` step.
enum PreparedCall {
    /// A final `(output, is_error)` result — no `Tool::execute` call is
    /// coming for this one.
    Done((String, bool)),
    /// Passed every synchronous check; `execute(args, &ctx)` on the named
    /// registry tool is the only remaining step.
    Ready {
        name: String,
        args: serde_json::Value,
    },
}

/// Marker prefix of the notice [`Agent::cap_tool_output`] appends to an
/// oversized tool result kept in `history` (the recorder receives the full
/// output BEFORE capping, so the sidecar never carries this). Only ever
/// applied when the D6/A7 supersession gate (`Self::run_loop`'s tool-result
/// push site) leaves capping ON — i.e. no recorder+policy pair is active, or
/// (legacy sidecars only) a reduction was minted before this gate existed.
/// Shared with `reduce::rehydrate`, whose capped-copy detection uses it to
/// decide when the recorded (sidecar) copy of a message provably supersedes
/// the capped history copy — keep the two in sync by construction, not by
/// convention.
pub(crate) const CAP_NOTICE_MARKER: &str = "\n\n[supercode: tool output truncated — ";

/// UX-26 (B7-warn): current wall-clock time as unix milliseconds, the same
/// unit [`crate::sidecar::rfc3339_to_ms`] parses session timestamps into —
/// lets [`Agent::build_request_messages`] compare "now" against a
/// cross-process signal (a loaded session's last message timestamp) on
/// equal footing with an in-process one (this agent's own last annotated
/// send). Saturates to 0 on a pre-epoch clock rather than panicking (never
/// happens on real hardware, but `duration_since` can theoretically error).
fn now_ms() -> i64 {
    std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|d| d.as_millis() as i64)
        .unwrap_or(0)
}

/// P5-3: process-wide sequence number backing [`next_subagent_id`] —
/// disambiguates two spawns landing in the same millisecond (which
/// `now_ms()` alone cannot).
static SUBAGENT_ID_SEQ: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);

/// P5-3: a fresh, process-unique child agent id (`"agent-<hex-ts>-<hex-seq>"`
/// — the native analog of Claude Code's `agent-<id>` naming, see
/// `crate::session::SessionMeta::agent_id`'s doc comment).
fn next_subagent_id() -> String {
    let seq = SUBAGENT_ID_SEQ.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
    format!("agent-{:x}-{:x}", now_ms(), seq)
}

/// P5-4: the shape [`Agent::child_approval_handler_factory`]/
/// [`Agent::set_child_approval_handler_factory`] share — factored into its
/// own alias (clippy `type_complexity`) rather than spelled out inline at
/// both use sites.
type ChildApprovalHandlerFactory = dyn Fn(
        String,
        std::sync::Arc<std::sync::Mutex<Vec<crate::subagents::QueuedApproval>>>,
    ) -> std::sync::Arc<dyn crate::permissions::PermissionsApprovalHandler>
    + Send
    + Sync;

/// A stateful agent: configuration, a model transport, a tool set, and the
/// running conversation. Drive it with [`Agent::send`].
pub struct Agent {
    config: Config,
    provider: std::sync::Arc<dyn Provider>,
    registry: ToolRegistry,
    history: Vec<ChatMessage>,
    ctx: ToolContext,
    /// Cumulative output (completion) tokens across every `send` on this agent.
    total_output_tokens: u64,
    /// Names of non-core tools discovered via `tool_search` (B6): advertised
    /// starting with the *next* request once populated.
    activated_tools: HashSet<String>,
    /// The live sidecar writer (A3), if this agent is recording. `None` is
    /// today's behavior, at zero cost: every append point becomes a no-op.
    recorder: Option<SidecarWriter>,
    /// Reversible reduction policy (A5/A7/A10). `None` is today's behavior,
    /// at zero cost: every provider request is built from `self.history`
    /// verbatim, exactly as before this landed.
    reduction_policy: Option<ReductionPolicy>,
    /// The accumulating reduction log (A5): fed back into
    /// [`reduce::project_messages`] on every request-build so already-applied
    /// reductions reproduce verbatim across turns and `send` calls (prefix
    /// stability). `history` itself is never touched by this — see
    /// [`Self::run_loop`].
    reduction_log: ReductionLog,
    /// B7: length of the stable, byte-identical-across-turns prefix at the
    /// front of [`Self::history`] — this agent's own system message plus
    /// every message of a previously-imported session — set by
    /// [`Self::load_session`]. `None` (the default) means no session has been
    /// loaded, so [`crate::provider::apply_cache_plan`] has nothing to
    /// annotate even under [`CachePlan::ImportedPrefix`].
    imported_prefix_len: Option<usize>,
    /// TR-7 (T20): the injectable side-call ([`reduce::summarize::SpanSummarizer`])
    /// used to summarize an A10 `TurnsCleared` span, if one is installed
    /// ([`Self::set_span_summarizer`]). `None` is today's behavior, at zero
    /// cost: [`Self::build_request_messages`] never calls
    /// [`reduce::prepare_cleared_turns_summary`] without one, so
    /// `policy.summarize_cleared_turns` being on with no summarizer
    /// installed behaves exactly like it being off (deterministic stub only)
    /// — never a panic, never a blocked request.
    span_summarizer: Option<std::sync::Arc<dyn reduce::summarize::SpanSummarizer + Send + Sync>>,
    /// TR-8 (T5): the tool-schema tier signature (global knob + per-tool
    /// overrides) as of the last request this agent built, or `None` before
    /// the first request. Compared against the CURRENT signature at the top
    /// of every [`Self::build_request_messages`] call so a tier change made
    /// mid-session (via [`Self::set_schema_tier`] /
    /// [`Self::set_tool_schema_tier`]) is detected and flagged to the B7
    /// cache planner as a cache-bust event (`provider::tier_change_is_cache_bust`).
    last_tool_schema_tier_signature: Option<u64>,
    /// PARITY-18 D4 — the target model's context-window size, if the caller
    /// has armed the guard via [`Self::set_context_limit`]. `None` (the
    /// default) means no guard: every request is sent unconditionally.
    /// CLI entry points arm it for their resolved model; direct SDK callers
    /// retain explicit control through [`Self::set_context_limit`].
    /// Once set, [`Self::run_loop`] re-checks
    /// [`crate::tokens::context_guard`] before EVERY request it builds —
    /// not just the first — so "never sends an over-context request" holds
    /// for the whole session, not only a one-shot preflight.
    context_limit: Option<u64>,
    /// PARITY-18 D3 — becomes `true` the first time [`Self::run_loop`]
    /// actually reaches its real send site (immediately before
    /// [`Provider::complete`]). Exposed via [`Self::request_issued`] so a
    /// caller can report "request sent" truthfully — never asserted ahead
    /// of time, so a pre-delivery failure (guard refusal, a build error) or
    /// an interactive session that quits before any turn completes is
    /// reported honestly as "not sent".
    requests_issued: bool,
    /// UX-26 (B7-warn): unix-ms wall-clock time this agent last knew the
    /// active [`CachePlan::ImportedPrefix`] breakpoint to be warm. Seeded by
    /// [`Self::load_session`] from the just-loaded session's OWN last
    /// message timestamp (`metadata["timestamp"]`, parsed via
    /// [`crate::sidecar::rfc3339_to_ms`]) — a cross-process signal: how long
    /// the resumed conversation has sat idle since ANY tool last touched it,
    /// which is exactly when Anthropic's server-side cache entry (if one
    /// ever existed) was last capable of being warm. Refreshed to "now"
    /// every time [`Self::run_loop`] actually sends a cache-annotated
    /// request (an in-process signal: idle time between this agent's own
    /// turns). `None` when no imported prefix exists yet, or the loaded
    /// session's last message carries no parseable timestamp — never
    /// guessed, so the TTL check in [`provider::cache_cold_reason`] simply
    /// doesn't fire rather than risk a false positive.
    last_cache_activity_ms: Option<i64>,
    /// UX-26: whether a PRIOR request already carried a cache_control
    /// annotation for the current [`Self::imported_prefix_len`] — i.e.
    /// whether reuse is genuinely "expected" on the NEXT annotated request.
    /// `false` until the first annotated request goes out (that one is
    /// establishing the cache entry, a legitimate write, never a "miss") and
    /// reset to `false` by [`Self::load_session`] whenever the imported
    /// prefix itself changes.
    cache_established: bool,
    /// UX-26 scratch: this turn's cache-warmth context, computed once at the
    /// top of [`Self::build_request_messages`] (before the request is sent,
    /// while `effective_cache_plan`/`busted` are in scope) and consumed once
    /// in [`Self::run_loop`] right after `usage` comes back — never read
    /// across turns, so a stale value can't leak. `(will_annotate,
    /// cache_established, idle_secs)` — see [`provider::cache_cold_reason`]
    /// for what each of the first two independently gates.
    pending_cache_turn: (bool, bool, Option<i64>),
    /// P4b: the injectable auto-title side-call ([`Self::set_session_titler`]),
    /// mirroring [`Self::span_summarizer`]'s "installing one alone changes
    /// nothing" contract — `Config::auto_title` is the actual gate a caller
    /// consults before invoking [`Self::auto_title`].
    session_titler: Option<std::sync::Arc<dyn crate::session_title::SessionTitler + Send + Sync>>,
    /// P4b (§1.6, catalog §4a "persisted per-turn usage records"): every
    /// [`crate::usage_log::UsageRecord`] recorded so far this agent's
    /// lifetime. Always accumulated (cheap, small) regardless of whether a
    /// caller ever persists it — see [`Self::usage_records`]/
    /// [`Self::save_usage_log`].
    usage_log: Vec<crate::usage_log::UsageRecord>,
    /// P4b: 0-based index of the NEXT model round-trip, for
    /// [`crate::usage_log::UsageRecord::turn`].
    turn_index: usize,
    /// P4b (§1.7/§3.1 `core.steering`, pi§3 semantics): queued mid-turn
    /// steering messages — drained at the top of [`Self::run_loop`]'s next
    /// iteration (pi's "steer = after current tool calls"). Empty by
    /// default, at zero cost: [`Self::run_loop`] skips the drain entirely
    /// when empty.
    steer_queue: std::sync::Arc<std::sync::Mutex<SteerInbox>>,
    /// P4b: queued follow-up messages — drained only once the loop is
    /// otherwise idle (pi's "follow-up = at idle"), i.e. exactly the point
    /// [`Self::run_loop`] would otherwise return a final answer.
    follow_up_queue: std::collections::VecDeque<String>,
    /// P4c (§5.2 P4 "doom-loop breaker", §3.1 `core.doom_loop_threshold`):
    /// `(tool name, canonical JSON args)` of the most recent tool call, if
    /// [`Config::doom_loop_threshold`] is armed — `None` before the first
    /// call this agent has run. See [`Self::check_doom_loop`].
    doom_loop_last_call: Option<(String, String)>,
    /// P4c: how many times [`Self::doom_loop_last_call`] has repeated
    /// consecutively so far (starts at 1 on the call that SET it).
    doom_loop_streak: u32,
    /// P4c (§1.10/§3.1 `core.model_switch.allow_switch`): every
    /// [`crate::model_change::ModelChangeRecord`] [`Self::switch_model`] has
    /// created so far this agent's lifetime. Always empty when
    /// `Config::model_switch_allow_switch` is off (the default) or no
    /// switch has happened yet.
    model_change_log: Vec<crate::model_change::ModelChangeRecord>,
    /// P4e (§1.6/§3.1 `core.session.git_metadata`, catalog:331): captured
    /// once at construction when [`Config::session_git_metadata`] is on;
    /// `None` when the gate is off (the default) or the best-effort git
    /// probe found nothing (not a repo, `git` missing). See
    /// [`Self::git_metadata`]/[`Self::save_git_metadata`].
    git_metadata: Option<crate::git_metadata::GitMetadataRecord>,
    /// P5-1 (§2.10, session-scoped "approve for session" cache): populated
    /// only when a [`crate::permissions::PermissionsApprovalHandler`]
    /// returns [`crate::permissions::ApprovalOutcome::AllowForSession`] —
    /// see [`Self::prepare_tool_call`]'s `Config::permissions_enabled`
    /// branch. Always constructed (cheap, empty) regardless of whether the
    /// engine is ever active — the same "zero cost when off" posture as
    /// [`Self::doom_loop_last_call`].
    permissions_approval_cache: crate::permissions::ApprovalCache,
    /// P5-1: the non-interactive decision seam a caller installs via
    /// [`Self::set_permissions_approval_handler`] — mirrors
    /// [`Self::span_summarizer`]/[`Self::session_titler`]'s "installing one
    /// alone changes nothing, `Config::permissions_enabled` is the actual
    /// gate" pattern. `None` (the default) means every `Ask`-tier decision
    /// is denied (fail-closed — see
    /// `crate::permissions::approval::PermissionsApprovalHandler`'s doc
    /// comment).
    permissions_approval_handler:
        Option<std::sync::Arc<dyn crate::permissions::PermissionsApprovalHandler>>,
    /// P5-2 (§2 module 15 D7 row 4 "prompts-as-commands"): MCP server
    /// prompts registered via [`Self::register_mcp_prompt`], keyed by their
    /// ALREADY-NAMESPACED command name (`mcp__<server>__<prompt>` — see
    /// [`crate::mcp::McpPromptSource`]'s doc comment for why that namespace
    /// is what keeps an untrusted server's prompt from ever colliding with
    /// a trusted `Config::prompts` entry). Empty by default, at zero cost:
    /// [`Self::expand_prompt_async`] only consults this after
    /// `Config::prompts` finds no match.
    mcp_prompts: std::collections::HashMap<String, Box<dyn crate::sdk::SdkPromptSource>>,
    /// P5-3 (§2 module 9): how deep in the spawn tree THIS agent is — `0`
    /// for a top-level agent. Set from [`Config::subagent_depth`] at
    /// construction; `Self::run_spawn_subagent` builds a child `Config`
    /// with `subagent_depth = self.subagent_depth + 1` and ALSO overwrites
    /// the freshly-built child `Agent`'s own field to match (belt-and-
    /// suspenders — the child never has to trust its own `Config` alone).
    subagent_depth: usize,
    /// P5-3 (resource bound, "must not fork-bomb"): the shared, tree-wide
    /// concurrency gauge every spawn (this agent's own, and every
    /// descendant's) increments/decrements against
    /// (`crate::subagents::try_acquire`/`ConcurrencyGuard`). A TOP-level
    /// agent gets a fresh `Arc::new(AtomicUsize::new(0))` at construction;
    /// `Self::run_spawn_subagent` clones this SAME `Arc` into every child it
    /// spawns (never a fresh one), so a cap of N holds across the WHOLE
    /// tree regardless of its branching shape — a parent with 3 children
    /// each spawning 3 more shares one counter, not nine independent ones.
    subagent_concurrency_gauge: std::sync::Arc<std::sync::atomic::AtomicUsize>,
    /// P5-3 (D3 "background+resume"): background subagents this agent has
    /// spawned and not yet reaped via `subagent_status`, keyed by their
    /// `child_agent_id`. Each entry's `JoinHandle` moves its own
    /// [`crate::subagents::ConcurrencyGuard`] into the spawned task, so the
    /// concurrency slot is held for exactly as long as the child is
    /// actually running, independent of whether/when the parent polls.
    background_subagents: std::collections::HashMap<String, BackgroundSubagent>,
    /// P5-3 (§2.2 C6 "parent-surfaced queue"): approval requests a
    /// `background_prompts = "parent"` child raised, queued here rather
    /// than blocking (see [`crate::subagents::QueuedApproval`]'s doc
    /// comment — each is already resolved `Deny` by the time it lands
    /// here). Exposed read-only via [`Self::pending_child_approvals`].
    /// Always constructed (cheap, empty) regardless of whether background
    /// spawning is ever used, same "zero cost when off" posture as
    /// [`Self::permissions_approval_cache`].
    pending_child_approvals:
        std::sync::Arc<std::sync::Mutex<Vec<crate::subagents::QueuedApproval>>>,
    /// P5-4 (tui, closes the P5-3 §2.2 C6 deferred chain — see
    /// [`crate::subagents::ParentQueueApprovalHandler`]'s doc comment for
    /// the "never blocks" contract this OVERRIDES only when a factory is
    /// installed): when `Some`, [`Self::run_spawn_subagent`] uses THIS
    /// factory — instead of constructing the default never-blocking
    /// [`crate::subagents::ParentQueueApprovalHandler`] — to build the
    /// `PermissionsApprovalHandler` a `background_prompts = "parent"`
    /// child gets. Installed via
    /// [`Self::set_child_approval_handler_factory`] by a `tui` embedder
    /// that wants queued child approvals to be genuinely ANSWERABLE
    /// (blocks the child's tool call until the parent resolves it, or
    /// denies if the factory's handler's channel is ever dropped/closed —
    /// still fail-closed, never a hang past process lifetime). `None` (the
    /// default) preserves P5-3's shipped behavior byte-for-byte: every
    /// `background_prompts = "parent"` child still gets the immediate-deny
    /// `ParentQueueApprovalHandler`, and [`Self::pending_child_approvals`]
    /// stays exactly the read-only audit view it already is.
    child_approval_handler_factory: Option<std::sync::Arc<ChildApprovalHandlerFactory>>,
    /// P5-3 (D5 "subagent transcripts… persisted + linked"): an optional
    /// `(store, this agent's own session name)` pair installed via
    /// [`Self::set_subagent_store`] — mirrors [`Self::set_recorder`]/
    /// [`Self::set_span_summarizer`]'s "installing one alone changes
    /// nothing" pattern. `None` (the default) means a spawned child's
    /// transcript/lineage is still joined back into THIS agent's context
    /// (the foreground/background mechanics work either way) but nothing
    /// is written to a [`crate::store::SessionStore`] — no behavior change
    /// for any caller that never installs one (e.g. every pre-P5-3 caller).
    subagent_store: Option<(std::sync::Arc<crate::store::SessionStore>, String)>,
    /// Imported Claude runtime state. The manifest can be paused or active,
    /// but this Agent contains no scheduler or timer handle; an embedding
    /// driver owns execution and persistence.
    claude_runtime_manifest: Option<crate::claude_runtime_state::ClaudeRuntimeManifest>,
    /// P5-6 (§2 module 4 `tools.background`, D10 "bg-manager"): background
    /// OS processes spawned via `background_exec`, keyed by job id, tracked
    /// until reaped (a terminal `background_status` poll, or an explicit
    /// `background_kill`) — see [`BackgroundJob`]'s doc comment. Always
    /// constructed (cheap, empty), same "zero cost when off" posture as
    /// [`Self::background_subagents`].
    background_jobs: std::collections::HashMap<String, BackgroundJob>,
    /// P5-6 (resource bound, mirroring [`Self::subagent_concurrency_gauge`]'s
    /// own precedent): the shared concurrency gauge every `background_exec`
    /// call on this agent increments/decrements against
    /// (`crate::subagents::try_acquire`/`ConcurrencyGuard` — reused
    /// verbatim, a second independent gauge instance scoped to background
    /// JOBS rather than subagent SPAWNS).
    background_concurrency_gauge: std::sync::Arc<std::sync::atomic::AtomicUsize>,
    /// P5-9 (§2 module 20 `checkpoint`): the write-path-interception
    /// observer installed on [`Self::ctx`]'s `write_observer` (as a
    /// `dyn WriteObserver`), held here ADDITIONALLY as its concrete type so
    /// [`Self::run_loop`] can call
    /// [`crate::checkpoint::CheckpointObserver::begin_turn`] once per turn.
    /// `None` when `Config::checkpoint_enabled` is `false` (the default) or
    /// the shadow store failed to open — see
    /// [`crate::checkpoint::observer_for_config`].
    checkpoint_observer: Option<std::sync::Arc<crate::checkpoint::CheckpointObserver>>,
    /// P5-11 (§2 module 28 `lsp`): the LSP server registry installed (via
    /// `crate::lsp::LspDiagnosticsObserver`) on `Self::ctx`'s
    /// `write_observer` chain, held here ADDITIONALLY as its concrete type
    /// so `impl Drop for Agent` can reach
    /// [`crate::lsp::LspManager::kill_all_sync`] (no orphaned language-
    /// server processes) and a clean-exit caller can reach
    /// [`crate::lsp::LspManager::shutdown_all`] for a graceful handshake.
    /// `None` when `Config::lsp_enabled` is `false` (the default).
    lsp_manager: Option<std::sync::Arc<crate::lsp::LspManager>>,
}

/// P5-6 (§2 module 4 `tools.background`): one background-spawned OS process
/// this agent is tracking, awaiting a `background_status`/`background_list`
/// poll (or `background_kill`/agent drop) to reap or terminate it.
///
/// **Real process, not a child agent.** Unlike [`BackgroundSubagent`] (which
/// wraps a whole recursive child [`Agent`] loop against the SAME mock/real
/// provider), this wraps a plain OS subprocess spawned via
/// [`crate::tools::build_sandboxed_sh`] — the exact function
/// [`crate::tools::BashTool::execute`] itself calls, so a background
/// command gets byte-identical sandboxing/cwd/env handling to a foreground
/// `bash` call (build brief: "reuse the bash tool's execution + sandbox
/// path").
struct BackgroundJob {
    /// The live process handle — kept directly on the job (not moved into a
    /// spawned task) so [`Agent::run_background_status`]/
    /// [`Agent::run_background_list`] can call the SYNCHRONOUS,
    /// non-blocking `Child::try_wait` to observe exit status, and
    /// [`Agent::run_background_kill`]/[`impl Drop for Agent`] can call the
    /// SYNCHRONOUS `Child::start_kill` for a REAL process kill — never just
    /// a `tokio::task::JoinHandle::abort` (which would only cancel a Rust
    /// future, not the OS process it spawned). `kill_on_drop(true)` was set
    /// at spawn time as defense-in-depth: even a `BackgroundJob` dropped
    /// through some path OTHER than the explicit kill call sites below
    /// still kills its child (a documented tokio behavior; a no-op if the
    /// process already exited).
    child: tokio::process::Child,
    /// The exact command text this job is running — the SAME text that was
    /// already checked against the permissions engine at spawn time (see
    /// [`Agent::background_permission_denial`]).
    command: String,
    /// The OS process id, captured once at spawn time (before `child` is
    /// ever mutated) — surfaced in every status/list/kill result, and the
    /// only thing an OUTSIDE observer (e.g. a test proving real
    /// termination) needs to check liveness independent of this process's
    /// own bookkeeping.
    pid: Option<u32>,
    /// Bounded, incrementally-appended combined stdout+stderr capture —
    /// written to by the reader tasks [`Agent::run_background_exec`] spawns
    /// right after `child.stdout`/`child.stderr` are taken, read by every
    /// status/list poll. Shared via `Arc` since the reader tasks outlive
    /// this method call.
    output: std::sync::Arc<crate::background::CapturedOutput>,
    /// Unix-ms wall-clock time the spawn happened.
    started_at_ms: i64,
    /// Set by [`Agent::run_background_kill`] — [`Agent::run_background_status`]/
    /// [`Agent::run_background_list`] report [`crate::background::JobStatus::Killed`]
    /// unconditionally once this is `true`, rather than racing
    /// `Child::try_wait` to see whether the kill signal has landed yet.
    killed: bool,
    /// The concurrency-gauge slot this job holds for as long as it remains
    /// in [`Agent::background_jobs`] — dropped (freeing the slot) when this
    /// `BackgroundJob` is removed from the map (a terminal reap, or an
    /// explicit kill), exactly mirroring [`BackgroundSubagent`]'s own
    /// "guard held for as long as it's tracked, not just while the process
    /// is alive" posture (§2 module 9 precedent, kept consistent here).
    _guard: crate::subagents::ConcurrencyGuard,
}

/// P5-6: the non-blocking status read [`Agent::run_background_status`]/
/// [`Agent::run_background_list`] share — `job.killed` (set by
/// [`Agent::run_background_kill`]) always wins over a fresh `try_wait`,
/// since a kill signal racing the OS reaping the process is otherwise
/// indistinguishable from "still running" for one poll cycle; reporting
/// `Killed` unconditionally once requested avoids that race entirely. A
/// `try_wait` error (would only happen if this job's id were somehow
/// double-reaped, which the map ownership below already prevents) is
/// treated as "no news yet" — `Running` — rather than inventing a made-up
/// exit code.
fn background_job_status(job: &mut BackgroundJob) -> crate::background::JobStatus {
    if job.killed {
        return crate::background::JobStatus::Killed;
    }
    match job.child.try_wait() {
        Ok(Some(status)) => crate::background::JobStatus::Exited(status.code()),
        Ok(None) | Err(_) => crate::background::JobStatus::Running,
    }
}

/// Fable-5 review (HIGH, "grandchildren orphaned on kill AND agent-drop"):
/// the shared real-kill body for both [`Agent::run_background_kill`] and
/// `impl Drop for Agent` — sends `SIGKILL` to `job`'s ENTIRE process group,
/// not just the one directly-tracked pid, so a surviving `&` job, pipeline
/// stage, or double-forking daemon spawned by the job is killed too, then
/// reaps the group leader so it doesn't linger as a zombie.
///
/// Relies on the spawn site (`Agent::run_background_exec`) having put the
/// job in its OWN new process group via `Command::process_group(0)` — which
/// makes the leader's pgid equal to its own pid, so `job.pid` doubles as the
/// group id here.
#[cfg(unix)]
fn kill_job_process_group(job: &mut BackgroundJob) {
    if let Some(pid) = job.pid {
        // SAFETY: `libc::kill` with a negative pid is `killpg` — it only
        // ever sends a signal (never dereferences memory), so this is safe
        // regardless of whether the group is still alive. A `-1`/`ESRCH`
        // return means the leader (and thus the whole group, since a group
        // can't outlive its leader) already exited — not an error, just
        // "already dead", exactly like `Child::start_kill`'s own documented
        // no-op-on-already-exited contract.
        unsafe {
            libc::kill(-(pid as libc::pid_t), libc::SIGKILL);
        }
    }
    // Belt-and-suspenders for the leader itself — `kill_on_drop(true)` set
    // at spawn time is the same outcome via a different (implicit) path —
    // then reap it so the SIGKILL we just delivered doesn't leave a zombie
    // behind.
    let _ = job.child.start_kill();
    let _ = job.child.try_wait();
}

/// Non-unix fallback: no portable process-group primitive is wired up here
/// (same posture as [`crate::tools::build_sandboxed_sh`]'s own platform
/// split) — falls back to the pre-fix per-child kill. A background job that
/// spawns a surviving grandchild process on a non-Unix target is a
/// documented residual, not silently claimed fixed by this cfg arm.
#[cfg(not(unix))]
fn kill_job_process_group(job: &mut BackgroundJob) {
    let _ = job.child.start_kill();
}

/// P5-6 (D1 "monitor/event feed", "output captured incrementally +
/// BOUNDED"): spawn a fire-and-forget reader task that continuously drains
/// `reader` (a piped `ChildStdout`/`ChildStderr`) into `output`, bounded at
/// `cap` bytes. Reading NEVER stops at the cap — only what's RETAINED is
/// bounded ([`crate::background::CapturedOutput::append`]'s own contract)
/// — because a background job's child process would otherwise block
/// forever writing to a full, undrained OS pipe once this stopped reading
/// it, silently hanging real work behind an apparently-"running" job. The
/// task exits on its own once the pipe reaches EOF (the process closed the
/// descriptor, whether by exiting or being killed) — no explicit
/// abort/cleanup call site is needed; a detached `tokio::spawn` this short-
/// lived is not the kind of orphaned-task risk `impl Drop for Agent`'s own
/// doc comment is about (that one concerns a whole recursive provider-
/// calling child AGENT loop, not a bounded byte-copy loop that ends the
/// instant its source pipe closes).
fn spawn_output_reader<R>(
    reader: R,
    output: std::sync::Arc<crate::background::CapturedOutput>,
    cap: usize,
) -> tokio::task::JoinHandle<()>
where
    R: tokio::io::AsyncRead + Unpin + Send + 'static,
{
    tokio::spawn(async move {
        use tokio::io::AsyncReadExt;
        let mut reader = reader;
        let mut buf = [0u8; 8192];
        loop {
            match reader.read(&mut buf).await {
                Ok(0) => break,
                Ok(n) => {
                    let chunk = String::from_utf8_lossy(&buf[..n]);
                    output.append(&chunk, cap);
                }
                Err(_) => break,
            }
        }
    })
}

/// P5-3: one background-spawned child this agent is tracking, awaiting a
/// `subagent_status` poll (or agent drop) to reap it.
struct BackgroundSubagent {
    /// Resolves to `(child_agent_id, child's final result, the child's own
    /// post-system-prompt history — for D5 transcript persistence once
    /// reaped)` — the concurrency-guard slot for this child is held INSIDE
    /// the spawned future (moved in at spawn time), so it releases the
    /// instant the child's own run loop finishes, not when the parent gets
    /// around to polling.
    handle: tokio::task::JoinHandle<(String, Result<String>, Vec<ChatMessage>)>,
    /// The task/prompt text the child was spawned with (surfaced by a
    /// `"pending"` status poll, since the handle alone can't answer "what
    /// is it doing").
    task: String,
    /// The named `agent_type` spawned, if any.
    agent_type: Option<String>,
    /// Unix-ms wall-clock time the spawn happened.
    started_at_ms: i64,
}

/// P5-3 safety hardening (Fable-5 review, MEDIUM-LOW "orphaned billed
/// spend"): a dropped parent must not leave a detached background child
/// running against a REAL provider. Without this, a parent dropped
/// mid-run (the caller's own process exits the scope, panics, or simply
/// stops polling) leaves every still-running [`BackgroundSubagent::handle`]
/// as an orphaned `tokio::spawn` task: nothing had ever awaited or
/// aborted it, so it runs to its own (`max_iterations`-bounded)
/// completion regardless — bounded but real provider spend nobody is
/// paying attention to.
///
/// `.abort()` on a [`tokio::task::JoinHandle`] is safe to call
/// unconditionally, including on an ALREADY-finished task (a documented
/// no-op there — see tokio's `JoinHandle::abort` docs) — so this never
/// needs to distinguish "still running" from "already done"; a background
/// child that already finished and is merely awaiting a `subagent_status`
/// reap is untouched in practice (aborting a finished task changes
/// nothing observable). For a task still mid-flight, tokio cancels it at
/// its next `.await` point, which drops that future in place — including
/// the `_guard: ConcurrencyGuard` moved into it at spawn time (see
/// [`Self::run_spawn_subagent`]'s `tokio::spawn` body) — so the
/// concurrency-gauge slot is released exactly the same way a normal
/// completion releases it (`ConcurrencyGuard`'s own `Drop`, in
/// `crate::subagents`). No separate cleanup call site to forget.
///
/// Deliberately does NOT touch [`Self::pending_child_approvals]` or
/// [`Self::subagent_store`] — this is purely "stop burning provider
/// calls on behalf of a caller who's gone", not a transcript-persistence
/// path (a child aborted mid-flight has no finished result to persist;
/// see this build's named residual on abort-time transcript loss).
impl Drop for Agent {
    fn drop(&mut self) {
        for (child_id, bg) in self.background_subagents.drain() {
            // Named, not silent: a child that was still running gets its
            // provider calls cut off here — worth a trace even though
            // there's no transcript left to persist (the future is
            // dropped mid-flight, before it ever returns a result).
            if !bg.handle.is_finished() {
                tracing::debug!(
                    child_id = %child_id,
                    "parent Agent dropped: aborting still-running background subagent \
                     to stop further provider spend"
                );
            }
            bg.handle.abort();
        }
        // P5-6 (§2 module 4 `tools.background`, build brief "on agent drop
        // / session end, jobs MUST be killed... real process kill via the
        // child handle's kill(), not just tokio task abort"): a REAL OS
        // process, not a Rust task — `Child::start_kill` (synchronous, no
        // `.await` needed, so callable from this non-async `Drop::drop`)
        // sends the actual kill signal; a no-op, per its own docs, on a
        // job that already exited. `kill_on_drop(true)` (set at spawn
        // time) is a second, independent line of defense for the same
        // outcome, but this explicit loop is what makes the guarantee
        // provable/traceable rather than relying solely on an implicit
        // tokio runtime behavior.
        for (job_id, mut job) in self.background_jobs.drain() {
            if !job.killed {
                tracing::debug!(
                    job_id = %job_id,
                    command = %job.command,
                    "parent Agent dropped: killing still-tracked background job's real \
                     OS process (and its whole process group — see \
                     `kill_job_process_group`)"
                );
            }
            kill_job_process_group(&mut job);
        }
        // P5-11 (§2 module 28 `lsp`, build brief "no orphaned language-
        // server processes"): a REAL OS process, same rationale as the
        // background-job loop just above — `kill_all_sync` is
        // synchronous (`Child::start_kill`, no `.await` needed, so
        // callable from this non-async `Drop::drop`), SIGKILLs each
        // server's WHOLE process group (unix — same `kill_job_process_group`
        // mechanism as the background-job loop above, so worker
        // grandchildren like rust-analyzer's proc-macro server or
        // typescript-language-server's `tsserver` are killed too, not just
        // the one directly-tracked pid), and is provable/traceable rather
        // than relying solely on `kill_on_drop(true)`'s implicit tokio
        // runtime behavior (which remains a second, independent line of
        // defense on every spawned `LspClient`).
        if let Some(lsp) = &self.lsp_manager {
            lsp.kill_all_sync();
        }
    }
}

/// P4 (§1.8 credential-helper indirection, D6 row): run an `api_key_cmd`
/// through the shell and return its trimmed stdout. Runs via `sh -c` (POSIX
/// shell, matching pi's `!command` precedent) so the configured string can
/// use pipes/substitution, e.g. `pass show api-key`. Never panics or
/// propagates an error: a spawn failure or non-zero exit is reported via
/// `tracing::warn!` and returns an empty `String`, which
/// [`Agent::new`]'s resolution chain treats exactly like an unset helper —
/// falling through to `Config::api_key_env`.
fn run_api_key_cmd(cmd: &str) -> String {
    match std::process::Command::new("sh").arg("-c").arg(cmd).output() {
        Ok(out) if out.status.success() => String::from_utf8_lossy(&out.stdout).trim().to_string(),
        Ok(out) => {
            tracing::warn!(
                "api_key_cmd exited with status {:?}; falling back to api_key_env",
                out.status.code()
            );
            String::new()
        }
        Err(e) => {
            tracing::warn!("api_key_cmd failed to run ({e}); falling back to api_key_env");
            String::new()
        }
    }
}

/// P4c (§1.2/§3.1 `core.shell_env_snapshot`, SPLIT CC+CX row, catalog:338):
/// capture the user's interactive login-shell environment ONCE, best-effort.
/// Runs `$SHELL -lc env` (falling back to `sh -lc env` when `$SHELL` is
/// unset) — a LOGIN shell (`-l`) sources the user's rc files, which is
/// exactly the sourcing `bash` calls should no longer need to repeat once
/// this snapshot is in hand. Never panics: any failure (spawn error,
/// non-zero exit, unparseable output) returns an empty map, which
/// `ToolContext::shell_env`'s "no-op when `None`/empty" contract already
/// treats as harmless.
fn capture_shell_env() -> std::collections::HashMap<String, String> {
    let shell = std::env::var("SHELL").unwrap_or_else(|_| "sh".to_string());
    let out = match std::process::Command::new(&shell)
        .arg("-lc")
        .arg("env")
        .output()
    {
        Ok(o) if o.status.success() => o.stdout,
        Ok(o) => {
            tracing::warn!(
                "shell_env_snapshot: `{shell} -lc env` exited with status {:?}; snapshot is empty",
                o.status.code()
            );
            return std::collections::HashMap::new();
        }
        Err(e) => {
            tracing::warn!(
                "shell_env_snapshot: failed to run `{shell} -lc env` ({e}); snapshot is empty"
            );
            return std::collections::HashMap::new();
        }
    };
    let text = String::from_utf8_lossy(&out);
    let mut map = std::collections::HashMap::new();
    for line in text.lines() {
        if let Some((k, v)) = line.split_once('=') {
            if !k.is_empty() {
                map.insert(k.to_string(), v.to_string());
            }
        }
    }
    map
}

/// Build the [`ToolContext`] an [`Agent`] hands to every tool call, folding
/// in every P4c per-tool config knob (§1.2) alongside the pre-existing
/// `cwd`/`sandbox` — shared by [`Agent::with_parts`]/[`Agent::with_provider_arc`]
/// so the two construction paths can never drift apart on which config
/// fields reach the context. Also builds (P5-9) the
/// [`crate::checkpoint::CheckpointObserver`], if `config.checkpoint_enabled`
/// — installed on the returned context's `write_observer` AND returned
/// separately (as the concrete type) so `Agent::run_loop` can call
/// [`crate::checkpoint::CheckpointObserver::begin_turn`] once per turn.
/// `None`/no-op end to end when the module is off — see
/// [`crate::checkpoint::observer_for_config`]'s own doc comment for the
/// default-off byte-identity guarantee.
///
/// P5-11 (§2 modules 28/29, D-5 "shared write-path interception seam"):
/// `crate::formatters::observer_for_config`/`crate::lsp::manager_for_config`
/// are folded into the SAME `write_observer` slot via
/// [`crate::tools::WriteObserverChain`], in the design's required order —
/// `checkpoint -> formatters -> lsp` (checkpoint's pre-image capture must
/// see the file before ANY mutation; lsp's diagnostics must see the file
/// AFTER formatting, never before). When 0 or 1 of the three modules is
/// active, this degrades to exactly what P5-9 shipped (`None`, or the
/// single concrete observer installed directly) — no chain wrapper is
/// introduced unless there is actually more than one observer to order,
/// keeping every single-module (or all-off) configuration byte-identical
/// to before this function grew multi-observer support. The `lsp` manager
/// is ALSO returned separately (like `checkpoint_observer`), so
/// `Agent`'s `Drop` impl can reach `crate::lsp::LspManager::kill_all_sync`
/// regardless of how the chain is shaped.
fn build_tool_context(
    config: &Config,
) -> (
    ToolContext,
    Option<std::sync::Arc<crate::checkpoint::CheckpointObserver>>,
    Option<std::sync::Arc<crate::lsp::LspManager>>,
) {
    let shell_env = if config.shell_env_snapshot {
        Some(std::sync::Arc::new(capture_shell_env()))
    } else {
        None
    };
    let checkpoint_observer = crate::checkpoint::observer_for_config(config);
    let format_observer = crate::formatters::observer_for_config(config);
    let lsp_manager = crate::lsp::manager_for_config(config);
    let lsp_observer = lsp_manager
        .clone()
        .map(|m| std::sync::Arc::new(crate::lsp::LspDiagnosticsObserver::new(m)));
    let mut observers: Vec<std::sync::Arc<dyn crate::tools::WriteObserver>> = Vec::new();
    if let Some(cp) = &checkpoint_observer {
        observers.push(cp.clone() as std::sync::Arc<dyn crate::tools::WriteObserver>);
    }
    if let Some(f) = &format_observer {
        observers.push(f.clone() as std::sync::Arc<dyn crate::tools::WriteObserver>);
    }
    if let Some(l) = &lsp_observer {
        observers.push(l.clone() as std::sync::Arc<dyn crate::tools::WriteObserver>);
    }
    let write_observer: Option<std::sync::Arc<dyn crate::tools::WriteObserver>> =
        match observers.len() {
            0 => None,
            1 => observers.into_iter().next(),
            _ => Some(std::sync::Arc::new(crate::tools::WriteObserverChain::new(
                observers,
            ))),
        };
    let ctx = ToolContext {
        cwd: config.cwd.clone(),
        sandbox: config.sandbox,
        multimodal_read: config.read_file_multimodal,
        require_read_before_edit: config.edit_file_require_read_before_edit,
        read_paths: std::sync::Arc::new(std::sync::Mutex::new(HashSet::new())),
        notebook_aware: config.edit_file_notebook_aware,
        shell_env,
        nested_instructions: config.nested_instructions,
        injected_instruction_dirs: std::sync::Arc::new(std::sync::Mutex::new(HashSet::new())),
        // P5-1 (§2 module 12 carry-forward): now sourced from real config
        // (`capabilities.permissions.sandbox.network.*`, wired by
        // `configfile::materialize_config`) instead of always `None`. `None`
        // (the default, unchanged when the config never sets it) is still
        // byte-identical to today's behavior.
        network_policy: config.network_policy.clone(),
        // P4e (S3.1 `core.tools.bash.timeout_secs`, S14): folds the `bash`
        // `ToolOverride`'s `timeout_secs`, if set, into the context every
        // `BashTool::execute` call receives -- `None` (no override
        // configured) is byte-identical to today's behavior.
        bash_timeout_secs: config
            .tool_overrides
            .get("bash")
            .and_then(|o| o.timeout_secs),
        write_observer,
        // P5-10 (§2 module 12): sourced from real config
        // (`capabilities.permissions.sandbox.{enabled,escalation,env_policy}`,
        // wired by `configfile::materialize_config`). `sandbox_approval_handler`
        // starts `None` here (no handler is installed yet at `Agent`
        // construction time) and is kept in sync by
        // `Agent::set_permissions_approval_handler` — see that method's doc
        // comment.
        sandbox_os_enabled: config.sandbox_os_enabled,
        sandbox_escalation: config.sandbox_escalation,
        sandbox_env_policy: config.sandbox_env_policy,
        sandbox_approval_handler: None,
    };
    (ctx, checkpoint_observer, lsp_manager)
}

/// P4b (§1.4/§3.1, catalog §4a "Global/user-level instruction file tier"):
/// where the user/global instruction tier lives — `$SUPERCODE_HOME`, else
/// `$XDG_CONFIG_HOME/supercode`, else `~/.config/supercode`. Deliberately
/// duplicates `crates/cli/src/userconfig.rs::config_home`'s exact precedence
/// rather than depending on the `cli` crate from `core` (wrong dependency
/// direction — `cli` depends on `core`, never the reverse). `pub(crate)`:
/// also the DEFAULT shadow-store root `crate::checkpoint::observer_for_config`
/// (P5-9) derives from when `Config::checkpoint_dir` is unset — one
/// `$SUPERCODE_HOME` resolver, not a second hand-rolled one.
pub(crate) fn global_instructions_dir() -> std::path::PathBuf {
    if let Ok(h) = std::env::var("SUPERCODE_HOME") {
        if !h.is_empty() {
            return std::path::PathBuf::from(h);
        }
    }
    if let Ok(xdg) = std::env::var("XDG_CONFIG_HOME") {
        if !xdg.is_empty() {
            return std::path::PathBuf::from(xdg).join("supercode");
        }
    }
    let home = std::env::var("HOME").unwrap_or_else(|_| ".".into());
    std::path::PathBuf::from(home)
        .join(".config")
        .join("supercode")
}

/// P4b (§1.4, catalog §4a "Instruction imports"): is `rel` (an `@`-import
/// target found inside a PROJECT-sourced instruction file) LEXICALLY safe to
/// resolve? Mirrors `configfile::is_safe_project_dir`'s posture (LOW-1
/// precedent): rejects absolute paths, `~`-relative paths, and any `..`
/// component — an untrusted repo's own CLAUDE.md/AGENTS.md must not be able
/// to `@import` its way to an arbitrary file on disk (e.g. `@/etc/passwd`,
/// `@../../.ssh/id_rsa`). Global-tier files (the user's own machine, same
/// trust level as the user's shell) are NOT run through this check.
///
/// This is a cheap PRE-FILTER only — it operates on the literal token text
/// and cannot see through a symlink committed in the repo whose *target*
/// escapes the root while the *link itself* has a clean, traversal-free
/// relative name (e.g. `@link.md` where `link.md -> /etc/passwd`). See
/// [`import_target_is_contained`] for the canonicalizing check that closes
/// that gap; the project-scoped resolution path runs both.
fn import_path_is_safe(rel: &str) -> bool {
    if rel.is_empty() || rel.contains('\0') {
        return false;
    }
    let path = std::path::Path::new(rel);
    if path.is_absolute() || rel.starts_with('~') {
        return false;
    }
    !path
        .components()
        .any(|c| matches!(c, std::path::Component::ParentDir))
}

/// P4b security fix (Fable-5 review, MEDIUM: symlink bypass of the
/// project-scoped `@`-import boundary): does `candidate` — after resolving
/// symlinks — stay inside `root` — also after resolving symlinks? This is
/// what actually enforces [`import_path_is_safe`]'s doc-comment guarantee
/// ("must not be able to `@import` its way to an arbitrary file on disk"):
/// the lexical check alone rejects `@/etc/passwd` and `@../../secret`, but a
/// repo can commit a symlink (e.g. `link.md -> /etc/passwd`) whose own
/// relative name is perfectly clean, defeating a purely lexical check.
///
/// Both sides are canonicalized before the comparison — not just
/// `candidate` — because `root` itself can legitimately be a symlink (a
/// tempdir under macOS's `/tmp` -> `/private/tmp`, or any other symlinked
/// project checkout); comparing a canonicalized candidate against a
/// non-canonicalized root would falsely reject genuinely-in-root files.
///
/// Fails CLOSED: a `canonicalize()` failure (broken symlink, a target that
/// doesn't exist, a permission error) returns `false` — never inlined,
/// mirroring [`expand_instruction_imports`]'s existing "unreadable file ⇒
/// left as literal text" posture rather than panicking or defaulting open.
pub(crate) fn import_target_is_contained(
    candidate: &std::path::Path,
    root: &std::path::Path,
) -> bool {
    let (Ok(real_root), Ok(real_candidate)) = (
        std::fs::canonicalize(root),
        std::fs::canonicalize(candidate),
    ) else {
        return false;
    };
    real_candidate.starts_with(&real_root)
}

/// P4b (§1.4/§3.1 `core.instruction_imports`, catalog:85): inline `@path`
/// import tokens found in `text` with the referenced file's own (trimmed)
/// content, resolved relative to `dir` (the directory the CONTAINING file
/// lives in — so a chain of imports each resolves relative to its own
/// location, not the original file's). `depth` bounds recursion (CC's own
/// default of 4, cited in the cc-parity preset) so a cyclical or
/// deeply-nested import chain can't blow the stack or loop forever.
/// `project_scoped` gates [`import_path_is_safe`] AND
/// [`import_target_is_contained`] — see their doc comments; `root` is the
/// containment boundary those checks canonicalize against (the SAME root
/// for every level of a nested import chain, even though `dir` itself walks
/// deeper with each level — an import three levels deep must still resolve
/// under the original project root, not merely under its own immediate
/// parent). Ignored when `!project_scoped` (the global/user tier, trusted,
/// unrestricted — see [`append_instruction_file`]'s doc comment).
/// Any token that isn't `@`-prefixed, doesn't resolve to a readable file, or
/// (project-scoped) fails the safety/containment check is left as literal
/// text — an import is best-effort, never a hard error that could make
/// instruction loading fail outright.
fn expand_instruction_imports(
    text: &str,
    dir: &std::path::Path,
    root: &std::path::Path,
    project_scoped: bool,
    depth: u8,
) -> String {
    if depth >= 4 {
        return text.to_string();
    }
    let mut out = String::with_capacity(text.len());
    for token in split_preserving_whitespace(text) {
        if let Some(rel) = token.strip_prefix('@') {
            if !rel.is_empty()
                && !rel.contains(char::is_whitespace)
                && (!project_scoped || import_path_is_safe(rel))
            {
                let candidate = dir.join(rel);
                if !project_scoped || import_target_is_contained(&candidate, root) {
                    if let Ok(imported) = std::fs::read_to_string(&candidate) {
                        let imported = imported.trim();
                        if !imported.is_empty() {
                            let imported_dir = candidate.parent().unwrap_or(dir);
                            out.push_str(&expand_instruction_imports(
                                imported,
                                imported_dir,
                                root,
                                project_scoped,
                                depth + 1,
                            ));
                            continue;
                        }
                    }
                }
            }
        }
        out.push_str(token);
    }
    out
}

/// Split `text` into tokens that, concatenated, reproduce it exactly —
/// alternating runs of non-whitespace and whitespace. Used by
/// [`expand_instruction_imports`] so `@import` tokens can be located and
/// replaced without disturbing surrounding formatting/whitespace.
fn split_preserving_whitespace(text: &str) -> Vec<&str> {
    let mut out = Vec::new();
    let mut start = 0;
    let mut in_ws = None;
    for (i, c) in text.char_indices() {
        let ws = c.is_whitespace();
        match in_ws {
            None => in_ws = Some(ws),
            Some(prev) if prev != ws => {
                out.push(&text[start..i]);
                start = i;
                in_ws = Some(ws);
            }
            _ => {}
        }
    }
    if start < text.len() {
        out.push(&text[start..]);
    }
    out
}

/// P4b (§1.4): append one instruction file's (trimmed, import-expanded)
/// content to `blob` as a labeled section, exactly like the pre-P4b inline
/// loop did — a no-op when `path` doesn't exist or is empty (the common
/// case). `project_scoped` distinguishes the project tier (imports bounded
/// to `root`, canonicalized-and-contained — see
/// [`import_target_is_contained`]) from the global tier (imports
/// unrestricted, same trust level as the user's own machine — `root` is
/// unused in that case). `root` is normally `path`'s own parent (the tier
/// root `path` was discovered under, e.g. `cwd` or an `additional_dirs`
/// entry) — see [`assemble_project_instructions`]'s call sites.
fn append_instruction_file(
    blob: &mut String,
    path: &std::path::Path,
    root: &std::path::Path,
    label: &str,
    imports_enabled: bool,
    project_scoped: bool,
) {
    let Ok(text) = std::fs::read_to_string(path) else {
        return;
    };
    let text = text.trim();
    if text.is_empty() {
        return;
    }
    let dir = path.parent().unwrap_or(std::path::Path::new("."));
    let content = if imports_enabled {
        expand_instruction_imports(text, dir, root, project_scoped, 0)
    } else {
        text.to_string()
    };
    blob.push_str(&format!("\n\n# {label}\n{content}"));
}

/// P4b (§1.4, obligation 4 assembly site): the full instruction-file blob —
/// global/user tier (catalog §4a "Global/user-level instruction file tier")
/// FIRST, then the existing cwd + `additional_dirs` project tier (root-first
/// ordering: nearer-to-cwd wins by appearing later, cx§2 precedent) — capped
/// by [`Config::project_doc_max_bytes`] if set (catalog §4a "hygiene caps
/// (`project_doc_max_bytes` analog)"). Byte-identical to the pre-P4b inline
/// loop when the global tier has no files, `instruction_imports` is off, and
/// `project_doc_max_bytes` is unset — i.e. for every config that doesn't
/// touch the new keys.
fn assemble_project_instructions(config: &Config) -> String {
    let mut blob = String::new();
    let global_dir = global_instructions_dir();
    for name in ["CLAUDE.md", "AGENTS.md"] {
        append_instruction_file(
            &mut blob,
            &global_dir.join(name),
            // Global tier is trusted/unrestricted (project_scoped=false
            // below) — `root` is never consulted, but pass `global_dir`
            // rather than a bogus value for clarity.
            &global_dir,
            name,
            config.instruction_imports,
            false,
        );
    }
    for root in std::iter::once(&config.cwd).chain(config.additional_dirs.iter()) {
        for name in ["CLAUDE.md", "AGENTS.md"] {
            append_instruction_file(
                &mut blob,
                &root.join(name),
                // Project tier: `@`-imports from THIS file must stay under
                // THIS root (canonicalized) — see
                // `import_target_is_contained`.
                root,
                name,
                config.instruction_imports,
                true,
            );
        }
    }
    if let Some(max) = config.project_doc_max_bytes {
        if blob.len() > max {
            let mut end = max;
            while end > 0 && !blob.is_char_boundary(end) {
                end -= 1;
            }
            blob.truncate(end);
            blob.push_str(
                "\n\n[supercode: instruction content truncated at core.project_doc_max_bytes]",
            );
        }
    }
    blob
}

/// P4b (§1.4/§3.1 `core.env_context`, catalog §4a "Environment context block
/// injection"): cwd, platform, date, and a best-effort git branch/dirty
/// status (silently absent when `cwd` isn't a git repo or `git` isn't on
/// `PATH` — never blocks agent construction).
fn env_context_block(config: &Config) -> String {
    let mut lines = vec![
        format!("cwd: {}", config.cwd.display()),
        format!("platform: {}", std::env::consts::OS),
        format!(
            "date: {}",
            crate::sidecar::now_rfc3339().get(..10).unwrap_or("")
        ),
    ];
    if let Some(status) = env_context_git_status(&config.cwd) {
        lines.push(status);
    }
    format!("\n\n# Environment\n{}", lines.join("\n"))
}

/// Best-effort `git branch (dirty|clean)` for [`env_context_block`]. `None`
/// on anything short of a clean success (not a repo, `git` missing, a
/// detached/errored state) — this is informational context, never worth
/// failing agent construction over.
fn env_context_git_status(cwd: &std::path::Path) -> Option<String> {
    let branch_out = std::process::Command::new("git")
        .args(["rev-parse", "--abbrev-ref", "HEAD"])
        .current_dir(cwd)
        .output()
        .ok()?;
    if !branch_out.status.success() {
        return None;
    }
    let branch = String::from_utf8_lossy(&branch_out.stdout)
        .trim()
        .to_string();
    if branch.is_empty() {
        return None;
    }
    let dirty = std::process::Command::new("git")
        .args(["status", "--porcelain"])
        .current_dir(cwd)
        .output()
        .ok()
        .map(|o| !o.stdout.is_empty())
        .unwrap_or(false);
    Some(format!(
        "git branch: {branch} ({})",
        if dirty { "dirty" } else { "clean" }
    ))
}

impl Agent {
    /// Build an agent backed by an OpenAI-compatible endpoint (OpenRouter by
    /// default). The API key is taken from [`Config::api_key`], then
    /// [`Config::api_key_cmd`] (P4: a credential-helper command, run via the
    /// shell — see [`run_api_key_cmd`]), then the configured environment
    /// variable ([`Config::api_key_env`]).
    pub fn new(config: Config) -> Result<Self> {
        let api_key = match &config.api_key {
            Some(k) if !k.is_empty() => k.clone(),
            _ => match config
                .api_key_cmd
                .as_deref()
                .filter(|c| !c.is_empty())
                .map(run_api_key_cmd)
            {
                // P4 (§1.8 credential-helper indirection, D6 row): the
                // helper ran and produced a non-empty key — use it. A
                // failed/empty helper falls through to `api_key_env` rather
                // than erroring outright, same "try the next source"
                // posture as every other layer in this resolution chain.
                Some(k) if !k.is_empty() => k,
                _ => std::env::var(&config.api_key_env)
                    .ok()
                    .filter(|k| !k.is_empty())
                    .ok_or_else(|| Error::MissingApiKey(config.api_key_env.clone()))?,
            },
        };
        // P4b (§1.1/§3.1 `core.retry`, pi§3 shape): `Config.retry_*` now
        // reaches the pre-existing transport-layer retry mechanism (see
        // `provider::HttpOptions::from_retry_config`'s doc comment for the
        // exact "byte-identical when unset" contract).
        let http_options = provider::HttpOptions::from_retry_config(
            config.retry_enabled,
            config.retry_max_retries,
            config.retry_base_delay_ms,
        );
        let provider = OpenAiProvider::new_with_options(
            config.base_url.clone(),
            api_key,
            config.extra_headers.clone(),
            http_options,
        );
        // P3 (design §5.2): `ToolRegistry::from_config` replaces the
        // unconditional `with_builtins()` call — a no-op when
        // `config.module_registry` is off (the default, §5.3 risk 2).
        let registry = ToolRegistry::from_config(&config);
        Ok(Self::with_parts(config, Box::new(provider), registry))
    }

    /// Build an agent with an explicit provider and the built-in tools. Handy
    /// for tests (inject a mock provider) or custom transports.
    pub fn with_provider(config: Config, provider: Box<dyn Provider>) -> Self {
        let registry = ToolRegistry::from_config(&config);
        Self::with_parts(config, provider, registry)
    }

    /// Build an agent from all three parts.
    pub fn with_parts(
        config: Config,
        provider: Box<dyn Provider>,
        mut registry: ToolRegistry,
    ) -> Self {
        // P5-12 (§2 module 18 `plugins`, D-10): register every trusted,
        // loaded plugin's declared tools — the same "unconditional, config-
        // gated" wiring `build_tool_context` just below gives
        // checkpoint/formatters/lsp. `crate::plugins::register_into` is a
        // true no-op (no filesystem read, no subprocess) whenever
        // `config.plugins_enabled` is `false` (the default) — byte-identical
        // to before this module existed. Runs here (the one tail every
        // `Agent` construction path funnels through — `new`/`with_provider`
        // both call this) rather than in `ToolRegistry::from_config`, so it
        // is NOT entangled with that function's unrelated `module_registry`
        // experimental gate.
        crate::plugins::register_into(&config, &mut registry);
        let (ctx, checkpoint_observer, lsp_manager) = build_tool_context(&config);
        // Auto-load project context files (CLAUDE.md / AGENTS.md) from the
        // working directory (and any extra roots), appending them to the system
        // prompt — the analog of how Claude Code / Codex discover them.
        // P4b (§1.4): also the global/user tier + instruction imports + the
        // `project_doc_max_bytes` hygiene cap — see `assemble_project_instructions`.
        let mut system = config.system_prompt.clone();
        if config.load_project_context {
            system.push_str(&assemble_project_instructions(&config));
        }
        // P4b (§1.4/§3.1 `core.env_context`, catalog §4a "Environment
        // context block injection"): `false` (the default) is a no-op —
        // byte-identical to today's behavior.
        if config.env_context {
            system.push_str(&env_context_block(&config));
        }
        // P4e (§1.4/§3.1 `core.context_injections`, catalog:91 "Synthetic
        // context-injection blocks"): same assembly site, right after
        // `env_context`. `false` (the default) is a no-op — byte-identical
        // to today's behavior; an empty `context_injection_blocks` list is
        // ALSO a no-op even with the gate on (nothing to append).
        if config.context_injections {
            for block in &config.context_injection_blocks {
                system.push_str(&format!("\n\n# {}\n{}", block.name, block.content));
            }
        }
        // P3 (design §5.2, §1.4 obligation 4, D-7): the skills prompt
        // section is a MODULE-GATED prompt section, the design's own
        // illustration of "a disabled module contributes no prompt
        // sections" — only assembled at all under
        // `[experimental] module_registry = true` (§5.3 risk 2: flag-off is
        // byte-for-byte today's behavior, and today's behavior never emits
        // this section, since it doesn't exist pre-P3). Gated further by
        // D-7 itself: `core.skills` requires a read pathway (`read_file` or
        // `bash`) — absent either, no section is appended, matching the
        // hard-dependency shape `configfile::validate_modules` enforces at
        // resolve time.
        if config.module_registry && config.skills_enabled {
            let has_read_pathway = config
                .core_tools_enabled
                .iter()
                .any(|t| t == "read_file" || t == "bash");
            if has_read_pathway {
                let mut names: Vec<&str> = config.prompts.keys().map(String::as_str).collect();
                names.sort_unstable();
                if !names.is_empty() {
                    system.push_str("\n\n# Skills\nAvailable skill/prompt templates (today: named `[core.prompts]` templates, D-7 — invoke via `/name args`):\n");
                    for name in names {
                        system.push_str(&format!("- {name}\n"));
                    }
                }
            }
        }
        let history = vec![ChatMessage::system(system)];
        // P4e (§1.6/§3.1 `core.session.git_metadata`, catalog:331): captured
        // once here, alongside `env_context`'s own git probe — `false` (the
        // default) is a no-op, byte-identical to today's behavior.
        let git_metadata = if config.session_git_metadata {
            crate::git_metadata::capture(&config.cwd, now_ms())
        } else {
            None
        };
        // P5-3: captured before `config` moves into the literal below (a
        // `usize` field READ, not a move, but it must happen before the
        // `config` shorthand field consumes the binding).
        let subagent_depth = config.subagent_depth;
        Agent {
            config,
            provider: std::sync::Arc::from(provider),
            registry,
            history,
            ctx,
            total_output_tokens: 0,
            activated_tools: HashSet::new(),
            recorder: None,
            reduction_policy: None,
            reduction_log: ReductionLog::default(),
            imported_prefix_len: None,
            span_summarizer: None,
            last_tool_schema_tier_signature: None,
            context_limit: None,
            requests_issued: false,
            last_cache_activity_ms: None,
            cache_established: false,
            pending_cache_turn: (false, false, None),
            session_titler: None,
            usage_log: Vec::new(),
            turn_index: 0,
            steer_queue: std::sync::Arc::new(std::sync::Mutex::new(SteerInbox::default())),
            follow_up_queue: std::collections::VecDeque::new(),
            doom_loop_last_call: None,
            doom_loop_streak: 0,
            model_change_log: Vec::new(),
            git_metadata,
            permissions_approval_cache: crate::permissions::ApprovalCache::new(),
            permissions_approval_handler: None,
            mcp_prompts: std::collections::HashMap::new(),
            subagent_depth,
            subagent_concurrency_gauge: std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0)),
            background_subagents: std::collections::HashMap::new(),
            pending_child_approvals: std::sync::Arc::new(std::sync::Mutex::new(Vec::new())),
            child_approval_handler_factory: None,
            subagent_store: None,
            claude_runtime_manifest: None,
            background_jobs: std::collections::HashMap::new(),
            background_concurrency_gauge: std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(
                0,
            )),
            checkpoint_observer,
            lsp_manager,
        }
    }

    /// A handle to this agent's model transport, for sharing with subagents.
    pub fn provider_arc(&self) -> std::sync::Arc<dyn Provider> {
        self.provider.clone()
    }

    /// Read-only access to this agent's resolved [`Config`] — e.g. so a
    /// caller (`crates/cli`'s `attach_mcp`) can consult
    /// [`Config::module_registry`]/[`Config::module_activation`] AFTER
    /// construction without having to separately thread the config through
    /// every call site that builds an `Agent` and later needs it again.
    /// Same trust boundary as every other already-public `Agent` accessor
    /// (`history`, `provider_arc`) — the caller is the same process that
    /// built this `Config` in the first place, not a new exposure surface.
    pub fn config(&self) -> &Config {
        &self.config
    }

    /// P5-9 (§2 module 20 `checkpoint`): this agent's checkpoint engine, if
    /// `Config::checkpoint_enabled` is `true` and the shadow store opened
    /// successfully — `None` otherwise (the default-off case, or a
    /// graceful-degrade after an I/O failure). A caller (CLI/TUI/embedder)
    /// uses this to `list`/`turn_diff`/`restore` WITHOUT re-deriving the
    /// shadow-store root itself. Deliberately named `checkpoint_observer`,
    /// not `checkpoint` — [`Self::checkpoint`] already names the unrelated
    /// in-memory conversation-position marker (see that method's doc
    /// comment).
    pub fn checkpoint_observer(&self) -> Option<&crate::checkpoint::CheckpointObserver> {
        self.checkpoint_observer.as_deref()
    }

    /// P5-11 (§2 module 28 `lsp`): this agent's LSP server registry, if
    /// `Config::lsp_enabled` is `true` — `None` otherwise (the default-off
    /// case). `impl Drop for Agent` already covers production teardown via
    /// [`crate::lsp::LspManager::kill_all_sync`] (a real, group-killing OS
    /// process kill — see `crate::lsp`'s module doc). This accessor exists
    /// for an OPTIONAL caller (CLI/TUI/embedder) that manages its own
    /// `Agent` lifecycle and additionally wants to reach
    /// [`crate::lsp::LspManager::shutdown_all`] for a graceful LSP
    /// `shutdown`/`exit` handshake BEFORE dropping the agent — nothing
    /// calls `shutdown_all` automatically today.
    pub fn lsp_manager(&self) -> Option<&crate::lsp::LspManager> {
        self.lsp_manager.as_deref()
    }

    /// Spawn a subagent that shares this agent's model transport, runs `task`
    /// to completion with its own fresh conversation (seeded with `system`), and
    /// returns its final answer. The analog of `Agent` / `spawn_agent`.
    pub async fn run_subagent(
        &self,
        system: impl Into<String>,
        task: impl Into<String>,
    ) -> Result<String> {
        let mut sub_config = Config::builder()
            .model(self.config.model.clone())
            .system_prompt(system)
            .cwd(self.config.cwd.clone())
            .sandbox(self.config.sandbox)
            .max_iterations(self.config.max_iterations)
            .build();
        sub_config.base_url = self.config.base_url.clone();
        let mut sub = Agent::with_provider_arc(sub_config, self.provider.clone());
        sub.send(task).await
    }

    /// Like [`Self::with_provider`] but sharing an existing transport handle.
    pub fn with_provider_arc(config: Config, provider: std::sync::Arc<dyn Provider>) -> Self {
        let (ctx, checkpoint_observer, lsp_manager) = build_tool_context(&config);
        let history = vec![ChatMessage::system(config.system_prompt.clone())];
        // P3 (design §5.2): see the `Self::new` doc note — a no-op when
        // `config.module_registry` is off (the default).
        let mut registry = ToolRegistry::from_config(&config);
        // P5-12: see `Self::with_parts`'s identical call — a no-op when
        // `config.plugins_enabled` is `false` (the default).
        crate::plugins::register_into(&config, &mut registry);
        // P4e: see `Self::with_parts`'s identical capture.
        let git_metadata = if config.session_git_metadata {
            crate::git_metadata::capture(&config.cwd, now_ms())
        } else {
            None
        };
        // P5-3: captured before `config` moves into the literal below (a
        // `usize` field READ, not a move, but it must happen before the
        // `config` shorthand field consumes the binding).
        let subagent_depth = config.subagent_depth;
        Agent {
            config,
            provider,
            registry,
            history,
            ctx,
            total_output_tokens: 0,
            activated_tools: HashSet::new(),
            recorder: None,
            reduction_policy: None,
            reduction_log: ReductionLog::default(),
            imported_prefix_len: None,
            span_summarizer: None,
            last_tool_schema_tier_signature: None,
            context_limit: None,
            requests_issued: false,
            last_cache_activity_ms: None,
            cache_established: false,
            pending_cache_turn: (false, false, None),
            session_titler: None,
            usage_log: Vec::new(),
            turn_index: 0,
            steer_queue: std::sync::Arc::new(std::sync::Mutex::new(SteerInbox::default())),
            follow_up_queue: std::collections::VecDeque::new(),
            doom_loop_last_call: None,
            doom_loop_streak: 0,
            model_change_log: Vec::new(),
            git_metadata,
            permissions_approval_cache: crate::permissions::ApprovalCache::new(),
            permissions_approval_handler: None,
            mcp_prompts: std::collections::HashMap::new(),
            subagent_depth,
            subagent_concurrency_gauge: std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0)),
            background_subagents: std::collections::HashMap::new(),
            pending_child_approvals: std::sync::Arc::new(std::sync::Mutex::new(Vec::new())),
            child_approval_handler_factory: None,
            subagent_store: None,
            claude_runtime_manifest: None,
            background_jobs: std::collections::HashMap::new(),
            background_concurrency_gauge: std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(
                0,
            )),
            checkpoint_observer,
            lsp_manager,
        }
    }

    /// Run a prompt on a background task, returning a handle that resolves to
    /// the final answer (and the agent, so the caller can continue it). The
    /// analog of background/async agent runs.
    pub fn run_in_background(
        mut self,
        prompt: impl Into<String>,
    ) -> tokio::task::JoinHandle<(Self, Result<String>)>
    where
        Self: Send + 'static,
    {
        let prompt = prompt.into();
        tokio::spawn(async move {
            let result = self.send(prompt).await;
            (self, result)
        })
    }

    /// Build an agent and seed it with a previously-recorded [`Session`] so it
    /// can continue where Claude Code or Codex left off.
    pub fn resume(config: Config, session: Session) -> Result<Self> {
        let mut agent = Agent::new(config)?;
        agent.load_session(session);
        Ok(agent)
    }

    /// Like [`Self::resume`], but also begins recording (A2/A3): a fresh
    /// native-v2 sidecar is created at `sidecar_path` from `session` (header +
    /// `session.raw` verbatim — the imported prefix's own fidelity), and every
    /// subsequent turn this agent produces is appended to it at full fidelity,
    /// independent of whatever `cap_tool_output`/`maybe_compact` (D6) do to
    /// `history`.
    ///
    /// Invariant this establishes ONLY once [`Self::set_reduction_policy`] is
    /// also called (the D6/A7 supersession gate, `Self::run_loop`): at any
    /// instant, `Session::from_native_str(sidecar).messages` equals
    /// `session.messages` (the imported prefix) followed by every message
    /// appended since — i.e. `self.history()[1..]` (`history[0]` is this
    /// agent's own system prompt, per [`Self::load_session`]; it is never
    /// part of `session` and is never written to the sidecar). Recording
    /// alone (no policy) leaves the gate off: `cap_tool_output` still runs on
    /// oversized tool results, and `history` can diverge from the sidecar for
    /// them — honestly, via the notice's "full output in session sidecar"
    /// label, never silently.
    pub fn resume_recorded(
        config: Config,
        session: Session,
        sidecar_path: &std::path::Path,
    ) -> Result<Self> {
        let mut agent = Agent::new(config)?;
        let recorder = SidecarWriter::create(sidecar_path, &session)?;
        agent.load_session(session);
        agent.recorder = Some(recorder);
        Ok(agent)
    }

    /// Install (or replace) this agent's sidecar recorder (A3).
    pub fn set_recorder(&mut self, w: SidecarWriter) {
        self.recorder = Some(w);
    }

    /// Install (or replace) this agent's reduction policy (A5/A7/A10). Once
    /// set, every provider request is built from a *projected* view of
    /// `history[1..]` (`reduce::project_messages`) rather than `history`
    /// verbatim — `history` itself is never shrunk or mutated by this; only
    /// the request view does.
    pub fn set_reduction_policy(&mut self, policy: ReductionPolicy) {
        self.reduction_policy = Some(policy);
    }

    /// This agent's reduction policy, if one is installed.
    pub fn reduction_policy(&self) -> Option<&ReductionPolicy> {
        self.reduction_policy.as_ref()
    }

    /// Change the global tool-schema tier (TR-8/T5) mid-session. Takes effect
    /// starting with the NEXT request this agent builds. Under
    /// [`CachePlan::ImportedPrefix`], the first request built after a change
    /// is flagged as a cache-bust event and its cache-control annotation is
    /// skipped for that one request (see [`provider::tier_change_is_cache_bust`],
    /// consulted in [`Self::build_request_messages`]) — normal annotation
    /// resumes on the next request if the tier doesn't change again.
    pub fn set_schema_tier(&mut self, tier: crate::tools::SchemaTier) {
        self.config.tool_schema_tier = tier;
    }

    /// Override the schema tier for a single tool (TR-8/T5) mid-session, same
    /// cache-bust interaction as [`Self::set_schema_tier`].
    pub fn set_tool_schema_tier(
        &mut self,
        name: impl Into<String>,
        tier: crate::tools::SchemaTier,
    ) {
        self.config
            .tool_overrides
            .entry(name.into())
            .or_default()
            .schema_tier = Some(tier);
    }

    /// A deterministic fingerprint of the current tool-schema tier
    /// configuration (global knob + every per-tool override), used to detect
    /// a mid-session tier change (TR-8/T5, dev/05). Order-independent over
    /// `tool_overrides` (sorted by name before hashing) so insertion order
    /// never spuriously changes the signature.
    fn schema_tier_signature(&self) -> u64 {
        use std::hash::{Hash, Hasher};
        let mut hasher = std::collections::hash_map::DefaultHasher::new();
        self.config.tool_schema_tier.hash(&mut hasher);
        let mut overrides: Vec<(&str, crate::tools::SchemaTier)> = self
            .config
            .tool_overrides
            .iter()
            .filter_map(|(name, o)| o.schema_tier.map(|t| (name.as_str(), t)))
            .collect();
        overrides.sort_by_key(|(name, _)| *name);
        for (name, tier) in overrides {
            name.hash(&mut hasher);
            tier.hash(&mut hasher);
        }
        hasher.finish()
    }

    /// Install (or replace) this agent's TR-7 span summarizer — the
    /// injectable side-call [`Self::build_request_messages`] uses to turn an
    /// A10 `TurnsCleared` span into an LLM-written summary paragraph when
    /// `policy.summarize_cleared_turns` is on. Installing one alone changes
    /// nothing: [`ReductionPolicy::summarize_cleared_turns`] (off by
    /// default) is the actual gate, so tests/callers that want the
    /// deterministic stub can simply never call this.
    pub fn set_span_summarizer(
        &mut self,
        summarizer: impl reduce::summarize::SpanSummarizer + Send + Sync + 'static,
    ) {
        self.span_summarizer = Some(std::sync::Arc::new(summarizer));
    }

    /// Prepare TR-7 metadata with this agent's installed summarizer for a
    /// projection performed by an outer driver before session history/log
    /// are loaded (the CLI foreign-resume preflight). `None` preserves the
    /// deterministic fallback when the gate is off, no summarizer exists,
    /// the span is below the cost floor, or the side-call fails.
    pub fn prepare_cleared_turns_summary(
        &self,
        msgs: &[ChatMessage],
        policy: &ReductionPolicy,
        prior: &ReductionLog,
    ) -> Option<reduce::PreparedClearSummary> {
        let summarizer = self.span_summarizer.as_deref()?;
        reduce::prepare_cleared_turns_summary(msgs, policy, prior, summarizer)
    }

    /// P5-4: install (or replace) this agent's [`crate::EventSink`] AFTER
    /// construction — `Config::event_sink` is otherwise only set at
    /// `Config`-build time (before `Agent::new`), which is too early for a
    /// `tui` embedder that only knows it's activating (and needs to
    /// replace whatever print-mode/REPL sink was already installed with
    /// one that feeds its own render loop instead of writing straight to
    /// stdout) once it already holds a live `Agent`. Mirrors [`Self::
    /// set_permissions_approval_handler`]'s "installing one alone changes
    /// nothing beyond what already consults `Config::event_sink`" pattern
    /// — this is a plain replacement, not a new activation gate.
    pub fn set_event_sink(&mut self, sink: crate::EventSink) {
        self.config.event_sink = Some(sink);
    }

    /// P5-1: install (or replace) this agent's permissions-engine approval
    /// handler — see [`crate::permissions::PermissionsApprovalHandler`].
    /// This is the non-interactive decision seam a CLI/TUI/SDK embedder
    /// implements for the `Ask`-tier prompt; the TUI's actual interactive
    /// UI is a separate module (P5 row 4), not built here. Installing one
    /// alone changes nothing: [`Config::permissions_enabled`] (off by
    /// default) is the actual gate — with no handler installed, every
    /// `Ask`-tier decision denies (fail-closed, see that trait's doc
    /// comment).
    ///
    /// P5-10 (§2 module 12, `escalation = "ask"`): the SAME handler also
    /// backs a sandbox-unenforceable `ask` decision
    /// (`crate::sandbox::decide_fs`'s `approval` parameter) — one installed
    /// seam serves both `permissions.rules`' `Ask` tier and
    /// `permissions.sandbox`'s `escalation = "ask"`, rather than requiring
    /// an embedder to install two near-identical handlers. Kept in sync on
    /// `self.ctx` (not just `self.permissions_approval_handler`) because
    /// `BashTool::execute`/`PersistentShellTool::execute` only ever see
    /// `&ToolContext`, never `&Agent` — see `ToolContext::
    /// sandbox_approval_handler`'s doc comment.
    pub fn set_permissions_approval_handler(
        &mut self,
        handler: impl crate::permissions::PermissionsApprovalHandler + 'static,
    ) {
        let handler: std::sync::Arc<dyn crate::permissions::PermissionsApprovalHandler> =
            std::sync::Arc::new(handler);
        self.permissions_approval_handler = Some(handler.clone());
        self.ctx.sandbox_approval_handler = Some(crate::sandbox::SandboxApprovalHandler(handler));
    }

    /// Install the compatibility approval seam used when the composable
    /// permissions engine is disabled. SDK-owned interactive frontends call
    /// this alongside [`Self::set_permissions_approval_handler`] so the same
    /// authenticated request channel works under either policy engine; the
    /// selected engine remains entirely a configuration decision.
    pub fn set_legacy_approval_handler(&mut self, handler: crate::config::ApprovalHandler) {
        self.config.approval_handler = Some(handler);
    }

    /// P5-3 (§2 module 9 D5 "subagent transcripts… persisted + linked"):
    /// install a [`crate::store::SessionStore`] (+ this agent's own session
    /// name in it) so `spawn_subagent` persists each child's transcript
    /// (via [`crate::store::SessionStore::save_subagent_transcript`]) and
    /// lineage record (via
    /// [`crate::store::SessionStore::save_subagent_lineage`]) once the
    /// child finishes. Installing one alone changes nothing about whether
    /// spawning WORKS — [`Config::subagents_enabled`] is the actual gate;
    /// this only controls whether a completed spawn's transcript additionally
    /// lands on disk.
    pub fn set_subagent_store(
        &mut self,
        store: std::sync::Arc<crate::store::SessionStore>,
        session_name: impl Into<String>,
    ) {
        self.subagent_store = Some((store, session_name.into()));
    }

    /// Seed the Claude runtime manifest reconstructed during resume.
    ///
    /// Installing state enables the matching Claude runtime tool schemas so
    /// a disk-reloaded continuation does not lose that vocabulary, but never
    /// starts a timer by itself. The supplied execution posture is preserved:
    /// an embedding scheduler may deliberately activate before installing it.
    pub fn set_claude_runtime_manifest(
        &mut self,
        manifest: crate::claude_runtime_state::ClaudeRuntimeManifest,
    ) {
        // A persisted manifest is itself the compatibility capability marker.
        // Reopening a Supercode session must not retain its timers while
        // silently dropping Claude's Cron*/ScheduleWakeup vocabulary.
        self.config.claude_runtime_tools_enabled = true;
        self.claude_runtime_manifest = Some(manifest);
    }

    /// Reinstall project-scoped Claude named-agent definitions when a
    /// Supercode continuation carrying a Claude runtime manifest is reopened
    /// from disk. The manifest is the durable capability marker; definitions
    /// themselves remain authoritative in `<cwd>/.claude/agents/*.md`.
    pub fn restore_claude_project_agents(&mut self) -> Result<usize> {
        let definitions = crate::claude_compat::load_project_agents(&self.config.cwd)?;
        crate::claude_compat::enable_claude_subagent_compatibility(&mut self.config);
        for imported in &definitions {
            self.config.subagents_definitions.insert(
                imported.definition.name.clone(),
                imported.definition.clone(),
            );
        }
        Ok(definitions.len())
    }

    /// Current imported Claude runtime state, including paused mutations made
    /// by `Cron*`/`ScheduleWakeup`, for persistence by the embedding loop.
    pub fn claude_runtime_manifest(
        &self,
    ) -> Option<&crate::claude_runtime_state::ClaudeRuntimeManifest> {
        self.claude_runtime_manifest.as_ref()
    }

    /// Mutable access for an embedding scheduler driver to atomically claim
    /// due events and persist the resulting manifest. Merely borrowing this
    /// state does not start a timer; execution remains the driver's explicit
    /// responsibility.
    pub fn claude_runtime_manifest_mut(
        &mut self,
    ) -> Option<&mut crate::claude_runtime_state::ClaudeRuntimeManifest> {
        self.claude_runtime_manifest.as_mut()
    }

    /// P5-4 (tui, closes the P5-3 §2.2 C6 deferred chain): install a
    /// factory this agent's [`Self::run_spawn_subagent`] calls (with the
    /// fresh child's own id and this agent's shared
    /// [`Self::pending_child_approvals`] queue) to build the
    /// `PermissionsApprovalHandler` a `background_prompts = "parent"`
    /// child gets, INSTEAD of the default
    /// [`crate::subagents::ParentQueueApprovalHandler`]. Installing one
    /// alone changes nothing about whether background spawning works —
    /// [`Config::subagents_background_prompts`] being
    /// [`crate::subagents::BackgroundPromptsPolicy::Parent`] is the actual
    /// gate that reaches this factory at all; a `Parent`-policy child
    /// spawned before this is installed (or on an agent that never installs
    /// it) still gets the immediate-deny default, unchanged.
    ///
    /// **Security note.** The factory only controls WHICH handler answers
    /// an `Ask`-tier request — it can never widen what gets asked in the
    /// first place: [`crate::permissions::approval::resolve_ask`] only
    /// calls a handler's `ask` when the rule engine has already resolved
    /// the call to `Ask` (`Deny` short-circuits before any handler is
    /// consulted; `Allow` never needs one), so a parent's "allow" answer
    /// here can only grant what the policy already routed to a prompt —
    /// never override a `Deny` the engine already decided.
    pub fn set_child_approval_handler_factory(
        &mut self,
        factory: impl Fn(
                String,
                std::sync::Arc<std::sync::Mutex<Vec<crate::subagents::QueuedApproval>>>,
            ) -> std::sync::Arc<dyn crate::permissions::PermissionsApprovalHandler>
            + Send
            + Sync
            + 'static,
    ) {
        self.child_approval_handler_factory = Some(std::sync::Arc::new(factory));
    }

    /// P5-3 (§2.2 C6 "parent-surfaced queue"): every approval request a
    /// `background_prompts = "parent"` child has raised so far, oldest
    /// first — a read-only audit view, not a mutable queue the caller
    /// answers. Under P5-3's own default handler (no P5-4 TUI factory
    /// installed) every entry here WAS already resolved `Deny` (a
    /// background call can't wait for an answer with no handler
    /// installed) — but once a `crate::tui::TuiChildApprovalHandler`
    /// factory is installed (P5-4,
    /// [`Self::set_child_approval_handler_factory`]), the underlying call
    /// genuinely blocks and may resolve `Allow`/`AllowForSession`; this
    /// method still records the SAME entry for the audit trail either
    /// way, so "queued here" no longer implies "was denied" in general —
    /// see [`crate::subagents::QueuedApproval`]'s doc comment.
    pub fn pending_child_approvals(&self) -> Vec<crate::subagents::QueuedApproval> {
        self.pending_child_approvals
            .lock()
            .map(|q| q.clone())
            .unwrap_or_default()
    }

    /// P4b: install (or replace) this agent's auto-title side-call — see
    /// [`crate::session_title::SessionTitler`]. Installing one alone changes
    /// nothing: [`Config::auto_title`] (off by default) is the actual gate a
    /// caller should consult before calling [`Self::auto_title`].
    pub fn set_session_titler(
        &mut self,
        titler: impl crate::session_title::SessionTitler + Send + Sync + 'static,
    ) {
        self.session_titler = Some(std::sync::Arc::new(titler));
    }

    /// P4b: produce a title for this agent's current conversation via the
    /// installed [`Self::set_session_titler`] side-call. Returns `None` (never
    /// panics, never blocks longer than the titler itself does) if no
    /// titler is installed, or the side-call itself declined (see
    /// [`crate::session_title::auto_title`]). Does NOT consult
    /// [`Config::auto_title`] itself — that gate is the caller's
    /// responsibility, matching [`Self::span_summarizer`]'s precedent of
    /// keeping the mechanism and the policy gate separate.
    pub fn auto_title(&self) -> Option<String> {
        let titler = self.session_titler.as_deref()?;
        crate::session_title::auto_title(&self.history, titler)
    }

    /// P4b (§1.6, catalog §4a "persisted per-turn usage records"): every
    /// [`crate::usage_log::UsageRecord`] this agent has accumulated so far.
    pub fn usage_records(&self) -> &[crate::usage_log::UsageRecord] {
        &self.usage_log
    }

    /// P4b: persist this agent's accumulated usage log to `store` under
    /// `name` — a thin wrapper over [`crate::store::SessionStore::save_usage_log`]
    /// so callers don't need to import both types.
    pub fn save_usage_log(&self, store: &crate::store::SessionStore, name: &str) -> Result<()> {
        store.save_usage_log(name, &self.usage_log)
    }

    /// P4b (§1.7, pi§3 semantics): queue a mid-turn steering message —
    /// delivered "after current tool calls" (pi's phrasing): at the top of
    /// [`Self::run_loop`]'s NEXT iteration, before the next model request is
    /// built, regardless of whether this turn is still mid-flight with
    /// pending tool calls. Drained per [`Config::steering_mode`].
    pub fn queue_steer(&self, message: impl Into<String>) {
        self.steer_queue
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .queue_unchecked(message.into());
    }

    /// Crate-internal shared steering handle used by the canonical SDK
    /// runtime. It remains writable while an active turn holds `&mut Agent`,
    /// allowing local and remote frontends to steer without owning the loop.
    pub(crate) fn steer_queue_handle(&self) -> std::sync::Arc<std::sync::Mutex<SteerInbox>> {
        self.steer_queue.clone()
    }

    /// P4b: queue a follow-up message — delivered "at idle" (pi's phrasing):
    /// only once [`Self::run_loop`] would otherwise return a final answer
    /// (no more tool calls pending). Drained per [`Config::follow_up_mode`].
    pub fn queue_follow_up(&mut self, message: impl Into<String>) {
        self.follow_up_queue.push_back(message.into());
    }

    /// P4b: how many steering messages are currently queued (mid-turn +
    /// follow-up combined) — mostly for tests/diagnostics.
    pub fn queued_steer_count(&self) -> usize {
        self.steer_queue
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .len()
            + self.follow_up_queue.len()
    }

    /// The accumulating reduction log (A5) — every reduction applied to any
    /// projected request view so far. Combined with a full-fidelity sidecar
    /// Session, this is enough to `reduce::invert` any projected view back to
    /// the exact original.
    pub fn reduction_log(&self) -> &ReductionLog {
        &self.reduction_log
    }

    /// PARITY-18 D4 — arm the per-send context guard: [`Self::run_loop`]
    /// will refuse (via [`Error::ContextLimitExceeded`]) to build and issue
    /// ANY request — the first or any later turn — whose
    /// [`crate::tokens::context_guard`] verdict is "does not fit" against
    /// `limit`. Call this once the target model's context-window size is
    /// known (`resume --reduced`'s preflight already computes it). Leaving
    /// this unset (the default) is a no-op: no guard runs, exactly today's
    /// pre-PARITY-18 behavior.
    pub fn set_context_limit(&mut self, limit: u64) {
        self.context_limit = Some(limit);
    }

    /// This agent's armed context limit, if [`Self::set_context_limit`] has
    /// been called.
    pub fn context_limit(&self) -> Option<u64> {
        self.context_limit
    }

    /// The model identifier this agent sends on its next request
    /// ([`Config::model`], as of construction/resume or the last
    /// [`Self::set_model`] call).
    pub fn model(&self) -> &str {
        &self.config.model
    }

    /// UX-30 dev/02 — switch the model this agent sends, starting with the
    /// NEXT request it builds (and every one after, until changed again).
    /// [`Self::run_loop`] reads `self.config.model` fresh on every request
    /// (see its `ChatRequest` construction), so this alone is enough —
    /// there is no cached/baked-in copy anywhere else to also update.
    /// Takes effect immediately; safe to call only between turns (the
    /// REPL's `/model` picker runs at the prompt, never mid-turn). Touches
    /// nothing else: history, the sidecar, and reduction state are exactly
    /// as untouched as [`Self::set_schema_tier`] leaves them for a
    /// mid-session tier change.
    ///
    /// P4c-review note: this is the LOW-LEVEL primitive — it swaps
    /// [`Config::model`] and nothing else. It does NOT run dep 8's
    /// reasoning-artifact filter
    /// ([`reduce::rehydrate::filter_reasoning_artifacts`]) and does NOT
    /// create a [`crate::model_change::ModelChangeRecord`], so calling it
    /// directly for a mid-session handoff between two DIFFERENT models
    /// leaves model-A's reasoning artifacts in `history` for model-B to
    /// inherit. [`Self::switch_model`] is the safe superset — gated by
    /// [`Config::model_switch_allow_switch`], it filters and records the
    /// switch before delegating to this method — and is what callers
    /// performing a governed mid-session model switch should use instead.
    pub fn set_model(&mut self, model: impl Into<String>) {
        self.config.model = model.into();
    }

    /// P4c (§1.10/§3.1 `core.model_switch.allow_switch`, D9 row, dep 8,
    /// design's "core NEW-significant" item): the mid-session model
    /// switch — a superset of [`Self::set_model`] gated by
    /// [`Config::model_switch_allow_switch`].
    ///
    /// **`allow_switch = false` (the default): EXACTLY [`Self::set_model`]**
    /// — same single field write, nothing else touched, no
    /// [`crate::model_change::ModelChangeRecord`] created. Byte-identical to
    /// calling `set_model` directly.
    ///
    /// **`allow_switch = true`:** additionally, before the swap takes
    /// effect, runs [`reduce::rehydrate::filter_reasoning_artifacts`] over
    /// [`Self::history`] — model-A's reasoning/thinking artifacts (any
    /// [`crate::message::ChatMessage::metadata`] key in
    /// [`reduce::rehydrate::REASONING_METADATA_KEYS`], any `content_parts`
    /// block whose `"type"` is in
    /// [`reduce::rehydrate::REASONING_CONTENT_PART_TYPES`]) are stripped
    /// BEFORE model-B ever builds a request from this history — then
    /// appends a typed, translatable [`crate::model_change::ModelChangeRecord`]
    /// to [`Self::model_change_records`] (persist it via
    /// [`Self::save_model_change_log`]). A switch TO the current model
    /// (`model == Self::model()`) is treated as a no-op — still exactly
    /// `set_model`'s mechanics, no record for a switch that didn't actually
    /// change anything (and nothing to filter FOR, since there was no
    /// handoff).
    pub fn switch_model(&mut self, model: impl Into<String>) {
        let to = model.into();
        if !self.config.model_switch_allow_switch || self.config.model == to {
            self.set_model(to);
            return;
        }
        let from = self.config.model.clone();
        let touched = reduce::rehydrate::filter_reasoning_artifacts(&mut self.history);
        self.set_model(to.clone());
        self.model_change_log
            .push(crate::model_change::ModelChangeRecord::new(
                self.turn_index,
                from,
                to,
                true,
                touched,
                now_ms(),
            ));
    }

    /// P4c: every [`crate::model_change::ModelChangeRecord`] this agent has
    /// accumulated so far (via [`Self::switch_model`] with `allow_switch`
    /// on). Empty when the knob is off or no switch has happened yet.
    pub fn model_change_records(&self) -> &[crate::model_change::ModelChangeRecord] {
        &self.model_change_log
    }

    /// P4c: persist this agent's accumulated model-change log to `store`
    /// under `name` — the [`crate::model_change::ModelChangeRecord`] analog
    /// of [`Self::save_usage_log`].
    pub fn save_model_change_log(
        &self,
        store: &crate::store::SessionStore,
        name: &str,
    ) -> Result<()> {
        store.save_model_change_log(name, &self.model_change_log)
    }

    /// P4e (§1.6/§3.1 `core.session.git_metadata`, catalog:331): this
    /// agent's captured git provenance, if [`Config::session_git_metadata`]
    /// was on at construction and the best-effort probe found a repo.
    pub fn git_metadata(&self) -> Option<&crate::git_metadata::GitMetadataRecord> {
        self.git_metadata.as_ref()
    }

    /// P4e: persist this agent's captured git metadata to `store` under
    /// `name` — a thin wrapper over
    /// [`crate::store::SessionStore::save_git_metadata`], the
    /// [`crate::git_metadata::GitMetadataRecord`] analog of
    /// [`Self::save_usage_log`]. A no-op (`Ok(())`, nothing written) when
    /// [`Self::git_metadata`] is `None`.
    pub fn save_git_metadata(&self, store: &crate::store::SessionStore, name: &str) -> Result<()> {
        match &self.git_metadata {
            Some(record) => store.save_git_metadata(name, record),
            None => Ok(()),
        }
    }

    /// P4e DEFECT-FIX (independent Fable-5 review of P4e: `core.session.persist`
    /// had a `Config` field and CLI plumbing at `ConfigProfile` → `Config` but
    /// no consumer at all): whether a CLI caller's session-store save sites
    /// (`persist_session`, `persist_full_view`) should actually write to
    /// disk. `true` (the default) is byte-identical to pre-fix behavior —
    /// every session persists. `false` makes a session ephemeral: it runs
    /// exactly as before, but no `<name>.jsonl`/sidecar family is ever
    /// written for it. A plain getter, same posture as [`Self::model`] —
    /// this crate itself never reads or enforces it; the CLI's save sites do.
    pub fn session_persist(&self) -> bool {
        self.config.session_persist
    }

    /// P4e DEFECT-FIX (independent Fable-5 review of P4e: `core.session.name`
    /// had a `Config` field and CLI plumbing but no consumer): the
    /// caller-configured session name, if `[core.session] name` was set.
    /// `None` (the default) leaves session naming exactly as before —
    /// `mint_session_name`'s auto-generated `<tag>-<adjective>-<noun>` shape.
    /// A plain getter, same posture as [`Self::session_persist`].
    pub fn session_name(&self) -> Option<&str> {
        self.config.session_name.as_deref()
    }

    /// PARITY-18 D3 — whether this agent has actually issued at least one
    /// live request to its [`Provider`] so far (set the instant
    /// [`Self::run_loop`] reaches its real send site, regardless of whether
    /// that call then succeeds or fails). Callers should report
    /// "request sent" from THIS, never from having merely passed the
    /// context guard or having called [`Self::send`] — either of those can
    /// happen with zero requests actually issued (a guard refusal, an
    /// interactive session quit before any turn completes).
    pub fn request_issued(&self) -> bool {
        self.requests_issued
    }

    /// P5-2 (§2.2 C2): whether this agent currently considers its
    /// [`CachePlan::ImportedPrefix`] cache entry warm — mirrors
    /// [`Self::request_issued`]'s read-only-observability precedent, so a
    /// caller (or a test) can confirm [`Self::register_tool`]'s C2
    /// invalidation actually took effect without reaching into private
    /// state.
    pub fn cache_established(&self) -> bool {
        self.cache_established
    }

    /// B7: length of the imported-prefix protected by [`CachePlan::ImportedPrefix`]
    /// (this agent's own system message plus every message of a
    /// previously-imported session), set by [`Self::load_session`]. `None`
    /// until a session has been loaded.
    pub fn imported_prefix_len(&self) -> Option<usize> {
        self.imported_prefix_len
    }

    /// Replace this agent's accumulating reduction log (C4: `/expand`/`/reduce`
    /// mutate the log directly via `reduce::invert_one`/`reduce::project_messages`
    /// and must feed the result back here so the *next* request build or
    /// persist sees the updated state instead of silently recomputing from an
    /// empty log). Also lets a caller (`resume_cmd`, C1) seed the log with the
    /// initial projection it already computed for the entry banner, so
    /// `reduction_log()` reflects reality even before this agent's first
    /// `send()` (which is otherwise the only place `build_request_messages`
    /// populates it).
    pub fn set_reduction_log(&mut self, log: ReductionLog) {
        self.reduction_log = log;
    }

    /// Replace the conversation with a loaded session, keeping this agent's own
    /// system prompt at the front. The session's own system/developer turns are
    /// preserved after it for context.
    pub fn load_session(&mut self, session: Session) {
        let system = self.history.first().cloned();
        self.history.clear();
        if let Some(sys) = system {
            self.history.push(sys);
        }
        self.history.extend(session.messages);
        // B7: the whole of `history` at this point — this agent's own system
        // message plus every imported message — is the stable prefix a
        // resumed session resends byte-identically every turn.
        self.imported_prefix_len = Some(self.history.len());
        // UX-26 (B7-warn): a freshly loaded prefix has no established cache
        // entry of THIS agent's own making yet (even if this agent was
        // resumed once before — that earlier prefix is gone). Seed the
        // activity clock from the loaded session's own last message
        // timestamp (walking backward past any trailing message that
        // carries none), so a session that's been sitting idle since
        // Claude Code/Codex/a prior supercode run last touched it is
        // correctly treated as already-cold on its very first turn here —
        // `None` (no timestamp anywhere in the loaded messages) leaves the
        // TTL check disarmed rather than guessing.
        self.cache_established = false;
        self.last_cache_activity_ms = self
            .history
            .iter()
            .rev()
            .find_map(|m| m.metadata.get("timestamp"))
            .and_then(|ts| crate::sidecar::rfc3339_to_ms(ts));
    }

    /// Append `msg` to the sidecar recorder (A3), if one is installed — a
    /// no-op, at zero cost, when `recorder` is `None` (today's behavior).
    fn record(&mut self, msg: &ChatMessage) -> Result<()> {
        if let Some(recorder) = self.recorder.as_mut() {
            recorder.append(msg)?;
        }
        Ok(())
    }

    /// Persist the live conversation to `path` as JSONL (one [`ChatMessage`]
    /// per line) so the session can be resumed later — supercode's own sessions
    /// become first-class, resumable artifacts.
    pub fn save_transcript(&self, path: impl AsRef<std::path::Path>) -> Result<()> {
        let mut out = String::new();
        for m in &self.history {
            out.push_str(&serde_json::to_string(m).map_err(Error::Decode)?);
            out.push('\n');
        }
        std::fs::write(path, out)?;
        Ok(())
    }

    /// Restore a conversation previously written with [`Self::save_transcript`],
    /// replacing the current history.
    pub fn load_transcript(&mut self, path: impl AsRef<std::path::Path>) -> Result<()> {
        let text = std::fs::read_to_string(path)?;
        let mut history = Vec::new();
        for line in text.lines().map(str::trim).filter(|l| !l.is_empty()) {
            history.push(serde_json::from_str::<ChatMessage>(line).map_err(Error::Decode)?);
        }
        self.history = history;
        Ok(())
    }

    /// Take a checkpoint of the current conversation position. Pass it to
    /// [`Self::rewind_to`] to discard everything sent since (the rewind/undo
    /// analog of `fork`/checkpoint).
    pub fn checkpoint(&self) -> usize {
        self.history.len()
    }

    /// Rewind the conversation to a [`Self::checkpoint`], discarding later turns.
    pub fn rewind_to(&mut self, checkpoint: usize) {
        self.history.truncate(checkpoint.min(self.history.len()));
    }

    /// Send a message with file inputs attached — the `--file` / `-i` analog.
    /// Each file's contents are injected into the prompt: UTF-8 text inline,
    /// binary (e.g. images) noted with a size marker. (Native image *vision*
    /// would additionally require multimodal content parts.)
    pub async fn send_with_files(
        &mut self,
        text: impl Into<String>,
        files: &[std::path::PathBuf],
    ) -> Result<String> {
        let mut prompt = text.into();
        for path in files {
            let block = match std::fs::read(path) {
                Ok(bytes) => match String::from_utf8(bytes.clone()) {
                    Ok(s) => format!("\n\n[file: {}]\n{}", path.display(), s),
                    Err(_) => format!(
                        "\n\n[file: {} — {} bytes, binary content omitted]",
                        path.display(),
                        bytes.len()
                    ),
                },
                Err(e) => format!("\n\n[file: {} — could not read: {e}]", path.display()),
            };
            prompt.push_str(&block);
        }
        let expanded = self.expand_prompt_async(&prompt).await;
        let msg = ChatMessage::user(expanded);
        self.guard_candidate_message(&msg)?;
        self.record(&msg)?;
        self.history.push(msg);
        self.run_loop().await
    }

    /// Send a message with image inputs to a vision model — the `-i/--image`
    /// analog. `image_urls` may be `https://…` links or `data:image/…;base64,…`
    /// URLs; they're attached as multimodal `image_url` content parts.
    pub async fn send_with_images(
        &mut self,
        text: impl Into<String>,
        image_urls: &[String],
    ) -> Result<String> {
        let expanded = self.expand_prompt_async(&text.into()).await;
        let msg = ChatMessage::user_with_images(expanded, image_urls);
        self.guard_candidate_message(&msg)?;
        self.record(&msg)?;
        self.history.push(msg);
        self.run_loop().await
    }

    /// Expand a `/<name> <args>` slash command against the registered prompt
    /// templates (`{args}` is replaced with the trailing text). Non-matching
    /// input is returned unchanged.
    pub fn expand_prompt(&self, input: &str) -> String {
        let trimmed = input.trim_start();
        let Some(rest) = trimmed.strip_prefix('/') else {
            return input.to_string();
        };
        let (name, args) = match rest.split_once(char::is_whitespace) {
            Some((n, a)) => (n, a.trim()),
            None => (rest, ""),
        };
        match self.config.prompts.get(name) {
            Some(template) => template.replace("{args}", args),
            None => input.to_string(),
        }
    }

    /// P5-2 (§2 module 15 D7 row 4 "prompts-as-commands"): like
    /// [`Self::expand_prompt`], but also consults MCP-server-sourced
    /// prompts registered via [`Self::register_mcp_prompt`] when the local
    /// `Config::prompts` table has no match — a live `prompts/get`
    /// round-trip, which is why this is async and [`Self::expand_prompt`]
    /// itself stays synchronous (its public sync signature is unchanged,
    /// for every existing caller that doesn't need MCP prompts).
    ///
    /// **Argument mapping (a scope decision, not a protocol requirement —
    /// the MCP spec leaves "how does free CLI text become named prompt
    /// arguments" to the client):** a prompt with zero or one declared
    /// arguments gets the whole trailing text (empty string if the prompt
    /// takes no arguments and none was given); a prompt with two or more
    /// declared arguments expects `key=value` pairs, whitespace-separated
    /// (`/mcp__server__prompt lang=rust topic=async`) — an unparseable pair
    /// (no `=`) is simply skipped, never a hard error (matches this
    /// method's "non-matching input passes through" fail-open posture for
    /// the LOCAL-prompt case above).
    pub async fn expand_prompt_async(&self, input: &str) -> String {
        let local = self.expand_prompt(input);
        if local != input {
            return local; // a local `Config::prompts` template matched
        }
        let trimmed = input.trim_start();
        let Some(rest) = trimmed.strip_prefix('/') else {
            return input.to_string();
        };
        let (name, args) = match rest.split_once(char::is_whitespace) {
            Some((n, a)) => (n, a.trim()),
            None => (rest, ""),
        };
        let Some(source) = self.mcp_prompts.get(name) else {
            return input.to_string();
        };
        let arg_map = match source.arg_names() {
            [] => std::collections::BTreeMap::new(),
            [single] => {
                let mut m = std::collections::BTreeMap::new();
                if !args.is_empty() {
                    m.insert(single.clone(), args.to_string());
                }
                m
            }
            _ => args
                .split_whitespace()
                .filter_map(|pair| pair.split_once('='))
                .map(|(k, v)| (k.to_string(), v.to_string()))
                .collect(),
        };
        match source.render(arg_map).await {
            Ok(rendered) => rendered,
            Err(e) => format!("Error: mcp prompt `{name}` failed: {e}"),
        }
    }

    /// P5-2 (§2 module 15 D7 row 4): register an MCP server's prompt as a
    /// slash-command source — `command_name` MUST already be the
    /// namespaced `mcp__<server>__<prompt>` form
    /// ([`crate::mcp::McpServerHandle::prompts`] produces exactly that
    /// shape); this method does not re-namespace or validate it, so a
    /// caller that hands it a bare name defeats the collision protection
    /// [`crate::mcp::McpPromptSource`]'s doc comment describes. Overwrites
    /// any prior registration under the same command name (re-attaching
    /// the same server replaces its own earlier prompt list; this can
    /// never touch a NON-`mcp__`-prefixed key, i.e. never a local
    /// `Config::prompts` entry).
    pub fn register_mcp_prompt(
        &mut self,
        command_name: impl Into<String>,
        source: impl crate::sdk::SdkPromptSource + 'static,
    ) {
        self.mcp_prompts
            .insert(command_name.into(), Box::new(source));
    }

    /// P5-2 (§2 module 15 D7 row 5 "instructions"): fold an MCP server's
    /// `initialize`-time instructions (or any other free-text note) into
    /// this agent's system message — the context-assembly site every other
    /// `core.*`/`capabilities.*` prompt-section append already uses
    /// (`Self::with_parts`), except this one fires AFTER construction
    /// (attaching MCP servers happens once the agent already exists — see
    /// `crates/cli/src/main.rs`'s `attach_mcp`). A no-op if `history` is
    /// somehow empty or its first message isn't a system message (never
    /// true for an `Agent` built via `Self::new`/`Self::with_parts`, but
    /// checked rather than assumed).
    pub fn append_system_note(&mut self, text: &str) {
        if let Some(system) = self.history.first_mut() {
            if system.role == Role::System {
                system
                    .content
                    .get_or_insert_with(String::new)
                    .push_str(text);
            }
        }
    }

    /// Compact the conversation if it has grown past the configured
    /// threshold.
    ///
    /// **Re-founded (A10):** with a [`ReductionPolicy`] installed
    /// ([`Self::set_reduction_policy`]), this no longer touches `self.history`
    /// at all. It derives `policy.clear_turns_older_than` from
    /// `compact_after_messages` so the *next* projected request view
    /// (`reduce::project_messages`, built in [`Self::run_loop`]) collapses the
    /// old turns into one reversible `TurnsCleared` stub instead —
    /// `history()` and the sidecar keep every message forever; only the view
    /// shrinks. Returns whether the live (unreduced) history currently
    /// exceeds the threshold, i.e. whether a clearing will actually be
    /// visible in the next projected view.
    ///
    /// **Legacy path (no policy) — LOSSY, kept only for byte-identical
    /// backward compatibility (D6):** destructively rewrites `self.history`,
    /// permanently discarding the dropped middle turns (replaced by a single
    /// non-reversible summary marker that becomes their SOLE remaining copy —
    /// exactly the lossy compaction this reduction layer differentiates
    /// against). Once a sidecar/recorder or a [`ReductionPolicy`] is in play,
    /// prefer installing a policy so this method takes the re-founded path
    /// above instead.
    pub fn maybe_compact(&mut self) -> bool {
        // P4e (§1.5/§3.1 `core.compaction.enabled`, "no master gate exists
        // yet"): checked FIRST, before either trigger — `false` disables
        // every auto-compaction trigger unconditionally (message-count AND
        // pressure), composing with them rather than replacing their own
        // logic. `true` (the default, matching today's pre-P4e behavior,
        // where nothing ever gated compaction) falls straight through to
        // the existing trigger checks below, unchanged.
        if !self.config.compaction_enabled {
            return false;
        }
        let threshold = self.config.compact_after_messages;
        // P4b (§1.5/§3.1 `core.compaction.reserve_tokens`, pi§2 shape): a
        // SECOND, independent trigger — context-window pressure — alongside
        // (not instead of) the message-count one above. `None` (the
        // default) is byte-identical to today's message-count-only
        // behavior; this whole block is a no-op then.
        let message_trigger = threshold.is_some_and(|t| self.history.len() > t);
        let pressure_trigger = self.compaction_pressure_triggered();
        if threshold.is_none() && self.config.compaction_reserve_tokens.is_none() {
            return false;
        }
        if !message_trigger && !pressure_trigger {
            return false;
        }
        if let Some(policy) = self.reduction_policy.as_mut() {
            if let Some(t) = threshold {
                policy.clear_turns_older_than = Some(t);
            }
            // P4b scope note: the token-PRESSURE trigger's "how much to
            // clear" derivation (below, for the legacy in-place path) has no
            // `ReductionPolicy`/A10 analog yet — that mechanism decides its
            // own clearing window once `clear_turns_older_than` is set, so
            // pressure firing alone (no message threshold configured) has
            // nothing new to hand it in this pass. Report the message-count
            // verdict only, matching today's pre-P4b behavior exactly when
            // only `threshold` is set.
            return message_trigger;
        }
        // Legacy in-place compaction (no `ReductionPolicy` installed) below.
        // `keep_recent`: the message-count trigger's own `threshold / 2`
        // shape when it's what fired (or both fired); otherwise (pressure
        // fired alone) a token-budget-derived count.
        let keep_recent = if message_trigger {
            (threshold.unwrap() / 2).max(2)
        } else {
            self.keep_recent_count_by_tokens()
        };
        if self.history.len() <= keep_recent {
            return false;
        }
        // Indices: 0 is the system prompt; collapse [first .. len-keep_recent).
        // `first` is 1 (only the system prompt is ever auto-preserved) unless
        // B7's coordination clamp widens it.
        let mut first = 1usize;
        // B7 coordination clamp: this legacy (no-`ReductionPolicy`) path
        // mutates `self.history` directly, so — unlike the re-founded A10
        // path (clamped inside `reduce::project_messages`, threaded from
        // `build_request_messages`) — it must clamp itself. Widening `first`
        // (not `cut`) is what actually protects the imported prefix: the
        // drop range is `[first, cut)`, so raising `cut` alone would only
        // drop MORE messages, not fewer. `imported_prefix_len` is already an
        // absolute `history` index count (it protects `history[0..len]`), so
        // no offset conversion is needed here.
        if matches!(self.config.cache_plan, CachePlan::ImportedPrefix) {
            if let Some(protected) = self.imported_prefix_len {
                first = first.max(protected);
            }
        }
        let mut cut = self.history.len() - keep_recent;
        if cut <= first {
            return false;
        }
        // Never begin the kept window on a tool result: its originating
        // assistant turn (with the matching `tool_calls`) is about to be
        // dropped, which would orphan the tool message and make the replayed
        // conversation invalid. Advance past any leading tool results.
        while cut < self.history.len() && self.history[cut].role == Role::Tool {
            cut += 1;
        }
        if cut >= self.history.len() {
            return false;
        }
        let dropped = cut - first;
        // P4b (§1.5/§3.1 `core.compaction.focus_instructions`, catalog D2
        // "no instruction steering" gap): appended to the marker whenever
        // set, regardless of which trigger fired. `None` (the default)
        // leaves this byte-identical to the pre-P4b marker text.
        let summary_text = match &self.config.compaction_focus_instructions {
            Some(focus) if !focus.is_empty() => format!(
                "[earlier conversation compacted: {dropped} message(s) summarized to save context]\n\nFocus: {focus}"
            ),
            _ => format!(
                "[earlier conversation compacted: {dropped} message(s) summarized to save context]"
            ),
        };
        let summary = ChatMessage::system(summary_text);
        let mut new_history = Vec::with_capacity(first + keep_recent + 2);
        new_history.extend(self.history[..first].iter().cloned());
        new_history.push(summary);
        new_history.extend(self.history.split_off(cut));
        self.history = new_history;
        true
    }

    /// P4b (§1.5/§3.1 `core.compaction.reserve_tokens`, pi§2 shape:
    /// `contextTokens > contextWindow - reserveTokens`): whether the
    /// estimated token size of the live history is within `reserve_tokens`
    /// of the model's context window. `false` when
    /// [`Config::compaction_reserve_tokens`] is unset (the default).
    fn compaction_pressure_triggered(&self) -> bool {
        let Some(reserve) = self.config.compaction_reserve_tokens else {
            return false;
        };
        let limit = provider::model_context_limit(&self.config.model)
            .unwrap_or(provider::UNKNOWN_MODEL_CONTEXT_FLOOR);
        let used = crate::tokens::estimate_view_tokens(&self.history);
        used.saturating_add(reserve) > limit
    }

    /// P4b (§1.5/§3.1 `core.compaction.keep_recent_tokens`): how many of the
    /// most recent messages (walking backward from the end of `self.history`,
    /// skipping the system prompt) fit within the configured token budget
    /// (default 20,000, pi§6 precedent). Always keeps at least 2 messages,
    /// matching the message-count trigger's own floor.
    fn keep_recent_count_by_tokens(&self) -> usize {
        let budget = self.config.compaction_keep_recent_tokens.unwrap_or(20_000);
        let mut used = 0u64;
        let mut count = 0usize;
        for msg in self.history.iter().skip(1).rev() {
            let t = crate::tokens::estimate_view_tokens(std::slice::from_ref(msg));
            if used.saturating_add(t) > budget && count > 0 {
                break;
            }
            used = used.saturating_add(t);
            count += 1;
        }
        count.max(2)
    }

    /// Register an additional tool (e.g. your own capability).
    ///
    /// P5-2 (§2.2 C2 "connect invalidates cache prefix"): registering a
    /// tool AFTER this agent has already issued a request
    /// ([`Self::request_issued`]) changes the tools schema every
    /// subsequent request carries — the exact prefix-churn shape C2
    /// describes, MCP-sourced or not. Resets [`Self::cache_established`] so
    /// the next cache-warmth check (`provider::cache_cold_reason`) doesn't
    /// wrongly assume the entry is still warm. A no-op call before the
    /// first request (the common case: `attach_mcp` registers tools once at
    /// startup, before any turn runs) changes nothing — byte-identical to
    /// today.
    pub fn register_tool(&mut self, tool: impl crate::tools::Tool + 'static) {
        self.registry.register(tool);
        if self.requests_issued {
            self.cache_established = false;
        }
    }

    /// The current conversation, including the system prompt.
    pub fn history(&self) -> &[ChatMessage] {
        &self.history
    }

    /// Send a user message and run the loop until the model produces a final
    /// answer (text with no tool calls) or the iteration budget is exhausted.
    pub async fn send(&mut self, user_input: impl Into<String>) -> Result<String> {
        let expanded = self.expand_prompt_async(&user_input.into()).await;
        let msg = ChatMessage::user(expanded);
        self.guard_candidate_message(&msg)?;
        self.record(&msg)?;
        self.history.push(msg);
        self.run_loop().await
    }

    /// Refuse an oversized new user turn before it mutates canonical history
    /// or an attached sidecar. The in-loop guard remains authoritative for
    /// every actual request; this preflight closes the first-request seam
    /// where `send*` used to record/push the message before that guard ran.
    fn guard_candidate_message(&self, msg: &ChatMessage) -> Result<()> {
        let Some(limit) = self.context_limit else {
            return Ok(());
        };

        let messages = match &self.reduction_policy {
            None => {
                let mut messages = self.history.clone();
                messages.push(msg.clone());
                messages
            }
            Some(policy) => {
                let has_system = self.history.first().is_some_and(|m| m.role == Role::System);
                let mut reducible = self.history[usize::from(has_system)..].to_vec();
                reducible.push(msg.clone());
                let mut prepared = policy.clone();
                reduce::prepare_read_freshness(&mut prepared, &reducible);
                let (view, _) =
                    reduce::project_messages(&reducible, &prepared, &self.reduction_log);
                let mut messages = Vec::with_capacity(view.len() + usize::from(has_system));
                if has_system {
                    messages.push(self.history[0].clone());
                }
                messages.extend(view);
                messages
            }
        };
        let messages =
            provider::apply_cache_plan(&messages, self.config.cache_plan, self.imported_prefix_len);
        let tools = self.tool_schemas();
        let (fits, projected_tokens) = crate::tokens::context_guard(&messages, &tools, limit);
        if !fits {
            return Err(Error::ContextLimitExceeded {
                projected_tokens,
                reserve_tokens: crate::tokens::CONTEXT_RESPONSE_RESERVE_TOKENS,
                context_limit: limit,
                model: self.config.model.clone(),
            });
        }
        Ok(())
    }

    /// The messages a provider request should carry for the CURRENT turn
    /// (A5/A7/A8/A10): with no [`ReductionPolicy`] installed, exactly
    /// `self.history.clone()` — byte-identical to every version of this
    /// method before reduction landed. With a policy installed, `history[0]`
    /// (this agent's own system prompt, never a reduction target) followed by
    /// [`reduce::project_messages`]'s projected view of `history[1..]`, fed
    /// with `self.reduction_log` so already-applied reductions reproduce
    /// verbatim across turns (prefix stability, A5) — the updated log is
    /// stored back onto `self` so the NEXT call (this turn, next turn, or a
    /// later `send`) sees the same accumulating state. `self.history` itself
    /// is never read back into or mutated by this: it stays the full
    /// canonical view, in lockstep with the sidecar (A3).
    ///
    /// When `policy.elide_stale_reads` is set, this re-runs
    /// [`reduce::probe_read_freshness`] (the one place A8's disk I/O happens)
    /// against `history[1..]` before projecting, so every request sees
    /// up-to-date freshness verdicts — `project_messages` itself stays pure.
    ///
    /// Finally, B7's [`provider::apply_cache_plan`] runs over the assembled
    /// view (regardless of whether a [`ReductionPolicy`] is installed) — a
    /// pure, cloning annotation step, so this method's `&mut self` mutations
    /// above (`self.reduction_log`) are already committed before it runs and
    /// its own output is never written back onto `self.history` or the log:
    /// purity for B7's cache breakpoints holds independently of A5's.
    fn build_request_messages(&mut self) -> Vec<ChatMessage> {
        let messages = match self.reduction_policy.clone() {
            None => self.history.clone(),
            Some(mut policy) => {
                reduce::prepare_read_freshness(&mut policy, &self.history[1..]);
                // B7 coordination clamp: while `CachePlan::ImportedPrefix` is
                // active, A10 turn-clearing must never establish a range
                // that dips into the imported prefix (protects the cache
                // breakpoint the request build will place there below).
                // `imported_prefix_len` counts `history[0]` (this agent's own
                // system message) plus the imported messages, but
                // `project_messages` only ever sees `history[1..]` — hence
                // the `- 1`.
                if matches!(self.config.cache_plan, CachePlan::ImportedPrefix) {
                    policy.protect_imported_prefix =
                        self.imported_prefix_len.map(|n| n.saturating_sub(1));
                }
                // TR-7 (T20): the one side-call site, run BEFORE
                // `project_messages` (which stays pure/I-O-free) — mirrors
                // `elide_stale_reads`/`probe_read_freshness` immediately
                // above. Only ever does anything when both the policy gate
                // AND a summarizer are present; either being absent means
                // `cleared_turns_summary` stays `None` and `project_messages`
                // renders the deterministic stub, same as before TR-7
                // existed.
                if policy.summarize_cleared_turns {
                    if let Some(summarizer) = self.span_summarizer.as_deref() {
                        policy.cleared_turns_summary = reduce::prepare_cleared_turns_summary(
                            &self.history[1..],
                            &policy,
                            &self.reduction_log,
                            summarizer,
                        );
                    }
                }
                let (view, log) =
                    reduce::project_messages(&self.history[1..], &policy, &self.reduction_log);
                self.reduction_log = log;
                let mut messages = Vec::with_capacity(view.len() + 1);
                messages.push(self.history[0].clone());
                messages.extend(view);
                messages
            }
        };
        // TR-8 (T5): a tool-schema tier change since the last request is a
        // cache-bust event under `CachePlan::ImportedPrefix` — the `tools`
        // array is part of the cache key alongside `messages`, so flag it by
        // skipping this one request's cache annotation rather than claiming
        // a prefix hit that won't actually land. Recorded unconditionally
        // (even under `CachePlan::Off`) so the signature stays current
        // regardless of which plan is active.
        let tier_sig = self.schema_tier_signature();
        let busted =
            provider::tier_change_is_cache_bust(self.last_tool_schema_tier_signature, tier_sig);
        self.last_tool_schema_tier_signature = Some(tier_sig);
        let effective_cache_plan = if busted {
            CachePlan::Off
        } else {
            self.config.cache_plan
        };
        // UX-26 (B7-warn): mirror `apply_cache_plan`'s own placement gate
        // (`ImportedPrefix` AND a non-zero prefix) to know whether THIS
        // request will actually carry a `cache_control` annotation. `busted`
        // requests (schema-tier change) and `CachePlan::Off` never annotate,
        // so `provider::cache_cold_reason` can never flag them — there was
        // nothing to reuse, by construction. `idle_secs` is computed
        // whenever a signal exists at all (even before this agent's first
        // annotated send — see `Self::last_cache_activity_ms`'s doc comment
        // on why the pre-establishment case matters); `cache_established`
        // additionally gates the usage-ratio check specifically (see
        // `provider::cache_cold_reason`'s doc comment for why those two
        // checks need independent gates).
        let will_annotate = matches!(effective_cache_plan, CachePlan::ImportedPrefix)
            && self.imported_prefix_len.is_some_and(|n| n > 0);
        let idle_secs = self
            .last_cache_activity_ms
            .map(|last| (now_ms() - last).max(0) / 1000);
        self.pending_cache_turn = (will_annotate, self.cache_established, idle_secs);
        provider::apply_cache_plan(&messages, effective_cache_plan, self.imported_prefix_len)
    }

    /// Run the model/tool loop over the current history until a final answer or
    /// the iteration budget is exhausted. (Shared by `send`, `send_with_files`,
    /// and `send_with_images`.)
    /// P4b (§1.7, pi§3 semantics): pop the next message(s) to deliver from
    /// `queue` per `mode` — `All` drains everything and joins it with a
    /// blank line, `OneAtATime` pops exactly one. `None` when `queue` is
    /// empty (the default state, at zero cost).
    fn drain_steer_queue(
        queue: &mut std::collections::VecDeque<String>,
        mode: SteeringMode,
    ) -> Option<String> {
        if queue.is_empty() {
            return None;
        }
        match mode {
            SteeringMode::All => Some(queue.drain(..).collect::<Vec<_>>().join("\n\n")),
            SteeringMode::OneAtATime => queue.pop_front(),
        }
    }

    async fn run_loop(&mut self) -> Result<String> {
        let _steer_turn = SteerTurnGuard::new(self.steer_queue.clone());
        let mut output_tokens_used: u64 = 0;

        // P5-9 (§2 module 20, cc's "per-prompt file-history-snapshot"):
        // open a fresh checkpoint for THIS turn — `run_loop` is called
        // exactly once per `send`/`send_with_files`/`send_with_images`
        // call (never recursively for the same turn), so this fires once
        // per user prompt, matching the design's per-prompt granularity.
        // `self.history.last()` is the user message that call just pushed.
        // `None` (`checkpoint_observer` unset, the default) is a no-op —
        // zero cost, no disk touched.
        if let Some(cp) = &self.checkpoint_observer {
            let label = self
                .history
                .last()
                .and_then(|m| m.content.as_deref())
                .unwrap_or("")
                .to_string();
            cp.begin_turn(&label);
        }

        for _ in 0..self.config.max_iterations {
            self.maybe_compact();
            // P4b (§1.7, pi§3 "steer = after current tool calls"): drain any
            // queued mid-turn steering message(s) BEFORE building the next
            // request — the top of every loop iteration is exactly "after
            // whatever tool calls the previous iteration just ran" (or, on
            // the very first iteration, before anything has happened yet,
            // which is an equally valid "deliver immediately" reading).
            // Empty queue (today's default state) is a no-op.
            let steer_msg = {
                let mut inbox = self
                    .steer_queue
                    .lock()
                    .unwrap_or_else(std::sync::PoisonError::into_inner);
                inbox.drain(self.config.steering_mode)
            };
            if let Some(steer_msg) = steer_msg {
                let msg = ChatMessage::user(steer_msg);
                self.record(&msg)?;
                self.history.push(msg);
            }
            // Recomputed every iteration (not hoisted): under `Deferred`
            // advertising, a `tool_search` call earlier in this same loop
            // activates tools that must be advertised starting with the very
            // next request (B6).
            let tools = self.tool_schemas();
            let messages = self.build_request_messages();

            // PARITY-18 D4 — re-check the context guard before EVERY
            // request this loop builds, not just the caller's one-shot
            // preflight: interactive turns 2+, `/expand all`, and any
            // mid-loop tool round-trip that grows `messages` can push a
            // barely-passing session over the limit between sends. Only
            // armed when a caller has opted in via `set_context_limit`.
            // Uses the exact same `tokens::context_guard`
            // formula the CLI preflight uses, so the two can never disagree.
            if let Some(limit) = self.context_limit {
                let (fits, projected_tokens) =
                    crate::tokens::context_guard(&messages, &tools, limit);
                if !fits {
                    return Err(Error::ContextLimitExceeded {
                        projected_tokens,
                        reserve_tokens: crate::tokens::CONTEXT_RESPONSE_RESERVE_TOKENS,
                        context_limit: limit,
                        model: self.config.model.clone(),
                    });
                }
            }

            let req = ChatRequest {
                model: self.config.model.clone(),
                messages,
                tools,
                temperature: self.config.temperature,
                max_tokens: self.config.max_tokens,
                effort: self.config.effort.clone(),
                response_format: self.config.response_format.clone(),
                extra_body: self.config.extra_body.clone(),
            };

            let (mut assistant, usage) = {
                let sink = self.config.event_sink.as_ref();
                let on_delta = move |s: &str| {
                    if let Some(sink) = sink {
                        sink(AgentEvent::TextDelta(s.to_string()));
                    }
                };
                // PARITY-18 D3 — the real send site: flip the flag
                // immediately before issuing the request, regardless of
                // whether `complete` then succeeds or fails, so
                // `request_issued()` truthfully reflects "a live request
                // was attempted" rather than "the run reached this line and
                // later succeeded."
                self.requests_issued = true;
                // P4b (§1.1/§3.1 `core.retry`, pi§3 shape): retry-with-
                // backoff already lives at the TRANSPORT layer
                // (`provider::OpenAiProvider::send_with_retry`, pre-existing
                // — connection failures and 5xx responses are retried
                // there); `Config.retry_*` (see `Agent::new`) makes that
                // EXISTING mechanism config-file-settable instead of
                // duplicating a second retry loop here, which would nest
                // retries confusingly on top of the transport's own.
                self.provider.complete(&req, &on_delta).await?
            };
            // Persist the actual generating model on the message itself.
            // A resumed foreign session keeps its original model in
            // `SessionMeta`; using only that session-level value on export
            // misattributes every Supercode continuation turn to the source
            // harness model. Per-message provenance lets native exporters
            // preserve the boundary accurately (for example, Claude history
            // followed by a GLM continuation).
            assistant
                .metadata
                .insert("model".to_string(), self.config.model.clone());
            output_tokens_used += usage.completion_tokens;
            self.total_output_tokens += usage.completion_tokens;

            // UX-26 (B7-warn): consult the verdict computed at build time
            // (before this request was sent) now that `usage` — the only
            // piece that couldn't be known pre-send — is in hand. Gated on
            // `Config::cache_warnings` (default on; `--no-cache-warnings` /
            // `SUPERCODE_CACHE_WARNINGS=0` at the CLI layer, dev/03) so this
            // stays a zero-behavior-change no-op for every caller that
            // hasn't opted into `CachePlan::ImportedPrefix` in the first
            // place (`pending_cache_turn.0` is `false` whenever
            // `CachePlan::Off`, so the predicate always returns `None` then
            // regardless of this flag).
            let (will_annotate, cache_established, idle_secs) = self.pending_cache_turn;
            // UX-26 T2 (accuracy fold-in): `CacheColdReason::message` asserts
            // Anthropic-specific facts (a fixed 5-minute ephemeral TTL, and
            // cache-read-ratio semantics that assume Anthropic's exact-count
            // billing) that are only true for Anthropic-family models. This
            // is a WARNING-only gate, deliberately not folded into
            // `will_annotate`/the breakpoint-placement gate above: whether a
            // `cache_control` breakpoint is safe/inert to send to a
            // non-Anthropic model through OpenRouter is a separate cache-
            // behavior question this ticket doesn't touch (see
            // `.volter/tracker/markdown/UX-26.md`'s T2 note) — narrowing only
            // the warning keeps this fix scoped to warning ACCURACY, with
            // zero change to what gets sent on the wire.
            let warning_applies_to_this_model =
                provider::is_anthropic_family_model(&self.config.model);
            if self.config.cache_warnings && warning_applies_to_this_model {
                if let Some(reason) =
                    provider::cache_cold_reason(will_annotate, cache_established, idle_secs, &usage)
                {
                    self.emit(AgentEvent::CacheWarning {
                        message: reason.message(),
                    });
                }
            }
            // Refresh the activity clock / establish-once flag for the NEXT
            // turn's comparison, but only when THIS request actually carried
            // the annotation — an unannotated (busted/Off) request neither
            // warms nor cools a cache entry it never touched.
            if will_annotate {
                self.last_cache_activity_ms = Some(now_ms());
                self.cache_established = true;
            }

            // UX-23: emitted before `TurnCompleted` so a `--trace`/
            // `stream-json` consumer sees "this round-trip cost N tokens"
            // land right alongside the round-trip it describes, rather than
            // needing to correlate it with a later event.
            self.emit(AgentEvent::Usage(usage.clone()));
            self.emit(AgentEvent::TurnCompleted);
            // P4b (§1.6, catalog §4a "persisted per-turn usage records"):
            // EventSink already streamed `Usage` above — this durably
            // accumulates the same data as a typed record (see
            // `Self::usage_records`/`Self::save_usage_log`), never a lossy
            // display-only channel.
            self.usage_log
                .push(crate::usage_log::UsageRecord::from_usage(
                    self.turn_index,
                    &self.config.model,
                    &usage,
                    now_ms(),
                ));
            self.turn_index += 1;
            self.record(&assistant)?;
            self.history.push(assistant.clone());

            let calls = assistant.tool_calls().to_vec();
            if calls.is_empty() {
                // Close steering acceptance under the same lock as the last
                // drain. A message accepted before this boundary extends the
                // current turn; anything later is rejected by the SDK and
                // can never leak into a future turn.
                let steer_msg = self
                    .steer_queue
                    .lock()
                    .unwrap_or_else(std::sync::PoisonError::into_inner)
                    .drain_or_close(self.config.steering_mode);
                if let Some(steer_msg) = steer_msg {
                    let msg = ChatMessage::user(steer_msg);
                    self.record(&msg)?;
                    self.history.push(msg);
                    continue;
                }
                // P4b (§1.7, pi§3 "follow-up = at idle"): a queued follow-up
                // message takes priority over the stop-gate — it's more
                // input to answer, not a veto of an answer already given.
                if let Some(follow_up_msg) =
                    Self::drain_steer_queue(&mut self.follow_up_queue, self.config.follow_up_mode)
                {
                    let msg = ChatMessage::user(follow_up_msg);
                    self.record(&msg)?;
                    self.history.push(msg);
                    continue;
                }
                // P4b (§1.9/§3.1 `[core] stop_gate`, D3 "stop/completion
                // gating"): consulted exactly once per iteration that would
                // otherwise return — computed into an owned `Option<String>`
                // so the immutable borrow of `self.config.stop_gate` ends
                // before the `self.record`/`self.history.push` calls below
                // need `&mut self`.
                let final_content = assistant.content.clone().unwrap_or_default();
                let veto_reason: Option<String> = self
                    .config
                    .stop_gate
                    .as_ref()
                    .and_then(|gate| gate(&final_content));
                if let Some(reason) = veto_reason {
                    let msg = ChatMessage::user(reason);
                    self.record(&msg)?;
                    self.history.push(msg);
                    continue;
                }
                return Ok(assistant.content.unwrap_or_default());
            }

            // Output-token budget (output only — input tokens are not counted,
            // so this does not bound cost): stop spawning further model turns
            // once the cumulative output-token budget for this `send` is
            // exhausted.
            if let Some(budget) = self.config.max_total_output_tokens {
                if output_tokens_used >= budget {
                    // The assistant turn we just pushed carries unanswered
                    // tool_calls. Leaving them dangling yields an invalid
                    // history (assistant tool_calls with no tool results) that
                    // the provider rejects on the next `send`/resume. Emit
                    // synthetic results so the transcript stays well-formed.
                    for call in &calls {
                        let msg = ChatMessage::tool_result(
                            call.id.clone(),
                            call.function.name.clone(),
                            "[skipped: output token budget reached]".to_string(),
                        );
                        self.record(&msg)?;
                        self.history.push(msg);
                    }
                    return Ok(assistant.content.clone().unwrap_or_default());
                }
            }

            // P4e (§3.1 `core.parallel_tool_calls`, catalog:59): off (the
            // default) or a single call takes the EXACT pre-P4e sequential
            // path below, byte-identical. Only `true` with 2+ calls in this
            // turn takes `Self::run_tools_concurrently` — see its doc
            // comment for exactly what does and doesn't run concurrently.
            if self.config.parallel_tool_calls && calls.len() > 1 {
                for call in &calls {
                    self.emit(AgentEvent::tool_started(call));
                }
                let results = self.run_tools_concurrently(&calls).await;
                for (call, (output, is_error)) in calls.iter().zip(results) {
                    self.emit(AgentEvent::ToolCallCompleted {
                        id: call.id.clone(),
                        name: call.function.name.clone(),
                        output: output.clone(),
                        is_error,
                    });
                    self.apply_tool_result(call, output, is_error)?;
                }
            } else {
                for call in &calls {
                    self.emit(AgentEvent::tool_started(call));
                    let (output, is_error) = self.run_tool(call).await;
                    self.emit(AgentEvent::ToolCallCompleted {
                        id: call.id.clone(),
                        name: call.function.name.clone(),
                        output: output.clone(),
                        is_error,
                    });
                    self.apply_tool_result(call, output, is_error)?;
                }
            }
        }

        Err(Error::MaxIterations(self.config.max_iterations))
    }

    /// The exact post-execution handling every tool result gets, regardless
    /// of whether it was produced by the sequential loop or
    /// [`Self::run_tools_concurrently`] — factored out of `Self::run_loop`'s
    /// tool-dispatch section (P4e) so both paths share one copy: multimodal
    /// image-marker detection, A7 output capping (gated exactly as before),
    /// TR-10 error stamping, and the `record`/`history` append. Always
    /// called in ORIGINAL call order, one call at a time, so the lossless
    /// sidecar's append-order invariant (S1.13) holds regardless of which
    /// dispatch path produced the result.
    fn apply_tool_result(
        &mut self,
        call: &crate::message::ToolCall,
        output: String,
        is_error: bool,
    ) -> Result<()> {
        // P4c (§1.2 `core.tools.read_file.multimodal` / `view_image`):
        // a successful tool result carrying the image-data-URL
        // marker becomes a `content_parts` image block instead of
        // plain text — checked BEFORE `cap_tool_output` (a data URL
        // is not meaningfully "capped" by a byte-length text notice)
        // and recorded identically on both the full and history
        // copies, mirroring `ImageRedacted`'s "images are their own
        // axis, orthogonal to A7 text truncation" treatment
        // (reduce.rs). An ERRORED call never carries the marker (a
        // tool only emits it on success), so `is_error` is not
        // re-checked here.
        if let Some(data_url) = output.strip_prefix(crate::tools::MULTIMODAL_IMAGE_MARKER) {
            let notice = format!("[{}: image content attached below]", call.function.name);
            let full_result = ChatMessage::tool_result_with_image(
                call.id.clone(),
                call.function.name.clone(),
                notice.clone(),
                data_url.to_string(),
            );
            let hist_result = ChatMessage::tool_result_with_image(
                call.id.clone(),
                call.function.name.clone(),
                notice,
                data_url.to_string(),
            );
            self.record(&full_result)?;
            self.history.push(hist_result);
            return Ok(());
        }
        // Record the FULL output before capping (A3): what the
        // sidecar keeps must never be the already-lossy, truncated
        // copy (#8/#40) — `history` alone governs what shrinks.
        let mut full_result =
            ChatMessage::tool_result(call.id.clone(), call.function.name.clone(), output.clone());
        // D6/A7 supersession gate (TR-12 land-blocker fix): `history`
        // is the exact slice `reduce::project_messages` mints A7/A10
        // reduction hashes from (`Self::build_request_messages`
        // below). Capping it here — as this unconditionally used to
        // do — would silently shrink the bytes those hashes cover, so
        // a hash minted now could never recompute the same way once
        // the sidecar is reloaded from disk later (`verify_log`/
        // `invert`, offline). Gate `cap_tool_output` off in exactly
        // the combination where reductions can be minted over
        // `history` AND the full bytes are durably retained: a
        // recorder AND a `ReductionPolicy` both installed. A7 then
        // owns tool-output bounding, reversibly, at projection time
        // (SPEC.md D6/A7) — `history`/the sidecar keep everything,
        // only the request view shrinks. With a policy but no
        // recorder (constructible via `set_reduction_policy` alone),
        // nothing durable backs the full bytes, so capping stays on —
        // the same honest-labeling spirit as `cap_tool_output`'s own
        // retention branch below, just applied at the gate instead of
        // the notice text. With no policy at all, this is untouched:
        // today's byte-identical legacy cap.
        let for_history = if self.recorder.is_some() && self.reduction_policy.is_some() {
            output
        } else {
            self.cap_tool_output(output)
        };
        let mut hist_result =
            ChatMessage::tool_result(call.id.clone(), call.function.name.clone(), for_history);
        if is_error {
            // TR-10: the reduction layer's success/failure boundary
            // (`ReductionKind::ToolInputElided` must never target an
            // errored call — TR-6's territory) has no other
            // structural signal on `ChatMessage`; stamp both the
            // recorded copy (so it survives a sidecar round-trip via
            // `NativeTurn`) and the live-history copy (so an
            // in-process `project_messages` sees it immediately).
            reduce::mark_tool_error(&mut full_result);
            reduce::mark_tool_error(&mut hist_result);
        }
        self.record(&full_result)?;
        self.history.push(hist_result);
        Ok(())
    }

    /// Truncate an oversized tool result so a single runaway command can't blow
    /// up the context window. Cuts on a char boundary and appends a notice.
    fn cap_tool_output(&self, output: String) -> String {
        let Some(max) = self.config.max_tool_output_bytes else {
            return output;
        };
        if max == 0 || output.len() <= max {
            return output;
        }
        // Find the largest char boundary <= max.
        let mut end = max;
        while end > 0 && !output.is_char_boundary(end) {
            end -= 1;
        }
        let total = output.len();
        let mut s = output[..end].to_string();
        // Honest retention labeling (D6, B10-AC4): only claim the sidecar has
        // the full output when a recorder is actually installed.
        let retention = if self.recorder.is_some() {
            "full output in session sidecar"
        } else {
            "full output not retained"
        };
        s.push_str(&format!(
            "{CAP_NOTICE_MARKER}{total} bytes total, showing first {end}; {retention}]"
        ));
        s
    }

    /// P5-3 note on the signature: written as a plain fn returning an
    /// explicitly boxed future (`Pin<Box<dyn Future + Send>>`) rather than
    /// as `async fn`. `spawn_subagent` makes this function genuinely
    /// recursive at the TYPE level: `run_tool` -> `run_spawn_subagent` ->
    /// (a child) `Agent::send` -> `run_loop` -> `run_tool` again — an
    /// `async fn`'s return type is an anonymous, compiler-inferred
    /// self-referential state machine, and inferring one that embeds
    /// itself (even indirectly, through several other functions) is a
    /// compile error (an infinitely-sized/cyclic opaque type). Declaring
    /// `run_tool`'s return type EXPLICITLY as a boxed trait object breaks
    /// the cycle: every other function on the call graph now embeds a
    /// concrete, already-known type here instead of one the compiler would
    /// otherwise need to (cyclically) infer. Callers are unaffected —
    /// `self.run_tool(call).await` reads identically either way.
    fn run_tool<'a>(
        &'a mut self,
        call: &'a crate::message::ToolCall,
    ) -> std::pin::Pin<Box<dyn std::future::Future<Output = (String, bool)> + Send + 'a>> {
        Box::pin(async move {
            let translated_builtin = if self.config.claude_runtime_tools_enabled {
                match self.translate_claude_builtin_call(call) {
                    Ok(translated) => translated,
                    Err(error) => return (format!("Error: {error}"), true),
                }
            } else {
                None
            };
            let call = translated_builtin.as_ref().unwrap_or(call);
            if self.config.claude_runtime_tools_enabled
                && matches!(
                    call.function.name.as_str(),
                    CLAUDE_CRON_CREATE
                        | CLAUDE_CRON_DELETE
                        | CLAUDE_CRON_LIST
                        | CLAUDE_SCHEDULE_WAKEUP
                )
            {
                return self.run_claude_runtime_tool(call);
            }
            // P5-3: `spawn_subagent`/`subagent_status` need full async
            // `&mut self` access (running a child agent's loop, or
            // awaiting an already-finished background `JoinHandle`) —
            // `prepare_tool_call` is purely synchronous, so these are
            // intercepted HERE, one level above it, rather than inside it
            // like `TOOL_SEARCH`/`EXPAND_REDUCTION`/`SIDECAR_SEARCH`.
            if call.function.name == CLAUDE_AGENT && self.config.subagents_claude_agent_alias {
                return match self.translate_claude_agent_call(call) {
                    Ok(translated) => self.run_spawn_subagent(&translated).await,
                    Err(error) => (format!("Error: {error}"), true),
                };
            }
            if call.function.name == SPAWN_SUBAGENT {
                return self.run_spawn_subagent(call).await;
            }
            // P5-3 safety-hardening fix (Fable-5 review, LOW "wrong error
            // when disabled"): gated on `subagents_enabled`, matching
            // `run_spawn_subagent`'s own already-correct disabled behavior
            // (that one gates INTERNALLY, at its own top; this one gates
            // HERE, at the interception point, because unlike
            // `spawn_subagent` it has no other reason to run any logic at
            // all when subagents are off). When disabled, a hallucinated
            // `subagent_status` call must NOT be intercepted — it falls
            // through to `prepare_tool_call`'s normal unknown-tool path
            // below, which returns `Error::UnknownTool("subagent_status")`,
            // byte-identical to the pre-P5-3 (and disabled-spawn_subagent)
            // error text — never `Error::SubagentNotFound`'s "unknown
            // subagent id" text, which would wrongly imply subagents are on
            // but this particular id is bogus.
            if call.function.name == SUBAGENT_STATUS && self.config.subagents_enabled {
                return self.run_subagent_status(call).await;
            }
            match self.prepare_tool_call(call) {
                PreparedCall::Done(result) => result,
                PreparedCall::Ready { name, args } => {
                    // `prepare_tool_call` already confirmed the registry has
                    // this tool.
                    let tool = self.registry.get(&name).expect("prepared as Ready");
                    let (output, is_error) = match tool.execute(args, &self.ctx).await {
                        Ok(out) => (out, false),
                        Err(e) => (format!("Error: {e}"), true),
                    };
                    if let Some(hook) = &self.config.post_tool_hook {
                        hook(&name, &output, is_error);
                    }
                    (output, is_error)
                }
            }
        })
    }

    /// P4e (§3.1 `core.parallel_tool_calls`, catalog:59): the SYNCHRONOUS
    /// half of dispatching one tool call — everything `Self::run_tool` did
    /// BEFORE its single `tool.execute(...).await`, factored out so
    /// [`Self::run_tools_concurrently`] can run these cheap, stateful,
    /// `&mut self` checks (agent intrinsics, unknown-tool, approval,
    /// doom-loop, pre-tool-hook) SEQUENTIALLY and in ORIGINAL call order —
    /// exactly as `run_tool` always has — before handing the remaining
    /// calls' `execute()` futures to `join_all`. `Self::run_tool` itself is
    /// now a thin wrapper over this (a pure refactor: byte-identical
    /// observable behavior, verified by the existing test suite).
    fn prepare_tool_call(&mut self, call: &crate::message::ToolCall) -> PreparedCall {
        let name = &call.function.name;
        if name == TOOL_SEARCH {
            // Agent intrinsic (B6): intercepted before registry lookup, since
            // `Tool::execute` has no access to the registry or `activated_tools`.
            return PreparedCall::Done(self.run_tool_search(call));
        }
        if name == EXPAND_REDUCTION {
            // Agent intrinsic (T12/TR-1): intercepted before registry lookup,
            // same reason — resolves against `self.reduction_log`/`self.history`,
            // which `Tool::execute` has no access to.
            return PreparedCall::Done(self.run_expand_reduction(call));
        }
        if name == SIDECAR_SEARCH {
            return PreparedCall::Done(self.run_sidecar_search(call));
        }
        // P5-6 (§2 module 4 `tools.background`): gated at the interception
        // point itself (not internally, at each method's own top) —
        // mirroring `SUBAGENT_STATUS`'s own fix (Fable-5 review, LOW "wrong
        // error when disabled"): a hallucinated call when the module is off
        // must fall through to the plain `Error::UnknownTool` path below,
        // never a background-specific error that would wrongly imply the
        // module is on. Unlike `SPAWN_SUBAGENT`/`SUBAGENT_STATUS`, none of
        // these four need async `&mut self` access (spawning a process,
        // `Child::try_wait`, and `Child::start_kill` are all synchronous),
        // so they're intercepted here in `prepare_tool_call` rather than in
        // `Self::run_tool`.
        if self.config.tools_background_enabled {
            if name == BACKGROUND_EXEC {
                return PreparedCall::Done(self.run_background_exec(call));
            }
            if name == BACKGROUND_STATUS {
                return PreparedCall::Done(self.run_background_status(call));
            }
            if name == BACKGROUND_LIST {
                return PreparedCall::Done(self.run_background_list(call));
            }
            if name == BACKGROUND_KILL {
                return PreparedCall::Done(self.run_background_kill(call));
            }
        }
        if self.registry.get(name).is_none() {
            let err = Error::UnknownTool(name.clone());
            return PreparedCall::Done((format!("Error: {err}"), true));
        }

        // P5-1 (§2 modules 10-11, integration point named in
        // COMPOSABLE-HARNESS-DESIGN.md's activation set): the permissions
        // ENGINE governs the gate when `capabilities.permissions.enabled`
        // is on; every other config resolves this to `false`
        // (`Config::default`), which takes the `else` branch below —
        // the EXACT pre-P5-1 code, untouched, so the default posture
        // (approval=never/sandbox=none) and every existing test's observed
        // behavior is byte-for-byte unchanged.
        if self.config.permissions_enabled {
            // The engine needs the command/path TEXT the legacy tool-name-
            // only gate below never looked at, so args must be parsed
            // BEFORE the gate here (not after, like the legacy branch).
            let args = match call.function.parsed_arguments() {
                Ok(v) => v,
                Err(e) => {
                    let err = Error::InvalidArguments {
                        tool: name.clone(),
                        message: e.to_string(),
                    };
                    return PreparedCall::Done((format!("Error: {err}"), true));
                }
            };
            if let Some(reason) = self.permissions_gate_denial(name, &args) {
                return PreparedCall::Done((format!("Error: {reason}"), true));
            }
            self.finish_prepare(name.clone(), args)
        } else {
            // ---- pre-P5-1 gate, byte-for-byte unchanged ----
            // Approval gate: if the policy requires it, consult the handler
            // (absent handler denies, so an OnRequest/Untrusted policy is
            // fail-closed).
            if self.config.needs_approval(name) {
                let approved = self
                    .config
                    .approval_handler
                    .as_ref()
                    .map(|h| h(call))
                    .unwrap_or(false);
                if !approved {
                    return PreparedCall::Done((
                        format!("Error: tool `{name}` was not approved for execution"),
                        true,
                    ));
                }
            }
            let args = match call.function.parsed_arguments() {
                Ok(v) => v,
                Err(e) => {
                    let err = Error::InvalidArguments {
                        tool: name.clone(),
                        message: e.to_string(),
                    };
                    return PreparedCall::Done((format!("Error: {err}"), true));
                }
            };
            self.finish_prepare(name.clone(), args)
        }
    }

    /// P5-1: the shared tail of [`Self::prepare_tool_call`] — doom-loop
    /// check, pre-tool hook, `Ready` construction — factored out so both
    /// the legacy gate and the new permissions-engine gate run the exact
    /// same downstream checks in the exact same order (§5.3 risk 1: the
    /// permissions engine changes WHO gets to run, never what happens once
    /// they're approved).
    fn finish_prepare(&mut self, name: String, args: serde_json::Value) -> PreparedCall {
        // P4c (§5.2 P4 "doom-loop breaker", oc UNIQUE `doom_loop` row,
        // catalog D3): a default, always-available veto point distinct from
        // `Config.pre_tool_hook` (a single user-installable slot — the
        // breaker must coexist with a caller's own hook, not compete for the
        // one slot). `None`/`Some(0|1)` is a no-op — byte-identical to
        // today (no repetition tracking, no call is ever refused on this
        // basis).
        if let Some(reason) = self.check_doom_loop(&name, &args) {
            return PreparedCall::Done((format!("Error: {reason}"), true));
        }
        // Pre-tool hook may block the call.
        if let Some(hook) = &self.config.pre_tool_hook {
            if let Some(reason) = hook(&name, &args) {
                return PreparedCall::Done((
                    format!("Error: blocked by pre-tool hook: {reason}"),
                    true,
                ));
            }
        }
        PreparedCall::Ready { name, args }
    }

    /// D-2 (Fable-5 delta review — LOW-MEDIUM, "over-grant residual"): the
    /// built-in tools whose `command`/`path`/`patch` arg IS semantically
    /// the whole call — the ONLY tools [`Self::permissions_gate_denial`]
    /// is allowed to turn into an [`permissions::ApprovalRequest::subject`]
    /// (see that method's own doc comment on the `subject` line for the
    /// full story). Bash-family (`bash`, and `shell` —
    /// [`crate::tools::builtins::PersistentShellTool`]'s registered name,
    /// what the review's "persistent-shell" refers to), the file tools
    /// (`read_file`/`write_file`/`edit_file`/`view_image`, whose `path` IS
    /// the subject), and `apply_patch` (whose `patch` envelope is handled
    /// separately but is unconditionally this tool only, see the `patch`
    /// local a few lines below). Deliberately NOT `list_dir`/`glob`/
    /// `search` — this crate's F2 fix (`ApprovalCache::key_for_request`)
    /// already falls back to a full-args digest for anything not on this
    /// list, which is a strictly SAFER (if slightly less cache-granular)
    /// default than guessing at more built-ins that weren't part of this
    /// finding.
    const SUBJECT_BEARING_BUILTIN_TOOLS: &'static [&'static str] = &[
        "bash",
        "shell",
        "read_file",
        "write_file",
        "edit_file",
        "view_image",
        "apply_patch",
    ];

    /// P5-1: evaluate `name`'s call (with parsed `args`) against the
    /// permissions engine (`crate::permissions`) — builds the
    /// [`crate::permissions::RuleSet`] from `Config`'s deny/ask/allow
    /// pattern lists (folding [`Config::permissions_protected_paths`] into
    /// the `deny` tier, module 13), picks a command-, path-, or name-only
    /// evaluation depending on what `args` carries, resolves an `Ask`
    /// decision via the session cache + THIS agent's own installed
    /// [`crate::permissions::PermissionsApprovalHandler`], and returns
    /// `Some(reason)` when the call is refused (`None` = proceed). Thin
    /// wrapper over [`Self::permissions_gate_denial_impl`] — see that
    /// method's doc comment for why the handler is a parameter there.
    fn permissions_gate_denial(&self, name: &str, args: &serde_json::Value) -> Option<String> {
        self.permissions_gate_denial_impl(name, args, self.permissions_approval_handler.as_deref())
    }

    /// P5-6 (§2.2 C6, build brief "wire to the P5-1 engine's non-
    /// interactive fail-closed path"): the SAME rule-evaluation body as
    /// [`Self::permissions_gate_denial`], but the approval `handler` is a
    /// PARAMETER instead of always reading `self.permissions_approval_handler`
    /// — `Agent::background_permission_denial` calls this with `handler:
    /// None` (or a [`crate::subagents::ParentQueueApprovalHandler`]) so a
    /// `background_exec` call's `Ask`-tier decisions resolve exactly like a
    /// P5-3 background child's do (`crate::permissions::resolve_ask`'s
    /// pre-existing "no handler ⇒ deny" contract), REGARDLESS of whether
    /// this agent itself has an interactive handler installed for its own
    /// foreground calls — a background job must never block on a prompt it
    /// has no way to answer, even if the agent hosting it could otherwise
    /// answer one. The rule SET and default-policy baseline are otherwise
    /// identical to a foreground call's — only how an `Ask` decision
    /// resolves ever differs, and only in the strictly-narrower direction
    /// (never escalates past what a foreground call of the same command is
    /// allowed).
    fn permissions_gate_denial_impl(
        &self,
        name: &str,
        args: &serde_json::Value,
        handler: Option<&dyn crate::permissions::PermissionsApprovalHandler>,
    ) -> Option<String> {
        use crate::permissions::{self, Decision, PathKind};

        // `Config.tool_deny_patterns`/`tool_allow_patterns` (P4a) ARE the
        // engine's deny/allow tiers — the same `capabilities.permissions.
        // rules.deny`/`.allow` keys, one source of truth, no duplication.
        // Protected paths (module 13) are an unconditional deny floor,
        // folded in here rather than checked separately, so they benefit
        // from the SAME first-match deny-wins priority every other deny
        // rule gets.
        let mut deny = self.config.tool_deny_patterns.clone();
        deny.extend(permissions::protected_path_deny_rules(
            &self.config.permissions_protected_paths,
        ));
        let rules = permissions::RuleSet {
            deny,
            ask: self.config.permissions_ask_patterns.clone(),
            allow: self.config.tool_allow_patterns.clone(),
        };

        // The baseline decision when NO rule matches at all — derived from
        // `ApprovalPolicy`, the same per-policy shape
        // `Config::needs_approval` uses for the legacy gate (see
        // `ApprovalPolicy::ModelRequested`'s doc comment for why this
        // richer gate approximates Codex's real "mostly silent" posture
        // instead of that method's conservative OnRequest-alike treatment
        // — explicit deny/ask rules still apply on top regardless).
        let default = match self.config.approval {
            crate::config::ApprovalPolicy::Never => Decision::Allow,
            crate::config::ApprovalPolicy::OnRequest => {
                if self.config.auto_approved_tools.contains(name) {
                    Decision::Allow
                } else {
                    Decision::Ask
                }
            }
            crate::config::ApprovalPolicy::Untrusted => Decision::Ask,
            crate::config::ApprovalPolicy::ModelRequested => Decision::Allow,
        };

        let command = args.get("command").and_then(|v| v.as_str());
        let path = args.get("path").and_then(|v| v.as_str());
        // F4 (Fable-5 adversarial review): `apply_patch`'s args carry a
        // patch ENVELOPE body (`args["patch"]`), not a `command` or a
        // `path` — the two branches above never fire for it, which is
        // exactly how a patch touching a protected path bypassed
        // `protected_paths` entirely. Only consulted for the `apply_patch`
        // tool specifically (a `patch`-shaped arg on some other tool is not
        // this envelope format and isn't given this treatment).
        let patch = (name == "apply_patch")
            .then(|| args.get("patch").and_then(|v| v.as_str()))
            .flatten();
        let decision = if let Some(command) = command {
            permissions::evaluate_command(&rules, name, command, default)
        } else if let Some(patch) = patch {
            // Same dual-check shape as the path branch below (pseudo-tool
            // `write(...)` rules from `protected_paths`, AND a rule
            // authored against the real `apply_patch` tool name), applied
            // to EVERY path the envelope's ops touch (`Add`/`Delete`/
            // `Update`'s `path`, plus `*** Move to:`). A patch that fails
            // to parse can't be proven to avoid a protected path — fail
            // closed to at least `Ask`, the same floor an unparseable bash
            // command gets in `permissions::evaluate_command`, rather than
            // silently let it through on `default`.
            let mut d = rules.evaluate(name, None).unwrap_or(default);
            match crate::tools::patch_target_paths(patch) {
                Ok(paths) => {
                    for p in &paths {
                        // SECURITY (CRITICAL fix): route both checks through
                        // the safe-path-resolving variants — a patch target
                        // like `x/../.git/config` must be caught exactly
                        // like a `write_file`/`edit_file` `path` argument
                        // would be (see `evaluate_path_safe`'s doc comment).
                        let pseudo = permissions::evaluate_path_safe(
                            &rules,
                            PathKind::Write,
                            &self.config.cwd,
                            p,
                            Decision::Allow,
                        );
                        let real_tool = permissions::evaluate_path_subject_safe(
                            &rules,
                            name,
                            &self.config.cwd,
                            p,
                            Decision::Allow,
                        );
                        d = d.stricter(pseudo).stricter(real_tool);
                    }
                }
                Err(_) => {
                    d = d.stricter(Decision::Ask);
                }
            }
            d
        } else if let Some(path) = path {
            let kind = if matches!(name, "write_file" | "edit_file") {
                PathKind::Write
            } else {
                PathKind::Read
            };
            // TWO independent sources of path-shaped rules can apply to the
            // same call, and BOTH must be checked:
            // (a) the `read(...)`/`write(...)` pseudo-tool (tool-agnostic —
            //     applies no matter WHICH tool touches the path; this is
            //     what `Config::permissions_protected_paths`/module 13
            //     expands into, via `protected_path_deny_rules`);
            // (b) a rule authored against the REAL tool name with the path
            //     as its subject — design §4.4's own oc-parity worked
            //     example writes exactly this shape (`"read_file(*.env)"`,
            //     not a pseudo-tool), matching how `bash(cmdglob)` rules
            //     are authored. `RuleSet::evaluate`'s bare-tool-name-glob
            //     branch (no parens) ALSO fires here regardless of
            //     `subject`, so this one call additionally covers a
            //     blanket "deny this tool entirely" rule — no separate
            //     `rules.evaluate(name, None)` call is needed.
            // SECURITY (CRITICAL fix, guarantor audit): both checks now
            // route through the safe-path-resolving variants (see
            // `evaluate_path_safe`'s doc comment) instead of glob-matching
            // the raw model-supplied `path` string directly — this is what
            // closes the traversal bypass (`write_file
            // path="x/../.git/config"`) and the analogous symlink escape.
            let pseudo_decision =
                permissions::evaluate_path_safe(&rules, kind, &self.config.cwd, path, default);
            let real_tool_decision = permissions::evaluate_path_subject_safe(
                &rules,
                name,
                &self.config.cwd,
                path,
                default,
            );
            pseudo_decision.stricter(real_tool_decision)
        } else {
            rules.evaluate(name, None).unwrap_or(default)
        };

        // D-2 (Fable-5 delta review — LOW-MEDIUM): `command`/`path`/`patch`
        // above are extracted (and used to DRIVE the decision above) for
        // ANY tool that happens to carry one of those arg names — that
        // part is unchanged and correct (a rule authored against, say, an
        // MCP tool's own name legitimately wants to glob-match its
        // `command`-shaped arg too). But the narrower single-field
        // `subject` handed to the cache/handler below must NOT do the
        // same for a non-built-in tool: an MCP (or other) tool's
        // `command`/`path` is just one field among potentially several
        // that together define what the call actually does — collapsing
        // an `AllowForSession` grant down to that one field would silently
        // auto-allow a later call with the SAME `command` but different
        // OTHER args (e.g. `{"command":"sync","target":"staging"}`
        // auto-allowing `{"command":"sync","target":"production"}`).
        // Restricting this to the known built-ins whose `subject` really
        // IS the whole call leaves every other tool with `subject: None`,
        // which routes it through `ApprovalCache::key_for_request`'s
        // full-args-digest fallback (F2) instead.
        let subject = Self::SUBJECT_BEARING_BUILTIN_TOOLS
            .contains(&name)
            .then(|| command.or(path).or(patch))
            .flatten();
        let req = permissions::ApprovalRequest {
            tool: name,
            subject,
            raw_args: args,
        };
        let approved = permissions::decision_to_approved(decision, || {
            permissions::resolve_ask(&self.permissions_approval_cache, handler, &req)
        });
        if approved {
            None
        } else {
            Some(format!(
                "tool `{name}` was not approved for execution (permissions engine: {decision:?})"
            ))
        }
    }

    /// P5-6 (§2.2 C6, build brief "a bg `rm -rf` subject to the same deny
    /// rules... must never escalate past what a foreground exec of the
    /// same command is allowed"): the permission gate `background_exec`
    /// runs BEFORE spawning anything. Evaluated against the tool name
    /// `"bash"` (not `"background_exec"`) deliberately — so any
    /// `bash(...)`-authored deny/ask/allow rule (or protected-path floor)
    /// applies to a background command byte-for-byte identically to a
    /// foreground `bash` call, the SAME rule set + default baseline
    /// [`Self::permissions_gate_denial`] would use for one.
    ///
    /// The one deliberate difference (C6 itself): an `Ask`-tier decision
    /// NEVER reaches an interactive handler here — a background job has no
    /// way to block on a prompt it can't answer. When
    /// [`Config::subagents_background_prompts`] is
    /// [`crate::subagents::BackgroundPromptsPolicy::Parent`], the denied
    /// request is additionally queued onto [`Self::pending_child_approvals`]
    /// (via [`crate::subagents::ParentQueueApprovalHandler`], reused
    /// verbatim — the SAME "parent-surfaced queue" §2.2 C6 names for
    /// `subagents.background`, with the job id standing in for a child
    /// agent id) for later inspection; any other configuration (including
    /// no `background_prompts` set at all) resolves via `handler: None` —
    /// [`crate::permissions::resolve_ask`]'s pre-existing "no handler ⇒
    /// deny" fail-closed default, identical to `subagents`'s own
    /// `AutoPolicy` reading. Either way, `Ask` always denies; only `Allow`
    /// (from the rule engine itself, or a PRIOR interactively-granted
    /// `AllowForSession` cache entry) ever lets a background command run —
    /// so this can only ever be as-or-more restrictive than a foreground
    /// call, never looser, regardless of configuration.
    ///
    /// Covers BOTH gate generations: when [`Config::permissions_enabled`]
    /// is on, the P5-1 engine (above) is used; otherwise the legacy
    /// [`Config::needs_approval`] gate is consulted but its
    /// `approval_handler` closure is NEVER invoked (that closure could
    /// itself block, e.g. a real interactive prompt) — an approval-required
    /// legacy policy simply denies a background command outright, the same
    /// never-hang guarantee under the older gate.
    fn background_permission_denial(&self, command: &str, job_id: &str) -> Option<String> {
        let args = serde_json::json!({ "command": command });
        if self.config.permissions_enabled {
            if let Some(crate::subagents::BackgroundPromptsPolicy::Parent) =
                self.config.subagents_background_prompts
            {
                let handler = crate::subagents::ParentQueueApprovalHandler {
                    child_agent_id: format!("bg:{job_id}"),
                    queue: self.pending_child_approvals.clone(),
                };
                self.permissions_gate_denial_impl("bash", &args, Some(&handler))
            } else {
                self.permissions_gate_denial_impl("bash", &args, None)
            }
        } else if self.config.needs_approval("bash") {
            Some(
                "tool `bash` requires approval, which a background job cannot request \
                 interactively (§2.2 C6: auto-policy denies)"
                    .to_string(),
            )
        } else {
            None
        }
    }

    /// P4e (§3.1 `core.parallel_tool_calls`, catalog:59 "Independent
    /// sibling calls run concurrently"): runs `calls`' `Tool::execute()`
    /// futures CONCURRENTLY via `futures::future::join_all`, for whichever
    /// calls [`Self::prepare_tool_call`] resolves to [`PreparedCall::Ready`]
    /// — i.e. every plain (non-intrinsic) registry-tool call that passes
    /// its synchronous approval/doom-loop/pre-tool-hook checks. A call that
    /// resolves to [`PreparedCall::Done`] (an intrinsic, an unknown tool, a
    /// denied/blocked call) is NOT parallelized — its result is already in
    /// hand from the synchronous prepare pass. Every prepare check still
    /// runs sequentially, in original call order, before ANY `execute()`
    /// future starts (only the actual tool I/O overlaps) — so doom-loop
    /// bookkeeping and pre-tool-hook vetoes see the exact same call order
    /// they would under the sequential path. Returns results in the SAME
    /// order as `calls`, so callers can always `zip` the two. Post-tool
    /// hooks fire per call, in original order, once every result is in
    /// hand — a caller-visible timing difference from the sequential path
    /// ONLY when this method runs at all (i.e. only when
    /// `Config::parallel_tool_calls` is on): hooks see "this batch
    /// finished" ordering rather than "this one call finished" ordering.
    /// Documented, not a bug.
    async fn run_tools_concurrently(
        &mut self,
        calls: &[crate::message::ToolCall],
    ) -> Vec<(String, bool)> {
        // P5-3: `spawn_subagent`/`subagent_status` need sequential `&mut
        // self` access `prepare_tool_call`'s synchronous-only signature
        // can't give them (see `Self::run_tool`'s identical interception).
        // A batch that includes one falls back to dispatching the WHOLE
        // batch sequentially via `Self::run_tool` — a documented, narrow
        // simplification (not a partial-parallelization attempt) rather
        // than restructuring `PreparedCall` to carry a future; a batch with
        // no subagent intrinsic is completely unaffected and still
        // parallelizes exactly as before.
        if calls.iter().any(|c| {
            c.function.name == SPAWN_SUBAGENT
                || c.function.name == SUBAGENT_STATUS
                || (self.config.claude_runtime_tools_enabled
                    && matches!(
                        c.function.name.as_str(),
                        CLAUDE_CRON_CREATE
                            | CLAUDE_CRON_DELETE
                            | CLAUDE_CRON_LIST
                            | CLAUDE_SCHEDULE_WAKEUP
                    ))
        }) {
            let mut out = Vec::with_capacity(calls.len());
            for call in calls {
                out.push(self.run_tool(call).await);
            }
            return out;
        }
        let prepared: Vec<PreparedCall> = calls.iter().map(|c| self.prepare_tool_call(c)).collect();
        let mut slots: Vec<Option<(String, bool)>> = prepared
            .iter()
            .map(|p| match p {
                PreparedCall::Done(r) => Some(r.clone()),
                PreparedCall::Ready { .. } => None,
            })
            .collect();

        let ready_idxs: Vec<usize> = prepared
            .iter()
            .enumerate()
            .filter(|(_, p)| matches!(p, PreparedCall::Ready { .. }))
            .map(|(i, _)| i)
            .collect();

        if !ready_idxs.is_empty() {
            let futs = ready_idxs.iter().map(|&i| {
                let PreparedCall::Ready { name, args } = &prepared[i] else {
                    unreachable!("filtered to Ready above")
                };
                // `self.registry.get` borrows `self.registry` immutably;
                // `self.ctx` is `Clone` (P4c precedent) so each future owns
                // its own copy rather than borrowing `self` across the
                // `.await` inside `join_all`.
                let tool = self.registry.get(name).expect("prepared as Ready");
                let args = args.clone();
                let ctx = self.ctx.clone();
                async move {
                    match tool.execute(args, &ctx).await {
                        Ok(out) => (out, false),
                        Err(e) => (format!("Error: {e}"), true),
                    }
                }
            });
            let results = futures::future::join_all(futs).await;
            for (idx, result) in ready_idxs.iter().zip(results) {
                slots[*idx] = Some(result);
            }
        }

        let out: Vec<(String, bool)> = slots
            .into_iter()
            .map(|s| s.expect("every call resolved to Some above"))
            .collect();
        // Post-tool hook, in original order — only for calls that actually
        // reached `execute()` (matches `run_tool`'s existing behavior: an
        // intrinsic/denied/blocked call never fires the post-tool hook).
        let ready_set: std::collections::HashSet<usize> = ready_idxs.into_iter().collect();
        for (i, call) in calls.iter().enumerate() {
            if !ready_set.contains(&i) {
                continue;
            }
            let (output, is_error) = &out[i];
            if let Some(hook) = &self.config.post_tool_hook {
                hook(&call.function.name, output, *is_error);
            }
        }
        out
    }

    /// P4c (§5.2 P4 "doom-loop breaker", §3.1 `core.doom_loop_threshold`):
    /// update the consecutive-identical-call streak for `(name, args)` and
    /// return `Some(reason)` the moment the streak reaches
    /// `Config.doom_loop_threshold` (a call whose name AND JSON-canonical
    /// arguments are byte-identical to the immediately preceding call
    /// extends the streak; anything else resets it to 1). `None`
    /// (`Config.doom_loop_threshold` unset, or `Some(n)` with `n < 2` — a
    /// threshold below 2 can never fire since the FIRST call already
    /// "repeats zero times") never touches the streak fields at all.
    fn check_doom_loop(&mut self, name: &str, args: &serde_json::Value) -> Option<String> {
        let threshold = self.config.doom_loop_threshold?;
        if threshold < 2 {
            return None;
        }
        // `serde_json::Value::Object` is a `BTreeMap` in this workspace (no
        // `preserve_order` feature), so `to_string()` is already
        // key-order-canonical — two calls that differ only in argument key
        // order are still treated as identical.
        let key = (name.to_string(), args.to_string());
        if self.doom_loop_last_call.as_ref() == Some(&key) {
            self.doom_loop_streak += 1;
        } else {
            self.doom_loop_last_call = Some(key);
            self.doom_loop_streak = 1;
        }
        if self.doom_loop_streak >= threshold {
            Some(format!(
                "doom-loop breaker: `{name}` called with identical arguments {} times in a row \
                 — try a different approach instead of repeating the same call",
                self.doom_loop_streak
            ))
        } else {
            None
        }
    }

    /// Whether `name` is in the eagerly-advertised "core" set for the current
    /// [`ToolAdvertising`] mode: every enabled tool under `Full`, or the
    /// explicit `core` allowlist under `Deferred`.
    fn is_core_tool(&self, name: &str) -> bool {
        match &self.config.tool_advertising {
            ToolAdvertising::Full => true,
            ToolAdvertising::Deferred { core } => core.iter().any(|c| c == name),
        }
    }

    /// The schema advertised on the wire for `t`: the raw (as-shipped)
    /// schema with TR-8/T5's per-tool schema tier applied. This is what
    /// [`Self::tool_schemas`] sends every request.
    fn schema_for(&self, t: &dyn crate::tools::Tool) -> ToolSchema {
        let raw = self.raw_schema_for(t);
        let tier = self.config.schema_tier_for(t.name());
        let (description, parameters) =
            crate::tools::tiers::minify(&raw.description, &raw.parameters, tier);
        ToolSchema {
            name: raw.name,
            description,
            parameters,
        }
    }

    /// The ORIGINAL, as-shipped schema for `t` — never tier-minified. This is
    /// the full contract [`Self::run_tool_search`] hands back on activation
    /// (TR-8/T5 dev/03: the B6 fetch path is the invert of tiering, so a
    /// model that fetched a tool via `tool_search` always sees the complete
    /// schema, byte-equal to `t.description()`/`t.parameters()` — modulo the
    /// pre-existing [`crate::Config::tool_description`] override, which is
    /// orthogonal to tiering).
    fn raw_schema_for(&self, t: &dyn crate::tools::Tool) -> ToolSchema {
        ToolSchema {
            name: t.name().to_string(),
            description: self
                .config
                .tool_description(t.name(), t.description())
                .to_string(),
            parameters: t.parameters(),
        }
    }

    /// The synthetic `tool_search` schema advertised under `Deferred` (B6).
    fn tool_search_schema() -> ToolSchema {
        ToolSchema {
            name: TOOL_SEARCH.to_string(),
            description: "Search for additional tools not currently advertised (the deferred \
                MCP surface and any other non-core tools). Matches keywords case-insensitively \
                against each tool's name and description. Matched tools become callable starting \
                with your NEXT message, not this one."
                .to_string(),
            parameters: serde_json::json!({
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "Keyword(s) to search for in tool names and descriptions."
                    },
                    "max_results": {
                        "type": "integer",
                        "description": "Maximum number of matching tools to return."
                    }
                },
                "required": ["query"],
                "additionalProperties": false
            }),
        }
    }

    /// The tool-schema array this agent would advertise on its NEXT
    /// request, exactly as [`Self::run_loop`] computes it. Public
    /// (PARITY-18 D1) so a caller can measure the real request-token cost
    /// of an agent's tool surface — including the current
    /// [`crate::config::ToolAdvertising`] mode's core/deferred split and
    /// the synthetic `tool_search`/`expand_reduction`/`sidecar_search`
    /// schemas — BEFORE ever calling [`Self::send`], e.g. for a preflight
    /// context-guard check.
    pub fn tool_schemas(&self) -> Vec<ToolSchema> {
        let mut out = match &self.config.tool_advertising {
            ToolAdvertising::Full => self
                .registry
                .iter()
                .filter(|t| self.config.tool_enabled(t.name()))
                .map(|t| self.schema_for(t))
                .collect(),
            ToolAdvertising::Deferred { .. } => {
                let mut out: Vec<ToolSchema> = self
                    .registry
                    .iter()
                    .filter(|t| self.config.tool_enabled(t.name()))
                    .filter(|t| {
                        self.is_core_tool(t.name()) || self.activated_tools.contains(t.name())
                    })
                    .map(|t| self.schema_for(t))
                    .collect();
                out.push(Self::tool_search_schema());
                out
            }
        };
        // T12/TR-1: `expand_reduction`/`sidecar_search` are orthogonal to
        // `tool_advertising` (which governs the ordinary tool surface) —
        // advertised whenever a `ReductionPolicy` is installed, regardless of
        // Full/Deferred, since only a reduced session ever has anything to
        // expand or search (SPEC.md TR-1 dev/01).
        if self.reduction_policy.is_some() {
            out.push(Self::expand_reduction_schema());
            out.push(Self::sidecar_search_schema());
        }
        // P5-3 (§2 module 9): `spawn_subagent`/`subagent_status` are
        // orthogonal to `tool_advertising` too, same reasoning as
        // `expand_reduction`/`sidecar_search` above — advertised whenever
        // `Config::subagents_enabled` is on, Full or Deferred alike.
        // `false` (the default) never appends either, so a config that
        // never turns the module on gets byte-identical tool schemas to
        // today.
        if self.config.subagents_enabled {
            out.push(self.spawn_subagent_schema());
            if self.config.subagents_claude_agent_alias {
                out.push(self.claude_agent_schema());
            }
            if self.config.subagents_background {
                out.push(Self::subagent_status_schema());
            }
        }
        if self.config.claude_runtime_tools_enabled {
            out.extend(self.claude_builtin_tool_schemas());
            out.push(Self::claude_cron_create_schema());
            out.push(Self::claude_cron_delete_schema());
            out.push(Self::claude_cron_list_schema());
            out.push(Self::claude_schedule_wakeup_schema());
        }
        // P5-6 (§2 module 4 `tools.background`): same orthogonal-to-
        // `tool_advertising` treatment, advertised whenever
        // `Config::tools_background_enabled` is on. `false` (the default)
        // never appends any of the four, so a config that never turns the
        // module on gets byte-identical tool schemas to today.
        if self.config.tools_background_enabled {
            out.push(Self::background_exec_schema());
            out.push(Self::background_status_schema());
            out.push(Self::background_list_schema());
            out.push(Self::background_kill_schema());
        }
        out
    }

    fn claude_builtin_tool_schemas(&self) -> Vec<ToolSchema> {
        let mut schemas = Vec::new();
        let mut push = |alias: &str, native: &str, description: &str, parameters| {
            if self.registry.get(native).is_some() && self.config.tool_enabled(native) {
                schemas.push(ToolSchema {
                    name: alias.to_string(),
                    description: description.to_string(),
                    parameters,
                });
            }
        };
        push(
            CLAUDE_BASH,
            "bash",
            "Claude Code-compatible shell command execution.",
            serde_json::json!({
                "type": "object",
                "properties": {
                    "command": {"type": "string"},
                    "timeout": {"type": "integer", "description": "Timeout in milliseconds."},
                    "description": {"type": "string"}
                },
                "required": ["command"],
                "additionalProperties": true
            }),
        );
        push(
            CLAUDE_READ,
            "read_file",
            "Claude Code-compatible file reader.",
            serde_json::json!({
                "type": "object",
                "properties": {
                    "file_path": {"type": "string"},
                    "offset": {"type": "integer"},
                    "limit": {"type": "integer"}
                },
                "required": ["file_path"],
                "additionalProperties": false
            }),
        );
        push(
            CLAUDE_WRITE,
            "write_file",
            "Claude Code-compatible file writer.",
            serde_json::json!({
                "type": "object",
                "properties": {"file_path": {"type": "string"}, "content": {"type": "string"}},
                "required": ["file_path", "content"],
                "additionalProperties": false
            }),
        );
        push(
            CLAUDE_EDIT,
            "edit_file",
            "Claude Code-compatible exact file edit.",
            serde_json::json!({
                "type": "object",
                "properties": {
                    "file_path": {"type": "string"},
                    "old_string": {"type": "string"},
                    "new_string": {"type": "string"},
                    "replace_all": {"type": "boolean"}
                },
                "required": ["file_path", "old_string", "new_string"],
                "additionalProperties": false
            }),
        );
        push(
            CLAUDE_GLOB,
            "glob",
            "Claude Code-compatible file glob.",
            serde_json::json!({
                "type": "object",
                "properties": {"pattern": {"type": "string"}, "path": {"type": "string"}},
                "required": ["pattern"],
                "additionalProperties": false
            }),
        );
        push(
            CLAUDE_GREP,
            "search",
            "Claude Code-compatible content search.",
            serde_json::json!({
                "type": "object",
                "properties": {"pattern": {"type": "string"}, "path": {"type": "string"}},
                "required": ["pattern"],
                "additionalProperties": true
            }),
        );
        schemas
    }

    fn translate_claude_builtin_call(
        &self,
        call: &crate::message::ToolCall,
    ) -> Result<Option<crate::message::ToolCall>> {
        let native = match call.function.name.as_str() {
            CLAUDE_BASH => "bash",
            CLAUDE_READ => "read_file",
            CLAUDE_WRITE => "write_file",
            CLAUDE_EDIT => "edit_file",
            CLAUDE_GLOB => "glob",
            CLAUDE_GREP => "search",
            _ => return Ok(None),
        };
        let mut args = call.function.parsed_arguments()?;
        let object = args
            .as_object_mut()
            .ok_or_else(|| Error::InvalidArguments {
                tool: call.function.name.clone(),
                message: "expected a JSON object".to_string(),
            })?;
        if let Some(path) = object.remove("file_path") {
            object.entry("path".to_string()).or_insert(path);
        }
        if call.function.name == CLAUDE_BASH {
            if let Some(timeout) = object.remove("timeout") {
                object.entry("timeout_ms".to_string()).or_insert(timeout);
            }
        }
        if call.function.name == CLAUDE_GLOB {
            if let Some(path) = object
                .remove("path")
                .and_then(|value| value.as_str().map(str::to_owned))
            {
                if let Some(pattern) = object.get_mut("pattern") {
                    if let Some(value) = pattern.as_str() {
                        if !std::path::Path::new(value).is_absolute() {
                            *pattern = serde_json::Value::String(
                                std::path::Path::new(&path)
                                    .join(value)
                                    .to_string_lossy()
                                    .into_owned(),
                            );
                        }
                    }
                }
            }
        }
        let mut translated = call.clone();
        translated.function.name = native.to_string();
        translated.function.arguments = serde_json::to_string(&args)?;
        Ok(Some(translated))
    }

    fn claude_cron_create_schema() -> ToolSchema {
        ToolSchema {
            name: CLAUDE_CRON_CREATE.to_string(),
            description: "Record a Claude-compatible cron job in the imported runtime manifest. \
                The job inherits the manifest's ACTIVE or PAUSED posture; an embedding scheduler, \
                not this agent loop, owns execution."
                .to_string(),
            parameters: serde_json::json!({
                "type": "object",
                "properties": {
                    "cron": {"type": "string", "description": "Cron expression to preserve."},
                    "prompt": {"type": "string", "description": "Prompt associated with the job."},
                    "recurring": {"type": "boolean", "default": false},
                    "durable": {"type": "boolean", "default": false}
                },
                "required": ["cron", "prompt"],
                "additionalProperties": false
            }),
        }
    }

    fn claude_cron_delete_schema() -> ToolSchema {
        ToolSchema {
            name: CLAUDE_CRON_DELETE.to_string(),
            description: "Delete a Claude-compatible cron job from the imported manifest. \
                This updates state only; an embedding scheduler owns execution."
                .to_string(),
            parameters: serde_json::json!({
                "type": "object",
                "properties": {"id": {"type": "string"}},
                "required": ["id"],
                "additionalProperties": false
            }),
        }
    }

    fn claude_cron_list_schema() -> ToolSchema {
        ToolSchema {
            name: CLAUDE_CRON_LIST.to_string(),
            description: "List imported Claude cron jobs and their explicit ACTIVE or PAUSED \
                manifest posture. This agent loop itself does not run a scheduler."
                .to_string(),
            parameters: serde_json::json!({
                "type": "object",
                "properties": {},
                "additionalProperties": false
            }),
        }
    }

    fn claude_schedule_wakeup_schema() -> ToolSchema {
        ToolSchema {
            name: CLAUDE_SCHEDULE_WAKEUP.to_string(),
            description: "Replace the one-shot wakeup stored in the imported Claude manifest. \
                The wakeup inherits the manifest's ACTIVE or PAUSED posture; an embedding scheduler \
                owns timer execution."
                .to_string(),
            parameters: serde_json::json!({
                "type": "object",
                "properties": {
                    "delaySeconds": {"type": "integer", "minimum": 0},
                    "reason": {"type": "string"},
                    "prompt": {"type": "string"}
                },
                "required": ["delaySeconds"],
                "additionalProperties": false
            }),
        }
    }

    /// The `background_exec` schema (P5-6, §2 module 4, D1 "background
    /// exec").
    fn background_exec_schema() -> ToolSchema {
        ToolSchema {
            name: BACKGROUND_EXEC.to_string(),
            description: "Run a shell command in the BACKGROUND: spawns it as a detached \
                process and returns a `job_id` IMMEDIATELY, before the command finishes — this \
                call never returns the command's output. Poll `background_status` with the \
                `job_id` to check progress and retrieve captured output; use `background_kill` \
                to cancel it early. The command goes through the exact same sandbox/permission \
                checks as a foreground `bash` call, and any check that would need an \
                interactive approval is denied automatically (a background job cannot wait for \
                one)."
                .to_string(),
            parameters: serde_json::json!({
                "type": "object",
                "properties": {
                    "command": {
                        "type": "string",
                        "description": "Shell command to run in the background via `sh -c`."
                    }
                },
                "required": ["command"],
                "additionalProperties": false
            }),
        }
    }

    /// The `background_status` schema (P5-6, D1 "monitor/event feed").
    fn background_status_schema() -> ToolSchema {
        ToolSchema {
            name: BACKGROUND_STATUS.to_string(),
            description: "Check on a background job spawned via background_exec: its \
                running/exited/killed status, exit code (once known), and the command's \
                captured stdout/stderr so far (bounded — very large output is truncated with a \
                marker). Once the job has exited or been killed, this call also reaps it (it \
                will no longer appear in background_list or accept further status polls)."
                .to_string(),
            parameters: serde_json::json!({
                "type": "object",
                "properties": {
                    "job_id": {
                        "type": "string",
                        "description": "The id `background_exec` returned when this job was \
                            started."
                    }
                },
                "required": ["job_id"],
                "additionalProperties": false
            }),
        }
    }

    /// The `background_list` schema (P5-6, D10 "bg-manager").
    fn background_list_schema() -> ToolSchema {
        ToolSchema {
            name: BACKGROUND_LIST.to_string(),
            description: "List every background job currently tracked (running, or finished \
                but not yet polled via background_status) — job id, command, status, pid, and \
                start time for each. Does not retrieve output or reap anything."
                .to_string(),
            parameters: serde_json::json!({
                "type": "object",
                "properties": {},
                "additionalProperties": false
            }),
        }
    }

    /// The `background_kill` schema (P5-6, D10 "bg-manager").
    fn background_kill_schema() -> ToolSchema {
        ToolSchema {
            name: BACKGROUND_KILL.to_string(),
            description: "Kill a background job's real process immediately (a no-op, not an \
                error, if it already exited on its own) and reap it."
                .to_string(),
            parameters: serde_json::json!({
                "type": "object",
                "properties": {
                    "job_id": {
                        "type": "string",
                        "description": "The id `background_exec` returned when this job was \
                            started."
                    }
                },
                "required": ["job_id"],
                "additionalProperties": false
            }),
        }
    }

    /// The `spawn_subagent` schema (P5-3, §2 module 9 D1 "spawn tool").
    /// Lists every configured `agent_type` name so the model knows what's
    /// available, but `agent_type` stays optional — an ad-hoc spawn with an
    /// inline `system_prompt` is always allowed too.
    fn spawn_subagent_schema(&self) -> ToolSchema {
        let mut names: Vec<&str> = self
            .config
            .subagents_definitions
            .keys()
            .map(String::as_str)
            .collect();
        names.sort_unstable();
        let agent_type_desc = if names.is_empty() {
            "Optional named subagent type to run (none configured — omit this and pass \
             `system_prompt` instead)."
                .to_string()
        } else {
            format!(
                "Optional named subagent type to run: {}. Omit to run an ad-hoc subagent with \
                 your own `system_prompt` instead.",
                names.join(", ")
            )
        };
        let background_desc = if self.config.subagents_background {
            "Run this subagent in the background instead of waiting for it — this call \
             returns immediately with a `subagent_id`; poll `subagent_status` with that id for \
             the result."
        } else {
            "Background subagents are disabled for this agent — this must be omitted or false."
        };
        ToolSchema {
            name: SPAWN_SUBAGENT.to_string(),
            description: "Spawn a subagent to work on a self-contained task and (by default) \
                wait for its final answer, which is returned as this call's result. The \
                subagent runs its own independent reasoning/tool loop; it does not see your \
                conversation except for the `task` text you give it here."
                .to_string(),
            parameters: serde_json::json!({
                "type": "object",
                "properties": {
                    "task": {
                        "type": "string",
                        "description": "The self-contained task/prompt for the subagent."
                    },
                    "agent_type": {
                        "type": "string",
                        "description": agent_type_desc
                    },
                    "system_prompt": {
                        "type": "string",
                        "description": "Inline system prompt for an ad-hoc subagent (ignored \
                            if `agent_type` is given — the named type's own prompt is used \
                            instead)."
                    },
                    "background": {
                        "type": "boolean",
                        "description": background_desc
                    }
                },
                "required": ["task"],
                "additionalProperties": false
            }),
        }
    }

    /// Claude Code-compatible alias for [`Self::spawn_subagent_schema`].
    fn claude_agent_schema(&self) -> ToolSchema {
        let mut names: Vec<String> = self.config.subagents_definitions.keys().cloned().collect();
        names.push("general-purpose".into());
        names.sort_unstable();
        names.dedup();
        ToolSchema {
            name: CLAUDE_AGENT.to_string(),
            description: "Claude Code-compatible subagent dispatcher. Runs a named or ad-hoc \
                child agent; children default to background execution in this compatibility mode."
                .to_string(),
            parameters: serde_json::json!({
                "type": "object",
                "properties": {
                    "prompt": {"type": "string", "description": "Self-contained child task."},
                    "subagent_type": {
                        "type": "string",
                        "description": format!("Named agent type. Available: {}", names.join(", "))
                    },
                    "description": {
                        "type": "string",
                        "description": "Short human-facing task label; preserved as descriptive input."
                    },
                    "model": {
                        "type": "string",
                        "description": "Optional model alias or full provider slug for this child."
                    },
                    "run_in_background": {
                        "type": "boolean",
                        "description": "Whether to return immediately with a child id (default true)."
                    }
                },
                "required": ["prompt"],
                "additionalProperties": false
            }),
        }
    }

    /// Translate Claude's `Agent` arguments to the native subagent intrinsic.
    fn translate_claude_agent_call(
        &self,
        call: &crate::message::ToolCall,
    ) -> Result<crate::message::ToolCall> {
        let args = call
            .function
            .parsed_arguments()
            .map_err(|error| Error::InvalidArguments {
                tool: CLAUDE_AGENT.to_string(),
                message: error.to_string(),
            })?;
        let object = args.as_object().ok_or_else(|| Error::InvalidArguments {
            tool: CLAUDE_AGENT.to_string(),
            message: "arguments must be an object".to_string(),
        })?;
        let mut translated = serde_json::Map::new();
        if let Some(value) = object.get("prompt") {
            translated.insert("task".to_string(), value.clone());
        }
        if let Some(value) = object.get("subagent_type") {
            // `general-purpose` is a built-in Claude agent, not a project
            // definition file. Supercode's equivalent is an ad-hoc child
            // using the inherited default system prompt, represented by an
            // omitted `agent_type`.
            if value.as_str() != Some("general-purpose") {
                translated.insert("agent_type".to_string(), value.clone());
            }
        }
        if let Some(value) = object.get("model") {
            translated.insert("model".to_string(), value.clone());
        }
        translated.insert(
            "background".to_string(),
            object
                .get("run_in_background")
                .cloned()
                .unwrap_or(serde_json::Value::Bool(true)),
        );
        Ok(crate::message::ToolCall {
            id: call.id.clone(),
            kind: call.kind.clone(),
            function: crate::message::FunctionCall {
                name: SPAWN_SUBAGENT.to_string(),
                arguments: serde_json::Value::Object(translated).to_string(),
            },
        })
    }

    /// Execute Claude's scheduling vocabulary against the imported manifest.
    ///
    /// This is intentionally a state editor, not a scheduler: it owns no
    /// timer/task handle and every successful response names the manifest's
    /// active/paused posture explicitly. Active creates get real UTC
    /// timestamps so an embedding scheduler can consume the persisted state.
    fn run_claude_runtime_tool(&mut self, call: &crate::message::ToolCall) -> (String, bool) {
        let args = match call.function.parsed_arguments() {
            Ok(value) if value.is_object() => value,
            Ok(_) => {
                return (
                    format!("Error: {} arguments must be an object", call.function.name),
                    true,
                )
            }
            Err(error) => return (format!("Error: {error}"), true),
        };
        let object = args.as_object().expect("checked object above");

        let Some(manifest) = self.claude_runtime_manifest.as_mut() else {
            return (
                "Error: Claude runtime compatibility was enabled without an imported runtime \
                 manifest; refusing to invent scheduler state"
                    .to_string(),
                true,
            );
        };
        let active = matches!(
            manifest.execution_state,
            crate::claude_runtime_state::ClaudeRuntimeExecutionState::Active
        );
        let state = if active { "active" } else { "paused" };
        let active_now_ms = active.then(now_ms);
        let active_now_unix = active_now_ms.map(|ms| ms.div_euclid(1_000));

        match call.function.name.as_str() {
            CLAUDE_CRON_LIST => {
                let jobs: Vec<serde_json::Value> = manifest
                    .active_crons
                    .iter()
                    .map(|job| {
                        serde_json::json!({
                            "id": job.id,
                            "cron": job.schedule,
                            "prompt": job.prompt,
                            "recurring": job.recurring,
                            "durable": job.durable_requested,
                            "state": state
                        })
                    })
                    .collect();
                let notice = if active {
                    "The manifest is active; an attached scheduler may claim due jobs."
                } else {
                    "Imported jobs are preserved but no scheduler is running."
                };
                (
                    serde_json::json!({
                        "execution_state": state,
                        "execution_notice": notice,
                        "jobs": jobs
                    })
                    .to_string(),
                    false,
                )
            }
            CLAUDE_CRON_CREATE => {
                let Some(schedule) = object.get("cron").and_then(serde_json::Value::as_str) else {
                    return ("Error: CronCreate requires string `cron`".to_string(), true);
                };
                let Some(prompt) = object.get("prompt").and_then(serde_json::Value::as_str) else {
                    return (
                        "Error: CronCreate requires string `prompt`".to_string(),
                        true,
                    );
                };
                let recurring = object
                    .get("recurring")
                    .and_then(serde_json::Value::as_bool)
                    .unwrap_or(false);
                let durable_requested = object
                    .get("durable")
                    .and_then(serde_json::Value::as_bool)
                    .unwrap_or(false);
                let mut sequence = 1_u64;
                let id = loop {
                    let candidate = format!("sc{sequence:06}");
                    if !manifest.active_crons.iter().any(|job| job.id == candidate) {
                        break candidate;
                    }
                    sequence += 1;
                };
                let kind = if recurring { "recurring " } else { "" };
                let created_at = active_now_ms.map(crate::sidecar::ms_to_rfc3339);
                let result = if active {
                    format!(
                        "Scheduled {kind}job {id} ({schedule}) in ACTIVE state. The job is \
                         eligible for execution by the attached scheduler."
                    )
                } else {
                    format!(
                        "Scheduled {kind}job {id} ({schedule}) in PAUSED state. The job is preserved \
                         in the continuation manifest but no scheduler is running and it will not execute."
                    )
                };
                manifest
                    .active_crons
                    .push(crate::claude_runtime_state::ClaudeCronJob {
                        id: id.clone(),
                        tool_use_id: call.id.clone(),
                        schedule: schedule.to_string(),
                        recurring,
                        durable_requested,
                        prompt: prompt.to_string(),
                        created_at,
                        expires_after_seconds: None,
                        creation_result: result.clone(),
                    });
                manifest
                    .active_crons
                    .sort_by(|left, right| left.id.cmp(&right.id));
                if let Some(now_unix) = active_now_unix {
                    let scheduler_before = manifest.scheduler.clone();
                    if let Err(error) = manifest.reconcile_scheduler(now_unix) {
                        manifest.active_crons.retain(|job| job.id != id);
                        manifest.scheduler = scheduler_before;
                        return (format!("Error: {error}"), true);
                    }
                }
                (result, false)
            }
            CLAUDE_CRON_DELETE => {
                let Some(id) = object.get("id").and_then(serde_json::Value::as_str) else {
                    return ("Error: CronDelete requires string `id`".to_string(), true);
                };
                let Some(index) = manifest.active_crons.iter().position(|job| job.id == id) else {
                    return (
                        format!("Error: unknown {state} Claude cron job `{id}`"),
                        true,
                    );
                };
                let removed = manifest.active_crons.remove(index);
                if let Some(now_unix) = active_now_unix {
                    let scheduler_before = manifest.scheduler.clone();
                    if let Err(error) = manifest.reconcile_scheduler(now_unix) {
                        manifest.active_crons.insert(index, removed);
                        manifest.scheduler = scheduler_before;
                        return (format!("Error: {error}"), true);
                    }
                }
                let result = if active {
                    format!("Cancelled job {id}. The job was removed from ACTIVE scheduler state.")
                } else {
                    format!("Cancelled job {id}. The job was PAUSED; no execution occurred.")
                };
                (result, false)
            }
            CLAUDE_SCHEDULE_WAKEUP => {
                let Some(delay_seconds) = object
                    .get("delaySeconds")
                    .and_then(serde_json::Value::as_u64)
                else {
                    return (
                        "Error: ScheduleWakeup requires integer `delaySeconds`".to_string(),
                        true,
                    );
                };
                let reason = object
                    .get("reason")
                    .and_then(serde_json::Value::as_str)
                    .map(str::to_string);
                let prompt = object
                    .get("prompt")
                    .and_then(serde_json::Value::as_str)
                    .map(str::to_string);
                let created_at = active_now_ms.map(crate::sidecar::ms_to_rfc3339);
                let scheduled_for = active_now_ms
                    .map(|now| {
                        let delay_ms = i64::try_from(delay_seconds)
                            .unwrap_or(i64::MAX)
                            .saturating_mul(1_000);
                        crate::sidecar::ms_to_rfc3339(now.saturating_add(delay_ms))
                    })
                    .unwrap_or_else(|| "PAUSED".to_string());
                let result = if active {
                    format!(
                        "Next wakeup scheduled for {scheduled_for} (in {delay_seconds}s). Runtime \
                         state is ACTIVE; the attached scheduler may execute it."
                    )
                } else {
                    format!(
                        "Next wakeup scheduled for PAUSED (in {delay_seconds}s). The request replaced \
                         the prior wakeup in the manifest, but no timer is running and it will not execute."
                    )
                };
                let previous_wakeups = if active {
                    Some(manifest.pending_wakeups.clone())
                } else {
                    None
                };
                manifest.pending_wakeups.clear();
                manifest
                    .pending_wakeups
                    .push(crate::claude_runtime_state::ClaudeWakeup {
                        tool_use_id: call.id.clone(),
                        delay_seconds,
                        reason,
                        prompt,
                        created_at,
                        scheduled_for: Some(scheduled_for),
                        creation_result: result.clone(),
                    });
                if let Some(now_unix) = active_now_unix {
                    let scheduler_before = manifest.scheduler.clone();
                    if let Err(error) = manifest.reconcile_scheduler(now_unix) {
                        manifest.pending_wakeups = previous_wakeups.unwrap_or_default();
                        manifest.scheduler = scheduler_before;
                        return (format!("Error: {error}"), true);
                    }
                }
                (result, false)
            }
            _ => unreachable!("runtime tool dispatch is name-gated"),
        }
    }

    /// The `subagent_status` schema (P5-3, D3 "background+resume").
    fn subagent_status_schema() -> ToolSchema {
        ToolSchema {
            name: SUBAGENT_STATUS.to_string(),
            description: "Check on (and, once finished, retrieve the result of) a background \
                subagent spawned via spawn_subagent with background=true. Pass the \
                `subagent_id` that spawn returned."
                .to_string(),
            parameters: serde_json::json!({
                "type": "object",
                "properties": {
                    "subagent_id": {
                        "type": "string",
                        "description": "The id `spawn_subagent` returned when this subagent \
                            was spawned."
                    }
                },
                "required": ["subagent_id"],
                "additionalProperties": false
            }),
        }
    }

    /// Build the CHILD `Config` a `spawn_subagent` call constructs its
    /// [`Agent`] from. The whole point of this method (§5.3-style
    /// "monotonic posture", build-brief "a subagent inherits or narrows —
    /// never widens — the parent's permission posture"): every field that
    /// governs what the child is ALLOWED to do (sandbox, approval,
    /// tool_overrides, deny/allow patterns, protected paths, the subagents
    /// caps themselves) is copied VERBATIM from `self.config` — never
    /// loosened — and the only NARROWING lever is `definition.tools`
    /// (intersected with whatever the parent already had enabled, never
    /// unioned in anything new).
    ///
    /// P5-3 safety hardening (Fable-5 review, LOW-MEDIUM "child safety-limit
    /// inheritance"): the monotonic-posture guarantee above was, before this
    /// fix, scoped to PERMISSION fields only — a child could still silently
    /// get a LOOSER safety BUDGET/BREAKER than its parent, because
    /// `max_total_output_tokens`/`max_tool_output_bytes`/`max_tokens`/
    /// `doom_loop_threshold`/`edit_file_require_read_before_edit` were never
    /// copied and so fell back to `Config::default()`'s (looser/uncapped)
    /// values on every spawn regardless of what the parent had configured.
    /// These are now copied verbatim alongside the permission-posture
    /// fields — a parent that capped its own output/tool-output/doom-loop
    /// exposure, or required read-before-edit, gets a child that is bound
    /// by the exact same ceiling, never a wider one.
    ///
    /// **Full field-by-field accounting** (every [`Config`] field, so this
    /// doc comment stays the single place that answers "did we forget
    /// one?"): fields already copied above/below this note (permission
    /// posture: `sandbox`/`approval`/`tool_overrides`/`auto_approved_tools`/
    /// `tool_deny_patterns`/`tool_allow_patterns`/`permissions_enabled`/
    /// `permissions_ask_patterns`/`permissions_protected_paths`/
    /// `network_policy`/`core_tools_enabled`/`module_registry`/
    /// `module_activation`/every `subagents_*` field; safety limits:
    /// `max_iterations`/`max_total_output_tokens`/`max_tool_output_bytes`/
    /// `max_tokens`/`doom_loop_threshold`/`edit_file_require_read_before_edit`;
    /// identity/transport: `model`/`system_prompt`/`cwd`/`base_url`/
    /// `api_key`/`api_key_env`/`api_key_cmd`) are the ones that gate
    /// harm/spend/hazard exposure. Every OTHER field is deliberately left at
    /// `Config::default()` because none of them is a safety ceiling the
    /// child could "loosen" by missing it:
    /// - `temperature`/`effort`/`response_format`/`extra_body`/`extra_headers`/
    ///   `tool_advertising`/`tool_schema_tier`/`cache_plan`/`cache_warnings`/
    ///   `reduction_policy`/
    ///   `session_*`/`small_model`/`model_fallback`/`env_context`/
    ///   `project_root_markers`/`project_doc_max_bytes`/`instruction_imports`/
    ///   `retry_*`/`compaction_*`/`auto_title`/`steering_mode`/
    ///   `follow_up_mode`/`read_file_multimodal`/`edit_file_notebook_aware`/
    ///   `shell_env_snapshot`/`nested_instructions`/`model_switch_allow_switch`/
    ///   `context_injections`/`context_injection_blocks`/`parallel_tool_calls`
    ///   are behavior/cost-shaping or presentation knobs, not hard guards —
    ///   a child defaulting on any of these can do LESS (e.g. no multimodal
    ///   read, no notebook-aware edits, no proactive compaction) or the same,
    ///   never something the parent hadn't already exposed it to. Several
    ///   default to their OFF/conservative state (`false`/`None`), which is
    ///   the tight direction, not the loose one.
    /// - `additional_dirs`: governs which extra roots are reachable at all
    ///   (`presets.rs`'s `[core] additional_dirs` note) — a child that
    ///   doesn't inherit it has FEWER reachable roots than its parent, i.e.
    ///   strictly tighter, never looser.
    /// - `load_project_context`: whether instruction files are auto-loaded
    ///   into the system prompt — a read-time convenience, not an access
    ///   grant (`sandbox`/`permissions_protected_paths` already gate actual
    ///   file access).
    /// - `prompts`: named `/slash` command templates for THIS agent's own
    ///   user-facing input surface, not something the model can invoke
    ///   against the child's tool surface.
    /// - `stop_gate`/`post_tool_hook`/`approval_handler`/`event_sink`:
    ///   code-only `Box<dyn Fn>` callbacks (see the `pre_tool_hook` note
    ///   immediately below — same non-`Clone` shape) that are observational
    ///   or terminate-only, not a call-time veto over what a tool is allowed
    ///   to do; `approval_handler` specifically is ALREADY documented at
    ///   this method's call site (`Self::run_spawn_subagent`) as
    ///   intentionally never set here — a foreground child gets no handler
    ///   by design, an embedder installs its own after spawn if it wants
    ///   one.
    ///
    /// **`pre_tool_hook` cannot propagate, and this is deliberate + named,
    /// not a silent gap**: `Config::pre_tool_hook` is a `Box<dyn Fn(&str,
    /// &serde_json::Value) -> Option<String> + Send + Sync>` — an
    /// embedder's own call-time veto over every tool call. `Box<dyn Fn>` is
    /// not `Clone` (there is no generic way to duplicate an opaque closure),
    /// so it genuinely CANNOT be copied into a child `Config` the way every
    /// `Clone`-able field above is — there is no fix that makes this one
    /// "verbatim copy" like the others. An embedder relying on a
    /// `pre_tool_hook` veto reaching spawned children as well as the parent
    /// MUST re-install one on the child explicitly (e.g. via a
    /// `spawn_subagent`-adjacent hook of their own, or by not relying on
    /// `pre_tool_hook` alone for anything safety-critical across a spawn
    /// boundary) — named here so this is a documented contract, not a gap
    /// an embedder discovers by a child silently misbehaving.
    fn build_child_config(
        &self,
        definition: Option<&crate::subagents::NamedAgentDefinition>,
        inline_system_prompt: Option<String>,
        model_override: Option<String>,
    ) -> Config {
        let system_prompt = definition
            .map(|d| d.system_prompt.clone())
            .filter(|s| !s.is_empty())
            .or(inline_system_prompt)
            .unwrap_or_else(|| self.config.system_prompt.clone());
        let model = model_override.unwrap_or_else(|| self.config.model.clone());

        let mut child = Config::builder()
            .model(model)
            .system_prompt(system_prompt)
            .cwd(self.config.cwd.clone())
            // Monotonic: verbatim, never loosened.
            .sandbox(self.config.sandbox)
            .approval(self.config.approval)
            .max_iterations(self.config.max_iterations)
            .build();
        child.base_url = self.config.base_url.clone();
        child.api_key = self.config.api_key.clone();
        child.api_key_env = self.config.api_key_env.clone();
        child.api_key_cmd = self.config.api_key_cmd.clone();
        // P5-3 safety hardening (Fable-5 review, LOW-MEDIUM "child
        // safety-limit inheritance"): the monotonic-posture spirit extends
        // to safety BUDGETS/BREAKERS, not just permissions — a child must
        // not get a looser cap/breaker than its parent by simply falling
        // back to `Config::default()`'s (looser) values. See this method's
        // doc comment for the full field-by-field accounting.
        child.max_total_output_tokens = self.config.max_total_output_tokens;
        child.max_tool_output_bytes = self.config.max_tool_output_bytes;
        child.max_tokens = self.config.max_tokens;
        child.doom_loop_threshold = self.config.doom_loop_threshold;
        child.edit_file_require_read_before_edit = self.config.edit_file_require_read_before_edit;
        // Monotonic tool posture: start from the PARENT's own overrides
        // (so anything the parent already disabled stays disabled), then
        // narrow further if a named definition restricts the tool set.
        child.tool_overrides = self.config.tool_overrides.clone();
        child.auto_approved_tools = self.config.auto_approved_tools.clone();
        child.tool_deny_patterns = self.config.tool_deny_patterns.clone();
        child.tool_allow_patterns = self.config.tool_allow_patterns.clone();
        child.permissions_enabled = self.config.permissions_enabled;
        child.permissions_ask_patterns = self.config.permissions_ask_patterns.clone();
        child.permissions_protected_paths = self.config.permissions_protected_paths.clone();
        child.network_policy = self.config.network_policy.clone();
        // P5-10 (§2 module 12): same monotonic-posture treatment as
        // `sandbox`/`approval` above — a subagent must inherit its
        // parent's OS-sandbox posture verbatim, never a looser
        // `Config::default()` fallback (`sandbox_os_enabled: None`,
        // `escalation: Deny`, `env_policy: Inherit` would otherwise be
        // right back to "confine only when the tier itself says so" for a
        // child whose parent explicitly forced the backstop on/off).
        child.sandbox_os_enabled = self.config.sandbox_os_enabled;
        child.sandbox_escalation = self.config.sandbox_escalation;
        child.sandbox_env_policy = self.config.sandbox_env_policy;
        if let Some(def) = definition {
            if let Some(allowed) = &def.tools {
                for name in &self.config.core_tools_enabled {
                    if !allowed.iter().any(|t| t == name) {
                        child
                            .tool_overrides
                            .entry(name.clone())
                            .or_default()
                            .enabled = Some(false);
                    }
                }
            }
        }
        child.core_tools_enabled = self.config.core_tools_enabled.clone();
        child.module_registry = self.config.module_registry;
        child.module_activation = self.config.module_activation.clone();
        // The subagents module itself never widens either: a child spawned
        // at depth d+1 inherits the SAME caps (never a looser depth/
        // concurrency/background posture than its own parent).
        child.subagents_enabled = self.config.subagents_enabled;
        child.subagents_max_depth = self.config.subagents_max_depth;
        child.subagents_max_concurrent = self.config.subagents_max_concurrent;
        child.subagents_background = self.config.subagents_background;
        child.subagents_background_prompts = self.config.subagents_background_prompts;
        child.subagents_claude_agent_alias = self.config.subagents_claude_agent_alias;
        child.subagents_definitions = self.config.subagents_definitions.clone();
        child.subagent_depth = self.subagent_depth + 1;
        child
    }

    /// Execute the `spawn_subagent` intrinsic (P5-3, §2 module 9). See
    /// `Self::build_child_config` for the monotonic-posture guarantee and
    /// `crate::subagents` for the depth/concurrency resource bounds and the
    /// §2.2 C6 background-policy enforcement.
    async fn run_spawn_subagent(&mut self, call: &crate::message::ToolCall) -> (String, bool) {
        if !self.config.subagents_enabled {
            let err = Error::UnknownTool(SPAWN_SUBAGENT.to_string());
            return (format!("Error: {err}"), true);
        }
        let args = match call.function.parsed_arguments() {
            Ok(v) => v,
            Err(e) => {
                let err = Error::InvalidArguments {
                    tool: SPAWN_SUBAGENT.to_string(),
                    message: e.to_string(),
                };
                return (format!("Error: {err}"), true);
            }
        };
        let task = args
            .get("task")
            .and_then(serde_json::Value::as_str)
            .unwrap_or("")
            .to_string();
        if task.is_empty() {
            let err = Error::InvalidArguments {
                tool: SPAWN_SUBAGENT.to_string(),
                message: "`task` is required and must be non-empty".to_string(),
            };
            return (format!("Error: {err}"), true);
        }
        let agent_type = args
            .get("agent_type")
            .and_then(serde_json::Value::as_str)
            .map(String::from);
        let inline_system_prompt = args
            .get("system_prompt")
            .and_then(serde_json::Value::as_str)
            .map(String::from);
        let background = args
            .get("background")
            .and_then(serde_json::Value::as_bool)
            .unwrap_or(false);
        let requested_model = args
            .get("model")
            .and_then(serde_json::Value::as_str)
            .map(|model| crate::model_catalog::resolve_alias(model, &[]));

        let definition = match &agent_type {
            Some(name) => match self.config.subagents_definitions.get(name) {
                Some(d) => Some(d.clone()),
                None => {
                    let err = Error::SubagentDefinitionNotFound(name.clone());
                    return (format!("Error: {err}"), true);
                }
            },
            None => None,
        };

        if background {
            if !self.config.subagents_background {
                let err = Error::tool(
                    SPAWN_SUBAGENT,
                    "background=true requires capabilities.subagents.background = true",
                );
                return (format!("Error: {err}"), true);
            }
            // §2.2 C6, defensive re-check (belt-and-suspenders — see
            // `Error::SubagentBackgroundPolicyMissing`'s doc comment for why
            // this can't just trust the resolver already checked it).
            if self.config.subagents_background_prompts.is_none() {
                let err = Error::SubagentBackgroundPolicyMissing;
                return (format!("Error: {err}"), true);
            }
        }

        // Resource bounds (fail-closed): depth first (cheap, no side
        // effect on failure), THEN concurrency (holds a slot — must be the
        // LAST check before actually spawning, so a refused spawn never
        // leaves a stray slot held).
        if let Err(e) =
            crate::subagents::check_depth(self.subagent_depth, self.config.subagents_max_depth)
        {
            return (format!("Error: {e}"), true);
        }
        let Some(guard) = crate::subagents::try_acquire(
            &self.subagent_concurrency_gauge,
            self.config.subagents_max_concurrent,
        ) else {
            let err = Error::SubagentConcurrencyExceeded {
                max_concurrent: self.config.subagents_max_concurrent,
            };
            return (format!("Error: {err}"), true);
        };

        let child_id = next_subagent_id();
        let child_config = self.build_child_config(
            definition.as_ref(),
            inline_system_prompt,
            requested_model.or_else(|| definition.as_ref().and_then(|d| d.model.clone())),
        );
        let child_model = child_config.model.clone();
        let mut child = Agent::with_provider_arc(child_config, self.provider.clone());
        child.subagent_depth = self.subagent_depth + 1;
        child.subagent_concurrency_gauge = self.subagent_concurrency_gauge.clone();

        // §2.2 C6: a background child NEVER gets a BLOCKING-BY-DEFAULT
        // interactive approval handler — either no handler at all
        // (`AutoPolicy`: the engine's pre-existing "no handler ⇒ deny"
        // fail-closed default), or (`Parent`) the never-blocking
        // `ParentQueueApprovalHandler`, UNLESS a `tui` embedder has
        // installed [`Self::child_approval_handler_factory`] (P5-4), in
        // which case THAT builds the handler instead — see
        // [`Self::set_child_approval_handler_factory`]'s doc comment for
        // why this can't escalate past what the rule engine already routed
        // to `Ask`. A foreground child also gets no handler here (today's
        // existing default posture; an embedder that wants an interactive
        // child installs its own via `set_permissions_approval_handler`
        // after this call returns, out of this method's scope).
        if background {
            if let Some(crate::subagents::BackgroundPromptsPolicy::Parent) =
                self.config.subagents_background_prompts
            {
                let handler: std::sync::Arc<dyn crate::permissions::PermissionsApprovalHandler> =
                    match &self.child_approval_handler_factory {
                        Some(factory) => {
                            factory(child_id.clone(), self.pending_child_approvals.clone())
                        }
                        None => std::sync::Arc::new(crate::subagents::ParentQueueApprovalHandler {
                            child_agent_id: child_id.clone(),
                            queue: self.pending_child_approvals.clone(),
                        }),
                    };
                child.ctx.sandbox_approval_handler =
                    Some(crate::sandbox::SandboxApprovalHandler(handler.clone()));
                child.permissions_approval_handler = Some(handler);
            }
        }

        let lineage = crate::subagents::SubagentLineage {
            child_agent_id: child_id.clone(),
            parent_session_id: self.subagent_store.as_ref().map(|(_, name)| name.clone()),
            parent_tool_use_id: call.id.clone(),
            depth: self.subagent_depth + 1,
            agent_type: agent_type.clone(),
            task: task.clone(),
            background,
            spawned_at_ms: now_ms(),
            model: child_model,
        };
        if let Some((store, parent_name)) = &self.subagent_store {
            let _ = store.save_subagent_lineage(parent_name, &child_id, &lineage);
        }

        if background {
            let spawned_task_text = task.clone();
            self.background_subagents.insert(
                child_id.clone(),
                BackgroundSubagent {
                    handle: tokio::spawn(async move {
                        // The concurrency slot lives for exactly as long as
                        // this future runs — moved in here, dropped when the
                        // child's `send` (and this future) finishes.
                        let _guard = guard;
                        let result = child.send(spawned_task_text).await;
                        let transcript = child.history()[1..].to_vec();
                        (child_id, result, transcript)
                    }),
                    task,
                    agent_type,
                    started_at_ms: lineage.spawned_at_ms,
                },
            );
            let out = serde_json::json!({
                "subagent_id": lineage.child_agent_id,
                "status": "spawned",
                "background": true,
            });
            return (out.to_string(), false);
        }

        // Foreground: run to completion now, guard held until this
        // function returns (then drops, freeing the slot).
        let result = child.send(task).await;
        let transcript = child.history()[1..].to_vec();
        self.persist_subagent_transcript(&child_id, &lineage, &transcript);
        drop(guard);
        match result {
            Ok(text) => (text, false),
            Err(e) => (format!("Error: subagent `{child_id}` failed: {e}"), true),
        }
    }

    /// Execute the `subagent_status` intrinsic (P5-3, D3
    /// "background+resume"): poll a background child; once its `JoinHandle`
    /// is finished, reap it (removing it from `Self::background_subagents`
    /// and persisting its transcript, same as the foreground path).
    async fn run_subagent_status(&mut self, call: &crate::message::ToolCall) -> (String, bool) {
        let args = match call.function.parsed_arguments() {
            Ok(v) => v,
            Err(e) => {
                let err = Error::InvalidArguments {
                    tool: SUBAGENT_STATUS.to_string(),
                    message: e.to_string(),
                };
                return (format!("Error: {err}"), true);
            }
        };
        let Some(id) = args.get("subagent_id").and_then(serde_json::Value::as_str) else {
            let err = Error::InvalidArguments {
                tool: SUBAGENT_STATUS.to_string(),
                message: "`subagent_id` is required".to_string(),
            };
            return (format!("Error: {err}"), true);
        };
        let Some(entry) = self.background_subagents.get(id) else {
            let err = Error::SubagentNotFound(id.to_string());
            return (format!("Error: {err}"), true);
        };
        if !entry.handle.is_finished() {
            let out = serde_json::json!({
                "subagent_id": id,
                "status": "pending",
                "task": entry.task,
                "agent_type": entry.agent_type,
                "started_at_ms": entry.started_at_ms,
            });
            return (out.to_string(), false);
        }
        // Finished — reap it. `.await` on an already-finished handle
        // resolves immediately (never actually blocks).
        let entry = self
            .background_subagents
            .remove(id)
            .expect("checked Some above");
        let (child_id, result, transcript) = match entry.handle.await {
            Ok(v) => v,
            Err(join_err) => {
                let err = Error::tool(
                    SUBAGENT_STATUS,
                    format!("subagent `{id}` task panicked: {join_err}"),
                );
                return (format!("Error: {err}"), true);
            }
        };
        // Re-derive the lineage record for persistence (cheap; the fields
        // are all still in hand) — mirrors the foreground path's single
        // `persist_subagent_transcript` call site.
        if let Some((store, parent_name)) = self.subagent_store.clone() {
            if let Ok(Some(lineage)) = store.load_subagent_lineage(&parent_name, &child_id) {
                self.persist_subagent_transcript(&child_id, &lineage, &transcript);
            }
        }
        match result {
            Ok(text) => {
                let out = serde_json::json!({
                    "subagent_id": child_id,
                    "status": "done",
                    "result": text,
                });
                (out.to_string(), false)
            }
            Err(e) => {
                let out = serde_json::json!({
                    "subagent_id": child_id,
                    "status": "error",
                    "message": e.to_string(),
                });
                (out.to_string(), true)
            }
        }
    }

    /// Execute the `background_exec` intrinsic (P5-6, §2 module 4, D1
    /// "background exec"): spawn `args.command` as a detached OS process
    /// via [`crate::tools::build_sandboxed_sh`] — the SAME sandboxed-spawn
    /// path [`crate::tools::BashTool::execute`] uses — and return its job
    /// id IMMEDIATELY, never the command's output. Gated by the same
    /// permission check a foreground `bash` call gets
    /// ([`Self::background_permission_denial`]), then a fail-closed
    /// concurrency cap ([`Config::tools_background_max_concurrent`]), THEN
    /// the actual spawn — in that order, so a refused call never holds a
    /// concurrency slot and never touches the process table.
    fn run_background_exec(&mut self, call: &crate::message::ToolCall) -> (String, bool) {
        let args = match call.function.parsed_arguments() {
            Ok(v) => v,
            Err(e) => {
                let err = Error::InvalidArguments {
                    tool: BACKGROUND_EXEC.to_string(),
                    message: e.to_string(),
                };
                return (format!("Error: {err}"), true);
            }
        };
        let command = args
            .get("command")
            .and_then(serde_json::Value::as_str)
            .unwrap_or("")
            .to_string();
        if command.is_empty() {
            let err = Error::InvalidArguments {
                tool: BACKGROUND_EXEC.to_string(),
                message: "`command` is required and must be non-empty".to_string(),
            };
            return (format!("Error: {err}"), true);
        }

        // A job id up front (before spawning) — used both as the audit
        // handle for a §2.2 C6 `Parent`-policy queued denial (this call may
        // never actually reach the spawn below) and, if the call proceeds,
        // as `Self::background_jobs`'s real key.
        let job_id = crate::background::next_job_id(now_ms());

        if let Some(reason) = self.background_permission_denial(&command, &job_id) {
            return (format!("Error: {reason}"), true);
        }

        // Fable-5 review (LOW, "pre_tool_hook + doom-loop don't cover
        // background_exec"): this intrinsic is intercepted in
        // `Self::prepare_tool_call` and returns before `Self::finish_prepare`
        // ever runs, so — unlike a foreground `bash` call — it was reaching
        // this real spawn below WITHOUT ever offering `Config.pre_tool_hook`
        // a chance to veto it. `background_exec` runs a REAL command (unlike
        // the purely in-process meta-intrinsics `tool_search`/
        // `expand_reduction`/`sidecar_search`, which have no such gap to
        // close), so it belongs behind the same security-relevant veto a
        // foreground call gets. Scoped to this one call site — the other
        // meta-intrinsics are unchanged. The doom-loop counter
        // (`Self::check_doom_loop`) is deliberately NOT wired here: it is a
        // foreground repetition breaker keyed on `(self.doom_loop_last_call,
        // self.doom_loop_streak)`, a single piece of state shared with the
        // ordinary tool-call loop — folding background jobs into that same
        // streak would make an interleaved foreground/background pattern
        // trip (or fail to trip) the breaker in ways that have nothing to
        // do with the foreground loop actually repeating itself; the
        // pre_tool_hook veto below is the security-relevant half of this
        // fix, the doom-loop breaker is not.
        if let Some(hook) = &self.config.pre_tool_hook {
            if let Some(reason) = hook(BACKGROUND_EXEC, &args) {
                return (format!("Error: blocked by pre-tool hook: {reason}"), true);
            }
        }

        let Some(guard) = crate::subagents::try_acquire(
            &self.background_concurrency_gauge,
            self.config.tools_background_max_concurrent,
        ) else {
            let err = Error::BackgroundJobConcurrencyExceeded {
                max_concurrent: self.config.tools_background_max_concurrent,
            };
            return (format!("Error: {err}"), true);
        };

        let mut cmd = match crate::tools::build_sandboxed_sh(&command, &self.ctx) {
            Ok(cmd) => cmd,
            Err(e) => return (format!("Error: {e}"), true),
        };
        cmd.current_dir(&self.ctx.cwd)
            .stdin(std::process::Stdio::null())
            .stdout(std::process::Stdio::piped())
            .stderr(std::process::Stdio::piped())
            // Defense-in-depth for the "must be killed on drop" guarantee —
            // see `impl Drop for Agent`'s doc comment; the EXPLICIT
            // `start_kill()` loop there is what makes the guarantee
            // provable, this is a second, independent line of defense for
            // the same outcome.
            .kill_on_drop(true);
        // Fable-5 review (HIGH, "grandchildren orphaned on kill AND
        // agent-drop"): `Child::start_kill` only signals the DIRECT child.
        // A background command that spawns a surviving subprocess (a `&`
        // job, a pipeline, a double-forking daemon — or, on macOS, the
        // `sandbox-exec` wrapper itself in `build_sandboxed_sh`, whose real
        // `sh` and ITS children are all grandchildren of the tracked pid)
        // leaves those processes running, reparented to init, after the
        // tracked job is "killed". Putting this job in its OWN new process
        // group (`pgid == its own pid`, since every descendant inherits the
        // group unless it explicitly opts out) lets `kill_job_process_group`
        // below signal the WHOLE tree at kill/drop time, not just the one
        // pid we happen to be tracking. No portable equivalent on Windows —
        // see `kill_job_process_group`'s `#[cfg(not(unix))]` fallback.
        #[cfg(unix)]
        cmd.process_group(0);
        // P4c (`core.shell_env_snapshot`)/P5-10 (`env_policy`):
        // `build_sandboxed_sh` (above) already applied both via its own
        // `apply_sandbox_env_policy` last step — no separate `ctx.shell_env`
        // application here (that would re-add a secret `Filtered`/`None`
        // just stripped, on top of the already-`env_clear`'d command).

        let mut child = match cmd.spawn() {
            Ok(c) => c,
            Err(e) => {
                drop(guard);
                let err = Error::tool(
                    BACKGROUND_EXEC,
                    format!("failed to spawn background command: {e}"),
                );
                return (format!("Error: {err}"), true);
            }
        };
        let pid = child.id();
        let output = std::sync::Arc::new(crate::background::CapturedOutput::new());
        let cap = self.config.tools_background_max_output_bytes;
        // Fire-and-forget: the reader tasks outlive this method call and
        // exit on their own at pipe EOF — see `spawn_output_reader`'s doc
        // comment. Bound to named (not `_`) locals only to keep clippy's
        // `let_underscore_future` lint quiet; neither handle is awaited or
        // aborted anywhere.
        if let Some(stdout) = child.stdout.take() {
            let _stdout_reader = spawn_output_reader(stdout, output.clone(), cap);
        }
        if let Some(stderr) = child.stderr.take() {
            let _stderr_reader = spawn_output_reader(stderr, output.clone(), cap);
        }

        let started_at_ms = now_ms();
        self.background_jobs.insert(
            job_id.clone(),
            BackgroundJob {
                child,
                command: command.clone(),
                pid,
                output,
                started_at_ms,
                killed: false,
                _guard: guard,
            },
        );

        let out = serde_json::json!({
            "job_id": job_id,
            "status": "running",
            "pid": pid,
            "command": command,
        });
        (out.to_string(), false)
    }

    /// Execute the `background_status` intrinsic (P5-6, D1 "monitor/event
    /// feed"): non-blocking poll of one job's run status (via
    /// `Child::try_wait`), drain its output captured since the LAST poll
    /// and emit it as an [`AgentEvent::BackgroundOutput`] event (the
    /// "event feed" — a real `EventSink` consumer sees each poll's new
    /// output live), and return the full captured output (bounded, per
    /// [`Config::tools_background_max_output_bytes`]) so far either way.
    /// Once the job is terminal (exited or killed), this reaps it — removes
    /// it from [`Self::background_jobs`], freeing its concurrency slot —
    /// same "poll once more to reap" contract [`Self::run_subagent_status`]
    /// already established for background subagents.
    fn run_background_status(&mut self, call: &crate::message::ToolCall) -> (String, bool) {
        let args = match call.function.parsed_arguments() {
            Ok(v) => v,
            Err(e) => {
                let err = Error::InvalidArguments {
                    tool: BACKGROUND_STATUS.to_string(),
                    message: e.to_string(),
                };
                return (format!("Error: {err}"), true);
            }
        };
        let Some(job_id) = args.get("job_id").and_then(serde_json::Value::as_str) else {
            let err = Error::InvalidArguments {
                tool: BACKGROUND_STATUS.to_string(),
                message: "`job_id` is required".to_string(),
            };
            return (format!("Error: {err}"), true);
        };
        let job_id = job_id.to_string();

        // Scoped so the mutable borrow of `self.background_jobs` ends
        // before `self.emit(...)`/`self.background_jobs.remove(...)` below
        // need their own (mutable) access to `self`.
        let (command, pid, started_at_ms, status, output_so_far, truncated, delta) = {
            let Some(job) = self.background_jobs.get_mut(&job_id) else {
                let err = Error::BackgroundJobNotFound(job_id);
                return (format!("Error: {err}"), true);
            };
            let status = background_job_status(job);
            let (output_so_far, truncated) = job.output.snapshot();
            let delta = job.output.drain_new();
            (
                job.command.clone(),
                job.pid,
                job.started_at_ms,
                status,
                output_so_far,
                truncated,
                delta,
            )
        };

        if !delta.is_empty() {
            self.emit(AgentEvent::BackgroundOutput {
                job_id: job_id.clone(),
                chunk: delta,
                truncated,
            });
        }

        let exit_code = match status {
            crate::background::JobStatus::Exited(code) => code,
            _ => None,
        };
        let out = serde_json::json!({
            "job_id": job_id,
            "command": command,
            "status": status.as_str(),
            "exit_code": exit_code,
            "pid": pid,
            "started_at_ms": started_at_ms,
            "output": output_so_far,
            "output_truncated": truncated,
        });
        if !matches!(status, crate::background::JobStatus::Running) {
            self.background_jobs.remove(&job_id);
        }
        (out.to_string(), false)
    }

    /// Execute the `background_list` intrinsic (P5-6, D10 "bg-manager"):
    /// list every background job this agent is currently tracking, without
    /// draining output or reaping anything (a read-only listing —
    /// `background_status` is the reaping poll).
    fn run_background_list(&mut self, _call: &crate::message::ToolCall) -> (String, bool) {
        let mut jobs = Vec::new();
        for (job_id, job) in self.background_jobs.iter_mut() {
            let status = background_job_status(job);
            jobs.push(serde_json::json!({
                "job_id": job_id,
                "command": job.command,
                "status": status.as_str(),
                "pid": job.pid,
                "started_at_ms": job.started_at_ms,
            }));
        }
        let out = serde_json::json!({ "jobs": jobs });
        (out.to_string(), false)
    }

    /// Execute the `background_kill` intrinsic (P5-6, D10 "bg-manager",
    /// build brief "kill/cancel a job"): request REAL termination of a
    /// background job's OS process AND its whole process group (see
    /// [`kill_job_process_group`] — Fable-5 review, HIGH, "grandchildren
    /// orphaned on kill"; a documented no-op if the process already
    /// exited) and reap it immediately, freeing its concurrency slot.
    fn run_background_kill(&mut self, call: &crate::message::ToolCall) -> (String, bool) {
        let args = match call.function.parsed_arguments() {
            Ok(v) => v,
            Err(e) => {
                let err = Error::InvalidArguments {
                    tool: BACKGROUND_KILL.to_string(),
                    message: e.to_string(),
                };
                return (format!("Error: {err}"), true);
            }
        };
        let Some(job_id) = args.get("job_id").and_then(serde_json::Value::as_str) else {
            let err = Error::InvalidArguments {
                tool: BACKGROUND_KILL.to_string(),
                message: "`job_id` is required".to_string(),
            };
            return (format!("Error: {err}"), true);
        };
        let job_id = job_id.to_string();
        let Some(mut job) = self.background_jobs.remove(&job_id) else {
            let err = Error::BackgroundJobNotFound(job_id);
            return (format!("Error: {err}"), true);
        };
        kill_job_process_group(&mut job);
        job.killed = true;
        let out = serde_json::json!({
            "job_id": job_id,
            "status": "killed",
            "pid": job.pid,
        });
        // `job` (and its `ConcurrencyGuard`) drops here, freeing the slot.
        (out.to_string(), false)
    }

    /// P5-3 (D5 "subagent transcripts… persisted + linked"): write a
    /// finished child's transcript to `Self::subagent_store`, if one is
    /// installed — a no-op otherwise (see that field's doc comment). Builds
    /// the child's `Session` the same way `to_native_jsonl_v2`'s doc
    /// comment describes (an empty imported prefix + `transcript` as
    /// `appended` `NativeTurn`s), with `meta.agent_id`/`parent_tool_use_id`/
    /// `lineage` populated from `lineage` so the native-v2 header carries
    /// the full lineage record on disk (see `Session::to_native_jsonl_v2`'s
    /// P5-3 doc note).
    ///
    /// P5-3 safety-hardening fix (Fable-5 review, LOW "translation-fidelity
    /// cosmetic"): `Session::from_claude_code_str("")` is used ONLY to get
    /// a blank `raw`/`messages` skeleton cheaply (an empty string parses
    /// identically under any loader) — it is NOT claiming this child's
    /// session actually came from Claude Code. Before this fix, that
    /// borrowed constructor's `meta.source` (`SessionSource::ClaudeCode`)
    /// leaked straight through to the persisted sidecar's `source` header,
    /// mislabeling a native `spawn_subagent` child as an imported CC
    /// session. Corrected to `SessionSource::Native` immediately after —
    /// see that variant's doc comment.
    fn persist_subagent_transcript(
        &self,
        child_id: &str,
        lineage: &crate::subagents::SubagentLineage,
        transcript: &[ChatMessage],
    ) {
        let Some((store, parent_name)) = &self.subagent_store else {
            return;
        };
        let mut session = match Session::from_claude_code_str("") {
            Ok(s) => s,
            Err(_) => return,
        };
        session.meta.source = crate::session::SessionSource::Native;
        session.meta.agent_id = Some(lineage.child_agent_id.clone());
        session.meta.parent_tool_use_id = Some(lineage.parent_tool_use_id.clone());
        session.meta.lineage = lineage.to_lineage_map();
        let sidecar_jsonl = session.to_native_jsonl_v2(transcript);
        let _ = store.save_subagent_transcript(parent_name, child_id, &sidecar_jsonl);
        let _ = store.save_subagent_lineage(parent_name, child_id, lineage);
    }

    /// Execute the `tool_search` intrinsic (B6): case-insensitive keyword
    /// match over `name` + `description` of every registered, enabled,
    /// non-core, not-yet-activated tool (builtin and `mcp__*` alike). Matches
    /// are activated (advertised starting with the next request) and
    /// returned as a JSON array of their full [`ToolSchema`]s.
    fn run_tool_search(&mut self, call: &crate::message::ToolCall) -> (String, bool) {
        let args = match call.function.parsed_arguments() {
            Ok(v) => v,
            Err(e) => {
                let err = Error::InvalidArguments {
                    tool: TOOL_SEARCH.to_string(),
                    message: e.to_string(),
                };
                return (format!("Error: {err}"), true);
            }
        };
        let query = args
            .get("query")
            .and_then(serde_json::Value::as_str)
            .unwrap_or("")
            .to_lowercase();
        let max_results = args
            .get("max_results")
            .and_then(serde_json::Value::as_u64)
            .map(|n| n as usize);

        let mut matches: Vec<ToolSchema> = self
            .registry
            .iter()
            .filter(|t| self.config.tool_enabled(t.name()))
            .filter(|t| !self.is_core_tool(t.name()))
            .filter(|t| !self.activated_tools.contains(t.name()))
            .filter(|t| {
                query.is_empty()
                    || t.name().to_lowercase().contains(&query)
                    || self
                        .config
                        .tool_description(t.name(), t.description())
                        .to_lowercase()
                        .contains(&query)
            })
            // TR-8/T5 dev/03: the on-demand fetch always returns the ORIGINAL
            // full schema, never the tier-minified one — that's the invert.
            .map(|t| self.raw_schema_for(t))
            .collect();

        if let Some(max) = max_results {
            matches.truncate(max);
        }

        for m in &matches {
            self.activated_tools.insert(m.name.clone());
        }

        let result = serde_json::to_string(&matches).unwrap_or_else(|_| "[]".to_string());
        (result, false)
    }

    /// The `expand_reduction` schema (T12/TR-1), advertised whenever a
    /// [`ReductionPolicy`] is installed.
    ///
    /// The description deliberately never spells the literal stub sentinel
    /// prefix: A11's export leak guard is unconditional, so an assistant
    /// turn that quoted a stub line verbatim (which teaching the syntax
    /// invites) would permanently fail export for that session. Stubs are
    /// described abstractly and the model is told to pass ids only.
    fn expand_reduction_schema() -> ToolSchema {
        ToolSchema {
            name: EXPAND_REDUCTION.to_string(),
            description: "Fetch back the original content hidden behind a reduction stub in \
                your current view — a truncated tool output, cleared old turns, or an elided \
                file read that was hidden to save context. Each stub line names a reduction id \
                like r0042-9f3c: pass ONLY that id here, and never quote or repeat a stub line \
                itself in your replies. The original is durably kept in the session sidecar. \
                Pass `byte_range` to fetch a slice of a large one at a time instead of all of \
                it at once; ranged results are prefixed with a `bytes start..end of total` \
                header so you can plan the next slice."
                .to_string(),
            parameters: serde_json::json!({
                "type": "object",
                "properties": {
                    "reduction_id": {
                        "type": "string",
                        "description": "The reduction id named in the stub line, e.g. \
                            \"r0042-9f3c\". Pass the id alone."
                    },
                    "byte_range": {
                        "type": "array",
                        "items": {"type": "integer"},
                        "minItems": 2,
                        "maxItems": 2,
                        "description": "Optional [start, end) byte offsets within the original \
                            content to fetch instead of all of it. Exactly two non-negative \
                            integers with start <= end."
                    }
                },
                "required": ["reduction_id"],
                "additionalProperties": false
            }),
        }
    }

    /// The `sidecar_search` schema (T12/TR-1), advertised whenever a
    /// [`ReductionPolicy`] is installed. Same no-literal-sentinel rule as
    /// [`Self::expand_reduction_schema`].
    fn sidecar_search_schema() -> ToolSchema {
        ToolSchema {
            name: SIDECAR_SEARCH.to_string(),
            description: "Search content currently hidden from your view by reduction stubs \
                (large tool outputs, cleared old turns, elided file reads) for a substring or \
                regex. Only hidden content is searched, never what you can already see. \
                Returns match snippets with each match's reduction_id for use with \
                expand_reduction; refer to results by their reduction id rather than quoting \
                stub lines. Results are capped — if `truncated` is true, narrow the query."
                .to_string(),
            parameters: serde_json::json!({
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "Non-empty substring or regex to search for \
                            (case-insensitive)."
                    }
                },
                "required": ["query"],
                "additionalProperties": false
            }),
        }
    }

    /// Reload the recorder's full recorded messages from disk (TR-1's
    /// `recorded` resolution source). Since TR-12's D6/A7 supersession gate
    /// (`Self::run_loop`), a `expand_reduction`/`sidecar_search` call only
    /// ever exists alongside an active [`ReductionPolicy`] (see
    /// [`EXPAND_REDUCTION`]'s doc), and pairing one with a recorder — as the
    /// CLI's reduced mode always does — means the gate is already on and
    /// `history[1..]` holds the same full bytes as this reload: this upgrade
    /// is then a dormant no-op (`reduce::rehydrate::prefer_recorded` sees
    /// `recorded == minted` and keeps `minted`). It stops being a no-op —
    /// defense in depth, not the common path — for a **legacy** sidecar
    /// recorded before this gate existed, or for a policy-without-recorder
    /// agent (gate off, so `history[1..]` still carries
    /// [`Self::cap_tool_output`]-capped copies): only there can `history[1..]`
    /// diverge from the sidecar, and only there does consulting this reload
    /// actually recover bytes `history[1..]` alone couldn't. `Ok(None)` when
    /// no recorder is attached (rehydration then resolves from history alone,
    /// whose capped copies — if any — carry their own honest cap notice). A
    /// disk-level reload is fine here regardless: these intrinsic calls are
    /// rare, model-initiated events, not per-request work.
    fn recorded_messages(&self) -> std::result::Result<Option<Vec<ChatMessage>>, String> {
        let Some(recorder) = &self.recorder else {
            return Ok(None);
        };
        let raw = std::fs::read_to_string(recorder.path())
            .map_err(|e| format!("failed to read the session sidecar: {e}"))?;
        let session = Session::from_sidecar_str(&raw)
            .map_err(|e| format!("failed to parse the session sidecar: {e}"))?;
        Ok(Some(session.messages))
    }

    /// Execute the `expand_reduction` intrinsic (T12/TR-1): resolves against
    /// `self.reduction_log` + `self.history[1..]` (the hash-minting source),
    /// upgraded to the recorder's full recorded bytes for cap-diverged
    /// content ([`Self::recorded_messages`]; the two-source contract is
    /// documented on `reduce::rehydrate`). `byte_range` is validated
    /// strictly — any malformed shape is a model-recoverable error naming
    /// the expected form and the original's true size, never a silent
    /// whole-content (or empty) return.
    fn run_expand_reduction(&mut self, call: &crate::message::ToolCall) -> (String, bool) {
        let args = match call.function.parsed_arguments() {
            Ok(v) => v,
            Err(e) => {
                let err = Error::InvalidArguments {
                    tool: EXPAND_REDUCTION.to_string(),
                    message: e.to_string(),
                };
                return (format!("Error: {err}"), true);
            }
        };
        let Some(id) = args.get("reduction_id").and_then(serde_json::Value::as_str) else {
            return (
                "Error: expand_reduction requires a `reduction_id` string argument".to_string(),
                true,
            );
        };
        let recorded = match self.recorded_messages() {
            Ok(r) => r,
            Err(e) => return (format!("Error: expand_reduction: {e}"), true),
        };
        let recorded = recorded.as_deref();

        // B3: strict shape validation — exactly two non-negative integers.
        // Anything else errors (with the true total when resolvable) rather
        // than silently degrading to a whole-content expand.
        let byte_range = match args.get("byte_range") {
            None | Some(serde_json::Value::Null) => None,
            Some(v) => {
                let parsed = v
                    .as_array()
                    .filter(|a| a.len() == 2)
                    .and_then(|a| Some((a[0].as_u64()? as usize, a[1].as_u64()? as usize)));
                match parsed {
                    Some(range) => Some(range),
                    None => {
                        let total = reduce::rehydrate::reduction_total_bytes(
                            &self.reduction_log,
                            &self.history[1..],
                            recorded,
                            id,
                        )
                        .map(|n| format!("; the original is {n} bytes"))
                        .unwrap_or_default();
                        return (
                            format!(
                                "Error: expand_reduction: malformed byte_range {v} — expected \
                                 [start, end): exactly two non-negative integers with \
                                 start <= end{total}"
                            ),
                            true,
                        );
                    }
                }
            }
        };
        match reduce::rehydrate::expand_reduction(
            &self.reduction_log,
            &self.history[1..],
            recorded,
            id,
            byte_range,
        ) {
            // A ranged result carries a provenance header naming the slice
            // and the true total, so the model can plan its next slice; a
            // whole-content expand stays byte-exact (TR-1 dev/01).
            Ok(outcome) => match outcome.range {
                Some((start, end)) => (
                    format!(
                        "[{id}: bytes {start}..{end} of {total}]\n{content}",
                        total = outcome.total_bytes,
                        content = outcome.content
                    ),
                    false,
                ),
                None => (outcome.content, false),
            },
            Err(e) => (format!("Error: {e}"), true),
        }
    }

    /// Execute the `sidecar_search` intrinsic (T12/TR-1); same two-source
    /// resolution as [`Self::run_expand_reduction`]. The result is bounded
    /// by construction (`reduce::rehydrate::SidecarSearchResult`'s caps), so
    /// a broad query can never re-inflate the context or bloat the sidecar
    /// the recorder appends this result to.
    fn run_sidecar_search(&mut self, call: &crate::message::ToolCall) -> (String, bool) {
        let args = match call.function.parsed_arguments() {
            Ok(v) => v,
            Err(e) => {
                let err = Error::InvalidArguments {
                    tool: SIDECAR_SEARCH.to_string(),
                    message: e.to_string(),
                };
                return (format!("Error: {err}"), true);
            }
        };
        let query = args
            .get("query")
            .and_then(serde_json::Value::as_str)
            .unwrap_or("");
        if query.trim().is_empty() {
            return (
                "Error: sidecar_search requires a non-empty `query` string argument".to_string(),
                true,
            );
        }
        let recorded = match self.recorded_messages() {
            Ok(r) => r,
            Err(e) => return (format!("Error: sidecar_search: {e}"), true),
        };
        match reduce::rehydrate::sidecar_search(
            &self.reduction_log,
            &self.history[1..],
            recorded.as_deref(),
            query,
        ) {
            Ok(result) => (
                serde_json::to_string(&result).unwrap_or_else(|_| "{}".to_string()),
                false,
            ),
            Err(e) => (format!("Error: {e}"), true),
        }
    }

    fn emit(&self, event: AgentEvent) {
        if let Some(sink) = &self.config.event_sink {
            sink(event);
        }
    }

    /// Number of non-system messages exchanged so far.
    pub fn turn_count(&self) -> usize {
        self.history
            .iter()
            .filter(|m| m.role != Role::System)
            .count()
    }

    /// Cumulative output (completion) tokens reported by the provider across
    /// every `send` on this agent. Zero if the provider reports no usage.
    pub fn total_output_tokens(&self) -> u64 {
        self.total_output_tokens
    }
}

#[cfg(test)]
mod api_key_cmd_tests {
    //! P4 (design §5.2, §1.8 D6 row): `api_key_cmd` credential-helper
    //! resolution. `Agent::new` never makes a network call, so these tests
    //! exercise the real resolution chain end-to-end without mocking.

    use super::*;

    /// Default-off: with no `api_key`/`api_key_cmd` set and an env var that
    /// isn't set either, resolution fails exactly as it always has —
    /// `api_key_cmd` being a brand-new field changes nothing when unset.
    #[test]
    fn default_none_falls_through_to_missing_api_key_error() {
        let config = Config::builder()
            .api_key_env("SUPERCODE_TEST_UNSET_VAR_API_KEY_CMD")
            .build();
        assert!(config.api_key.is_none());
        assert!(config.api_key_cmd.is_none());
        let err = Agent::new(config).err().expect("no key source configured");
        assert!(matches!(err, Error::MissingApiKey(_)));
    }

    /// Happy path: `api_key_cmd` alone (no `api_key`, no matching env var)
    /// is enough for `Agent::new` to succeed — the helper's stdout is
    /// resolved and used.
    #[test]
    fn api_key_cmd_alone_resolves_successfully() {
        let config = Config::builder()
            .api_key_cmd("echo sk-test-from-helper")
            .api_key_env("SUPERCODE_TEST_UNSET_VAR_API_KEY_CMD_2")
            .build();
        assert!(Agent::new(config).is_ok());
    }

    /// A failing helper command (non-zero exit, or empty stdout) falls
    /// through to `api_key_env` rather than propagating the helper's own
    /// failure — same "try the next source" posture as every other layer.
    #[test]
    fn api_key_cmd_failure_falls_through_to_env() {
        std::env::set_var(
            "SUPERCODE_TEST_API_KEY_CMD_FALLBACK",
            "sk-from-env-fallback",
        );
        let config = Config::builder()
            .api_key_cmd("exit 1")
            .api_key_env("SUPERCODE_TEST_API_KEY_CMD_FALLBACK")
            .build();
        assert!(Agent::new(config).is_ok());
        std::env::remove_var("SUPERCODE_TEST_API_KEY_CMD_FALLBACK");
    }

    /// A failing helper AND no fallback env var still produces the same
    /// `MissingApiKey` error today's no-key path always produced — the new
    /// source never turns a hard failure into a silent empty key.
    #[test]
    fn api_key_cmd_failure_with_no_fallback_still_errors() {
        let config = Config::builder()
            .api_key_cmd("exit 1")
            .api_key_env("SUPERCODE_TEST_UNSET_VAR_API_KEY_CMD_3")
            .build();
        let err = Agent::new(config)
            .err()
            .expect("helper failed, no env fallback");
        assert!(matches!(err, Error::MissingApiKey(_)));
    }

    /// `run_api_key_cmd` directly: happy path trims trailing whitespace/
    /// newline from the command's stdout.
    #[test]
    fn run_api_key_cmd_trims_output() {
        assert_eq!(run_api_key_cmd("echo '  sk-abc123  '"), "sk-abc123");
    }

    /// `run_api_key_cmd` directly: a nonexistent binary fails to spawn and
    /// returns an empty string rather than panicking.
    #[test]
    fn run_api_key_cmd_spawn_failure_returns_empty() {
        // `sh -c` itself always spawns; feed it a command that can't run.
        assert_eq!(
            run_api_key_cmd("/no/such/binary/at/all --flag"),
            String::new()
        );
    }
}