agent_engine/runtime/

mod.rs

use crate::{Result, RuntimeError, ToolRegistry};
use futures::stream::Stream;
use reqwest::Client;
use serde_json::{json, Value};
use std::path::{Path, PathBuf};
use std::pin::Pin;
use std::sync::Arc;
use std::sync::Mutex;
use std::time::Duration;
use tokio::sync::{mpsc, RwLock};
use tokio_util::sync::CancellationToken;

mod anthropic_quota;
mod api;
mod api_sync;
pub mod attachments;
mod auth;
#[cfg(test)]
mod body_golden;
pub mod budget;
mod cache_diagnosis;
pub(crate) mod cloud_invoke;
pub mod compaction;
pub mod context;
pub mod continuation;
pub mod google_gemini;
pub mod google_vertex;
pub(crate) mod helpers;
mod http_phases;
mod image_limits;
mod image_retention;
pub mod keepwarm;
pub mod tool_keepwarm;
/// The cache-simulating provider of the tool-round keep-warm tests
/// (`runtime::stream`, `tools::subagent`; host paths:
/// `tests/worker_tool_keepwarm.rs`).
#[cfg(test)]
#[path = "../../tests/support/cache_sim.rs"]
pub(crate) mod cache_sim;
pub mod message_time;
pub mod openai;
pub mod relay;
mod request;
mod request_encoding;
mod sse;
mod sse_types;
mod stream;
mod chat_tools;
pub mod subagent;
pub mod subagent_watch;
pub mod telemetry;
mod read_dedupe;
pub mod tool_spill;
pub mod trace;
pub mod worker_context;
pub(crate) mod transport;
mod types;

use api::ApiMethods;
use auth::AuthMethods;
use helpers::HelperMethods;
use stream::StreamMethods;
use types::AuthState;
pub use types::{AgentEvent, CancelCause, LlmEvent, SessionEvent, StreamEvent, TurnCompletion};
pub use stream::activation_policy;

/// Result of resolving before_tool_call extension policy.
pub enum BeforeToolCallDecision {
    Continue { input: Value },
    Block { reason: String },
}

/// Emit a `before_tool_call` event and include the runtime tool name when it
/// differs from the API-safe name. `session_id` is the owning conversation
/// (`None` for workers → JSON `null`, unchanged). `cwd` is the folder the
/// call runs in (`ToolCapabilities.cwd`; `None` = the process folder).
pub async fn emit_before_tool_call(
    hook_bus: &Arc<crate::extensions::hooks::HookBus>,
    tool_name: &str,
    runtime_tool_name: Option<&str>,
    input: Value,
    session_id: Option<&str>,
    cwd: Option<&Path>,
) -> crate::extensions::hooks::events::HookResult {
    let mut event = crate::extensions::hooks::events::HookEvent::before_tool_call(tool_name, input)
        .with_session(session_id)
        .with_cwd(cwd);
    if let Some(runtime_tool_name) = runtime_tool_name {
        event.tool_runtime_name = Some(runtime_tool_name.to_string());
    }
    hook_bus.emit(&event).await
}

/// Answer the Confirm dialog sends for its "Allow all from <ext>" button.
/// `resolve_before_tool_call_result` treats it as an allow AND latches the
/// confirm's source extension(s) in the session's
/// [`Runtime::session_allow_all`] set, so later Confirm results from those
/// extensions (only) are auto-approved without a prompt.
pub use crate::tools::CONFIRM_ANSWER_ALLOW_ALL;

/// Longest extension `message` shown in a Confirm prompt (chars / lines).
/// The host-built header and the dialog's buttons and scope line must stay
/// visible, so a long or hostile message is cut with a marker.
const MAX_CONFIRM_MESSAGE_CHARS: usize = 1200;
const MAX_CONFIRM_MESSAGE_LINES: usize = 12;

/// Sanitize, redact and cap an extension's confirm message for display:
/// control / bidi / zero-width characters are stripped first (prA I-R4-3:
/// they could move the cursor or disguise text in plain-text clients),
/// then credentials are redacted (prA I-R4-1: confirmers echo the
/// rewritten command), then the length is capped.
fn cap_confirm_message(message: &str) -> String {
    let message = redact_secret_text(&crate::session::sanitize_prompt_text(message));
    let lines: Vec<&str> = message.lines().collect();
    let over_lines = lines.len() > MAX_CONFIRM_MESSAGE_LINES;
    let over_chars = message.chars().count() > MAX_CONFIRM_MESSAGE_CHARS;
    if !over_lines && !over_chars {
        return message;
    }
    let kept = lines[..lines.len().min(MAX_CONFIRM_MESSAGE_LINES)].join("\n");
    let kept: String = kept.chars().take(MAX_CONFIRM_MESSAGE_CHARS).collect();
    format!("{kept}\n…[message truncated by host]")
}

/// Lines / chars of the final-input preview shown when the input was
/// rewritten by an earlier extension.
const MAX_REWRITE_PREVIEW_LINES: usize = 6;
const MAX_REWRITE_PREVIEW_CHARS: usize = 400;

/// Scrub credential shapes out of free text (a shell command, a header
/// line, an extension's confirm message) for the Confirm prompt only. Uses
/// the AGGRESSIVE `trace::export::redact_command_text` (the shared
/// high-precision rules plus recall-first ones). Stored text the model or
/// user reads back (context archive, memory, traces) keeps the precise
/// shared scrub (prA I-R5-1). The prompt is broadcast to every attached
/// client (observers included) and replayed on attach, so it must never
/// show a secret.
fn redact_secret_text(text: &str) -> String {
    crate::runtime::trace::export::redact_command_text(text)
}

/// `redact_value` (secret-named keys, secret-shaped strings) plus
/// [`redact_secret_text`] on every remaining string, recursively.
fn redact_preview_value(value: &mut Value) {
    crate::runtime::trace::export::redact_value(value);
    match value {
        Value::Object(map) => map.values_mut().for_each(redact_preview_value),
        Value::Array(items) => items.iter_mut().for_each(redact_preview_value),
        Value::String(s) => *s = redact_secret_text(s),
        _ => {}
    }
}

/// Host-rendered, capped, REDACTED preview of the input that will actually
/// run: a shell-style `command` string as-is, anything else as pretty JSON.
fn final_input_preview(input: &Value) -> String {
    let mut input = input.clone();
    redact_preview_value(&mut input);
    let full = match input.get("command").and_then(Value::as_str) {
        Some(cmd) => cmd.to_string(),
        None => serde_json::to_string_pretty(&input).unwrap_or_else(|_| input.to_string()),
    };
    let lines: Vec<&str> = full.lines().collect();
    let mut out: String = lines[..lines.len().min(MAX_REWRITE_PREVIEW_LINES)]
        .iter()
        .map(|l| format!("  {l}"))
        .collect::<Vec<_>>()
        .join("\n");
    let mut cut = lines.len() > MAX_REWRITE_PREVIEW_LINES;
    if out.chars().count() > MAX_REWRITE_PREVIEW_CHARS {
        out = out.chars().take(MAX_REWRITE_PREVIEW_CHARS).collect();
        cut = true;
    }
    if cut {
        out.push_str("\n  …[preview truncated]");
    }
    out
}

/// Prompt body for an extension tool-call Confirm: the host-known source and
/// tool lead, so an extension's own `message` cannot hide who is asking.
/// When earlier extensions rewrote the input (prA N3) the header also says
/// so, by whom, and previews the FINAL input — the one approval runs —
/// before the extension's own message (which is cut first on small screens).
fn confirm_prompt_body(
    message: &str,
    sources: &[String],
    tool: Option<&str>,
    rewrite: Option<(&[String], &Value)>,
) -> String {
    let source = if sources.is_empty() {
        "(unknown)".to_string()
    } else {
        sources.join(", ")
    };
    let rewritten = match rewrite {
        Some((by, input)) => format!(
            "Input rewritten by: {}\nInput that will run:\n{}\n",
            if by.is_empty() { "(unknown)".to_string() } else { by.join(", ") },
            final_input_preview(input)
        ),
        None => String::new(),
    };
    // The extension's own text is labelled, so it can't pass itself off as
    // the host-built header (e.g. a fake "Input that will run:" block).
    // Ids and tool names are host-chosen, but sanitize the whole body too.
    crate::session::sanitize_prompt_text(&format!(
        "Extension: {source}\nTool: {}\n{rewritten}\nMessage from {source}:\n{}\n\nAllow this tool call?",
        tool.unwrap_or("(unknown)"),
        cap_confirm_message(message)
    ))
}

/// What an approved Confirm runs: the input as modified earlier in the hook
/// chain, if any (`Modify`), else the original (`Continue`).
fn approved(
    modified_input: Option<Value>,
) -> crate::extensions::hooks::events::HookResult {
    match modified_input {
        Some(input) => crate::extensions::hooks::events::HookResult::Modify { input },
        None => crate::extensions::hooks::events::HookResult::Continue,
    }
}

/// Resolve a before_tool_call result that may request user confirmation.
///
/// When `auto_approve_confirms` is true, or every source extension of the
/// Confirm (same load generation) has been latched by an earlier "Allow
/// all" answer, `Confirm` is short-circuited to approval. Headless /
/// non-interactive callers with neither fail closed. "Allow all" is only
/// offered (a `ConfirmScope` is attached) when this call site has a latch
/// to arm; otherwise the dialog offers Allow / Deny only.
pub async fn resolve_before_tool_call_result(
    hook_result: crate::extensions::hooks::events::HookResult,
    secret_prompt: Option<&crate::tools::SecretPromptHandle>,
    auto_approve_confirms: bool,
    session_allow_all: Option<&crate::tools::SessionConfirmLatch>,
) -> crate::extensions::hooks::events::HookResult {
    match hook_result {
        crate::extensions::hooks::events::HookResult::Confirm {
            message,
            sources,
            generations,
            tool,
            modified_input,
            modified_by,
        } => {
            let keys = crate::tools::latch_keys(&sources, &generations);
            if auto_approve_confirms {
                tracing::info!(message = %message, "confirm auto-approved (auto_approve_confirms=true)");
                return approved(modified_input);
            }
            if session_allow_all.is_some_and(|latch| latch.allows_all(&keys)) {
                tracing::info!(
                    message = %message,
                    extensions = ?sources,
                    "confirm auto-approved (extension allow-all latched)"
                );
                return approved(modified_input);
            }

            let Some(prompt) = secret_prompt else {
                return crate::extensions::hooks::events::HookResult::Block {
                    reason: format!(
                        "Tool call requires confirmation but no interactive prompt is available: {}",
                        redact_secret_text(&crate::session::sanitize_prompt_text(&message))
                    ),
                };
            };

            // No latch on this path → no scope → no Allow-all button, so the
            // dialog and the "armed" notice never promise what can't happen.
            let scope = (session_allow_all.is_some() && !sources.is_empty()).then(|| {
                crate::session::ConfirmScope {
                    extensions: sources.clone(),
                    tool: tool.clone(),
                    rewritten_by: if modified_input.is_some() {
                        modified_by.clone()
                    } else {
                        Vec::new()
                    },
                }
            });
            let response = prompt
                .prompt_scoped(
                    crate::session::CONFIRM_PROMPT_TITLE.to_string(),
                    confirm_prompt_body(
                        &message,
                        &sources,
                        tool.as_deref(),
                        modified_input.as_ref().map(|i| (modified_by.as_slice(), i)),
                    ),
                    scope,
                )
                .await;

            match response.as_deref().map(str::trim) {
                Some(answer)
                    if answer.eq_ignore_ascii_case("yes") || answer.eq_ignore_ascii_case("y") =>
                {
                    approved(modified_input)
                }
                Some(answer) if answer.eq_ignore_ascii_case(CONFIRM_ANSWER_ALLOW_ALL) => {
                    match session_allow_all {
                        Some(latch) if latch.allow(&keys) => tracing::warn!(
                            extensions = ?sources,
                            "user chose Allow all: confirms from these extensions are auto-approved for the rest of this conversation"
                        ),
                        Some(_) => tracing::warn!(
                            "user chose Allow all on an unattributed or revoked confirm; approving this call only"
                        ),
                        None => tracing::warn!(
                            "user chose Allow all but this path has no session latch; approving this call only"
                        ),
                    }
                    approved(modified_input)
                }
                _ => crate::extensions::hooks::events::HookResult::Block {
                    reason: format!(
                        "Tool call confirmation denied: {}",
                        redact_secret_text(&crate::session::sanitize_prompt_text(&message))
                    ),
                },
            }
        }
        other => other,
    }
}

/// Resolve before_tool_call policy into executable input or a block reason.
pub async fn resolve_before_tool_call_decision(
    original_input: Value,
    hook_result: crate::extensions::hooks::events::HookResult,
    secret_prompt: Option<&crate::tools::SecretPromptHandle>,
    auto_approve_confirms: bool,
    session_allow_all: Option<&crate::tools::SessionConfirmLatch>,
) -> BeforeToolCallDecision {
    match resolve_before_tool_call_result(
        hook_result,
        secret_prompt,
        auto_approve_confirms,
        session_allow_all,
    )
    .await
    {
        crate::extensions::hooks::events::HookResult::Block { reason } => {
            BeforeToolCallDecision::Block { reason }
        }
        crate::extensions::hooks::events::HookResult::Modify { input } => {
            BeforeToolCallDecision::Continue { input }
        }
        _ => BeforeToolCallDecision::Continue {
            input: original_input,
        },
    }
}

/// Maximum size (bytes) of a `Replace` output an extension may substitute.
/// A transform returning more than this is clamped (with a warning) to bound
/// memory between deserialization and the downstream tool-output truncation.
/// Generous on purpose — legitimate compression/redaction shrinks output.
const MAX_REPLACE_OUTPUT: usize = 1024 * 1024; // 1 MiB

/// Emit an `after_tool_call` event and return the tool output to record in
/// history — either the original `output`, or a `Replace { output }`
/// substituted by an extension transform hook (compression, redaction,
/// summarization). The runtime tool name is included when it differs from the
/// API-safe name.
///
/// Fail-safe: any non-`Replace` result — `Continue`, a result rejected by the
/// permission matrix, a crashed/timed-out handler, or any future variant —
/// returns the original `output` unchanged. A misbehaving extension can never
/// drop or corrupt a tool's output.
///
/// Note: the event delivered to the extension carries `tool_output` as a
/// size-limited preview for large outputs — a ~256 KB prefix plus a
/// `…[truncated, N total bytes]` marker (see [`HookEvent::after_tool_call`]),
/// matching `max_tool_buffer`. The final returned string is then truncated to
/// `max_tool_output` (the context budget) — compress-then-truncate ordering,
/// so a transform extension sees the full buffered output and decides what to
/// keep before the hard cap is applied.
#[allow(clippy::too_many_arguments)] // the hook event's fields
pub async fn emit_after_tool_call(
    hook_bus: &Arc<crate::extensions::hooks::HookBus>,
    tool_name: &str,
    runtime_tool_name: Option<&str>,
    input: Value,
    output: String,
    max_tool_output: usize,
    session_id: Option<&str>,
    cwd: Option<&Path>,
) -> String {
    emit_after_tool_call_outcome(
        hook_bus,
        tool_name,
        runtime_tool_name,
        input,
        output,
        max_tool_output,
        session_id,
        cwd,
        false,
    )
    .await
    .output
}

/// Distinguish an authoritative hook replacement from ordinary budget truncation.
/// Byte equality with the summary cannot establish whether the delta lane is stale.
pub(super) struct AfterToolCallOutcome {
    /// Post-hook output cut to `max_tool_output` (the legacy history text).
    pub output: String,
    pub replaced: bool,
    /// The post-hook output BEFORE that cut, kept only when asked for
    /// (`keep_uncut`, i.e. tool output virtualisation is on) and only when
    /// the cut removed bytes: a spill must hold the whole result.
    pub uncut: Option<String>,
}

impl AfterToolCallOutcome {
    /// The whole post-hook output: what a virtualised result spills.
    pub fn full_text(&self) -> &str {
        self.uncut.as_deref().unwrap_or(&self.output)
    }
}

#[allow(clippy::too_many_arguments)] // the hook event's fields plus the history policy
pub(super) async fn emit_after_tool_call_outcome(
    hook_bus: &Arc<crate::extensions::hooks::HookBus>,
    tool_name: &str,
    runtime_tool_name: Option<&str>,
    input: Value,
    output: String,
    max_tool_output: usize,
    session_id: Option<&str>,
    cwd: Option<&Path>,
    keep_uncut: bool,
) -> AfterToolCallOutcome {
    use crate::extensions::hooks::events::HookResult;
    // Keep the original to return verbatim if no transform fires.
    let original = output.clone();
    let mut event =
        crate::extensions::hooks::events::HookEvent::after_tool_call(tool_name, input, output)
            .with_session(session_id)
            .with_cwd(cwd);
    if let Some(runtime_tool_name) = runtime_tool_name {
        event.tool_runtime_name = Some(runtime_tool_name.to_string());
    }
    let mut replaced = false;
    let post_hook = match hook_bus.emit(&event).await {
        HookResult::Replace { mut output } => {
            replaced = true;
            if output.len() > MAX_REPLACE_OUTPUT {
                tracing::warn!(
                    tool = %tool_name,
                    len = output.len(),
                    cap = MAX_REPLACE_OUTPUT,
                    "Extension Replace output exceeds cap — clamping",
                );
                // Floor to a UTF-8 char boundary before truncating.
                let mut boundary = MAX_REPLACE_OUTPUT;
                while boundary > 0 && !output.is_char_boundary(boundary) {
                    boundary -= 1;
                }
                output.truncate(boundary);
            }
            output
        }
        HookResult::Continue => original,
        other => {
            tracing::warn!(
                tool = %tool_name,
                ?other,
                "Unexpected after_tool_call result — using original output",
            );
            original
        }
    };
    // Compress-then-truncate: apply the context-budget cap AFTER the hook.
    // Mirrors `HelperMethods::truncate_tool_result` byte-for-byte so the
    // no-extension path is behavior-identical to the legacy ordering.
    let output =
        crate::runtime::helpers::HelperMethods::truncate_tool_result(&post_hook, max_tool_output);
    let uncut = (keep_uncut && post_hook.len() > max_tool_output).then_some(post_hook);
    AfterToolCallOutcome {
        output,
        replaced,
        uncut,
    }
}

/// A reasoning-level substitution performed during a model change because the
/// newly selected model does not support the previously active level.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct ReasoningClamp {
    pub from: agent_core::reasoning::ReasoningLevel,
    pub to: agent_core::reasoning::ReasoningLevel,
}

/// The core runtime — manages API communication, tool execution, authentication,
/// and streaming for all SynapsCLI binaries (chat, chatui, server, rpc, daemon).
#[derive(Clone)]
struct PromptReloadSource {
    manifest: PathBuf,
    context: agent_core::prompt::SelectionContext,
    user_module: Option<agent_core::prompt::PromptModule>,
    delegation_policy_digest: Option<String>,
}

pub struct Runtime {
    client: Client,
    anthropic_stream_client: Arc<std::sync::OnceLock<Client>>,
    auth: Arc<RwLock<AuthState>>,
    model: String,
    tools: Arc<RwLock<ToolRegistry>>,
    system_prompt: Option<String>,
    /// Compiled, content-safe effective prompt metadata retained for inspection/reload.
    effective_prompt: Option<agent_core::prompt::PromptStack>,
    prompt_generation: u64,
    /// Inputs retained from boot so callable reloads compile the same manifest selection.
    prompt_reload_source: Option<PromptReloadSource>,
    thinking_budget: u32,
    /// Named reasoning level — canonical for Codex and future providers.
    /// When set, overrides what `thinking_level()` returns. For legacy
    /// Anthropic models this is derived from `thinking_budget` at read time.
    named_level: Option<agent_core::reasoning::ReasoningLevel>,
    /// True when the user has explicitly chosen a reasoning level (via command
    /// or config). False for derived/default values applied by set_model or
    /// session restore. Controls whether set_model overwrites the level with
    /// the new model's default.
    explicit_reasoning: bool,
    /// Foreground by default; central subagent construction marks Worker so
    /// logical Ultra can never recursively activate proactive orchestration.
    codex_request_role: crate::runtime::openai::catalog::CodexRequestRole,
    /// User override for context window size (tokens). When set, takes
    /// precedence over the model's auto-detected window from
    /// `models::context_window_for_model`. Lets users cap context at e.g.
    /// 200k even on models that natively support 1M.
    context_window_override: Option<u64>,
    /// Opt-in context windows within a stable logical session (#112).
    /// DARK by default: `ContextManagementMode::Off`.
    pub(crate) continuation: continuation::SharedContinuation,
    /// Subagent workers only: automatic context rollover into the worker's
    /// archive directory (see `worker_context`). Inert by default and on
    /// every foreground runtime.
    worker_context: worker_context::WorkerContext,
    /// Model used for compaction. Falls back to claude-sonnet-4-6 if not set.
    compaction_model: Option<String>,
    /// Where compaction summarization runs (spec §9.4).
    compaction_mode: agent_core::compaction::CompactionMode,
    /// Content classes excluded from remote compaction disclosure.
    compaction_exclusions: Vec<agent_core::compaction::ContentClass>,
    /// Transport-construction seam (spec §9.4 / CP-12 M3): incremented at
    /// the SINGLE remote-summarization entry point before any preflight,
    /// auth, or HTTP request construction. Shared across clones so the
    /// local-only zero-network proof observes every path.
    remote_summarization_attempts: Arc<std::sync::atomic::AtomicU64>,
    /// Shared registry for reactive subagent handles.
    subagent_registry: Arc<Mutex<crate::runtime::subagent::SubagentRegistry>>,
    /// Session-scoped orchestration enforcement installed during boot.
    orchestration: Option<Arc<crate::orchestration::OrchestrationRuntime>>,
    /// Shared event queue — for Event Bus tooling.
    event_queue: Arc<crate::events::EventQueue>,
    // New configurable fields
    max_tool_output: usize,
    /// Tool output virtualisation thresholds (`tool_output_spill_*`). OFF
    /// on construction — `Runtime::new()` is also the embedding/test path
    /// and must not write under the base dir unless configured; every
    /// production boot applies the config (default on), workers through
    /// `apply_subagent_runtime_policy`.
    tool_spill_policy: tool_spill::ToolSpillPolicy,
    /// `<base>/spill`, bound once at construction (like `capture_dir`).
    tool_spill_root: PathBuf,
    /// Session-cumulative virtualisation counters; shared by clones.
    tool_spill_stats: Arc<tool_spill::ToolSpillStats>,
    /// Replace re-reads of unchanged files with a stub and note write/edit
    /// results. Off until a config is applied (production default on).
    read_dedupe: bool,
    /// `thinking_display`: the foreground session's `thinking.display`
    /// ([`Self::wire_thinking_display`]). Workers always send summarized.
    thinking_display: crate::config::ThinkingDisplay,
    /// Keep the cache warm while a tool round runs (`cache_keepwarm_tools`,
    /// `runtime::tool_keepwarm`). Off until a config is applied
    /// (production default on).
    tool_keepwarm: bool,
    bash_timeout: u64,
    bash_max_timeout: u64,
    subagent_timeout: u64,
    api_retries: u32,
    refusal_retries: u32,
    /// Telemetry level for structured per-request API logging (opt-in).
    telemetry_level: crate::runtime::telemetry::TelemetryLevel,
    /// Rotated generations kept per log (`telemetry_log_generations`), read
    /// when the session writer starts.
    telemetry_log_generations: u32,
    /// Session-shared bounded observability writer (Task 11, spec §6.5).
    /// `Some` iff `telemetry_level` is Basic/Full. Cloned runtimes
    /// (subagents) share the same handle — one worker per session, never
    /// per request. Dropping the last handle detaches the worker (it drains
    /// and exits on its own), so `Drop` can never hang.
    telemetry_writer: Option<crate::runtime::telemetry::TelemetryWriter>,
    /// Trace context handed to every `ApiOptions` site. Enabled (writer
    /// sink) iff `telemetry_writer` is `Some`; otherwise the no-op sink.
    /// Shared across clones so all requests of the session carry the same
    /// session-scoped context IDs.
    trace_ctx: trace::TraceContext,
    /// Explicit one-shot trace controls (Task 12): `/trace next` arms
    /// exactly the next outgoing provider request, even when telemetry is
    /// Off, then auto-disarms. Shared across clones.
    trace_controls: std::sync::Arc<trace::TraceControls>,
    /// The chat's memory binding: the captured project of its folder (#112).
    /// Until it is bound, the project of the process folder, as before.
    memory_backend: crate::memory_backend::MemoryBinding,
    /// Whether the binding has been made ([`Self::bind_memory`]) or given
    /// ([`Self::inherit_memory_binding`]). It is made once and never
    /// replaced.
    memory_bound: bool,
    /// Writer handle backing the most recent armed one-shot ephemeral
    /// trace context (telemetry Off + `/trace next`). Retained here — not
    /// only inside the request's cloned context — so the session exit
    /// epilogue (`shutdown_observability*`) can drain the armed record
    /// even if the process exits right after the request. Replaced on
    /// re-arm; never touched on the request path beyond one mutex store.
    /// Shared across clones.
    one_shot_trace_writer:
        std::sync::Arc<std::sync::Mutex<Option<crate::runtime::telemetry::TelemetryWriter>>>,
    /// Content-capture root, bound ONCE at construction (fix1 I2b): every
    /// sweep — startup, status, sync and async shutdown epilogues — and the
    /// one-shot content capture use THIS path, never a late ambient
    /// `SYNAPS_BASE_DIR` read, so post-construction env churn (parallel
    /// tests, profile switches) can never redirect capture I/O.
    capture_dir: std::path::PathBuf,
    /// Opt into the cache-diagnosis beta (`cache-diagnosis-2026-04-07`).
    cache_diagnostics: bool,
    /// Configured prompt-cache TTL (`auto` default | 5m | 1h | hybrid).
    /// Threaded into every request via `ApiOptions`; `auto` is resolved per
    /// request from the Anthropic credential kind (1h OAuth, 5m API key).
    cache_ttl: crate::core::config::CacheTtlSetting,
    /// gzip policy for Anthropic request bodies (`request_compression`,
    /// `request_compression_level`). Threaded into every request via
    /// `ApiOptions`; its refusal latch is process-wide.
    request_compression: request_encoding::RequestCompression,
    /// One-time-per-session latch for the silent 1h-downgrade notice
    /// (spec §3.4.1). Shared into `ApiOptions` for every request.
    ttl_downgrade_notified: std::sync::Arc<std::sync::atomic::AtomicBool>,
    /// Session-scoped "1h honored at least once" latch (spec §3.4.1) —
    /// suppresses the downgrade notice on healthy Hybrid turns where the 1h
    /// prefix is already cached. Shared into `ApiOptions` for every request.
    saw_1h_honored: std::sync::Arc<std::sync::atomic::AtomicBool>,
    /// When this runtime's foreground request chain last had an Anthropic
    /// request accepted (task #444 keep-warm scheduling). Fresh per clone,
    /// shared by `share_ttl_latches`, like the TTL latches above.
    cache_anchor: std::sync::Arc<keepwarm::CacheAnchorCell>,
    /// Last foreground Anthropic message id (`msg_...`) — threaded into the
    /// next request's `diagnostics.previous_message_id` when diagnostics is
    /// on. Fresh per clone, shared by `share_ttl_latches`.
    last_msg_id: Arc<Mutex<Option<String>>>,
    /// Set once the API rejected cache diagnostics: off for the session.
    /// Shared across `Clone` (subagents never re-probe a rejecting API).
    cache_diag_rejected: Arc<std::sync::atomic::AtomicBool>,
    session_manager: std::sync::Arc<crate::tools::shell::SessionManager>,
    /// Extension hook bus for dispatching events to extensions.
    hook_bus: Arc<crate::extensions::hooks::HookBus>,
    // Held to keep the reaper task alive for the Runtime's lifetime; never read directly.
    #[allow(dead_code)]
    reaper_handle: Option<tokio::task::JoinHandle<()>>,
    #[allow(dead_code)]
    reaper_cancel: Option<tokio_util::sync::CancellationToken>,
    /// How provider credentials are resolved: `Local` (read/refresh auth.json —
    /// the default) or `Remote` (fetch short-lived access tokens from a broker).
    /// Set from config in `apply_config`. See task #157.
    credential_source: crate::auth::CredentialSource,
    /// In-memory cache of broker-fetched access tokens (Remote source only).
    /// Cheap to clone (Arc inside). Never persisted to disk.
    token_cache: crate::auth::TokenCache,
    /// Exact operator-trusted worker identities loaded from `favorite_models`.
    /// These seed the session policy and are replayed when a manifestless
    /// foreground model change replaces that policy snapshot.
    trusted_worker_models: Vec<agent_core::prompt::QualifiedModelId>,
    /// Subagent fan-out limits (`subagent.*`) used whenever this runtime
    /// installs a manifestless delegation policy (boot, new session, model
    /// change). Changing them affects the next install, not running workers.
    subagent_limits: crate::config::SubagentLimits,
    /// True when this stream should expose only its session projection rather
    /// than the legacy full tool schema. Opt-in and false by default so the
    /// flag-off request bytes stay unchanged (Task 18).
    progressive_tool_disclosure: bool,
    /// `tools.always_loaded`: tool ids joined to the progressive session core
    /// (`SessionToolSet::progressive_core_with`). Inert without disclosure.
    tools_always_loaded: Arc<[crate::tools::catalog::ToolId]>,
    /// `tools.deferred_loading`: Anthropic `defer_loading` + `tool_reference`
    /// for progressive disclosure, so activations keep the cached prefix.
    deferred_tool_loading: bool,
    /// `tools.activation_confirm` host policy for model-initiated
    /// `activate_tools` (auto | prompt | deny). Default `auto`.
    activation_confirm: agent_core::config::ActivationConfirm,
    /// Current worker handle for bounded delegation-tree accounting. `None`
    /// for foreground roots.
    delegation_parent: Option<String>,
    /// Session-scoped, per-extension "Allow all" latch for extension
    /// `Confirm` gates (`before_tool_call`). An extension id is added when
    /// the user picks "Allow all from <ext>" in the Confirm dialog; shared by
    /// clones and delegated workers in the same session. Never persisted.
    session_allow_all: std::sync::Arc<crate::tools::SessionConfirmLatch>,
    /// Shared exact MCP lease manager (Task 19). Installed at engine boot
    /// when MCP exact mode is active; streams mint per-session capabilities
    /// and RAII guards from it.
    mcp_runtime: Option<std::sync::Arc<crate::mcp::McpRuntimeManager>>,
    /// Durable shared session-scope guard (Task 19 review): terminates this
    /// runtime session's MCP leases only when the LAST owner (runtime clone
    /// or in-flight stream) drops — never per provider turn.
    mcp_session_scope: Option<std::sync::Arc<crate::mcp::McpSessionEndGuard>>,
    /// Shared exact EXTENSION lease manager (Task 20). Installed at engine
    /// boot under progressive disclosure; streams mint per-session
    /// capabilities and hold the durable scope below.
    extension_runtime: Option<std::sync::Arc<crate::extensions::lease::ExtensionRuntimeManager>>,
    /// Durable shared session-scope guard for extension leases (Task 20):
    /// terminates this runtime session's extension leases only when the
    /// LAST owner (runtime clone or in-flight stream) drops.
    extension_session_scope:
        Option<std::sync::Arc<crate::extensions::lease::ExtensionSessionEndGuard>>,
    /// Per-turn budget (Task 23, spec §8.1). Resolved from role + typed
    /// config at boot; every stream turn is metered against it.
    turn_budget: crate::runtime::budget::TurnBudget,
    /// Set while an event-started turn's stream starts (like the event
    /// budget in `turn_budget`): the owner's steer moves that stream onto
    /// the session budget. `None` for every other stream.
    owner_steer_budget: Option<Arc<crate::runtime::budget::OwnerSteerBudget>>,
    /// Source of the per-message `_synaps_at` times (see `message_time`).
    message_clock: crate::runtime::message_time::MessageClock,
    /// Private runtime-scoped tool-session identity (Task 16). Scopes the
    /// per-stream `SessionToolSet` the execution gate authorizes against.
    /// Minted fresh per constructed `Runtime` — two independently
    /// constructed runtimes can never share session grants — and shared by
    /// `Clone` because clones share the same live tool registry (the
    /// existing shared-session behavior). Never persisted; unrelated to
    /// saved session IDs.
    host_tool_session: crate::tools::activation::SessionId,
    /// Frozen system-prompt catalog of activatable tools (progressive
    /// disclosure): rendered once per session on first use, then reused
    /// byte-for-byte so activations and late registrations never change the
    /// cached prompt prefix. Outer `None` = not rendered yet.
    tool_catalog_prompt: Arc<std::sync::Mutex<Option<Option<String>>>>,
    /// The `SessionToolSet` retained for `host_tool_session` across stream
    /// turns (exact activations are session-scoped). Minted empty with the
    /// runtime and shared by `Clone`, exactly like `host_tool_session`.
    retained_tool_set: crate::tools::activation::RetainedSessionToolSet,
    /// Activations carried over from this conversation's previous runtime
    /// (daemon reload, unpark), restored at the first progressive turn
    /// start. Shared by `Clone` like `retained_tool_set`.
    pending_tool_restore: crate::tools::activation::PendingToolRestore,
    /// Conversation/session identity this runtime serves. Keys the
    /// `on_session_start` hook injection (Phase 2 keys everything).
    /// `None` = unkeyed (workers, tests) — reads no injection.
    session_id: Option<String>,
    /// Per-session working directory (Phase 2 daemon mode). `None` = process
    /// cwd — every in-process host leaves it `None`, so `ToolCapabilities.cwd`
    /// stays `None` exactly as before. The daemon sets it per session.
    cwd: Option<PathBuf>,
    /// Per-session environment snapshot (session-identity T1). `None` =
    /// inherit process env (in-process hosts). Daemon sessions carry the
    /// creator client's env; tools apply `env_clear().envs()` when `Some`.
    env: Option<crate::session::types::ToolEnv>,
    /// Names of env vars stripped as secrets (T5).
    env_stripped: Vec<String>,
    /// Shared per-session "already warned" set for env-stripped notices (T5).
    /// Cloned into every `ToolCapabilities` so the dedup survives across
    /// tool calls within the same session.
    env_warned: std::sync::Arc<std::sync::Mutex<std::collections::HashSet<String>>>,
}

/// Mint a fresh runtime-scoped tool-session identity. Process id + UUIDv4
/// keeps it unique across runtimes and restarts; the parse cannot fail on
/// this generated shape.
fn fresh_host_tool_session() -> crate::tools::activation::SessionId {
    crate::tools::activation::SessionId::parse(&format!(
        "runtime-{}-{}",
        std::process::id(),
        uuid::Uuid::new_v4()
    ))
    .expect("generated runtime session id is always valid")
}

/// Idle timeout for the runtime HTTP client: how long a request may go
/// without receiving *any* bytes (headers or body chunks) before it is
/// killed. Resets on every received chunk, so healthy long-running streams
/// are never interrupted.
const HTTP_READ_TIMEOUT: Duration = Duration::from_secs(90);

/// Build the runtime's shared HTTP client.
///
/// Timeout design (incident: session 20260714-025948-3dab):
/// * `connect_timeout(10s)` — connection-establishment ceiling.
/// * `read_timeout` — **idle** detector; resets after each successful read
///   (reqwest `ReadTimeoutBody`). A hung request surfaces in seconds
///   instead of minutes, while an actively-streaming turn can run
///   indefinitely.
/// * Deliberately **no** total `.timeout(…)`: that was a wall-clock
///   deadline that kept ticking while bytes flowed, killing any healthy
///   stream longer than 300s and taking 300s to notice a dead connection.
///   Turn-level lifecycle is owned by cancellation tokens, not the client.
fn build_http_client(read_timeout: Duration) -> reqwest::Result<Client> {
    Client::builder()
        .tls_built_in_webpki_certs(true)
        .tls_built_in_native_certs(true)
        .connect_timeout(Duration::from_secs(10))
        .read_timeout(read_timeout)
        .build()
}

/// The process-global HTTP client for `EngineHost` (same builder as
/// `Runtime::new`).
/// The host's HTTP client builder (also used by the attach-TUI client, A4).
pub fn build_host_http_client() -> Result<Client> {
    build_http_client(HTTP_READ_TIMEOUT)
        .map_err(|e| RuntimeError::Config(format!("Failed to build HTTP client: {}", e)))
}

/// Internal: every field of `Runtime` in one place so `new()`,
/// `new_headless()`, `foreground_runtime()` and `worker_runtime()` cannot
/// drift. `Runtime::new()` is: build fresh parts → `from_parts`.
pub(crate) struct RuntimeParts {
    pub host: crate::host::HostParts,
    pub session_manager: Arc<crate::tools::shell::SessionManager>,
    pub reaper: Option<(tokio::task::JoinHandle<()>, CancellationToken)>,
}

impl RuntimeParts {
    /// Fresh shell session manager + idle reaper (needs a tokio runtime).
    pub(crate) fn with_reaper(host: crate::host::HostParts) -> Self {
        let session_manager = fresh_session_manager();
        let cancel = CancellationToken::new();
        let handle =
            crate::tools::shell::session::start_reaper(session_manager.clone(), cancel.clone());
        Self {
            host,
            session_manager,
            reaper: Some((handle, cancel)),
        }
    }

    /// Fresh shell session manager, no reaper (offline / headless).
    #[cfg_attr(not(any(test, feature = "testing")), allow(dead_code))]
    pub(crate) fn without_reaper(host: crate::host::HostParts) -> Self {
        Self {
            host,
            session_manager: fresh_session_manager(),
            reaper: None,
        }
    }
}

fn fresh_session_manager() -> Arc<crate::tools::shell::SessionManager> {
    let config = crate::tools::shell::ShellConfig::default();
    crate::tools::shell::SessionManager::new(config)
}

/// Fresh, process-local host parts — exactly what `Runtime::new()` used to
/// construct inline: new client, new registry, new hook bus, Local creds,
/// new token cache, no lease managers.
fn fresh_host_parts() -> Result<crate::host::HostParts> {
    let client = build_host_http_client()?;
    // Operational retention (Task 12): physically remove expired
    // content-capture bundles at session startup — bounded, fail-soft,
    // confined to the private capture dir. The root resolved here is
    // the SAME value bound into `capture_dir` (fix1 I2b).
    let capture_dir = trace::default_capture_dir();
    let _ = trace::sweep_expired_captures(&capture_dir);
    // Same for tool-output spill dirs past their age (bound here too).
    let spill_root = sweep_spill_root();
    Ok(crate::host::HostParts {
        client,
        tools: Arc::new(RwLock::new(ToolRegistry::new())),
        hook_bus: Arc::new(crate::extensions::hooks::HookBus::new()),
        credential_source: crate::auth::CredentialSource::Local,
        token_cache: crate::auth::TokenCache::new(),
        mcp_runtime: None,
        extension_runtime: None,
        capture_dir,
        spill_root,
        progressive_tool_disclosure: false,
        tools_always_loaded: Arc::from([]),
    })
}

/// Resolve `<base>/spill` once and remove session spill dirs past
/// [`agent_core::core::tool_spill::SPILL_MAX_AGE`] — bounded, fail-soft,
/// confined to the spill root, like the capture sweep beside it.
pub(crate) fn sweep_spill_root() -> PathBuf {
    let root = agent_core::core::tool_spill::default_spill_root();
    let _ = agent_core::core::tool_spill::sweep_stale_spill(
        &root,
        agent_core::core::tool_spill::SPILL_MAX_AGE,
        std::time::SystemTime::now(),
    );
    root
}

/// Preserve compatibility with favorite IDs written before Anthropic used its
/// runtime-qualified provider name. Authorization always stores the canonical
/// exact identity; unrelated bare values remain invalid and are ignored.
fn canonical_trusted_worker_model(model: &str) -> String {
    let model = model.trim();
    if let Some(id) = model.strip_prefix("claude/") {
        format!("anthropic/{id}")
    } else if model.starts_with("claude-") {
        format!("anthropic/{model}")
    } else {
        model.to_owned()
    }
}

/// The owning runtime stops its shell-session reaper when it goes away
/// (clones carry `None`). Without this every dropped `Runtime` — a parked
/// session, a finished worker — left a 30 s ticker holding the manager.
impl Drop for Runtime {
    fn drop(&mut self) {
        if let Some(c) = &self.reaper_cancel {
            c.cancel();
        }
    }
}

impl Runtime {
    pub async fn new() -> Result<Self> {
        // UNCHANGED semantics: fresh everything (tests).
        Ok(Self::from_parts(RuntimeParts::with_reaper(
            fresh_host_parts()?,
        )))
    }

    /// The single struct-literal site for `Runtime`. Host-owned parts come
    /// from `parts.host`; everything else is fresh.
    pub(crate) fn from_parts(parts: RuntimeParts) -> Self {
        // Runtime construction is credential-blind. Credentials are acquired
        // lazily through the broker abstraction after configuration is applied;
        // this layer never opens auth.json or consults a secret environment var.
        let (auth_token, auth_type, refresh_token, token_expires) =
            (String::new(), "oauth".to_string(), None, Some(0));
        let RuntimeParts {
            host,
            session_manager,
            reaper,
        } = parts;
        let (reaper_handle, reaper_cancel) = match reaper {
            Some((h, c)) => (Some(h), Some(c)),
            None => (None, None),
        };
        let mcp_runtime = host.mcp_runtime;
        let extension_runtime = host.extension_runtime;

        let mut runtime = Runtime {
            client: host.client,
            anthropic_stream_client: Default::default(),
            auth: Arc::new(RwLock::new(AuthState {
                auth_token,
                auth_type,
                refresh_token,
                token_expires,
                auth_file_mtime: None,
                bound_credential: None,
            })),
            model: crate::models::default_model().to_string(),
            tools: host.tools,
            system_prompt: None,
            effective_prompt: None,
            prompt_generation: 0,
            prompt_reload_source: None,
            thinking_budget: 4096,
            named_level: None,
            explicit_reasoning: false,
            codex_request_role: crate::runtime::openai::catalog::CodexRequestRole::Foreground,
            context_window_override: None,
            compaction_model: None,
            compaction_mode: agent_core::compaction::CompactionMode::default(),
            compaction_exclusions: Vec::new(),
            remote_summarization_attempts: Arc::new(std::sync::atomic::AtomicU64::new(0)),
            subagent_registry: Arc::new(Mutex::new(
                crate::runtime::subagent::SubagentRegistry::new(),
            )),
            orchestration: None,
            event_queue: Arc::new(crate::events::EventQueue::new(1000)),
            max_tool_output: 30000,
            tool_spill_policy: tool_spill::ToolSpillPolicy::OFF,
            tool_spill_root: host.spill_root,
            tool_spill_stats: Arc::default(),
            read_dedupe: false,
            thinking_display: crate::config::ThinkingDisplay::Summarized,
            tool_keepwarm: false,
            bash_timeout: 30,
            bash_max_timeout: 300,
            subagent_timeout: 300,
            api_retries: 3,
            refusal_retries: 2,
            telemetry_level: crate::runtime::telemetry::TelemetryLevel::Off,
            telemetry_log_generations: crate::runtime::telemetry::DEFAULT_MAX_GENERATIONS,
            telemetry_writer: None,
            trace_ctx: trace::TraceContext::disabled(),
            trace_controls: std::sync::Arc::new(trace::TraceControls::new()),
            // Unbound until `bind_memory`, which replaces this binding with the
            // project of the folder given by `set_cwd`; until then, the process
            // folder's (as `legacy_current()` did). Continuation off (#112).
            memory_backend: crate::memory_backend::MemoryBinding::for_cwd(
                std::env::current_dir().ok(),
            ),
            memory_bound: false,
            continuation: Arc::new(Mutex::new(continuation::ContinuationState::default())),
            worker_context: Default::default(),
            one_shot_trace_writer: std::sync::Arc::new(std::sync::Mutex::new(None)),
            capture_dir: host.capture_dir,
            cache_diagnostics: false,
            cache_ttl: crate::core::config::CacheTtlSetting::default(),
            request_compression: Default::default(),
            ttl_downgrade_notified: std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)),
            saw_1h_honored: std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)),
            cache_anchor: Default::default(),
            last_msg_id: Arc::new(Mutex::new(None)),
            cache_diag_rejected: Default::default(),
            session_manager,
            hook_bus: host.hook_bus,
            reaper_handle,
            reaper_cancel,
            credential_source: host.credential_source,
            token_cache: host.token_cache,
            trusted_worker_models: Vec::new(),
            subagent_limits: crate::config::SubagentLimits::default(),
            progressive_tool_disclosure: host.progressive_tool_disclosure,
            tools_always_loaded: host.tools_always_loaded,
            deferred_tool_loading: true,
            activation_confirm: agent_core::config::ActivationConfirm::default(),
            delegation_parent: None,
            session_allow_all: std::sync::Arc::new(crate::tools::SessionConfirmLatch::new()),
            mcp_runtime: None,
            mcp_session_scope: None,
            extension_runtime: None,
            extension_session_scope: None,
            turn_budget: crate::runtime::budget::TurnBudget::for_role(
                crate::runtime::budget::TurnRole::Foreground,
            ),
            owner_steer_budget: None,
            message_clock: Default::default(),
            host_tool_session: fresh_host_tool_session(),
            tool_catalog_prompt: Arc::new(std::sync::Mutex::new(None)),
            retained_tool_set: Default::default(),
            pending_tool_restore: Default::default(),
            session_id: None,
            cwd: None,
            env: None,
            env_stripped: Vec::new(),
            env_warned: Default::default(),
        };
        // Lease managers are installed through the same seams boot used, so
        // the per-runtime durable session-scope guards are minted exactly as
        // before (keyed by this runtime's fresh host tool session).
        if let Some(m) = mcp_runtime {
            runtime.install_mcp_runtime(m);
        }
        if let Some(m) = extension_runtime {
            runtime.install_extension_runtime(m);
        }
        runtime
    }


    /// Offline construction seam for headless test harnesses (P4).
    ///
    /// Identical to [`Runtime::new`] except:
    /// - auth is a stub token — no `auth.json` read, no keychain, no network
    /// - the idle shell-session reaper is not spawned — no tokio runtime
    ///   required at construction time
    ///
    /// Every accessor (`model()`, `thinking_level()`, tool registry, …) works
    /// normally; anything that would hit the Anthropic API fails at call time,
    /// which is the correct behavior for a harness that only drives the UI.
    #[cfg(any(test, feature = "testing"))]
    pub fn new_headless() -> Self {
        let client = build_http_client(HTTP_READ_TIMEOUT)
            .expect("reqwest client construction is infallible with built-in roots");
        let host = crate::host::HostParts {
            client,
            tools: Arc::new(RwLock::new(ToolRegistry::new())),
            hook_bus: Arc::new(crate::extensions::hooks::HookBus::new()),
            credential_source: crate::auth::CredentialSource::Local,
            token_cache: crate::auth::TokenCache::new(),
            mcp_runtime: None,
            extension_runtime: None,
            capture_dir: trace::default_capture_dir(),
            spill_root: agent_core::core::tool_spill::default_spill_root(),
            progressive_tool_disclosure: false,
            tools_always_loaded: Arc::from([]),
        };
        let mut runtime = Self::from_parts(RuntimeParts::without_reaper(host));
        runtime.auth = Arc::new(RwLock::new(AuthState {
            auth_token: "test-token".to_string(),
            auth_type: "api_key".to_string(),
            refresh_token: None,
            token_expires: None,
            auth_file_mtime: None,
            bound_credential: None,
        }));
        runtime
    }

    pub fn set_system_prompt(&mut self, prompt: String) {
        self.system_prompt = Some(prompt);
    }

    pub fn system_prompt(&self) -> Option<&str> {
        self.system_prompt.as_deref()
    }

    /// The system prompt that actually goes on the wire for the next
    /// request: the configured base plus any builtin orchestration adapter
    /// (see [`agent_core::prompt::builtin_orchestration_adapters`]) selected
    /// by the *current* model identity, so the doctrine follows mid-session
    /// model switches. Composition is skipped when:
    /// - a typed prompt manifest is active — the manifest author owns the
    ///   full stack;
    /// - the runtime exposes no subagent tools (worker runtimes) — the
    ///   doctrine would be unactionable noise;
    /// - the model identity cannot be canonicalized — fail closed to the
    ///   unmodified base.
    pub async fn effective_system_prompt(&self) -> Option<String> {
        let base = self.orchestrated_system_prompt().await;
        // A typed prompt manifest owns the full stack: nothing is appended.
        if self.effective_prompt.is_some() {
            return base;
        }
        match self.frozen_tool_catalog().await {
            Some(catalog) => Some(match base {
                Some(base) if !base.is_empty() => format!("{base}\n\n{catalog}"),
                _ => catalog,
            }),
            None => base,
        }
    }

    /// The progressive-disclosure catalog of activatable tools for the
    /// system prompt, frozen per session (see `tool_catalog_prompt`). `None`
    /// when progressive disclosure is off, model-initiated activation is
    /// denied, the runtime cannot activate tools (workers), or nothing
    /// outside the session core is activatable.
    pub async fn frozen_tool_catalog(&self) -> Option<String> {
        if !self.progressive_tool_disclosure
            || self.activation_confirm == agent_core::config::ActivationConfirm::Deny
        {
            return None;
        }
        if let Some(frozen) = self
            .tool_catalog_prompt
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .clone()
        {
            return frozen;
        }
        let rendered = {
            let registry = self.tools.read().await;
            if registry.get("activate_tools").is_none() {
                None
            } else {
                let core_set = chat_tools::chat_tool_set(
                    self.host_tool_session.clone(),
                    registry.catalog(),
                    true,
                    self.context_management_enabled(),
                    &self.tools_always_loaded,
                );
                let core: std::collections::HashSet<String> = core_set
                    .core_ids()
                    .map(|id| id.as_str().to_string())
                    .collect();
                // What this chat is not offered it cannot activate either.
                crate::tools::prompt_catalog::render(
                    &registry.catalog().without(core_set.withheld()),
                    &core,
                    crate::tools::activation::is_activatable_record,
                )
            }
        };
        let mut slot = self
            .tool_catalog_prompt
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        // First render wins (a concurrent caller may have frozen it already).
        slot.get_or_insert(rendered).clone()
    }

    /// The configured base prompt plus the builtin orchestration doctrine
    /// for the current model (what `effective_system_prompt` returned before
    /// the tool catalog was appended).
    async fn orchestrated_system_prompt(&self) -> Option<String> {
        if self.effective_prompt.is_some() {
            return self.system_prompt.clone();
        }
        if self.tools.read().await.get("subagent_start").is_none() {
            return self.system_prompt.clone();
        }
        let Ok(model) = crate::orchestration::canonical_foreground_identity(&self.model) else {
            return self.system_prompt.clone();
        };
        let workflow_mode = match self.reasoning_level() {
            agent_core::reasoning::ReasoningLevel::UltraCode => {
                Some(agent_core::prompt::WorkflowMode::UltraCode)
            }
            agent_core::reasoning::ReasoningLevel::Max => {
                Some(agent_core::prompt::WorkflowMode::Max)
            }
            agent_core::reasoning::ReasoningLevel::XHigh => {
                Some(agent_core::prompt::WorkflowMode::XHigh)
            }
            _ => None,
        };
        let Ok(context) = agent_core::prompt::SelectionContext::new(model, None)
            .map(|context| context.with_workflow_mode(workflow_mode))
        else {
            return self.system_prompt.clone();
        };
        agent_core::prompt::compose_orchestration_prompt(self.system_prompt.as_deref(), &context)
    }

    pub fn effective_prompt(&self) -> Option<&agent_core::prompt::PromptStack> {
        self.effective_prompt.as_ref()
    }

    pub fn prompt_generation(&self) -> u64 {
        self.prompt_generation
    }

    pub fn retain_prompt_reload_source(
        &mut self,
        manifest: PathBuf,
        context: agent_core::prompt::SelectionContext,
        user_module: Option<agent_core::prompt::PromptModule>,
        delegation_policy_digest: Option<String>,
    ) {
        self.prompt_reload_source = Some(PromptReloadSource {
            manifest,
            context,
            user_module,
            delegation_policy_digest,
        });
    }

    /// Recompile retained manifest inputs, validate, then atomically install the candidate.
    pub fn reload_prompt(&mut self) -> std::result::Result<u64, agent_core::prompt::PromptError> {
        let source = self.prompt_reload_source.clone().ok_or_else(|| {
            agent_core::prompt::PromptError::Invalid("no prompt manifest is active".into())
        })?;
        let raw = std::fs::read_to_string(&source.manifest).map_err(|_| {
            agent_core::prompt::PromptError::Invalid("prompt manifest is unavailable".into())
        })?;
        let manifest = agent_core::prompt::PromptManifest::parse(&raw)?;
        let reload_catalog = crate::orchestration::OrchestrationRuntime::trusted_catalog(
            source.context.model(),
            manifest.delegation_catalog_candidates(),
        )
        .map_err(|error| agent_core::prompt::PromptError::Invalid(error.into()))?;
        let candidate_policy_digest = manifest
            .delegation_policy(source.context.model().clone(), &reload_catalog)?
            .map(|policy| policy.digest());
        if candidate_policy_digest != source.delegation_policy_digest {
            return Err(agent_core::prompt::PromptError::Invalid(
                "hot reload cannot safely change delegation policy".into(),
            ));
        }
        let registry = manifest.registry(source.manifest.parent())?;
        let candidate = agent_core::prompt::compile_prompt_stack(
            &manifest,
            &registry,
            &source.context,
            source.user_module,
        )?;
        self.apply_prompt_stack(candidate)
    }

    /// Atomically validate and install a compiled stack. Failed validation leaves all state intact.
    pub fn apply_prompt_stack(
        &mut self,
        candidate: agent_core::prompt::PromptStack,
    ) -> std::result::Result<u64, agent_core::prompt::PromptError> {
        if let Some(current) = &self.effective_prompt {
            current.validate_hot_reload(&candidate)?;
        }
        let composed = candidate.composed().to_owned();
        self.effective_prompt = Some(candidate);
        self.system_prompt = Some(composed);
        self.prompt_generation = self.prompt_generation.saturating_add(1);
        Ok(self.prompt_generation)
    }

    pub fn prompt_inspection_json(&self) -> Option<String> {
        let mode = self
            .orchestration
            .as_ref()
            .map(|runtime| runtime.enforcement_mode())
            .unwrap_or(agent_core::orchestration::EnforcementMode::Off);
        self.effective_prompt.as_ref().and_then(|stack| {
            serde_json::to_string(&serde_json::json!({
                "generation": self.prompt_generation,
                "effective": stack.inspect(mode),
                "token_estimate": stack.composed().len().div_ceil(4),
            }))
            .ok()
        })
    }

    /// Install the shared exact MCP lease manager (Task 19, engine boot).
    /// Also mints ONE durable session-scope guard for the runtime's tool
    /// session, shared by every clone and stream; only the LAST owner's
    /// drop terminates leases.
    pub fn install_mcp_runtime(&mut self, manager: std::sync::Arc<crate::mcp::McpRuntimeManager>) {
        self.mcp_session_scope = Some(std::sync::Arc::new(crate::mcp::McpSessionEndGuard::new(
            self.host_tool_session.clone(),
            std::sync::Arc::clone(&manager),
        )));
        self.mcp_runtime = Some(manager);
    }

    /// Set this runtime's per-turn budget (Task 23). The engine resolves
    /// role + typed config at boot; subagent constructions pass
    /// their role's budget explicitly.
    pub fn set_turn_budget(&mut self, budget: crate::runtime::budget::TurnBudget) {
        self.turn_budget = budget;
    }

    /// Hand the next stream an [`OwnerSteerBudget`](crate::runtime::budget::OwnerSteerBudget)
    /// (`None` clears it). Like [`Self::set_turn_budget`], the stream takes it
    /// when it starts.
    pub fn set_owner_steer_budget(
        &mut self,
        lift: Option<Arc<crate::runtime::budget::OwnerSteerBudget>>,
    ) {
        self.owner_steer_budget = lift;
    }

    /// Where message times come from (tests inject a clock).
    pub fn message_clock(&self) -> &crate::runtime::message_time::MessageClock {
        &self.message_clock
    }

    pub fn set_message_clock(&mut self, clock: crate::runtime::message_time::MessageClock) {
        self.message_clock = clock;
    }

    /// The currently configured per-turn budget.
    pub fn turn_budget(&self) -> &crate::runtime::budget::TurnBudget {
        &self.turn_budget
    }

    /// Install the shared exact EXTENSION lease manager (Task 20, engine
    /// boot). Also mints ONE durable session-scope guard for the runtime's
    /// tool session, shared by every clone and stream; only the LAST
    /// owner's drop terminates leases.
    pub fn install_extension_runtime(
        &mut self,
        manager: std::sync::Arc<crate::extensions::lease::ExtensionRuntimeManager>,
    ) {
        // Bind the handler host scope to THIS runtime's durable tool
        // session: a Mixed extension's tool leases and hook/provider/user
        // handler leases share one key — ONE shared child per plugin.
        manager.bind_host_scope(self.host_tool_session.clone());
        self.extension_session_scope = Some(std::sync::Arc::new(
            crate::extensions::lease::ExtensionSessionEndGuard::new(
                self.host_tool_session.clone(),
                std::sync::Arc::clone(&manager),
            ),
        ));
        self.extension_runtime = Some(manager);
    }

    pub fn install_orchestration(
        &mut self,
        runtime: Arc<crate::orchestration::OrchestrationRuntime>,
    ) {
        for model in &self.trusted_worker_models {
            if let Err(error) = runtime.grant_worker_model(model.as_str()) {
                tracing::warn!(
                    model = model.as_str(),
                    error = %error,
                    "failed to apply configured worker-model trust"
                );
            }
        }
        self.orchestration = Some(runtime);
    }

    /// A runtime spawned for a worker shares only this session's bounded
    /// dispatch/tree authority. It must not replay process-global favorite
    /// models: doing so would turn a child construction into a fresh grant
    /// source. The child can dispatch only identities already authorized by
    /// this installed session policy.
    pub fn install_worker_orchestration(
        &mut self,
        runtime: Arc<crate::orchestration::OrchestrationRuntime>,
    ) {
        self.orchestration = Some(runtime);
    }

    /// The `subagent.*` limits this runtime sizes delegation policies with.
    pub fn subagent_limits(&self) -> crate::config::SubagentLimits {
        self.subagent_limits
    }

    pub fn set_delegation_parent(&mut self, parent: Option<String>) {
        self.delegation_parent = parent;
    }

    pub fn orchestration(&self) -> Option<&Arc<crate::orchestration::OrchestrationRuntime>> {
        self.orchestration.as_ref()
    }

    /// Extends the live session delegation policy with one explicitly
    /// user-trusted worker model (e.g. favorited mid-session in the models
    /// picker). Mid-session trust grants were always meant to be honored;
    /// the policy snapshot is not pinned against user decisions.
    pub fn grant_worker_model(&self, model: &str) -> std::result::Result<(), String> {
        self.orchestration
            .as_ref()
            .ok_or_else(|| "delegation policy unavailable".to_string())?
            .grant_worker_model(model)
    }

    /// Infallible model change. Returns the reasoning clamp applied (if any)
    /// when the current level is not supported by the new model; errors from
    /// `try_set_model` are swallowed (model left unchanged).
    pub fn set_model(&mut self, model: String) -> Option<ReasoningClamp> {
        self.try_set_model(model).ok().flatten()
    }

    /// Apply a model change while preserving orchestration lifecycle invariants.
    /// Returns an error instead of mutating either model or policy when active
    /// workers still require collection/reconciliation or when the replacement
    /// foreground cannot produce a trusted manifestless policy snapshot.
    /// On success, returns `Some(ReasoningClamp)` when an explicit reasoning
    /// level had to be substituted because the new model does not support it.
    pub fn try_set_model(
        &mut self,
        model: String,
    ) -> std::result::Result<Option<ReasoningClamp>, String> {
        // Older model pickers passed the rendered health row back here, e.g.
        // `✅  339ms  groq/llama-3.3-70b`. Remove that exact decoration shape,
        // rather than searching inside the ID: provider-qualified IDs may
        // legitimately contain `claude-` after their slash.
        let trimmed = model.trim();
        let cleaned = trimmed
            .split_once(char::is_whitespace)
            .and_then(|(_, rest)| {
                let rest = rest.trim_start();
                let (latency, candidate) = rest.split_once(char::is_whitespace)?;
                let millis = latency.strip_suffix("ms")?;
                if !millis.is_empty() && millis.chars().all(|c| c.is_ascii_digit()) {
                    let candidate = candidate.trim();
                    if !candidate.is_empty() && !candidate.chars().any(char::is_whitespace) {
                        return Some(candidate);
                    }
                }
                None
            })
            .unwrap_or(trimmed);
        // Manifestless orchestration snapshots are tied to the exact foreground
        // identity. Refuse the model mutation while a policy worker remains
        // unreconciled; replacing its registry would erase the completion-gate
        // remediation path, while changing only the runtime model would create a
        // stale authorization snapshot.
        if self.prompt_reload_source.is_none()
            && self
                .orchestration
                .as_ref()
                .is_some_and(|current| !current.unreconciled_runtime_handles().is_empty())
        {
            return Err("model change blocked: workers require collection/reconciliation".into());
        }
        let orchestration_replacement = if self.prompt_reload_source.is_none()
            && self.orchestration.is_some()
        {
            let foreground = crate::orchestration::canonical_foreground_identity(cleaned)
                .map_err(|_| "model change blocked: unresolved foreground model".to_string())?;
            let replacement = crate::orchestration::OrchestrationRuntime::from_limits(
                foreground,
                self.subagent_limits,
            )
            .map_err(|_| "model change blocked: trusted worker catalog unavailable".to_string())?;
            for trusted in &self.trusted_worker_models {
                replacement
                    .grant_worker_model(trusted.as_str())
                    .map_err(|_| {
                        "model change blocked: configured worker trust invalid".to_string()
                    })?;
            }
            Some(Arc::new(replacement))
        } else {
            None
        };
        self.model = cleaned.to_owned();
        // Replace the manifestless policy atomically after a safe model change so
        // inheritance and exact same-provider choices follow the new foreground.
        // Typed manifest policies are immutable and are not rewritten here.
        if let Some(replacement) = orchestration_replacement {
            self.orchestration = Some(replacement);
        }
        // Apply the new model's default reasoning level from exact capability
        // metadata (Codex catalog default, xAI documented default/Adaptive)
        // unless the user has explicitly chosen a level (explicit_reasoning is
        // true — set via command, config, or explicit session restore).
        // Derived/default levels are overwritten by the new model's default.
        if !self.explicit_reasoning {
            if let Some(level) =
                crate::runtime::openai::catalog::validation::default_level_for_model(cleaned)
            {
                self.set_reasoning_level(level);
            }
        }
        // An explicit level that the new model rejects would otherwise make
        // every subsequent turn fail pre-network until the user runs
        // `/thinking`. Substitute the nearest documented level instead, keeping
        // the explicit provenance flag so later switches still honor the user.
        Ok(self.clamp_reasoning_to_model())
    }

    /// If the current effective reasoning level fails validation against the
    /// current model, replace it with the model's documented default (or the
    /// first documented option that validates). The `explicit_reasoning` flag
    /// is preserved. Returns the substitution performed, if any; leaves state
    /// untouched (and returns `None`) when nothing validates.
    fn clamp_reasoning_to_model(&mut self) -> Option<ReasoningClamp> {
        use crate::runtime::openai::catalog::validation::{
            default_level_for_model, thinking_options_for_model, validate_reasoning_mutation,
        };
        let model = self.model.as_str();
        let from = self.reasoning_level();
        if validate_reasoning_mutation(model, from).is_ok() {
            return None;
        }
        let to = default_level_for_model(model)
            .filter(|level| validate_reasoning_mutation(model, *level).is_ok())
            .or_else(|| {
                thinking_options_for_model(model)
                    .iter()
                    .filter_map(|opt| agent_core::reasoning::ReasoningLevel::parse(opt))
                    .find(|level| validate_reasoning_mutation(model, *level).is_ok())
            })?;
        let explicit = self.explicit_reasoning;
        self.set_reasoning_level(to);
        self.explicit_reasoning = explicit;
        Some(ReasoningClamp { from, to })
    }

    pub fn set_tools(&mut self, tools: ToolRegistry) {
        self.tools = Arc::new(RwLock::new(tools));
    }

    pub fn subagent_registry(&self) -> &Arc<Mutex<crate::runtime::subagent::SubagentRegistry>> {
        &self.subagent_registry
    }

    pub fn event_queue(&self) -> &Arc<crate::events::EventQueue> {
        &self.event_queue
    }

    /// Install a session-lifetime queue (B3 unpark: the queue outlives the
    /// `Runtime` so `synaps send` keeps resolving while parked). Must be
    /// called before the first turn — in-flight streams hold a clone.
    pub fn set_event_queue(&mut self, queue: Arc<crate::events::EventQueue>) {
        self.event_queue = queue;
    }

    /// Get a shared reference to the extension hook bus.
    pub fn hook_bus(&self) -> &Arc<crate::extensions::hooks::HookBus> {
        &self.hook_bus
    }

    /// Replace the hook bus before the first turn (in-flight streams keep
    /// the bus they started with). Subagent spawn only: the legacy
    /// fresh-runtime path installs the host's worker view, and
    /// `subagent_transform_hooks = off` detaches it.
    pub(crate) fn set_hook_bus(&mut self, bus: Arc<crate::extensions::hooks::HookBus>) {
        self.hook_bus = bus;
    }

    /// Background shell/PTY sessions owned by this runtime (`Checkpoint`
    /// and `Parked` close them via `shutdown_all`).
    pub fn session_manager(&self) -> &Arc<crate::tools::shell::SessionManager> {
        &self.session_manager
    }

    /// Runtime-scoped tool-session identity used by the stream execution
    /// gate (Task 16). Shared by clones (which share the tool registry);
    /// fresh per independently constructed `Runtime`.
    pub fn host_tool_session_id(&self) -> &crate::tools::activation::SessionId {
        &self.host_tool_session
    }

    /// Conversation/session identity this runtime serves (keys hook
    /// injection). `None` = unkeyed (workers). Does not reset consent: linked
    /// compaction successors remain the same conversation. Use
    /// [`Self::begin_conversation`] for explicit new/resume boundaries.
    pub fn set_session_id(&mut self, id: Option<String>) {
        self.session_id = id;
    }

    /// Start an explicitly new or resumed conversation with fresh Confirm consent.
    /// Revoke the old set first, so workers still running from the previous
    /// conversation stop auto-approving, then replace the Arc so old
    /// stream/worker clones cannot grant consent to the next conversation.
    /// Ordinary turns and linked compaction successors must not call this.
    pub fn begin_conversation(&mut self, id: Option<String>) {
        self.session_allow_all.revoke();
        self.session_allow_all = std::sync::Arc::new(crate::tools::SessionConfirmLatch::new());
        // New conversation, new cache-diagnosis chain (cleared in place so
        // shared driver/ping clones see it too).
        *self
            .last_msg_id
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner) = None;
        self.set_session_id(id);
    }

    pub fn session_id(&self) -> Option<&str> {
        self.session_id.as_deref()
    }

    /// Per-session working directory handed to tools via
    /// `ToolCapabilities.cwd`. `None` = process cwd.
    pub fn set_cwd(&mut self, cwd: Option<PathBuf>) {
        self.cwd = cwd;
    }

    pub fn cwd(&self) -> Option<&Path> {
        self.cwd.as_deref()
    }

    /// Per-session environment snapshot handed to tools via
    /// `ToolCapabilities.env`. `None` = inherit process env.
    pub fn set_env(&mut self, env: Option<crate::session::types::SessionEnv>) {
        self.env = env.map(Into::into);
    }

    /// [`Self::set_env`] with the forwarded secrets an owned session's tools
    /// also see (kept apart, merged at spawn).
    pub fn set_tool_env(&mut self, env: Option<crate::session::types::ToolEnv>) {
        self.env = env;
    }

    pub fn set_env_stripped(&mut self, stripped: Vec<String>) {
        self.env_stripped = stripped;
    }

    /// The session env snapshot (without any forwarded secrets).
    pub fn env(&self) -> Option<&crate::session::types::SessionEnv> {
        self.env.as_ref().map(|e| &e.vars)
    }

    /// What a slash command typed in this chat is run with: this runtime's
    /// own id, folder, memory binding and tool environment. The one place a
    /// command gets them from, so it cannot be handed the host's.
    pub fn command_chat(&self) -> crate::skills::commands::CommandChat<'_> {
        crate::skills::commands::CommandChat {
            session_id: self.session_id(),
            cwd: self.cwd(),
            memory_backend: &self.memory_backend,
            env: self.env.as_ref(),
        }
    }

    pub fn env_stripped(&self) -> &[String] {
        &self.env_stripped
    }

    /// Test seam: the work phase and the note this chat's rollover state
    /// holds, as its last `context_checkpoint` left them.
    #[cfg(any(test, feature = "testing"))]
    pub fn context_checkpoint_for_harness(&self) -> (&'static str, String) {
        let state = self
            .continuation
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        (state.policy.phase().as_str(), state.note.clone())
    }

    /// Get a shared reference to the tool registry (for MCP lazy loading).
    pub fn tools_shared(&self) -> Arc<RwLock<ToolRegistry>> {
        Arc::clone(&self.tools)
    }

    /// The registry's tool schema as this chat is offered it: without the
    /// tools its turns withhold (`chat_tools::withheld_tools`). The registry is
    /// the host's and holds every chat's tools, so its own
    /// [`ToolRegistry::tools_schema`] is not what any one chat sends.
    pub async fn offered_tools_schema(&self) -> Arc<Vec<Value>> {
        let withheld = chat_tools::withheld_tools(self.context_management_enabled());
        let registry = self.tools.read().await;
        registry
            .tools_schema_without(&withheld)
            .unwrap_or_else(|| registry.tools_schema())
    }

    pub fn model(&self) -> &str {
        &self.model
    }

    pub fn http_client(&self) -> &Client {
        &self.client
    }

    /// Load provider capability evidence for `model` into THIS process before
    /// a model / reasoning mutation is validated against it. Only Meta Muse
    /// needs this: its exact effort set comes from a live, server-driven
    /// catalog held in a process-local cache, which a daemon-hosted session
    /// does not share with the TUI that listed the model. Uses this
    /// runtime's own credential source (local or remote broker); bounded,
    /// backed off, and never fails the caller.
    pub async fn prepare_model_capabilities(&self, model: &str) {
        if model.starts_with("muse/") {
            let broker = crate::auth::broker_from_source(
                &self.credential_source,
                &self.token_cache,
                self.client.clone(),
            );
            crate::runtime::openai::catalog::ensure_muse_catalog_for(model, &broker).await;
        }
    }

    /// Whether this runtime projects a progressive (core-only) tool schema.
    pub fn progressive_tool_disclosure(&self) -> bool {
        self.progressive_tool_disclosure
    }

    /// This conversation's exact tool activations as plain data, for the
    /// next runtime that serves it (daemon reload, park → unpark). The
    /// retained set's activations once a progressive turn ran; before that,
    /// whatever this runtime was itself handed and has not applied yet — so
    /// a reload or park before the first turn does not lose them either.
    pub fn carried_tool_activations(&self) -> Vec<crate::tools::activation::CarriedActivation> {
        let retained = self
            .retained_tool_set
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .clone();
        if let Some(shared) = retained {
            return shared
                .read()
                .unwrap_or_else(std::sync::PoisonError::into_inner)
                .carried_activations();
        }
        self.pending_tool_restore
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .clone()
            .unwrap_or_default()
    }

    /// Hand this runtime the activations the same conversation held in its
    /// previous runtime. Nothing is granted here: they are re-validated and
    /// restored (or dropped and reported) at the first progressive turn
    /// start, against that moment's catalog and this runtime's
    /// `tools.activation_confirm` (see
    /// `runtime::stream::restore_pending_activations`). An empty list
    /// clears nothing that is already pending.
    pub fn restore_tool_activations(
        &self,
        carried: Vec<crate::tools::activation::CarriedActivation>,
    ) {
        if carried.is_empty() {
            return;
        }
        let mut pending = self
            .pending_tool_restore
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        match pending.as_mut() {
            Some(list) => {
                for c in carried {
                    if !list.iter().any(|p| p.tool_id == c.tool_id) {
                        list.push(c);
                    }
                }
            }
            None => *pending = Some(carried),
        }
    }

    /// `tools.activation_confirm` policy in force for this runtime.
    pub fn activation_confirm(&self) -> agent_core::config::ActivationConfirm {
        self.activation_confirm
    }

    /// Override the `tools.activation_confirm` policy (hosts/tests that do
    /// not go through `apply_config`).
    pub fn set_activation_confirm(&mut self, mode: agent_core::config::ActivationConfirm) {
        self.activation_confirm = mode;
    }

    /// Identity of the underlying connection pool: two `Client` handles that
    /// clone the same `Arc` share one pool. `reqwest::Client` is a single
    /// `Arc<ClientRef>`, so its bytes ARE the pool pointer.
    #[cfg(any(test, feature = "testing"))]
    pub fn http_client_pool_id(&self) -> usize {
        const _: () = assert!(std::mem::size_of::<Client>() == std::mem::size_of::<usize>());
        // SAFETY: size asserted above; only the bit pattern is read, no
        // ownership is taken (transmute_copy does not move out of `self`).
        unsafe { std::mem::transmute_copy::<Client, usize>(&self.client) }
    }
    pub fn set_thinking_budget(&mut self, budget: u32) {
        self.thinking_budget = budget;
        // Sync named_level from budget so the two fields stay consistent.
        self.named_level = Some(agent_core::reasoning::ReasoningLevel::from_legacy_budget(
            budget,
        ));
        // Config/restore path — not an explicit user choice.
        self.explicit_reasoning = false;
    }

    /// Set the named reasoning level (config/restore path). Updates
    /// `thinking_budget` from the level's canonical budget when one exists;
    /// for Max/Ultra (no numeric budget), leaves `thinking_budget` unchanged.
    /// Marks `explicit_reasoning = false` — use `set_reasoning_level_explicit`
    /// for user commands and settings.
    pub fn set_reasoning_level(&mut self, level: agent_core::reasoning::ReasoningLevel) {
        self.named_level = Some(level);
        if let Some(budget) = level.to_legacy_budget() {
            self.thinking_budget = budget;
        }
        // Max/Ultra: do NOT overwrite thinking_budget with u32::MAX.
        self.explicit_reasoning = false;
    }

    /// Set the named reasoning level as an **explicit user choice** (slash
    /// commands, settings panel). Identical to `set_reasoning_level` but sets
    /// `explicit_reasoning = true` so that a subsequent `set_model` call does
    /// not overwrite the level with the new model's default.
    pub fn set_reasoning_level_explicit(&mut self, level: agent_core::reasoning::ReasoningLevel) {
        self.set_reasoning_level(level);
        self.explicit_reasoning = true;
    }

    /// Restore a saved session's `thinking_level` string as an explicit user
    /// choice, then clamp it against the current model. Call **after**
    /// `set_model(session.model)`: re-applying the saved level would otherwise
    /// undo the clamp `set_model` performed (old grok+xhigh session files).
    /// Returns the clamp applied, if any, so callers can surface it.
    pub fn restore_session_reasoning(&mut self, thinking_level: &str) -> Option<ReasoningClamp> {
        if let Some(level) = agent_core::reasoning::ReasoningLevel::parse(thinking_level) {
            self.set_reasoning_level_explicit(level);
        } else if let Some(budget) = crate::models::budget_for_thinking_level(thinking_level) {
            self.set_thinking_budget_explicit(budget);
        } else {
            return None;
        }
        self.clamp_reasoning_to_model()
    }

    /// Set a custom numeric thinking budget as an **explicit user choice**
    /// (e.g. `/thinking 8192`). Retains the exact budget in `thinking_budget`
    /// while syncing `named_level` to the nearest named level for display.
    /// Sets `explicit_reasoning = true`.
    pub fn set_thinking_budget_explicit(&mut self, budget: u32) {
        self.thinking_budget = budget;
        self.named_level = Some(agent_core::reasoning::ReasoningLevel::from_legacy_budget(
            budget,
        ));
        self.explicit_reasoning = true;
    }

    /// Expose the `explicit_reasoning` provenance flag (test/introspection).
    pub fn is_reasoning_explicit(&self) -> bool {
        self.explicit_reasoning
    }

    /// Raw thinking budget value (for testing and legacy request building).
    pub fn thinking_budget_raw(&self) -> u32 {
        self.thinking_budget
    }

    /// Validate `level` against the current model's capability metadata via
    /// the shared per-provider validator (cache then exact static tables),
    /// then apply it. Returns `Err(user-facing message)` and leaves runtime
    /// state unchanged if the level is unsupported.
    pub fn set_reasoning_level_checked(
        &mut self,
        level: agent_core::reasoning::ReasoningLevel,
    ) -> std::result::Result<(), String> {
        crate::runtime::openai::catalog::validation::validate_reasoning_mutation(
            &self.model,
            level,
        )?;
        if self.model.starts_with("anthropic/")
            && level == agent_core::reasoning::ReasoningLevel::UltraCode
        {
            if self.codex_request_role()
                != crate::runtime::openai::catalog::ExecutionRole::Foreground
            {
                return Err("ultracode_requires_foreground".into());
            }
            if self.effective_prompt.is_some() {
                return Err("typed_prompt_manifest_blocks_required_doctrine".into());
            }
            if self.orchestration.is_none() {
                return Err("ultracode_requires_orchestration".into());
            }
            let tools = self.tools.try_read().map_err(|_| {
                "ultracode prerequisite state is busy; refusing mutation".to_string()
            })?;
            let required = ["subagent_start", "subagent_status", "subagent_collect"];
            if !required.iter().all(|name| {
                tools
                    .get(name)
                    .is_some_and(|tool| tool.extension_id().is_none())
            }) {
                return Err("ultracode_requires_lifecycle_tools".into());
            }
        }
        self.set_reasoning_level_explicit(level);
        Ok(())
    }

    /// Validate then apply a custom numeric thinking budget (`/thinking <N>`).
    /// The budget's derived `ReasoningLevel` runs through the same exact-model
    /// mutation validator as named levels (Anthropic fixed-budget models pass:
    /// derived levels never exceed XHigh and thinking-capable descriptors
    /// accept them). On `Ok` the exact budget is retained — never downgraded
    /// to a named-level canonical budget. On `Err` state is unchanged.
    pub fn set_thinking_budget_checked(&mut self, budget: u32) -> std::result::Result<(), String> {
        let level = agent_core::reasoning::ReasoningLevel::from_legacy_budget(budget);
        crate::runtime::openai::catalog::validation::validate_reasoning_mutation(
            &self.model,
            level,
        )?;
        self.set_thinking_budget_explicit(budget);
        Ok(())
    }

    pub fn reasoning_level(&self) -> agent_core::reasoning::ReasoningLevel {
        self.named_level.unwrap_or_else(|| {
            agent_core::reasoning::ReasoningLevel::from_legacy_budget(self.thinking_budget)
        })
    }

    pub(crate) fn set_codex_request_role(
        &mut self,
        role: crate::runtime::openai::catalog::CodexRequestRole,
    ) {
        self.codex_request_role = role;
    }

    pub(crate) fn codex_request_role(&self) -> crate::runtime::openai::catalog::CodexRequestRole {
        self.codex_request_role
    }

    pub(crate) fn codex_delegation_plan(
        model: &str,
        level: agent_core::reasoning::ReasoningLevel,
        role: crate::runtime::openai::catalog::CodexRequestRole,
    ) -> Option<crate::runtime::openai::catalog::CodexExecutionPlan> {
        use crate::runtime::openai::catalog::{plan_codex_execution, CodexRequestRole};
        if role != CodexRequestRole::Foreground
            || level != agent_core::reasoning::ReasoningLevel::Ultra
            || !model.starts_with("openai-codex/")
        {
            return None;
        }
        plan_codex_execution(model, level, role, None).ok()
    }

    async fn authorized_anthropic_plan(
        &self,
    ) -> Result<Option<crate::runtime::openai::catalog::AnthropicExecutionPlan>> {
        if !self.model.starts_with("anthropic/") {
            return Ok(None);
        }
        let tools = self.tools.read().await;
        let builtin = |name: &str| {
            tools
                .get(name)
                .is_some_and(|tool| tool.extension_id().is_none())
        };
        let readiness = self
            .orchestration
            .as_ref()
            .and_then(|runtime| runtime.ultracode_readiness(&self.model).ok());
        let prerequisites = crate::runtime::openai::catalog::AnthropicPlanPrerequisites {
            orchestration_policy: self.orchestration.is_some(),
            foreground_worker_authorized: readiness.is_some(),
            concurrent_limit: readiness.map_or(0, |limits| limits.0),
            total_limit: readiness.map_or(0, |limits| limits.1),
            lifecycle_start: builtin("subagent_start"),
            lifecycle_status: builtin("subagent_status"),
            lifecycle_steer: builtin("subagent_steer"),
            lifecycle_collect: builtin("subagent_collect"),
            lifecycle_resume: builtin("subagent_resume"),
        };
        crate::runtime::openai::catalog::plan_anthropic_execution(
            &self.model,
            self.reasoning_level(),
            self.codex_request_role(),
            prerequisites,
            crate::runtime::openai::catalog::capability_cache::get(&self.model).and_then(|entry| {
                match entry.reasoning {
                    crate::runtime::openai::catalog::ReasoningSupport::AnthropicAdaptive {
                        adaptive,
                    } => Some(adaptive),
                    crate::runtime::openai::catalog::ReasoningSupport::None => Some(false),
                    _ => None,
                }
            }),
        )
        .map(Some)
        .map_err(|error| RuntimeError::Config(error.to_string()))
    }

    /// Validate exact-model reasoning and Ultra orchestration prerequisites
    /// before refresh, broker access, or provider network work.
    async fn validate_request_preflight(&self) -> Result<()> {
        self.validate_request_preflight_for(&self.model, self.codex_request_role)
            .await
    }

    /// Session-driver preflight: same validation as request preflight.
    /// merge(112): added for session_driver.rs; thin delegation.
    pub(crate) async fn validate_session_driver_preflight(&self) -> Result<()> {
        self.validate_request_preflight().await
    }

    async fn validate_request_preflight_for(
        &self,
        model: &str,
        role: crate::runtime::openai::catalog::CodexRequestRole,
    ) -> Result<()> {
        let level = self.reasoning_level();
        if model.starts_with("anthropic/")
            && level == agent_core::reasoning::ReasoningLevel::UltraCode
            && self.effective_prompt.is_some()
        {
            return Err(RuntimeError::Config(
                "Anthropic execution plan rejected: typed_prompt_manifest_blocks_required_doctrine"
                    .into(),
            ));
        }
        if model.starts_with("anthropic/")
            && matches!(
                level,
                agent_core::reasoning::ReasoningLevel::Max
                    | agent_core::reasoning::ReasoningLevel::UltraCode
                    | agent_core::reasoning::ReasoningLevel::Ultra
            )
        {
            let tools = self.tools.read().await;
            let builtin = |name: &str| {
                tools
                    .get(name)
                    .is_some_and(|tool| tool.extension_id().is_none())
            };
            let readiness = self
                .orchestration
                .as_ref()
                .and_then(|runtime| runtime.ultracode_readiness(model).ok());
            let prerequisites = crate::runtime::openai::catalog::AnthropicPlanPrerequisites {
                orchestration_policy: self.orchestration.is_some(),
                foreground_worker_authorized: readiness.is_some(),
                concurrent_limit: readiness.map_or(0, |limits| limits.0),
                total_limit: readiness.map_or(0, |limits| limits.1),
                lifecycle_start: builtin("subagent_start"),
                lifecycle_status: builtin("subagent_status"),
                lifecycle_steer: builtin("subagent_steer"),
                lifecycle_collect: builtin("subagent_collect"),
                lifecycle_resume: builtin("subagent_resume"),
            };
            drop(tools);
            let result = crate::runtime::openai::catalog::plan_anthropic_execution(
                model,
                level,
                role,
                prerequisites,
                crate::runtime::openai::catalog::capability_cache::get(model).and_then(|entry| {
                    match entry.reasoning {
                        crate::runtime::openai::catalog::ReasoningSupport::AnthropicAdaptive {
                            adaptive,
                        } => Some(adaptive),
                        crate::runtime::openai::catalog::ReasoningSupport::None => Some(false),
                        _ => None,
                    }
                }),
            );
            match result {
                Ok(plan) => {
                    tracing::debug!(event = "anthropic_mode_plan", qualified_model = %model, requested_level = %level, runtime_role = role.as_str(), execution_mode = ?plan.mode, wire_effort = plan.wire_effort.map_or("omitted", |effort| effort.as_str()), workflow = ?plan.workflow, credentials_attempted = false, network_attempted = false);
                    return Ok(());
                }
                Err(error) => {
                    tracing::debug!(event = "anthropic_mode_plan", qualified_model = %model, requested_level = %level, runtime_role = role.as_str(), decision = "deny", deny_code = error.code().as_str(), credentials_attempted = false, network_attempted = false);
                    return Err(RuntimeError::Config(error.to_string()));
                }
            }
        }
        if model.starts_with("openai-codex/") {
            let plan =
                crate::runtime::openai::catalog::plan_codex_execution(model, level, role, None)
                    .map_err(|error| {
                        tracing::debug!(
                            event = "codex_mode_plan",
                            qualified_model = %model,
                            requested_level = %level,
                            runtime_role = role.as_str(),
                            decision = "deny",
                            deny_code = error.code().as_str(),
                            network_attempted = false,
                            "Codex request preflight denied"
                        );
                        RuntimeError::Config(error.to_string())
                    })?;

            if plan.automatic_delegation() {
                if self.orchestration.is_none() {
                    tracing::debug!(
                        event = "codex_mode_plan",
                        qualified_model = %model,
                        requested_level = %level,
                        runtime_role = role.as_str(),
                        decision = "deny",
                        deny_code = "ultra_requires_orchestration",
                        network_attempted = false,
                        "Codex request preflight denied"
                    );
                    return Err(RuntimeError::Config(
                        "Ultra requires an installed orchestration policy".to_string(),
                    ));
                }

                let tools = self.tools.read().await;
                let required = ["subagent_start", "subagent_status", "subagent_collect"];
                let tools_ready = required.iter().all(|name| {
                    tools
                        .get(name)
                        .is_some_and(|tool| tool.extension_id().is_none())
                });
                if !tools_ready {
                    tracing::debug!(
                        event = "codex_mode_plan",
                        qualified_model = %model,
                        requested_level = %level,
                        runtime_role = role.as_str(),
                        decision = "deny",
                        deny_code = "ultra_requires_subagent_tools",
                        network_attempted = false,
                        "Codex request preflight denied"
                    );
                    return Err(RuntimeError::Config(
                        "Ultra requires built-in subagent_start, subagent_status, and subagent_collect tools"
                            .to_string(),
                    ));
                }
            }
            return Ok(());
        }

        crate::runtime::openai::catalog::validation::validate_reasoning_mutation(model, level)
            .map_err(|error| {
                tracing::debug!(
                    event = "codex_mode_plan",
                    qualified_model = %model,
                    requested_level = %level,
                    runtime_role = role.as_str(),
                    decision = "deny",
                    deny_code = "unsupported_reasoning_level",
                    network_attempted = false,
                    "request preflight denied"
                );
                RuntimeError::Config(error)
            })
    }

    pub fn set_compaction_model(&mut self, model: Option<String>) {
        self.compaction_model = model;
    }

    /// Set where compaction summarization runs (spec §9.4).
    pub fn set_compaction_mode(&mut self, mode: agent_core::compaction::CompactionMode) {
        self.compaction_mode = mode;
    }

    /// Set the content classes withheld from remote compaction disclosure.
    pub fn set_compaction_exclusions(
        &mut self,
        exclude: Vec<agent_core::compaction::ContentClass>,
    ) {
        self.compaction_exclusions = exclude;
    }

    /// Number of times the remote-summarization transport seam was entered
    /// (CP-12 M3). Local-only compaction must leave this at zero.
    pub fn remote_summarization_attempts(&self) -> u64 {
        self.remote_summarization_attempts
            .load(std::sync::atomic::Ordering::SeqCst)
    }

    /// The session's compaction disclosure policy (spec §9.4) — consumed by
    /// `runtime::compaction` for both the preview and the dispatch path.
    pub fn compaction_policy(&self) -> crate::runtime::compaction::DisclosurePolicy {
        crate::runtime::compaction::DisclosurePolicy {
            mode: self.compaction_mode,
            exclude: self.compaction_exclusions.clone(),
        }
    }

    pub fn set_context_window(&mut self, window: Option<u64>) {
        self.context_window_override = window;
    }

    /// The configured window override (`None` = the model default).
    pub fn context_window_override(&self) -> Option<u64> {
        self.context_window_override
    }

    /// Effective context window for the current model — user override if set,
    /// otherwise the model's native window from `models::context_window_for_model`.
    pub fn compaction_model(&self) -> &str {
        self.compaction_model
            .as_deref()
            .unwrap_or("claude-sonnet-4-6")
    }

    pub fn context_window(&self) -> u64 {
        self.context_window_override
            .unwrap_or_else(|| crate::models::context_window_for_model(&self.model))
    }

    /// Task 29 (spec §9.1): the ONE request-aware context assessment every
    /// frontend consumes on its compaction trigger path. Reads the segments
    /// the next request will actually carry — the effective system prompt,
    /// the exposed tool-schema set, the supplied history — and the runtime's
    /// configured reserves (thinking budget, tool-result cap, the expected
    /// output within the model's `max_tokens`, provider window with the
    /// documented safety margin).
    ///
    /// Skill bodies that reach the request do so through the system prompt
    /// or history today, so they are already accounted there; the skill lane
    /// is fed once loaders hand the engine distinct segments. Nothing in
    /// core feeds the memory lane any more: it stays empty.
    pub async fn assess_context(
        &self,
        messages: &[crate::SharedMessage],
    ) -> context::ContextAssessment {
        let system = self.effective_system_prompt().await;
        let schema = self.offered_tools_schema().await;
        let anchor = self
            .continuation
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .usage_anchor_for(&self.model)
            .cloned();
        context::assess(&context::ContextBudgetInputs {
            model: &self.model,
            provider_window: self.context_window(),
            system_prompt: system.as_deref(),
            tools_schema: &schema,
            messages,
            skill_contents: &[],
            memory_contents: &[],
            thinking_budget_tokens: self.thinking_budget as u64,
            next_tool_result_bytes: self.max_tool_output as u64,
            max_output_tokens: HelperMethods::max_tokens_for_model(&self.model),
        })
        .anchored(anchor.as_ref(), messages)
    }

    /// Give a worker its parent's binding: the parent's base and project,
    /// wherever the worker's own folder is, with a fresh worker execution
    /// author. Oneshot, start, and resume share this path; even resume must
    /// not reuse a prior actor. The runtime is then bound: a later
    /// `apply_config` on it does not replace the binding with one for the
    /// worker's folder. Tool grants stay untouched.
    pub(crate) fn inherit_memory_binding(&mut self, binding: &crate::memory_backend::MemoryBinding) {
        self.memory_backend = binding.for_worker();
        self.memory_bound = true;
    }

    /// Bind this runtime's memory to the project of its folder, once: a
    /// runtime that is already bound keeps its binding. Standalone host
    /// entry points call it without applying unrelated settings;
    /// `apply_config` calls it too, so a runtime is given its folder
    /// (`set_cwd`) before its first `apply_config`.
    pub fn bind_memory(&mut self) {
        if self.memory_bound {
            return;
        }
        // merge(112) §4: under the daemon the session cwd differs from
        // process cwd; use it when available so project-scope resolution
        // lands in the right repo.
        self.memory_backend = crate::memory_backend::MemoryBinding::for_cwd(
            self.cwd.clone().or_else(|| std::env::current_dir().ok()),
        );
        self.memory_bound = true;
    }

    #[cfg(test)]
    #[allow(dead_code)] // merge(112): consumed in phase 5
    pub(crate) fn memory_backend_for_test(&self) -> crate::memory_backend::MemoryBinding {
        self.memory_backend.clone()
    }

    /// Apply a parsed config file to this runtime (model, thinking budget, etc.)
    /// Includes the `disabled_tools` pass on this runtime's registry — the
    /// fresh-`Runtime::new()` path, where the registry holds builtins only.
    pub fn apply_config(&mut self, config: &crate::config::SynapsConfig) {
        self.apply_config_inner(config, true);
    }

    /// `apply_config` WITHOUT the `disabled_tools` pass. Used by
    /// `EngineHost::foreground_runtime`: the host already disabled builtins
    /// on the fresh registry before skills/MCP registered (the old boot
    /// point), and a second pass here would also strip `load_skill`,
    /// `search_skills` and the descriptor-backed MCP tools — which
    /// the old boot never did.
    pub(crate) fn apply_config_keep_tools(&mut self, config: &crate::config::SynapsConfig) {
        self.apply_config_inner(config, false);
    }

    fn apply_config_inner(&mut self, config: &crate::config::SynapsConfig, disable_tools: bool) {
        // #112: bind memory early (the binding is immutable once made).
        self.bind_memory();
        if let Some(ref model) = config.model {
            self.set_model(model.clone());
        }
        // Named level takes priority over raw budget.
        // Config-specified thinking is an explicit user choice — preserve across model switches.
        if let Some(level) = config.thinking_level {
            self.set_reasoning_level_explicit(level);
        } else if let Some(budget) = config.thinking_budget {
            self.set_thinking_budget_explicit(budget);
        }
        // A configured level the configured model rejects would boot into a
        // permanently failing state; clamp exactly as `/model` does.
        if let Some(clamp) = self.clamp_reasoning_to_model() {
            tracing::warn!(
                from = clamp.from.as_str(),
                to = clamp.to.as_str(),
                model = %self.model,
                "configured reasoning level not supported by model; clamped"
            );
        }
        self.context_window_override = config.context_window;
        self.continuation
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .config = config.context_management;
        self.compaction_model = config.compaction_model.clone();
        self.compaction_mode = config.compaction_mode;
        self.compaction_exclusions = config.compaction_exclude.clone();
        self.max_tool_output = config.max_tool_output;
        self.tool_spill_policy = tool_spill::ToolSpillPolicy::from_config(config);
        self.read_dedupe = config.read_dedupe;
        self.thinking_display = config.thinking_display;
        self.tool_keepwarm = config.cache_keepwarm_tools;
        self.bash_timeout = config.bash_timeout;
        self.bash_max_timeout = config.bash_max_timeout;
        self.subagent_timeout = config.subagent_timeout;
        self.api_retries = config.api_retries;
        self.refusal_retries = config.refusal_retries;
        self.telemetry_level =
            crate::runtime::telemetry::TelemetryLevel::from_str_key(&config.telemetry);
        self.telemetry_log_generations = config.telemetry_log_generations;
        self.sync_observability();
        self.cache_diagnostics = config.cache_diagnostics;
        self.cache_ttl = config.cache_ttl;
        self.request_compression = request_encoding::RequestCompression::from_config(
            config.request_compression,
            config.request_compression_level,
        );
        self.progressive_tool_disclosure = config.progressive_tool_disclosure;
        self.tools_always_loaded = Arc::from(config.tools_always_loaded.as_slice());
        self.deferred_tool_loading = config.tools_deferred_loading;
        self.activation_confirm = config.tools_activation_confirm;
        tracing::info!(
            mode = config.tools_activation_confirm.as_str(),
            auto_approve_confirms = config.server.auto_approve_confirms,
            "tools.activation_confirm: model-initiated activate_tools policy"
        );
        self.trusted_worker_models = config
            .favorite_models
            .iter()
            .filter_map(|model| {
                let canonical = canonical_trusted_worker_model(model);
                match agent_core::prompt::QualifiedModelId::parse(canonical) {
                    Ok(model) => Some(model),
                    Err(_) => {
                        tracing::warn!(
                            model = model.as_str(),
                            "ignoring invalid favorite worker-model identity"
                        );
                        None
                    }
                }
            })
            .collect();
        self.trusted_worker_models.sort();
        self.trusted_worker_models.dedup();
        self.subagent_limits = config.subagent;
        if let Some(orchestration) = &self.orchestration {
            for model in &self.trusted_worker_models {
                if let Err(error) = orchestration.grant_worker_model(model.as_str()) {
                    tracing::warn!(
                        model = model.as_str(),
                        error = %error,
                        "failed to apply configured worker-model trust"
                    );
                }
            }
        }
        self.apply_auth_config(config);

        // Remove any built-in tools the user disabled via `disabled_tools`.
        // try_write is safe here: this runs on a fresh runtime's private
        // registry before it is shared with other tasks.
        if disable_tools && !config.disabled_tools.is_empty() {
            if let Ok(mut reg) = self.tools.try_write() {
                reg.disable(&config.disabled_tools);
            }
        }
    }

    /// Apply only the credential-source portion of config. Used by subagent
    /// spawns that build a fresh `Runtime::new()` (which defaults to Local) and
    /// must inherit the user's Remote/broker setup. (#158 A3)
    ///
    /// A6: invalidates the token cache if the source/endpoint changed.
    /// A2: scrubs `AuthState` when Remote so no local `auth.json` credential is
    /// used or held (invariant 1).
    pub fn apply_auth_config(&mut self, config: &crate::config::SynapsConfig) {
        let new_source = config.auth.credential_source();
        if new_source != self.credential_source {
            self.token_cache.invalidate("anthropic");
        }
        self.credential_source = new_source;
        if self.credential_source.is_remote() {
            if let Ok(mut auth) = self.auth.try_write() {
                AuthMethods::scrub_for_remote(&mut auth);
            }
        }
        // Account switch (G5): if the in-memory Anthropic OAuth token is
        // bound to a credential other than the one this config now selects
        // (`auth.account.anthropic` / `SYNAPS_ACCOUNT_ANTHROPIC`, or a
        // source change), drop it so the next turn re-vends through the
        // broker — the previous account is never silently kept. Stateless:
        // compares the token's recorded binding to the new selection.
        if let Ok(mut auth) = self.auth.try_write() {
            if auth.auth_type == "oauth" && !auth.auth_token.is_empty() {
                let selector = config
                    .auth
                    .account_selector(crate::auth::OAuthProviderId::Anthropic);
                let check = crate::runtime::auth::check_binding(
                    &self.credential_source,
                    &selector,
                    auth.bound_credential.as_deref(),
                    crate::runtime::auth::RefreshPoint::TurnStart,
                );
                if check != crate::runtime::auth::BindingCheck::Matches {
                    AuthMethods::scrub_for_account_switch(&mut auth);
                    self.token_cache.invalidate("anthropic");
                }
            }
        }
        // Install the process-wide credential broker matching this source so
        // every request path (streams, pings, catalog, TUI status) resolves
        // credentials through the same boundary. Local sources get the
        // in-process broker — no separately launched daemon required.
        crate::auth::set_global_broker(crate::auth::broker_from_source(
            &self.credential_source,
            &self.token_cache,
            self.client.clone(),
        ));
    }

    pub fn thinking_budget(&self) -> u32 {
        self.thinking_budget
    }

    pub fn max_tool_output(&self) -> usize {
        self.max_tool_output
    }

    pub fn bash_timeout(&self) -> u64 {
        self.bash_timeout
    }

    pub fn bash_max_timeout(&self) -> u64 {
        self.bash_max_timeout
    }

    pub fn subagent_timeout(&self) -> u64 {
        self.subagent_timeout
    }

    pub fn api_retries(&self) -> u32 {
        self.api_retries
    }

    pub fn set_max_tool_output(&mut self, v: usize) {
        self.max_tool_output = v;
    }

    pub fn tool_spill_policy(&self) -> tool_spill::ToolSpillPolicy {
        self.tool_spill_policy
    }

    pub fn set_tool_spill_policy(&mut self, policy: tool_spill::ToolSpillPolicy) {
        self.tool_spill_policy = policy;
    }

    pub fn read_dedupe(&self) -> bool {
        self.read_dedupe
    }

    /// Read dedupe + write/edit notes; off = legacy tool-result bytes.
    pub fn set_read_dedupe(&mut self, enabled: bool) {
        self.read_dedupe = enabled;
    }

    /// The configured `thinking_display` (whatever this runtime's role).
    pub fn thinking_display(&self) -> crate::config::ThinkingDisplay {
        self.thinking_display
    }

    pub fn set_thinking_display(&mut self, display: crate::config::ThinkingDisplay) {
        self.thinking_display = display;
    }

    /// The `thinking.display` this runtime's requests send: the configured
    /// value on the foreground session; always summarized for a worker or an
    /// internal request. A worker's thinking summaries are its record: they
    /// stream into `/watch`, and its rollover transcripts and the parent's
    /// archived worker history keep them for review, and no one is
    /// waiting on its first text token.
    pub(crate) fn wire_thinking_display(&self) -> crate::config::ThinkingDisplay {
        match self.codex_request_role {
            crate::runtime::openai::catalog::CodexRequestRole::Foreground => self.thinking_display,
            _ => crate::config::ThinkingDisplay::Summarized,
        }
    }

    pub fn tool_keepwarm(&self) -> bool {
        self.tool_keepwarm
    }

    /// Keep the cache warm while a tool round runs (`cache_keepwarm_tools`).
    pub fn set_tool_keepwarm(&mut self, enabled: bool) {
        self.tool_keepwarm = enabled;
    }

    /// Session totals of tool output virtualisation (count, bytes kept out
    /// of history).
    pub fn tool_spill_totals(&self) -> tool_spill::ToolSpillTotals {
        self.tool_spill_stats.totals()
    }

    /// The virtualiser one stream uses. Its dir is keyed by the persisted
    /// conversation id, so retention removes it with the session; a worker
    /// (no conversation id) is keyed by its own runtime trace session.
    pub(crate) fn tool_spill(&self) -> tool_spill::ToolSpill {
        if !self.tool_spill_policy.is_enabled() {
            return tool_spill::ToolSpill::off();
        }
        let key = self
            .session_id
            .clone()
            .unwrap_or_else(|| self.trace_ctx.session_id().as_str().to_string());
        tool_spill::ToolSpill::new(
            self.tool_spill_policy,
            &self.tool_spill_root,
            &key,
            Arc::clone(&self.tool_spill_stats),
        )
    }

    pub fn set_bash_timeout(&mut self, v: u64) {
        self.bash_timeout = v;
    }

    pub fn set_bash_max_timeout(&mut self, v: u64) {
        self.bash_max_timeout = v;
    }

    pub fn set_subagent_timeout(&mut self, v: u64) {
        self.subagent_timeout = v;
    }

    /// Copy only retry policy; worker construction must not re-install brokers
    /// or acquire foreground capabilities by calling full apply_config.
    pub(crate) fn apply_retry_config(&mut self, config: &crate::SynapsConfig) {
        self.api_retries = config.api_retries;
        self.refusal_retries = config.refusal_retries;
    }

    pub fn refusal_retries(&self) -> u32 {
        self.refusal_retries
    }

    pub fn set_api_retries(&mut self, v: u32) {
        self.api_retries = v;
    }

    pub fn telemetry_level(&self) -> crate::runtime::telemetry::TelemetryLevel {
        self.telemetry_level
    }

    pub fn set_telemetry_level(&mut self, level: crate::runtime::telemetry::TelemetryLevel) {
        self.telemetry_level = level;
        self.sync_observability();
    }

    /// This session's telemetry, for the subagents it spawns.
    pub fn worker_telemetry(&self) -> crate::runtime::telemetry::WorkerTelemetry {
        crate::runtime::telemetry::WorkerTelemetry {
            level: self.telemetry_level,
            writer: self.telemetry_writer.clone(),
            parent_session_id: self.session_id.clone(),
            parent_runtime_session_id: Some(self.trace_ctx.session_id().as_str().to_string()),
            ..Default::default()
        }
    }

    /// Record a worker's requests at its parent's telemetry level, through
    /// the parent's session writer. Worker runtimes start `Off`
    /// (`from_parts`), so without this every subagent request was missing
    /// from `api-log.jsonl`.
    ///
    /// The writer is shared, not rebuilt: it owns its files from one thread,
    /// and a writer per worker would race rotation. The trace context is
    /// fresh, though: a worker is its own trace session with its own
    /// cache-break snapshots, so its requests never read as breaks in the
    /// parent's conversation. A parent that is `Off` leaves the worker `Off`.
    pub fn inherit_telemetry(&mut self, parent: &crate::runtime::telemetry::WorkerTelemetry) {
        match (&parent.writer, parent.level.enabled()) {
            (Some(writer), true) => {
                self.telemetry_level = parent.level;
                self.trace_ctx = trace::TraceContext::with_sink(Arc::new(
                    crate::runtime::telemetry::WriterTraceSink::new(writer.clone()),
                ));
                // Same writer, tagged: the worker's records carry its agent,
                // handle and the parent's session id.
                self.telemetry_writer = parent.worker_writer();
            }
            _ => self.set_telemetry_level(crate::runtime::telemetry::TelemetryLevel::Off),
        }
    }

    /// Reconcile the shared observability writer + trace sink with the
    /// telemetry level (Task 11 config rule, documented until Task 12 adds
    /// explicit trace config/UI): `basic`/`full` enables BOTH legacy
    /// telemetry persistence and metadata-only trace persistence through
    /// one bounded session writer; `off` disables both. The trace schema is
    /// structurally metadata-only, so this never persists raw content.
    fn sync_observability(&mut self) {
        if self.telemetry_level.enabled() {
            if self.telemetry_writer.is_none() {
                let writer = crate::runtime::telemetry::TelemetryWriter::new(
                    crate::runtime::telemetry::WriterOptions {
                        max_generations: self.telemetry_log_generations,
                        ..Default::default()
                    },
                );
                self.trace_ctx = trace::TraceContext::with_sink(Arc::new(
                    crate::runtime::telemetry::WriterTraceSink::new(writer.clone()),
                ));
                self.telemetry_writer = Some(writer);
            }
        } else {
            // Dropping this handle detaches the worker: it drains whatever
            // is queued and exits on its own — never a hang. In-flight
            // requests holding a cloned handle keep enqueueing harmlessly.
            self.telemetry_writer = None;
            self.trace_ctx = trace::TraceContext::disabled();
        }
    }

    /// The session trace context handed to every request (cheap clone).
    pub fn trace_context(&self) -> trace::TraceContext {
        self.trace_ctx.clone()
    }

    /// Effective trace context for the next outgoing model request:
    /// consumes a pending one-shot arm (`/trace next [content]`, Task 12).
    ///
    /// One-shot semantics (B1): the arm covers exactly one *logical
    /// request* — the returned armed context carries a one-shot request
    /// gate consumed inside `RequestTracer::begin`, so the first request
    /// through it (all retry attempts included) emits records and every
    /// subsequent request sharing the same `ApiOptions` (tool-loop
    /// continuations) is disabled. Normal Basic/Full session contexts are
    /// never gated.
    ///
    /// When the session sink is disabled the one-shot record rides an
    /// ephemeral writer forked from the session context (same session ID,
    /// digest key, degradation counters, and §6.6 cache snapshot store —
    /// so `/context` sees the armed request's diagnostics). The writer
    /// handle is retained in `one_shot_trace_writer` until replaced or
    /// drained by the shutdown epilogue, guaranteeing the armed record
    /// flushes on session exit. With the session sink already enabled
    /// (telemetry Basic/Full) the arm only adds the optional content
    /// capture.
    pub(crate) fn effective_trace_context(&self) -> trace::TraceContext {
        match self.trace_controls.consume() {
            None => self.trace_ctx.clone(),
            Some(with_content) => {
                let base = if self.trace_ctx.enabled() {
                    self.trace_ctx.clone()
                } else {
                    let writer = crate::runtime::telemetry::TelemetryWriter::new(
                        crate::runtime::telemetry::WriterOptions::default(),
                    );
                    // Retain the writer for the exit epilogue (drops any
                    // previously retained one-shot writer, whose worker
                    // keeps draining in the background).
                    *self
                        .one_shot_trace_writer
                        .lock()
                        .expect("one-shot trace writer lock poisoned") = Some(writer.clone());
                    self.trace_ctx
                        .fork_with_sink(Arc::new(crate::runtime::telemetry::WriterTraceSink::new(
                            writer,
                        )))
                        .with_one_shot_request_gate()
                };
                if with_content {
                    base.with_content_capture(Arc::new(trace::ContentCapture::new(
                        self.capture_dir.clone(),
                    )))
                } else {
                    base
                }
            }
        }
    }

    /// Arm tracing for exactly the next outgoing provider request
    /// (`/trace next`; `with_content` adds the one-request redacted
    /// content capture for `/trace next content`).
    pub fn trace_arm_next(&self, with_content: bool) {
        self.trace_controls.arm_next(with_content);
    }

    /// Metadata-only `/trace status` report: mode, persistence path,
    /// counters. Never secrets or content. Also performs the opportunistic
    /// expired-capture sweep (B2): status is a trace interaction, so stale
    /// content-capture bundles are physically removed here too.
    pub fn trace_status(&self) -> trace::TraceStatusReport {
        let _ = trace::sweep_expired_captures(&self.capture_dir);
        trace::TraceStatusReport {
            persistent_enabled: self.telemetry_writer.is_some(),
            arm: self.trace_controls.peek(),
            trace_path: crate::runtime::telemetry::default_trace_log_path(),
            writer_stats: self.telemetry_writer.as_ref().map(|w| w.stats()),
            degraded_records: self.trace_ctx.degraded_records(),
        }
    }

    /// Structured `/context` report (Task 12): counts, byte lengths, cache
    /// change/reuse estimates, and writer counters — never content.
    ///
    /// `history` is the conversation owned by the calling surface (TUI /
    /// headless); when it is not provided the history lines honestly read
    /// `unavailable` rather than fabricating zeros. Loaded skills/memories
    /// are session-surface state the runtime cannot enumerate, so they are
    /// reported `unavailable` with that provenance.
    pub fn context_report(&self, history: Option<&[crate::SharedMessage]>) -> trace::ContextReport {
        use trace::ReportValue;
        let (history_messages, history_bytes) = match history {
            Some(msgs) => (
                ReportValue::Count(msgs.len() as u64),
                ReportValue::Count(
                    msgs.iter()
                        .map(|m| {
                            serde_json::to_vec(&**m)
                                .map(|v| v.len() as u64)
                                .unwrap_or(0)
                        })
                        .sum(),
                ),
            ),
            None => (ReportValue::Unavailable, ReportValue::Unavailable),
        };
        // Non-blocking view of the tool registry: if it is momentarily
        // write-locked, report honestly instead of blocking or guessing.
        let tool_count = match self.tools.try_read() {
            Ok(tools) => ReportValue::Count(tools.iter_tools_sorted().len() as u64),
            Err(_) => ReportValue::Unavailable,
        };
        trace::ContextReport {
            model: self.model.clone(),
            system_prompt_bytes: ReportValue::Count(
                self.system_prompt().map(|s| s.len() as u64).unwrap_or(0),
            ),
            tool_count,
            history_messages,
            history_bytes,
            loaded_skills: ReportValue::Unavailable,
            loaded_memories: ReportValue::Unavailable,
            cache: self.trace_ctx.cache_snapshots().last_activity(),
            trace_enabled: self.trace_ctx.enabled(),
            writer_stats: self.telemetry_writer.as_ref().map(|w| w.stats()),
            degraded_records: self.trace_ctx.degraded_records(),
        }
    }

    /// The shared observability writer, when telemetry is enabled.
    pub fn telemetry_writer(&self) -> Option<crate::runtime::telemetry::TelemetryWriter> {
        self.telemetry_writer.clone()
    }

    /// The writer handle for this runtime's requests: stamps the persisted
    /// conversation id (`session_id`) on every api-log record, keeping any
    /// worker identity the handle already carries.
    pub(crate) fn request_telemetry(&self) -> Option<crate::runtime::telemetry::TelemetryWriter> {
        self.telemetry_writer
            .as_ref()
            .map(|w| w.with_session(self.session_id.clone()))
    }

    /// Bounded shutdown flush of the observability writer: stop intake,
    /// drain until `timeout`, return typed stats (`None` when telemetry is
    /// off). Sync and bounded — safe from shutdown paths that cannot await;
    /// async callers should use [`Self::shutdown_observability_async`].
    ///
    /// Semantics (Task 11):
    /// - telemetry `off` → no writer exists → `None`, a true no-op;
    /// - "flushed" means every queued record was appended into OS file
    ///   buffers (`write(2)` returned) — deliberately no `fsync`, these are
    ///   best-effort diagnostic logs;
    /// - on timeout the worker stays detached and keeps draining in the
    ///   background; the caller logs metadata-only stats and continues —
    ///   trace loss must never fail or abort a clean exit.
    pub fn shutdown_observability(
        &self,
        timeout: std::time::Duration,
    ) -> Option<crate::runtime::telemetry::ShutdownOutcome> {
        // Operational retention (Task 12): the exit epilogue also removes
        // expired content-capture bundles — bounded, fail-soft, so a stale
        // bundle never has to wait for the next trace interaction. Swept at
        // the CONSTRUCTION-BOUND root (fix1 I2b), immune to env churn.
        let _ = trace::sweep_expired_captures(&self.capture_dir);
        // Drain a retained one-shot ephemeral writer first (Task 12 fix:
        // an armed `/trace next` record written with telemetry Off must
        // survive session exit). Its outcome is returned only when no
        // session writer exists — the session writer's stats remain the
        // primary signal.
        let ephemeral = self
            .one_shot_trace_writer
            .lock()
            .expect("one-shot trace writer lock poisoned")
            .take();
        let ephemeral_outcome = ephemeral.map(|w| w.shutdown(timeout));
        self.telemetry_writer
            .as_ref()
            .map(|w| w.shutdown(timeout))
            .or(ephemeral_outcome)
    }

    /// Async epilogue helper for every clean process/runtime exit path that
    /// owns a `Runtime` (headless chat, TUI teardown,
    /// RPC/server graceful shutdown): clones the writer handle and runs the
    /// bounded [`Self::shutdown_observability`] on the blocking pool, so an
    /// executor thread is never parked. Same `None`/no-fsync/timeout
    /// semantics as the sync variant. Idempotent — a second call finds the
    /// intake already closed and returns immediately.
    ///
    /// Note: the writer is shared across `Runtime` clones (subagents). Call
    /// this only from the session owner's exit epilogue, after in-flight
    /// work has drained — cloned runtimes still serving requests would see
    /// their subsequent records counted as drops.
    pub async fn shutdown_observability_async(
        &self,
        timeout: std::time::Duration,
    ) -> Option<crate::runtime::telemetry::ShutdownOutcome> {
        // Same operational retention sweep as the sync epilogue, off the
        // executor (bounded filesystem work, fail-soft on a failed spawn).
        // Construction-bound root (fix1 I2b).
        let capture_dir = self.capture_dir.clone();
        let _ =
            tokio::task::spawn_blocking(move || trace::sweep_expired_captures(&capture_dir)).await;
        // Same one-shot ephemeral drain as the sync variant.
        let ephemeral = self
            .one_shot_trace_writer
            .lock()
            .expect("one-shot trace writer lock poisoned")
            .take();
        let ephemeral_outcome = match ephemeral {
            Some(writer) => Some(writer.shutdown_async(timeout).await),
            None => None,
        };
        match self.telemetry_writer.clone() {
            Some(writer) => Some(writer.shutdown_async(timeout).await),
            None => ephemeral_outcome,
        }
    }

    /// Test seam: install a custom writer (e.g. temp paths, artificial
    /// write delay) as the session observability sink, exactly as
    /// `sync_observability` would for a production writer.
    #[cfg(any(test, feature = "testing"))]
    pub fn install_observability_for_tests(
        &mut self,
        writer: crate::runtime::telemetry::TelemetryWriter,
    ) {
        self.trace_ctx = trace::TraceContext::with_sink(Arc::new(
            crate::runtime::telemetry::WriterTraceSink::new(writer.clone()),
        ));
        self.telemetry_writer = Some(writer);
    }

    pub fn cache_diagnostics(&self) -> bool {
        self.cache_diagnostics
    }

    pub fn set_cache_diagnostics(&mut self, v: bool) {
        self.cache_diagnostics = v;
    }

    /// `request_compression` / `request_compression_level`, as `apply_config`
    /// sets them (the process-wide refusal latch included).
    pub fn set_request_compression(
        &mut self,
        mode: agent_core::config::RequestCompressionMode,
        level: u32,
    ) {
        self.request_compression = request_encoding::RequestCompression::from_config(mode, level);
    }

    /// `(mode, level)` of the gzip policy this runtime's requests carry.
    #[cfg(test)]
    pub(crate) fn request_compression_for_test(
        &self,
    ) -> (Option<agent_core::config::RequestCompressionMode>, u32) {
        self.request_compression.configured()
    }

    /// The configured cache TTL (`auto` unless config or a caller fixed it).
    /// What a request actually sends is [`Self::resolved_cache_ttl`].
    pub fn cache_ttl(&self) -> crate::core::config::CacheTtlSetting {
        self.cache_ttl
    }

    /// The wire TTL the next Anthropic request will carry: the configured
    /// setting resolved against the credential kind currently bound
    /// (`auto`: 1h on OAuth/subscription, 5m on an API key). Same function
    /// the request builder applies, so keep-warm scheduling and status
    /// agree with the wire. Depends only on the kind, never on which
    /// account: failover between seats of one kind leaves it unchanged.
    pub async fn resolved_cache_ttl(&self) -> crate::core::config::CacheTtl {
        let auth = self.auth.read().await;
        self.cache_ttl.resolve_for_auth_type(&auth.auth_type)
    }

    /// Thread a parent runtime's TTL atomics onto this (cloned) runtime so
    /// that a driver-owned turn shares the session's single downgrade latch
    /// instead of firing a redundant 1h→5m notice.
    pub fn share_ttl_latches(&mut self, parent: &Runtime) {
        self.ttl_downgrade_notified = parent.ttl_downgrade_notified.clone();
        self.saw_1h_honored = parent.saw_1h_honored.clone();
        self.cache_anchor = parent.cache_anchor.clone();
        // Same session, same diagnostics chain: a driver turn continues it,
        // and a ping sees the same rejection latch (it never moves the chain).
        self.last_msg_id = parent.last_msg_id.clone();
        self.cache_diag_rejected = parent.cache_diag_rejected.clone();
    }

    /// The cache-diagnosis chain for this runtime's requests, `None` when
    /// `cache_diagnostics` is off (wire unchanged).
    pub(crate) fn cache_diagnosis(&self) -> Option<cache_diagnosis::CacheDiagnosis> {
        self.cache_diagnostics.then(|| {
            cache_diagnosis::CacheDiagnosis::new(
                Arc::clone(&self.last_msg_id),
                Arc::clone(&self.cache_diag_rejected),
            )
        })
    }

    /// Test seam: set the bound Anthropic credential kind (`oauth` |
    /// `api_key` | `none`) so `cache_ttl = auto` resolution can be exercised
    /// without a real login.
    #[cfg(test)]
    pub(crate) async fn set_auth_kind_for_test(&self, kind: &str) {
        self.auth.write().await.auth_type = kind.to_string();
    }

    /// Change the cache TTL strategy mid-session. The next request re-marks
    /// with the new TTL; the old prefix expires naturally (single-last
    /// strategy never prunes old markers, so no invalidation logic needed).
    /// Accepts a fixed [`crate::core::config::CacheTtl`] or a full
    /// [`crate::core::config::CacheTtlSetting`] (including `auto`).
    pub fn set_cache_ttl(&mut self, ttl: impl Into<crate::core::config::CacheTtlSetting>) {
        self.cache_ttl = ttl.into();
    }

    /// Worker rollover policy from config (`apply_subagent_runtime_policy`).
    /// Inert until [`Self::attach_worker_transcripts`] gives it somewhere to
    /// keep evicted history.
    pub fn set_worker_context_settings(
        &mut self,
        settings: Option<worker_context::WorkerContextSettings>,
    ) {
        self.worker_context.settings = settings;
    }

    /// Where this worker's evicted history goes (its archive directory and
    /// handle), plus who to tell. Set per spawn, before the run starts.
    pub fn attach_worker_transcripts(&mut self, sink: worker_context::WorkerTranscriptSink) {
        self.worker_context.sink = Some(sink);
    }

    pub fn worker_context(&self) -> &worker_context::WorkerContext {
        &self.worker_context
    }

    pub fn thinking_level(&self) -> &str {
        // For Max/Ultra the named_level wins; for legacy budget-only levels
        // we fall through to the legacy bucketing.
        match self.named_level {
            Some(level) => level.as_str(),
            None => crate::core::models::thinking_level_for_budget(self.thinking_budget),
        }
    }

    /// Check if the OAuth token is expired and refresh it if needed.
    pub async fn refresh_if_needed(&self) -> Result<()> {
        self.refresh_if_needed_for_model(&self.model).await
    }

    /// `/auth reload`: drop the in-memory Anthropic token and re-vend it
    /// through the credential broker (re-reads auth.json locally, forcing a
    /// refresh if the file still holds the token we had). Returns a
    /// token-free status line.
    pub async fn reload_auth(&self) -> Result<String> {
        AuthMethods::reload_after_rejection(
            Arc::clone(&self.auth),
            &self.client,
            &self.credential_source,
            &self.token_cache,
            None,
        )
        .await?;
        let expires = self.auth.read().await.token_expires;
        let source = if self.credential_source.is_remote() {
            "remote broker"
        } else {
            "auth.json"
        };
        Ok(match expires {
            Some(exp) => {
                let mins = exp.saturating_sub(crate::epoch_millis()) / 60_000;
                format!("Anthropic credentials reloaded from {source} (valid ~{mins} min)")
            }
            None => format!("Anthropic credentials reloaded from {source}"),
        })
    }

    /// A cache keep-warm ping's refresh: keeps the bound account or fails,
    /// never selects or switches one ([`auth::RefreshPoint::KeepWarm`]).
    pub(crate) async fn refresh_for_keepwarm(&self) -> Result<()> {
        if !crate::runtime::auth::model_is_anthropic(&self.model) {
            return Ok(());
        }
        AuthMethods::refresh_if_needed(
            Arc::clone(&self.auth),
            &self.client,
            &self.credential_source,
            &self.token_cache,
            Some(&self.model),
            crate::runtime::auth::RefreshPoint::KeepWarm,
        )
        .await
    }

    async fn refresh_if_needed_for_model(&self, model: &str) -> Result<()> {
        // Non-Anthropic models resolve their own provider auth in the OpenAI
        // path (incl. via the broker), so skip the Anthropic pre-fetch. (#158 #7)
        if !crate::runtime::auth::model_is_anthropic(model) {
            return Ok(());
        }
        AuthMethods::refresh_if_needed(
            Arc::clone(&self.auth),
            &self.client,
            &self.credential_source,
            &self.token_cache,
            Some(model),
            crate::runtime::auth::RefreshPoint::TurnStart,
        )
        .await
    }

    /// Make a simple non-streaming API call for compaction (no tools).
    ///
    /// Uses a dedicated summarization system prompt (not the user's), omits
    /// all tools, and returns the raw text response. Caller supplies the
    /// full message array including the serialized conversation.
    pub async fn compact_call(&self, messages: Vec<crate::SharedMessage>) -> Result<String> {
        // Transport-construction seam: counted BEFORE preflight, auth
        // refresh, or any request assembly (CP-12 M3).
        self.remote_summarization_attempts
            .fetch_add(1, std::sync::atomic::Ordering::SeqCst);
        let model = self.compaction_model();
        self.validate_request_preflight_for(
            model,
            crate::runtime::openai::catalog::CodexRequestRole::Internal,
        )
        .await?;
        self.refresh_if_needed_for_model(model).await?;

        use crate::runtime::compaction::COMPACTION_SYSTEM_PROMPT;

        ApiMethods::call_api_simple(
            &self.auth,
            &self.client,
            model,
            COMPACTION_SYSTEM_PROMPT,
            self.thinking_budget,
            self.reasoning_level(),
            &messages,
            self.api_retries,
            // Default options preserve this path's historical beta gating
            // and endpoint; only the session observability seams are wired
            // (Task 11) so compaction requests trace/persist like any other.
            // Deliberately the BASE context: an internal compaction request
            // must not consume a user's one-shot `/trace next` arm.
            &api::ApiOptions {
                trace: self.trace_ctx.clone(),
                telemetry: self.telemetry_writer.clone(),
                // The largest body Synaps sends: the whole transcript.
                request_compression: self.request_compression.clone(),
                ..api::ApiOptions::default()
            },
        )
        .await
    }

    /// Run a prompt as a cancellable stream of [`StreamEvent`]s. Convenience wrapper
    /// around [`run_stream_with_messages`] for single-turn usage.
    pub async fn run_stream(
        &self,
        prompt: String,
        cancel: CancellationToken,
    ) -> Pin<Box<dyn Stream<Item = StreamEvent> + Send>> {
        self.run_stream_with_messages(
            vec![std::sync::Arc::new(
                json!({"role": "user", "content": prompt}),
            )],
            cancel,
            None,
            None,
            false,
        )
        .await
    }

    /// Session-scoped "Allow all" latch for extension Confirm gates: the
    /// extension ids the user picked "Allow all from <ext>" for.
    pub fn session_allow_all(&self) -> &std::sync::Arc<crate::tools::SessionConfirmLatch> {
        &self.session_allow_all
    }

    /// Bind a worker to the calling session, not the process-wide host.
    pub(crate) fn inherit_session_allow_all(
        &mut self,
        parent: Option<&std::sync::Arc<crate::tools::SessionConfirmLatch>>,
    ) {
        if let Some(parent) = parent {
            self.session_allow_all = parent.clone();
        }
    }

    /// Run a multi-turn conversation as a cancellable stream of [`StreamEvent`]s.
    /// This is the main entry point for chat UIs and agents. Handles tool execution,
    /// API retries, and dynamic tool registration (MCP) internally.
    pub async fn run_stream_with_messages(
        &self,
        messages: Vec<crate::SharedMessage>,
        cancel: CancellationToken,
        steering_rx: Option<mpsc::UnboundedReceiver<String>>,
        secret_prompt: Option<crate::tools::SecretPromptHandle>,
        auto_approve_confirms: bool,
    ) -> Pin<Box<dyn Stream<Item = StreamEvent> + Send>> {
        self.run_stream_tracked(
            messages,
            cancel,
            steering_rx,
            secret_prompt,
            auto_approve_confirms,
            TurnCompletion::new(),
        )
        .await
    }

    /// [`Runtime::run_stream_with_messages`] that also reports, through
    /// `completion`, whether the turn reached its normal end (see
    /// [`TurnCompletion`]).
    pub async fn run_stream_tracked(
        &self,
        messages: Vec<crate::SharedMessage>,
        cancel: CancellationToken,
        steering_rx: Option<mpsc::UnboundedReceiver<String>>,
        secret_prompt: Option<crate::tools::SecretPromptHandle>,
        auto_approve_confirms: bool,
        completion: TurnCompletion,
    ) -> Pin<Box<dyn Stream<Item = StreamEvent> + Send>> {
        // CP-11 fix-2 (A): the caller-facing boundary is BOUNDED. The
        // internal producer keeps an unbounded sender for API stability;
        // the relay drains it eagerly, enforces the fixed preview-delta
        // retention budget, and cancels the turn when the caller stream
        // is dropped (releasing provider tasks).
        let (tx, internal_rx) = mpsc::unbounded_channel();
        let bounded_rx =
            crate::runtime::relay::spawn_bounded_stream_relay(internal_rx, cancel.clone());

        // One correlation ID per turn: carried by the typed terminal outcome
        // (spec §5.2) so every frontend can tie the failure to trace lines.
        let turn_correlation_id = agent_core::next_turn_correlation_id();

        if self
            .continuation
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .durability_blocked
        {
            let error = RuntimeError::Session(
                "context head save is unresolved; reload the session before further inference"
                    .into(),
            );
            let _ = tx.send(StreamEvent::Session(SessionEvent::Error(
                helpers::turn_error_for(&error, &turn_correlation_id),
            )));
            let _ = tx.send(StreamEvent::Session(SessionEvent::Done));
            return Box::pin(tokio_stream::wrappers::ReceiverStream::new(bounded_rx));
        }

        if let Err(error) = self.validate_request_preflight().await {
            let _ = tx.send(StreamEvent::Session(SessionEvent::Error(
                helpers::turn_error_for(&error, &turn_correlation_id),
            )));
            let _ = tx.send(StreamEvent::Session(SessionEvent::Done));
            return Box::pin(tokio_stream::wrappers::ReceiverStream::new(bounded_rx));
        }

        let anthropic_execution_plan = match self.authorized_anthropic_plan().await {
            Ok(plan) => plan,
            Err(error) => {
                let _ = tx.send(StreamEvent::Session(SessionEvent::Error(
                    helpers::turn_error_for(&error, &turn_correlation_id),
                )));
                let _ = tx.send(StreamEvent::Session(SessionEvent::Done));
                return Box::pin(tokio_stream::wrappers::ReceiverStream::new(bounded_rx));
            }
        };

        // Refresh OAuth token if expired after capability preflight.
        if let Err(e) = self.refresh_if_needed().await {
            let _ = tx.send(StreamEvent::Session(SessionEvent::Error(
                helpers::turn_error_for(&e, &turn_correlation_id),
            )));
            let _ = tx.send(StreamEvent::Session(SessionEvent::Done));
            return Box::pin(tokio_stream::wrappers::ReceiverStream::new(bounded_rx));
        }

        // Clone the Arc, not the whole Runtime — the spawned task shares the
        // same AuthState so mid-loop token refreshes are visible immediately.
        let auth = Arc::clone(&self.auth);
        let client = self.client.clone();
        let credential_source = self.credential_source.clone();
        let token_cache = self.token_cache.clone();
        let model = self.model.clone();
        let tools = self.tools.clone();
        let system_prompt = self.effective_system_prompt().await;
        let thinking_budget = self.thinking_budget;
        let reasoning_level = self.reasoning_level();
        let max_tool_output = self.max_tool_output;
        let bash_timeout = self.bash_timeout;
        let bash_max_timeout = self.bash_max_timeout;
        let subagent_timeout = self.subagent_timeout;
        let api_retries = self.api_retries;
        let refusal_retries = self.refusal_retries;
        let session_manager = self.session_manager.clone();
        // Opt into the 1M-context beta header only when the user explicitly
        // requested 1M (via context_window setting). Default 200k matches
        // Anthropic's claude-code default and gives smarter inference.
        let subagent_registry = self.subagent_registry.clone();
        // Extra Arc clone for the reaper hook — session takes ownership of the
        // original clone above; this one is captured separately by the spawn closure.
        let reaper_registry = Arc::clone(&subagent_registry);
        let reaper_orchestration = self.orchestration.clone();
        let event_queue = self.event_queue.clone();
        let options = api::ApiOptions {
            // Placeholder: the stream loop installs the tool context its
            // own tools are made from before the first provider round.
            tool_context: None,
            use_1m_context: self.context_window_override == Some(1_000_000),
            cache_ttl: self.cache_ttl,
            ttl_downgrade_notified: self.ttl_downgrade_notified.clone(),
            saw_1h_honored: self.saw_1h_honored.clone(),
            credential_source: self.credential_source.clone(),
            token_cache: self.token_cache.clone(),
            anthropic_base_url: None,
            anthropic_stream_client: self.anthropic_stream_client.clone(),
            request_compression: self.request_compression.clone(),
            anthropic_execution_plan,
            codex_request_role: self.codex_request_role(),
            thinking_display: self.wire_thinking_display(),
            // Placeholder: the stream loop sets it per round once it has
            // applied request-local image retention.
            retention_frontier: None,
            trace: self.effective_trace_context(),
            request_correlation: None,
            suppress_stream_deltas: false,
            telemetry: self.request_telemetry(),
            // Threads the runtime-scoped gate identity into extension-
            // provider interior tool loops (Task 16).
            tool_session_id: Some(self.host_tool_session.clone()),
            // Placeholder: the stream loop installs its RETAINED shared
            // session tool set handle and request schema projection before
            // the first provider round.
            session_tool_set: None,
            request_tools_schema: None,
            anthropic_tool_split: None,
            usage_counters: None,
            // The foreground turn's accepted requests are what keep-warm
            // schedules from.
            cache_anchor: Some(self.cache_anchor.clone()),
            keepwarm: None,
            // The foreground chain: the only requests that thread
            // `diagnostics.previous_message_id`.
            cache_diagnosis: self.cache_diagnosis(),
            context_rollover: None,
            // Placeholder: the stream loop installs one per provider round.
            round_signal: None,
        };

        let session = crate::runtime::stream::StreamSession {
            memory_backend: self.memory_backend.clone(),
            turn_completion: completion,
            context_window: self.context_window(),
            continuation: self.continuation.clone(),
            worker_context: self.worker_context.clone(),
            auth,
            client,
            credential_source,
            token_cache,
            options,
            api_retries,
            refusal_retries,
            model,
            tools,
            system_prompt,
            thinking_budget,
            reasoning_level,
            tx: tx.clone(),
            cancel,
            steering_rx,
            max_tool_output,
            tool_spill: self.tool_spill(),
            read_dedupe: self.read_dedupe,
            tool_keepwarm: self.tool_keepwarm.then(|| tool_keepwarm::ToolKeepWarm::new(
                self.trace_ctx.clone(),
                tool_keepwarm::interval_override(),
            )),
            bash_timeout,
            bash_max_timeout,
            subagent_timeout,
            session_manager,
            subagent_registry,
            event_queue,
            secret_prompt,
            hook_bus: self.hook_bus.clone(),
            session_id: self.session_id.clone(),
            cwd: self.cwd.clone(),
            env: self.env.clone(),
            env_stripped: self.env_stripped.clone(),
            env_warned: self.env_warned.clone(),
            auto_approve_confirms,
            session_allow_all: self.session_allow_all.clone(),
            telemetry_level: self.telemetry_level,
            orchestration: self.orchestration.clone(),
            delegation_parent: self.delegation_parent.clone(),
            turn_correlation_id: turn_correlation_id.clone(),
            progressive_tool_disclosure: self.progressive_tool_disclosure,
            tools_always_loaded: Arc::clone(&self.tools_always_loaded),
            deferred_tool_loading: self.deferred_tool_loading,
            activation_confirm: self.activation_confirm,
            tool_session_id: self.host_tool_session.clone(),
            retained_tool_set: std::sync::Arc::clone(&self.retained_tool_set),
            pending_tool_restore: std::sync::Arc::clone(&self.pending_tool_restore),
            mcp_runtime: self.mcp_runtime.clone(),
            mcp_session_scope: self.mcp_session_scope.clone(),
            extension_runtime: self.extension_runtime.clone(),
            extension_session_scope: self.extension_session_scope.clone(),
            turn_budget: self.turn_budget.clone(),
            owner_steer_budget: self.owner_steer_budget.clone(),
            message_clock: self.message_clock.clone(),
        };

        tokio::spawn(async move {
            if let Err(e) = StreamMethods::run_stream_internal(session, messages).await {
                let _ = tx.send(StreamEvent::Session(SessionEvent::Error(
                    helpers::turn_error_for(&e, &turn_correlation_id),
                )));
            }
            // Engine-owned housekeeping: reap finished subagent handles before
            // signalling Done.  Runs on the tokio thread pool — no public sync
            // caller becomes async.  Poison-safe via reap_finished internals.
            crate::runtime::subagent::reap_finished(
                &reaper_registry,
                reaper_orchestration.as_deref(),
            );
            agent_core::core::memstat::log_turn_memory();
            let _ = tx.send(StreamEvent::Session(SessionEvent::Done));
        });

        Box::pin(tokio_stream::wrappers::ReceiverStream::new(bounded_rx))
    }
}

impl Clone for Runtime {
    fn clone(&self) -> Self {
        Self {
            client: self.client.clone(),
            anthropic_stream_client: self.anthropic_stream_client.clone(),
            auth: Arc::clone(&self.auth),
            model: self.model.clone(),
            tools: self.tools.clone(),
            system_prompt: self.system_prompt.clone(),
            effective_prompt: self.effective_prompt.clone(),
            prompt_generation: self.prompt_generation,
            prompt_reload_source: self.prompt_reload_source.clone(),
            thinking_budget: self.thinking_budget,
            named_level: self.named_level,
            explicit_reasoning: self.explicit_reasoning,
            codex_request_role: self.codex_request_role,
            context_window_override: self.context_window_override,
            continuation: Arc::clone(&self.continuation),
            worker_context: self.worker_context.clone(),
            compaction_model: self.compaction_model.clone(),
            compaction_mode: self.compaction_mode,
            compaction_exclusions: self.compaction_exclusions.clone(),
            remote_summarization_attempts: Arc::clone(&self.remote_summarization_attempts),
            subagent_registry: self.subagent_registry.clone(),
            orchestration: self.orchestration.clone(),
            event_queue: self.event_queue.clone(),
            max_tool_output: self.max_tool_output,
            tool_spill_policy: self.tool_spill_policy,
            tool_spill_root: self.tool_spill_root.clone(),
            tool_spill_stats: Arc::clone(&self.tool_spill_stats),
            read_dedupe: self.read_dedupe,
            thinking_display: self.thinking_display,
            tool_keepwarm: self.tool_keepwarm,
            bash_timeout: self.bash_timeout,
            bash_max_timeout: self.bash_max_timeout,
            subagent_timeout: self.subagent_timeout,
            api_retries: self.api_retries,
            refusal_retries: self.refusal_retries,
            telemetry_level: self.telemetry_level,
            telemetry_log_generations: self.telemetry_log_generations,
            // Shared session observability: clones (subagents) enqueue into
            // the SAME bounded writer and trace context — one worker and one
            // set of session-scoped context IDs per session.
            telemetry_writer: self.telemetry_writer.clone(),
            trace_ctx: self.trace_ctx.clone(),
            trace_controls: Arc::clone(&self.trace_controls),
            memory_backend: self.memory_backend.clone(),
            memory_bound: self.memory_bound,
            one_shot_trace_writer: Arc::clone(&self.one_shot_trace_writer),
            capture_dir: self.capture_dir.clone(),
            cache_diagnostics: self.cache_diagnostics,
            cache_ttl: self.cache_ttl,
            request_compression: self.request_compression.clone(),
            // Subagents are their own session — fresh latches so a downgrade
            // in the subagent's chain surfaces its own (single) notice and
            // honored-state isn't inherited from the parent.
            ttl_downgrade_notified: std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)),
            saw_1h_honored: std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)),
            cache_anchor: Default::default(),
            // Subagents start their own request chain — inheriting the parent's
            // last msg_id would produce bogus `messages_changed` diagnostics.
            last_msg_id: Arc::new(Mutex::new(None)),
            cache_diag_rejected: Arc::clone(&self.cache_diag_rejected),
            session_manager: self.session_manager.clone(),
            hook_bus: self.hook_bus.clone(),
            reaper_handle: None, // Cloned runtimes don't own the reaper
            reaper_cancel: None, // Cloned runtimes don't own the reaper
            credential_source: self.credential_source.clone(),
            token_cache: self.token_cache.clone(), // shares the same cache (Arc inside)
            trusted_worker_models: self.trusted_worker_models.clone(),
            subagent_limits: self.subagent_limits,
            progressive_tool_disclosure: self.progressive_tool_disclosure,
            tools_always_loaded: Arc::clone(&self.tools_always_loaded),
            deferred_tool_loading: self.deferred_tool_loading,
            activation_confirm: self.activation_confirm,
            delegation_parent: self.delegation_parent.clone(),
            session_allow_all: self.session_allow_all.clone(),
            mcp_runtime: self.mcp_runtime.clone(),
            // Clones SHARE the durable session scope: dropping one clone or
            // one stream can never kill a sibling's leases.
            mcp_session_scope: self.mcp_session_scope.clone(),
            extension_runtime: self.extension_runtime.clone(),
            // Same durable shared-scope rule for extension leases.
            extension_session_scope: self.extension_session_scope.clone(),
            turn_budget: self.turn_budget.clone(),
            owner_steer_budget: self.owner_steer_budget.clone(),
            message_clock: self.message_clock.clone(),
            // Clones share the live tool registry, so they share the SAME
            // host tool session (matching existing shared-session behavior);
            // independently constructed runtimes mint fresh identities and
            // can never share session grants.
            host_tool_session: self.host_tool_session.clone(),
            tool_catalog_prompt: Arc::clone(&self.tool_catalog_prompt),
            retained_tool_set: std::sync::Arc::clone(&self.retained_tool_set),
            pending_tool_restore: std::sync::Arc::clone(&self.pending_tool_restore),
            // Clones serve the same conversation.
            session_id: self.session_id.clone(),
            cwd: self.cwd.clone(),
            env: self.env.clone(),
            env_stripped: self.env_stripped.clone(),
            env_warned: self.env_warned.clone(),
        }
    }
}

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

    /// Task 16 session identity semantics: clones of one Runtime share the
    /// live tool registry, so they share the same host tool session;
    /// independently constructed runtimes mint fresh identities and can
    /// never share session grants.
    #[test]
    fn host_tool_session_shared_by_clones_fresh_per_runtime() {
        let rt = Runtime::new_headless();
        let clone = rt.clone();
        assert_eq!(
            rt.host_tool_session_id(),
            clone.host_tool_session_id(),
            "clones share the live registry and therefore the host session"
        );

        let other = Runtime::new_headless();
        assert_ne!(
            rt.host_tool_session_id(),
            other.host_tool_session_id(),
            "independently constructed runtimes must never share a session"
        );
    }

    /// Activations handed to a runtime before its first turn are what it
    /// hands on (a second reload, or a park, before any turn loses
    /// nothing); clones share them; another runtime does not see them.
    #[test]
    fn handed_activations_are_carried_on_before_the_first_turn() {
        use crate::tools::activation::{CarriedActivation, CarriedProvenance};
        let probe = |name: &str| CarriedActivation {
            tool_id: format!("builtin:{name}"),
            schema_digest: "ab".repeat(32),
            provenance: CarriedProvenance::BuiltinRuntime,
        };
        let rt = Runtime::new_headless();
        assert!(rt.carried_tool_activations().is_empty());
        rt.restore_tool_activations(vec![probe("subagent_start")]);
        rt.restore_tool_activations(Vec::new());
        rt.restore_tool_activations(vec![probe("subagent_start"), probe("subagent_status")]);
        assert_eq!(
            rt.carried_tool_activations(),
            vec![probe("subagent_start"), probe("subagent_status")],
            "merged without duplicates; an empty hand-off clears nothing"
        );
        assert_eq!(rt.clone().carried_tool_activations().len(), 2);
        assert!(Runtime::new_headless().carried_tool_activations().is_empty());
    }

    /// Task 11 config rule (documented until Task 12 adds explicit trace
    /// config): telemetry `basic`/`full` enables the shared session writer
    /// sink — legacy telemetry AND metadata-only trace persistence; `off`
    /// disables both (trace context reverts to the no-op sink).
    #[test]
    fn telemetry_level_gates_shared_observability_sink() {
        let mut rt = Runtime::new_headless();
        assert!(!rt.trace_context().enabled(), "off by default");
        assert!(rt.telemetry_writer().is_none());

        rt.set_telemetry_level(crate::runtime::telemetry::TelemetryLevel::Basic);
        assert!(rt.trace_context().enabled(), "basic enables the trace sink");
        assert!(rt.telemetry_writer().is_some());

        // Same level again: the writer/session context is not recreated.
        let ctx_before = format!("{:?}", rt.trace_context());
        rt.set_telemetry_level(crate::runtime::telemetry::TelemetryLevel::Full);
        assert_eq!(format!("{:?}", rt.trace_context()), ctx_before);

        // Clones (subagents) share the same writer sink + session context.
        let clone = rt.clone();
        assert!(clone.trace_context().enabled());
        assert_eq!(format!("{:?}", clone.trace_context()), ctx_before);
        assert!(clone.telemetry_writer().is_some());

        rt.set_telemetry_level(crate::runtime::telemetry::TelemetryLevel::Off);
        assert!(!rt.trace_context().enabled(), "off disables both");
        assert!(rt.telemetry_writer().is_none());
        assert!(rt
            .shutdown_observability(std::time::Duration::ZERO)
            .is_none());
    }

    /// Owner-epilogue contract (Task 11): with telemetry Off the async
    /// flush is a `None` no-op and idempotent; with a writer installed, a
    /// chat-like epilogue (final record queued, then bounded flush) must
    /// persist the record; and a slow writer must return within the budget
    /// (TimedOut) instead of blocking exit.
    #[tokio::test]
    async fn shutdown_observability_async_epilogue_contract() {
        // Off → None, twice (idempotent no-op, no writer ever created).
        let rt = Runtime::new_headless();
        let budget = crate::runtime::telemetry::DEFAULT_SHUTDOWN_FLUSH_TIMEOUT;
        assert!(rt.shutdown_observability_async(budget).await.is_none());
        assert!(rt.shutdown_observability_async(budget).await.is_none());

        // Chat-like shutdown: the final queued record is persisted.
        let tmp = tempfile::tempdir().unwrap();
        let telemetry_path = tmp.path().join("synaps/api-log.jsonl");
        let mut rt = Runtime::new_headless();
        let writer = crate::runtime::telemetry::TelemetryWriter::new(
            crate::runtime::telemetry::WriterOptions {
                telemetry_path: Some(telemetry_path.clone()),
                trace_path: Some(tmp.path().join("synaps/request-trace.jsonl")),
                ..Default::default()
            },
        );
        rt.install_observability_for_tests(writer.clone());
        writer.enqueue_telemetry(crate::runtime::telemetry::TelemetryRecord {
            ts: 7,
            model: "claude-sonnet-4-6".to_string(),
            ..Default::default()
        });
        let outcome = rt
            .shutdown_observability_async(budget)
            .await
            .expect("writer installed");
        assert!(outcome.is_flushed());
        assert_eq!(outcome.stats().written, 1);
        assert_eq!(
            std::fs::read_to_string(&telemetry_path)
                .unwrap()
                .lines()
                .count(),
            1,
            "queued final record must be on disk after the epilogue flush"
        );
        // Idempotent second call: intake already closed, still Flushed.
        assert!(rt
            .shutdown_observability_async(budget)
            .await
            .expect("writer still installed")
            .is_flushed());

        // Slow writer: the epilogue returns within its bounded budget.
        let mut rt = Runtime::new_headless();
        let slow = crate::runtime::telemetry::TelemetryWriter::new(
            crate::runtime::telemetry::WriterOptions {
                telemetry_path: Some(tmp.path().join("slow/api-log.jsonl")),
                trace_path: Some(tmp.path().join("slow/request-trace.jsonl")),
                write_delay: Some(std::time::Duration::from_millis(300)),
                ..Default::default()
            },
        );
        rt.install_observability_for_tests(slow.clone());
        for _ in 0..10 {
            slow.enqueue_telemetry(crate::runtime::telemetry::TelemetryRecord::default());
        }
        let start = std::time::Instant::now();
        let outcome = rt
            .shutdown_observability_async(std::time::Duration::from_millis(200))
            .await
            .expect("writer installed");
        assert!(!outcome.is_flushed(), "10×300ms cannot drain in 200ms");
        // Generous ceiling (no timing fragility): well under the 3s a full
        // drain would need, proving the deadline was honored.
        assert!(
            start.elapsed() < std::time::Duration::from_millis(1500),
            "bounded epilogue must not block exit: {:?}",
            start.elapsed()
        );
    }

    /// Task 12 fix: an armed one-shot ephemeral writer (telemetry Off +
    /// `/trace next`) is retained by the runtime and drained by the same
    /// shutdown epilogue as the session writer, so the armed record cannot
    /// be lost to an exit racing the background worker.
    #[tokio::test]
    async fn one_shot_ephemeral_writer_is_retained_and_drained_at_shutdown() {
        let rt = Runtime::new_headless();
        assert!(!rt.trace_context().enabled(), "telemetry off");
        rt.trace_arm_next(false);
        let armed = rt.effective_trace_context();
        assert!(armed.enabled(), "armed ephemeral context traces");
        // No session writer, but the epilogue still finds (and drains) the
        // retained one-shot writer.
        let budget = crate::runtime::telemetry::DEFAULT_SHUTDOWN_FLUSH_TIMEOUT;
        let outcome = rt
            .shutdown_observability_async(budget)
            .await
            .expect("retained one-shot writer must be drained by the epilogue");
        assert!(outcome.is_flushed());
        // Consumed: a second epilogue call finds nothing.
        assert!(rt.shutdown_observability_async(budget).await.is_none());
    }

    /// Task 12 operational expiry: stale content-capture bundles are
    /// physically removed at Runtime startup and by BOTH shutdown
    /// epilogues (sync and async), not only on trace interactions.
    /// Combined into one serialized test: all three paths resolve the
    /// capture dir through `SYNAPS_BASE_DIR`.
    #[tokio::test]
    #[serial_test::serial(synaps_base_dir)]
    async fn startup_and_shutdown_paths_sweep_expired_captures() {
        struct BaseDirGuard(Option<String>);
        impl Drop for BaseDirGuard {
            fn drop(&mut self) {
                match self.0.take() {
                    Some(old) => std::env::set_var("SYNAPS_BASE_DIR", old),
                    None => std::env::remove_var("SYNAPS_BASE_DIR"),
                }
            }
        }
        let tmp = tempfile::tempdir().unwrap();
        let guard = BaseDirGuard(std::env::var("SYNAPS_BASE_DIR").ok());
        std::env::set_var("SYNAPS_BASE_DIR", tmp.path());

        let cap_dir = trace::default_capture_dir();
        assert!(
            cap_dir.starts_with(tmp.path()),
            "capture dir must be private to the test"
        );
        let stale_id = trace::TraceId::new("req-stale").unwrap();
        let plant_stale = || {
            agent_core::core::private_fs::ensure_private_dir(&cap_dir).unwrap();
            let stale = trace::controls::ContentCaptureBundle {
                schema: trace::CONTENT_CAPTURE_SCHEMA.to_string(),
                request_id: stale_id.clone(),
                created_unix_ms: 1_000,
                expires_unix_ms: 2_000, // long past
                redacted: true,
                over_budget: false,
                body: Some(serde_json::json!({"old": true})),
            };
            let path = trace::controls::capture_path(&cap_dir, &stale_id);
            std::fs::write(&path, serde_json::to_vec(&stale).unwrap()).unwrap();
            path
        };

        // 1) Runtime startup sweeps.
        let stale_path = plant_stale();
        let rt = Runtime::new().await.expect("runtime constructs");
        assert!(
            !stale_path.exists(),
            "Runtime startup must physically remove expired capture bundles"
        );

        // 2) Sync shutdown epilogue sweeps (telemetry off → None outcome).
        let stale_path = plant_stale();
        let _ = rt.shutdown_observability(std::time::Duration::ZERO);
        assert!(
            !stale_path.exists(),
            "sync shutdown epilogue must remove expired capture bundles"
        );

        // 3) Async shutdown epilogue sweeps.
        let stale_path = plant_stale();
        let _ = rt
            .shutdown_observability_async(std::time::Duration::from_millis(100))
            .await;
        assert!(
            !stale_path.exists(),
            "async shutdown epilogue must remove expired capture bundles"
        );
        drop(guard);
    }

    /// fix1 I2b: the capture root binds at CONSTRUCTION. Ambient
    /// SYNAPS_BASE_DIR churn after construction must not redirect the
    /// sweep — each runtime keeps sweeping its own root (concurrent
    /// roots), so a parallel test flipping env can never race the epilogue.
    #[tokio::test]
    #[serial_test::serial(synaps_base_dir)]
    async fn capture_root_binds_at_construction_and_survives_env_churn() {
        let base_a = tempfile::tempdir().unwrap();
        let base_b = tempfile::tempdir().unwrap();
        let old = std::env::var("SYNAPS_BASE_DIR").ok();

        std::env::set_var("SYNAPS_BASE_DIR", base_a.path());
        let rt_a = Runtime::new().await.expect("runtime A");
        std::env::set_var("SYNAPS_BASE_DIR", base_b.path());
        let rt_b = Runtime::new().await.expect("runtime B");

        let cap_a = base_a.path().join("trace").join("capture");
        let cap_b = base_b.path().join("trace").join("capture");
        let plant = |dir: &std::path::Path, name: &str| {
            agent_core::core::private_fs::ensure_private_dir(dir).unwrap();
            let id = trace::TraceId::new(name).unwrap();
            let stale = trace::controls::ContentCaptureBundle {
                schema: trace::CONTENT_CAPTURE_SCHEMA.to_string(),
                request_id: id.clone(),
                created_unix_ms: 1_000,
                expires_unix_ms: 2_000,
                redacted: true,
                over_budget: false,
                body: Some(serde_json::json!({"old": true})),
            };
            let path = trace::controls::capture_path(dir, &id);
            std::fs::write(&path, serde_json::to_vec(&stale).unwrap()).unwrap();
            path
        };

        // Point ambient env somewhere else entirely — the bound roots must win.
        let decoy = tempfile::tempdir().unwrap();
        std::env::set_var("SYNAPS_BASE_DIR", decoy.path());

        let stale_a = plant(&cap_a, "req-stale-a");
        let stale_b = plant(&cap_b, "req-stale-b");
        let _ = rt_a.shutdown_observability(std::time::Duration::ZERO);
        assert!(
            !stale_a.exists(),
            "runtime A must sweep the root bound at ITS construction"
        );
        assert!(
            stale_b.exists(),
            "runtime A must never sweep another runtime's root"
        );
        let _ = rt_b
            .shutdown_observability_async(std::time::Duration::from_millis(100))
            .await;
        assert!(
            !stale_b.exists(),
            "runtime B must sweep the root bound at ITS construction"
        );

        match old {
            Some(v) => std::env::set_var("SYNAPS_BASE_DIR", v),
            None => std::env::remove_var("SYNAPS_BASE_DIR"),
        }
    }

    #[test]
    fn reload_rejects_policy_change_without_partial_apply() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("prompt.yaml");
        let manifest = |total: &str, content: &str| {
            format!(
            "schema: synaps-prompt/1\nkernel: kernel\nmodules:\n  - id: kernel\n    version: v1\n    source: builtin\n    priority: 0\n    selectors: {{}}\n    mutability: mutable_guidance\n    content: {content}\npolicies:\n  delegation:\n    mode: enforced\n    allowed_models: [anthropic/claude-sonnet-4-6]\n    max_concurrent_workers: 1\n    max_total_workers: {total}\n"
        )
        };
        std::fs::write(&path, manifest("1", "before")).unwrap();
        let parsed =
            agent_core::prompt::PromptManifest::parse(&std::fs::read_to_string(&path).unwrap())
                .unwrap();
        let model =
            agent_core::prompt::QualifiedModelId::parse("anthropic/claude-sonnet-4-6").unwrap();
        let context = agent_core::prompt::SelectionContext::new(model.clone(), None).unwrap();
        let stack = agent_core::prompt::compile_prompt_stack(
            &parsed,
            &parsed.registry(path.parent()).unwrap(),
            &context,
            None,
        )
        .unwrap();
        let catalog = crate::orchestration::OrchestrationRuntime::trusted_catalog(
            &model,
            parsed.delegation_catalog_candidates(),
        )
        .unwrap();
        let digest = parsed
            .delegation_policy(model, &catalog)
            .unwrap()
            .map(|p| p.digest());
        let mut runtime = Runtime::new_headless();
        runtime.apply_prompt_stack(stack).unwrap();
        runtime.retain_prompt_reload_source(path.clone(), context, None, digest.clone());
        let generation = runtime.prompt_generation();
        let composed = runtime.effective_prompt().unwrap().composed().to_owned();

        std::fs::write(&path, manifest("2", "after")).unwrap();
        let error = runtime.reload_prompt().unwrap_err().to_string();
        assert!(error.contains("cannot safely change delegation policy"));
        assert_eq!(runtime.prompt_generation(), generation);
        assert_eq!(runtime.effective_prompt().unwrap().composed(), composed);
        assert_eq!(
            runtime
                .prompt_reload_source
                .as_ref()
                .unwrap()
                .delegation_policy_digest,
            digest
        );
    }

    #[test]
    fn worker_install_does_not_replay_process_global_favorite_grants() {
        let foreground =
            crate::orchestration::canonical_foreground_identity("anthropic/claude-fable-5")
                .unwrap();
        let session = Arc::new(
            crate::orchestration::OrchestrationRuntime::baseline(foreground, 4, 8).unwrap(),
        );
        let favorite = "openai-codex/gpt-5.6-sol";
        assert!(
            session.preflight(favorite).is_err(),
            "precondition: the parent session did not grant this identity"
        );

        let mut worker = Runtime::new_headless();
        worker.apply_config(&crate::config::SynapsConfig {
            favorite_models: vec![favorite.to_owned()],
            ..Default::default()
        });
        worker.install_worker_orchestration(Arc::clone(&session));

        assert!(
            session.preflight(favorite).is_err(),
            "constructing a child must not mint/replay a favorite-model grant"
        );
        assert!(!session.effective_choices().contains(&favorite.to_owned()));
    }

    /// Tool output virtualisation is off on a bare runtime (embedding and
    /// test path), follows the config once applied, and spills a foreground
    /// session into `<base>/spill/<conversation id>` so retention can remove
    /// it with the session.
    #[test]
    #[serial_test::serial(synaps_base_dir)]
    fn tool_spill_follows_config_and_is_keyed_by_the_conversation() {
        let base = crate::test_env::BaseDirGuard::new();
        let mut runtime = Runtime::new_headless();
        assert!(!runtime.tool_spill_policy().is_enabled());
        assert!(!runtime.tool_spill().is_enabled());
        runtime.apply_config(&crate::config::SynapsConfig::default());
        assert_eq!(
            runtime.tool_spill_policy(),
            tool_spill::ToolSpillPolicy {
                threshold_bytes: 12 * 1024,
                head_bytes: 4 * 1024,
                tail_bytes: 3 * 1024,
            }
        );
        runtime.set_session_id(Some("20261029-120000-ab12".into()));
        assert_eq!(
            runtime.tool_spill().dir(),
            Some(base.path().join("spill").join("20261029-120000-ab12").as_path())
        );
        // Clones (the same session) share the counters.
        let clone = runtime.clone();
        let text = "line\n".repeat(4_000);
        let virtualised = clone
            .tool_spill()
            .virtualise("toolu_1", &tool_spill::SpillTarget::Spill, &text)
            .expect("over the threshold");
        assert_eq!(runtime.tool_spill_totals().virtualised, 1);
        assert_eq!(
            runtime.tool_spill_totals().kept_out_bytes(),
            (text.len() - virtualised.preview.len()) as u64
        );
        runtime.apply_config(&agent_core::config::load_config_from_str(
            "tool_output_spill_bytes = 0\n",
        ));
        assert!(!runtime.tool_spill().is_enabled(), "0 turns it off");
    }

    #[test]
    fn read_dedupe_is_off_until_configured_then_follows_config() {
        let mut runtime = Runtime::new_headless();
        assert!(!runtime.read_dedupe(), "off until a config is applied");
        runtime.apply_config(&agent_core::config::load_config_from_str(""));
        assert!(runtime.read_dedupe(), "the config default is on");
        assert!(runtime.clone().read_dedupe(), "clones keep it");
        runtime.apply_config(&agent_core::config::load_config_from_str("read_dedupe = off\n"));
        assert!(!runtime.read_dedupe());
    }

    /// `thinking_display` reaches the foreground's requests; a worker (and
    /// an internal request) sends summarized whatever the config says, even
    /// one cloned from an omitted foreground.
    #[test]
    fn thinking_display_is_foreground_only() {
        use crate::config::ThinkingDisplay;
        use crate::runtime::openai::catalog::CodexRequestRole;
        let mut runtime = Runtime::new_headless();
        assert_eq!(runtime.wire_thinking_display(), ThinkingDisplay::Summarized);
        runtime.apply_config(&agent_core::config::load_config_from_str("thinking_display = omitted\n"));
        assert_eq!(runtime.thinking_display(), ThinkingDisplay::Omitted);
        assert_eq!(runtime.wire_thinking_display(), ThinkingDisplay::Omitted);
        assert_eq!(runtime.clone().wire_thinking_display(), ThinkingDisplay::Omitted);

        // What `apply_subagent_runtime_policy` does to every worker runtime.
        let mut worker = runtime.clone();
        worker.set_codex_request_role(CodexRequestRole::Worker);
        assert_eq!(worker.thinking_display(), ThinkingDisplay::Omitted, "config kept");
        assert_eq!(worker.wire_thinking_display(), ThinkingDisplay::Summarized, "wire forced");
        let mut internal = runtime.clone();
        internal.set_codex_request_role(CodexRequestRole::Internal);
        assert_eq!(internal.wire_thinking_display(), ThinkingDisplay::Summarized);

        runtime.apply_config(&agent_core::config::load_config_from_str(""));
        assert_eq!(runtime.wire_thinking_display(), ThinkingDisplay::Summarized, "default");
    }

    #[test]
    fn tool_keepwarm_is_off_until_configured_then_follows_config() {
        let mut runtime = Runtime::new_headless();
        assert!(!runtime.tool_keepwarm(), "off until a config is applied");
        runtime.apply_config(&agent_core::config::load_config_from_str(""));
        assert!(runtime.tool_keepwarm(), "the config default is on");
        assert!(runtime.clone().tool_keepwarm(), "clones keep it");
        runtime.apply_config(&agent_core::config::load_config_from_str(
            "cache_keepwarm_tools = off\n",
        ));
        assert!(!runtime.tool_keepwarm());
    }

    #[test]
    fn host_install_may_apply_explicit_operator_favorites() {
        let mut runtime = Runtime::new_headless();
        let config = crate::config::SynapsConfig {
            favorite_models: vec![
                "openai-codex/gpt-5.6-luna".to_owned(),
                "anthropic/claude-opus-4-6".to_owned(),
                // Legacy favorite spelling must retain its existing compatibility.
                "claude/claude-fable-5".to_owned(),
                // Malformed persisted values must fail closed without bricking boot.
                "not-qualified".to_owned(),
            ],
            ..Default::default()
        };
        runtime.apply_config(&config);
        let foreground =
            agent_core::prompt::QualifiedModelId::parse("openai-codex/gpt-5.6-sol").unwrap();
        runtime.install_orchestration(std::sync::Arc::new(
            crate::orchestration::OrchestrationRuntime::baseline(foreground, 3, 8).unwrap(),
        ));

        let orchestration = runtime.orchestration().unwrap();
        for trusted in [
            "openai-codex/gpt-5.6-sol",
            "openai-codex/gpt-5.6-luna",
            "anthropic/claude-opus-4-6",
            "anthropic/claude-fable-5",
        ] {
            assert!(
                orchestration
                    .effective_choices()
                    .contains(&trusted.to_owned()),
                "missing configured worker choice {trusted}: {:?}",
                orchestration.effective_choices()
            );
        }
        let authorized = orchestration
            .resolve_and_authorize("sa_cross_provider", Some("anthropic/claude-opus-4-6"))
            .expect("an explicitly favorited cross-provider model must be authorized");
        assert_eq!(authorized.model.as_str(), "anthropic/claude-opus-4-6");
        assert!(!orchestration
            .effective_choices()
            .contains(&"not-qualified".to_owned()));
    }

    #[test]
    fn configured_worker_choices_survive_manifestless_foreground_changes() {
        let mut runtime = Runtime::new_headless();
        runtime.apply_config(&crate::config::SynapsConfig {
            favorite_models: vec!["anthropic/claude-opus-4-6".to_owned()],
            ..Default::default()
        });
        let foreground =
            agent_core::prompt::QualifiedModelId::parse("openai-codex/gpt-5.6-sol").unwrap();
        runtime.install_orchestration(std::sync::Arc::new(
            crate::orchestration::OrchestrationRuntime::baseline(foreground, 3, 8).unwrap(),
        ));

        runtime
            .try_set_model("xai-auth/grok-4.5-latest".to_owned())
            .unwrap();

        let orchestration = runtime.orchestration().unwrap();
        assert_eq!(orchestration.foreground_model(), "xai-auth/grok-4.5-latest");
        assert!(orchestration
            .effective_choices()
            .contains(&"anthropic/claude-opus-4-6".to_owned()));
        orchestration
            .resolve_and_authorize("sa_after_switch", Some("anthropic/claude-opus-4-6"))
            .expect("configured worker trust must survive policy replacement");
    }

    #[test]
    fn configured_subagent_limits_size_the_policy_rebuilt_on_model_change() {
        let mut runtime = Runtime::new_headless();
        assert_eq!(
            runtime.subagent_limits(),
            crate::config::SubagentLimits::default(),
            "defaults before any config"
        );
        runtime.apply_config(&crate::config::SynapsConfig {
            subagent: crate::config::SubagentLimits {
                max_concurrent: 24,
                max_total: 96,
                max_depth: 2,
            },
            ..Default::default()
        });
        let foreground =
            agent_core::prompt::QualifiedModelId::parse("openai-codex/gpt-5.6-sol").unwrap();
        runtime.install_orchestration(std::sync::Arc::new(
            crate::orchestration::OrchestrationRuntime::from_limits(
                foreground,
                runtime.subagent_limits(),
            )
            .unwrap(),
        ));
        runtime
            .try_set_model("xai-auth/grok-4.5-latest".to_owned())
            .unwrap();

        let orchestration = runtime.orchestration().unwrap();
        assert_eq!(orchestration.foreground_model(), "xai-auth/grok-4.5-latest");
        assert_eq!(
            orchestration.ultracode_readiness("xai-auth/grok-4.5-latest"),
            Ok((24, 96)),
            "the replacement policy uses the configured limits, not a hardcoded 8/64"
        );
        assert_eq!(orchestration.reserve_delegation("a", None), Ok(1));
        assert_eq!(orchestration.reserve_delegation("b", Some("a")), Ok(2));
        assert_eq!(
            orchestration.reserve_delegation("c", Some("b")),
            Err(crate::orchestration::DelegationTreeDenied::DepthLimit { max_depth: 2 })
        );
    }

    #[test]
    fn set_model_replaces_manifestless_orchestration_foreground_snapshot() {
        let mut runtime = Runtime::new_headless();
        let glm = agent_core::prompt::QualifiedModelId::parse("openrouter/z-ai/glm-5.2").unwrap();
        runtime.install_orchestration(std::sync::Arc::new(
            crate::orchestration::OrchestrationRuntime::baseline(glm, 3, 8).unwrap(),
        ));

        runtime.set_model("openrouter/deepseek/deepseek-v4-pro".to_owned());

        let orchestration = runtime.orchestration().unwrap();
        assert_eq!(
            orchestration.foreground_model(),
            "openrouter/deepseek/deepseek-v4-pro"
        );
        assert!(orchestration
            .effective_choices()
            .contains(&"openrouter/moonshotai/kimi-k2.7-code".to_owned()));
        assert!(orchestration
            .resolve_and_authorize("sa_kimi", Some("openrouter/moonshotai/kimi-k2.7-code"))
            .is_ok());
    }

    /// Regression ("model change blocked: trusted worker catalog
    /// unavailable"): a daemon-hosted session has no live Muse catalog in
    /// its process (the picker fetched it in the TUI process). Switching to
    /// a Muse id must not depend on that process-local cache.
    #[test]
    fn set_model_to_uncached_muse_id_rebuilds_orchestration() {
        let _guard = crate::runtime::openai::catalog::muse::test_cache_guard();
        crate::runtime::openai::catalog::capability_cache::replace_provider("muse", &[]);
        let mut runtime = Runtime::new_headless();
        let glm = agent_core::prompt::QualifiedModelId::parse("openrouter/z-ai/glm-5.2").unwrap();
        runtime.install_orchestration(std::sync::Arc::new(
            crate::orchestration::OrchestrationRuntime::baseline(glm, 3, 8).unwrap(),
        ));

        for model in ["muse/muse-wonder-1.0", "muse/muse-live-only-zz5"] {
            runtime
                .try_set_model(model.to_owned())
                .unwrap_or_else(|e| panic!("{model}: {e}"));
            assert_eq!(runtime.model(), model);
            assert_eq!(runtime.orchestration().unwrap().foreground_model(), model);
        }
        // Malformed ids still fail closed.
        assert!(runtime.try_set_model("muse/Bad Id".to_owned()).is_err());
        assert_eq!(runtime.model(), "muse/muse-live-only-zz5");
    }

    #[test]
    fn set_model_refuses_to_orphan_unreconciled_policy_workers() {
        let mut runtime = Runtime::new_headless();
        let glm = agent_core::prompt::QualifiedModelId::parse("openrouter/z-ai/glm-5.2").unwrap();
        let orchestration = std::sync::Arc::new(
            crate::orchestration::OrchestrationRuntime::baseline(glm, 3, 8).unwrap(),
        );
        orchestration
            .resolve_and_authorize("sa_active", None)
            .unwrap();
        runtime.install_orchestration(orchestration.clone());
        runtime.model = "openrouter/z-ai/glm-5.2".to_owned();

        let error = runtime
            .try_set_model("openrouter/deepseek/deepseek-v4-pro".to_owned())
            .unwrap_err();

        assert!(error.contains("collection/reconciliation"));
        assert_eq!(runtime.model(), "openrouter/z-ai/glm-5.2");
        assert!(std::sync::Arc::ptr_eq(
            runtime.orchestration().unwrap(),
            &orchestration
        ));
        assert!(orchestration.is_unreconciled("sa_active"));
    }

    #[test]
    fn set_model_keeps_active_typed_manifest_orchestration_unchanged() {
        let dir = tempfile::tempdir().unwrap();
        let manifest_path = dir.path().join("prompt.yaml");
        let glm = agent_core::prompt::QualifiedModelId::parse("openrouter/z-ai/glm-5.2").unwrap();
        let mut runtime = Runtime::new_headless();
        let orchestration = std::sync::Arc::new(
            crate::orchestration::OrchestrationRuntime::baseline(glm.clone(), 3, 8).unwrap(),
        );
        runtime.install_orchestration(orchestration.clone());
        runtime.retain_prompt_reload_source(
            manifest_path,
            agent_core::prompt::SelectionContext::new(glm, None).unwrap(),
            None,
            None,
        );

        runtime.set_model("openrouter/deepseek/deepseek-v4-pro".to_owned());

        assert!(std::sync::Arc::ptr_eq(
            runtime.orchestration().unwrap(),
            &orchestration
        ));
        assert_eq!(
            runtime.orchestration().unwrap().foreground_model(),
            "openrouter/z-ai/glm-5.2"
        );
    }

    #[test]
    fn set_model_preserves_qualified_provider_ids_and_bare_claude() {
        let cases = [
            (
                "github-copilot/claude-opus-4.8",
                "github-copilot/claude-opus-4.8",
            ),
            ("anthropic/claude-sonnet-4-6", "anthropic/claude-sonnet-4-6"),
            ("google/gemini-2.5-pro", "google/gemini-2.5-pro"),
            ("openai/gpt-5", "openai/gpt-5"),
            ("claude-opus-4-8", "claude-opus-4-8"),
        ];

        for (input, expected) in cases {
            let mut runtime = Runtime::new_headless();
            runtime.set_model(input.to_owned());
            assert_eq!(runtime.model(), expected, "input: {input}");
        }
    }

    #[test]
    fn set_model_strips_only_legacy_health_status_decoration() {
        let cases = [
            ("✅  339ms  groq/llama-3.3-70b", "groq/llama-3.3-70b"),
            (
                "✅  339ms  github-copilot/claude-opus-4.8",
                "github-copilot/claude-opus-4.8",
            ),
            (
                "⚠️  1200ms  anthropic/claude-sonnet-4-6",
                "anthropic/claude-sonnet-4-6",
            ),
        ];

        for (input, expected) in cases {
            let mut runtime = Runtime::new_headless();
            runtime.set_model(input.to_owned());
            assert_eq!(runtime.model(), expected, "input: {input}");
        }
    }

    #[tokio::test]
    async fn confirm_without_prompt_fails_closed() {
        let result = resolve_before_tool_call_result(
            crate::extensions::hooks::events::HookResult::confirm("Run deploy?"),
            None,
            false,
            None,
        )
        .await;

        assert!(matches!(
            result,
            crate::extensions::hooks::events::HookResult::Block { reason }
                if reason.contains("requires confirmation") && reason.contains("Run deploy?")
        ));
    }

    #[tokio::test]
    async fn modify_result_replaces_tool_input() {
        let result = resolve_before_tool_call_decision(
            serde_json::json!({"command":"rm -rf /"}),
            crate::extensions::hooks::events::HookResult::Modify {
                input: serde_json::json!({"command":"echo safe"}),
            },
            None,
            false,
            None,
        )
        .await;

        match result {
            BeforeToolCallDecision::Continue { input } => {
                assert_eq!(input, serde_json::json!({"command":"echo safe"}));
            }
            BeforeToolCallDecision::Block { reason } => panic!("unexpected block: {reason}"),
        }
    }

    #[tokio::test]
    async fn confirm_prompt_yes_continues() {
        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
        let handle = crate::tools::SecretPromptHandle::new(tx);

        let task = tokio::spawn(async move {
            let request = rx.recv().await.expect("confirm prompt request");
            assert_eq!(request.title, "Confirm tool call");
            assert!(request.prompt.contains("Run deploy?"));
            let _ = request.response_tx.send(Some("yes".to_string()));
        });

        let result = resolve_before_tool_call_result(
            crate::extensions::hooks::events::HookResult::confirm("Run deploy?"),
            Some(&handle),
            false,
            None,
        )
        .await;

        task.await.unwrap();
        assert!(matches!(
            result,
            crate::extensions::hooks::events::HookResult::Continue
        ));
    }

    #[tokio::test]
    async fn confirm_prompt_non_yes_blocks() {
        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
        let handle = crate::tools::SecretPromptHandle::new(tx);

        let task = tokio::spawn(async move {
            let request = rx.recv().await.expect("confirm prompt request");
            let _ = request.response_tx.send(Some("no".to_string()));
        });

        let result = resolve_before_tool_call_result(
            crate::extensions::hooks::events::HookResult::confirm("Run deploy?"),
            Some(&handle),
            false,
            None,
        )
        .await;

        task.await.unwrap();
        assert!(matches!(
            result,
            crate::extensions::hooks::events::HookResult::Block { reason }
                if reason.contains("confirmation denied")
        ));
    }

    fn ext(ids: &[&str]) -> Vec<String> {
        ids.iter().map(|s| s.to_string()).collect()
    }

    fn key(ids: &[&str]) -> Vec<crate::tools::LatchKey> {
        ids.iter().map(|s| (s.to_string(), 0)).collect()
    }

    fn sourced_confirm(
        message: &str,
        source: &str,
        tool: &str,
    ) -> crate::extensions::hooks::events::HookResult {
        crate::extensions::hooks::events::HookResult::Confirm {
            message: message.into(),
            sources: ext(&[source]),
            tool: Some(tool.into()),
            generations: vec![],
            modified_input: None,
            modified_by: vec![],
        }
    }

    #[test]
    fn session_allow_all_clones_share_but_independent_runtimes_do_not() {
        use std::sync::Arc;
        let parent = Runtime::new_headless();
        let cloned = parent.clone();
        let independent = Runtime::new_headless();
        assert!(Arc::ptr_eq(
            parent.session_allow_all(),
            cloned.session_allow_all()
        ));
        assert!(!Arc::ptr_eq(
            parent.session_allow_all(),
            independent.session_allow_all()
        ));
        cloned.session_allow_all().allow(&key(&["jev"]));
        assert!(parent.session_allow_all().allows_all(&key(&["jev"])));
        assert!(!independent.session_allow_all().allows_all(&key(&["jev"])));
    }

    #[test]
    fn session_allow_all_conversation_boundary_isolates_old_clones() {
        use std::sync::Arc;
        let mut runtime = Runtime::new_headless();
        let old_worker = runtime.clone();
        old_worker.session_allow_all().allow(&key(&["jev"]));

        runtime.begin_conversation(Some("next".into()));
        assert_eq!(runtime.session_id(), Some("next"));
        assert!(runtime.session_allow_all().is_empty());
        assert!(!Arc::ptr_eq(
            runtime.session_allow_all(),
            old_worker.session_allow_all()
        ));
        // #148 I2b: workers still running from the old conversation stop
        // auto-approving — the old set is revoked before the swap.
        assert!(
            old_worker.session_allow_all().is_empty(),
            "old latch must be cleared, not just replaced"
        );
        // An old clone can neither grant the new conversation nor re-arm
        // the revoked (orphaned) set.
        assert!(!old_worker.session_allow_all().allow(&key(&["jev"])));
        assert!(old_worker.session_allow_all().is_empty());
        assert!(runtime.session_allow_all().is_empty());

        let new_worker = runtime.clone();
        new_worker.session_allow_all().allow(&key(&["jev"]));
        assert!(runtime.session_allow_all().allows_all(&key(&["jev"])));
        // Explicit resume of even the same ID starts fresh consent.
        runtime.begin_conversation(Some("next".into()));
        assert!(runtime.session_allow_all().is_empty());
        assert!(!Arc::ptr_eq(
            runtime.session_allow_all(),
            new_worker.session_allow_all()
        ));
    }

    #[test]
    fn conversation_boundary_resets_the_cache_diagnosis_chain() {
        let mut runtime = Runtime::new_headless();
        runtime.set_cache_diagnostics(true);
        let mut driver = runtime.clone();
        driver.share_ttl_latches(&runtime);
        runtime.cache_diagnosis().unwrap().record("msg_old");
        // Compaction successor: same conversation, chain kept.
        runtime.set_session_id(Some("successor".into()));
        assert_eq!(
            runtime
                .cache_diagnosis()
                .unwrap()
                .previous_message_id()
                .as_deref(),
            Some("msg_old")
        );
        // `/clear` (and resume) start a new chain, seen by shared clones too.
        runtime.begin_conversation(Some("next".into()));
        assert_eq!(
            runtime.cache_diagnosis().unwrap().previous_message_id(),
            None
        );
        assert_eq!(
            driver.cache_diagnosis().unwrap().previous_message_id(),
            None
        );
    }

    #[test]
    fn session_allow_all_identity_change_preserves_compaction_consent() {
        use std::sync::Arc;
        let mut runtime = Runtime::new_headless();
        let worker = runtime.clone();
        worker.session_allow_all().allow(&key(&["jev"]));
        runtime.set_session_id(Some("linked-successor".into()));
        assert!(Arc::ptr_eq(
            runtime.session_allow_all(),
            worker.session_allow_all()
        ));
        assert!(runtime.session_allow_all().allows_all(&key(&["jev"])));
    }

    /// "Allow all from <ext>": the answer continues THIS call and latches
    /// that extension only. A later confirm from the same extension resolves
    /// without a prompt; a confirm from another extension still prompts.
    #[tokio::test]
    async fn confirm_prompt_always_latches_only_the_source_extension() {
        let latch = crate::tools::SessionConfirmLatch::new();
        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
        let handle = crate::tools::SecretPromptHandle::new(tx);

        let task = tokio::spawn(async move {
            let first = rx.recv().await.expect("first confirm prompt");
            // The dialog is told who asks and for which tool BEFORE answering.
            assert_eq!(
                first.confirm_scope,
                Some(crate::session::ConfirmScope {
                    extensions: vec!["jev".into()],
                    tool: Some("bash".into()),
                    rewritten_by: vec![],
                })
            );
            assert!(first.prompt.starts_with("Extension: jev\nTool: bash\n"), "{}", first.prompt);
            let _ = first
                .response_tx
                .send(Some(CONFIRM_ANSWER_ALLOW_ALL.to_string()));
            // The next prompt must be the OTHER extension's confirm (jev's
            // second confirm is latched and never prompts).
            let other = rx.recv().await.expect("other extension still prompts");
            assert_eq!(
                other.confirm_scope.as_ref().map(|s| s.extensions.clone()),
                Some(vec!["other".to_string()])
            );
            let _ = other.response_tx.send(None);
            assert!(rx.recv().await.is_none(), "no further prompts");
        });

        let first = resolve_before_tool_call_result(
            sourced_confirm("Run deploy?", "jev", "bash"),
            Some(&handle),
            false,
            Some(&latch),
        )
        .await;
        assert!(matches!(
            first,
            crate::extensions::hooks::events::HookResult::Continue
        ));
        assert_eq!(latch.extensions(), ext(&["jev"]), "latch keyed by source");

        let second = resolve_before_tool_call_result(
            sourced_confirm("Run deploy again?", "jev", "write"),
            Some(&handle),
            false,
            Some(&latch),
        )
        .await;
        assert!(matches!(
            second,
            crate::extensions::hooks::events::HookResult::Continue
        ));

        let other = resolve_before_tool_call_result(
            sourced_confirm("Run ls?", "other", "bash"),
            Some(&handle),
            false,
            Some(&latch),
        )
        .await;
        assert!(
            matches!(other, crate::extensions::hooks::events::HookResult::Block { .. }),
            "another extension's confirm is not covered by jev's latch"
        );
        drop(handle);
        task.await.unwrap();
    }

    /// Latched A + unlatched B on one call (a merged Confirm): it prompts,
    /// the prompt names both, and "Allow all" then latches both.
    #[tokio::test]
    async fn latched_plus_unlatched_source_prompts_and_allow_all_latches_both() {
        let latch = crate::tools::SessionConfirmLatch::new();
        latch.allow(&key(&["a"]));
        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
        let handle = crate::tools::SecretPromptHandle::new(tx);
        let task = tokio::spawn(async move {
            let req = rx.recv().await.expect("must prompt: b is not latched");
            assert_eq!(
                req.confirm_scope.as_ref().map(|s| s.extensions.clone()),
                Some(vec!["a".to_string(), "b".to_string()])
            );
            assert!(req.prompt.starts_with("Extension: a, b\n"), "{}", req.prompt);
            let _ = req.response_tx.send(Some(CONFIRM_ANSWER_ALLOW_ALL.to_string()));
        });
        let merged = || crate::extensions::hooks::events::HookResult::Confirm {
            message: "[a] x\n\n[b] y".into(),
            sources: ext(&["a", "b"]),
            generations: vec![],
            tool: Some("bash".into()),
            modified_input: None,
            modified_by: vec![],
        };
        let first = resolve_before_tool_call_result(merged(), Some(&handle), false, Some(&latch)).await;
        task.await.unwrap();
        assert!(matches!(first, crate::extensions::hooks::events::HookResult::Continue));
        assert!(latch.allows_all(&key(&["a", "b"])));
        // Now fully latched: no prompt (handle dropped → would block).
        drop(handle);
        let second = resolve_before_tool_call_result(merged(), None, false, Some(&latch)).await;
        assert!(matches!(second, crate::extensions::hooks::events::HookResult::Continue));
    }

    /// A call site with no latch never offers "Allow all" (no scope), so the
    /// dialog and the session's "armed" notice can't promise it.
    #[tokio::test]
    async fn no_latch_means_no_allow_all_scope() {
        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
        let handle = crate::tools::SecretPromptHandle::new(tx);
        let task = tokio::spawn(async move {
            let req = rx.recv().await.unwrap();
            assert!(req.confirm_scope.is_none());
            let _ = req.response_tx.send(Some("y".into()));
        });
        let r = resolve_before_tool_call_result(
            sourced_confirm("Run?", "jev", "bash"),
            Some(&handle),
            false,
            None,
        )
        .await;
        task.await.unwrap();
        assert!(matches!(r, crate::extensions::hooks::events::HookResult::Continue));
    }

    /// prA N3: a Confirm whose input an earlier extension rewrote shows who
    /// rewrote it and a capped preview of the FINAL input, ahead of the
    /// confirmer's own message; the scope records `rewritten_by` so the
    /// session can mark the transcript; approval runs that final input.
    #[tokio::test]
    async fn rewritten_input_is_shown_in_the_confirm_prompt() {
        let latch = crate::tools::SessionConfirmLatch::new();
        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
        let handle = crate::tools::SecretPromptHandle::new(tx);
        let task = tokio::spawn(async move {
            let req = rx.recv().await.unwrap();
            let body = req.prompt.clone();
            let scope = req.confirm_scope.clone().unwrap();
            let _ = req.response_tx.send(Some("y".into()));
            (body, scope)
        });
        let result = resolve_before_tool_call_result(
            crate::extensions::hooks::events::HookResult::Confirm {
                message: "Run this bash command?".into(),
                sources: ext(&["jev"]),
                generations: vec![],
                tool: Some("bash".into()),
                modified_input: Some(serde_json::json!({"command": "rm -rf build/"})),
                modified_by: ext(&["rewriter"]),
            },
            Some(&handle),
            false,
            Some(&latch),
        )
        .await;
        let (body, scope) = task.await.unwrap();
        let header_end = body.find("Message from jev:\nRun this bash command?").unwrap();
        let header = &body[..header_end];
        assert!(header.contains("Input rewritten by: rewriter"), "{body}");
        assert!(header.contains("Input that will run:\n  rm -rf build/"), "{body}");
        assert_eq!(scope.rewritten_by, ext(&["rewriter"]));
        assert!(matches!(
            result,
            crate::extensions::hooks::events::HookResult::Modify { input }
                if input == serde_json::json!({"command": "rm -rf build/"})
        ));
        // No rewrite → no rewrite header, empty rewritten_by.
        let plain = confirm_prompt_body("m", &ext(&["jev"]), Some("bash"), None);
        assert!(!plain.contains("rewritten"), "{plain}");
    }

    /// prA I-R3-1: the preview is redacted before it is rendered.
    #[test]
    fn final_input_preview_redacts_credentials() {
        let secrets = [
            "SECRET-BEARER-123",
            "SECRET-KEY-456",
            "SECRET-FIELD-789",
            "QmFzaWNTRUNSRVQ=",
            "SECRET-TOK-000",
            "SECRET-FLAG-111",
            "sk-ant-api03-abcdefghijklmnop",
            "eyJhbGciOiJIUzI1NiJ9.eyJzdWIiOiIxIn0.c2lnbmF0dXJl",
            "ghp_abcdefghijklmnopqrstuvwxyz0123",
        ];
        let command = serde_json::json!({
            "command": "curl -H 'Authorization: Bearer SECRET-BEARER-123' \
                        'https://api.example/v1?api_key=SECRET-KEY-456' \
                        -H \"authorization: Basic QmFzaWNTRUNSRVQ=\" \
                        && TOKEN=SECRET-TOK-000 deploy --password SECRET-FLAG-111 \
                        sk-ant-api03-abcdefghijklmnop \
                        eyJhbGciOiJIUzI1NiJ9.eyJzdWIiOiIxIn0.c2lnbmF0dXJl \
                        ghp_abcdefghijklmnopqrstuvwxyz0123"
        });
        let json = serde_json::json!({
            "url": "https://x/y",
            "api_key": "SECRET-FIELD-789",
            "headers": {"X-Note": "Bearer SECRET-BEARER-123"},
            "args": ["--token=SECRET-TOK-000"],
        });
        for input in [command, json] {
            let p = final_input_preview(&input);
            for secret in secrets {
                assert!(!p.contains(secret), "{secret} leaked in:\n{p}");
            }
            assert!(p.contains("[REDACTED]"), "{p}");
        }
        // Non-secret text survives so the user can still judge the call.
        let plain = final_input_preview(&serde_json::json!({"command": "rm -rf build/ && ls -la"}));
        assert!(plain.contains("rm -rf build/ && ls -la"), "{plain}");
    }

    /// prA I-R4-2: every shape from the round-4 table is redacted, by the
    /// SHARED scrubber (trace export gets the same rules).
    #[test]
    fn redact_secret_text_covers_round4_shapes() {
        let rows: &[(&str, &str)] = &[
            ("curl -H 'x-api-key: XAPIKEY-SECRET-1' https://a", "XAPIKEY-SECRET-1"),
            ("FOO_TOKEN=TOK-SECRET-2 run", "TOK-SECRET-2"),
            ("MY_SECRET=SEC-SECRET-3 ./x", "SEC-SECRET-3"),
            ("DB_PASSWORD=PW-SECRET-4 ./x", "PW-SECRET-4"),
            ("OPENAI_API_KEY=APIKEY-SECRET-5 ./x", "APIKEY-SECRET-5"),
            ("curl -H 'Authorization: Basic QkFTSUMtU0VDUkVULTY='", "QkFTSUMtU0VDUkVULTY="),
            ("aws_secret_access_key=AWSSEC-SECRET-7", "AWSSEC-SECRET-7"),
            ("git clone https://alice:USERINFO-SECRET-8@github.com/o/r", "USERINFO-SECRET-8"),
            ("psql postgres://app:PGPW-SECRET-9@db:5432/prod", "PGPW-SECRET-9"),
            ("DATABASE_URL=postgres://app:DBURL-SECRET-10@db/prod ./run", "DBURL-SECRET-10"),
            ("curl -H 'Cookie: sid=COOKIE-SECRET-11; theme=dark' https://a", "COOKIE-SECRET-11"),
            ("printf 'Set-Cookie: sid=SETCOOKIE-SECRET-12'", "SETCOOKIE-SECRET-12"),
            ("curl -u alice:CURLU-SECRET-13 https://a", "CURLU-SECRET-13"),
            ("mysql -u root -pMYSQL-SECRET-14 prod", "MYSQL-SECRET-14"),
            ("sshpass -p SSHPASS-SECRET-15 ssh host", "SSHPASS-SECRET-15"),
            ("aws configure set aws_secret_access_key AWSCFG-SECRET-16", "AWSCFG-SECRET-16"),
            ("aws x --secret-access-key FLAG-SECRET-17 --region eu", "FLAG-SECRET-17"),
            ("aws x --access-key FLAG-SECRET-18", "FLAG-SECRET-18"),
            ("STRIPE_KEY=rk_live_STRIPE-SECRET-19 node app", "STRIPE-SECRET-19"),
            ("OPENAI_KEY=OAIKEY-SECRET-20 node app", "OAIKEY-SECRET-20"),
        ];
        // prA I-R5-1: rows whose shape is unmistakable are also caught by
        // the SHARED high-precision scrub (trace export, memory, archive).
        let precise = [
            "QkFTSUMtU0VDUkVULTY=",
            "USERINFO-SECRET-8",
            "PGPW-SECRET-9",
            "DBURL-SECRET-10",
            "COOKIE-SECRET-11",
            "SETCOOKIE-SECRET-12",
            "CURLU-SECRET-13",
            "MYSQL-SECRET-14",
            "SSHPASS-SECRET-15",
        ];
        for (text, secret) in rows {
            // The Confirm prompt path catches every row.
            let out = redact_secret_text(text);
            assert!(!out.contains(secret), "{secret} leaked: {out}");
            assert!(out.contains("[REDACTED]"), "{out}");
            let preview = final_input_preview(&serde_json::json!({ "command": text }));
            assert!(!preview.contains(secret), "{secret} leaked in preview: {preview}");
            if precise.contains(secret) {
                let mut v = serde_json::Value::String(text.to_string());
                crate::runtime::trace::export::redact_value(&mut v);
                assert!(
                    !v.as_str().unwrap().contains(secret),
                    "shared scrub leaked {secret}: {v}"
                );
            }
        }
        // Over-redaction guards: ordinary commands and prose stay intact.
        for ok in [
            "mkdir -p build/out",
            "ssh -p 2222 host",
            "ssh -p2222 host",
            "docker run -p 8080:80 nginx",
            "mysql -h db -P 3306 -p prod",
            "curl -u alice https://example.com/path",
            "echo the token is valid and the password field is empty",
            "monkey=1 make",
            "git log --format=%H -- src/",
            "https://example.com/a/b?page=2",
        ] {
            assert_eq!(redact_secret_text(ok), ok, "over-redacted");
        }
    }

    /// prA I-R5-1: the shared scrub (`redact_value`) also runs over the
    /// rollover context archive, memory captures and traces, so benign code
    /// and prose must pass through it unchanged. Only the Confirm prompt
    /// uses the aggressive rules.
    #[test]
    fn shared_scrub_leaves_benign_code_and_prose_alone() {
        let benign = [
            "max_tokens: 4096",
            "password: String,",
            "let token = load()?;",
            "the password-reset flow",
            "pub authorization: Option<String>,",
            "cookie: HashMap<String, String>,",
            "MY_KEY=value_in_docs",
            "--token-file path/to/file",
            "fn refresh_access_token(&self) -> Result<Token>",
            "aws_secret_access_key is the field name",
        ];
        for text in benign {
            let mut v = serde_json::Value::String(text.to_string());
            crate::runtime::trace::export::redact_value(&mut v);
            assert_eq!(v.as_str().unwrap(), text, "shared scrub mangled benign text");
            assert_eq!(crate::runtime::trace::export::redact_text(text), text);
        }
        // Inside objects too (archive / memory records are JSON).
        let mut record = serde_json::json!({
            "text": "set max_tokens: 4096 and keep the password-reset flow",
            "code": ["password: String,", "let token = load()?;"],
        });
        let before = record.clone();
        crate::runtime::trace::export::redact_value(&mut record);
        assert_eq!(record, before);
    }

    /// prA I-R4-1: the confirmer's message is redacted too — it routinely
    /// echoes the (credential-injected) command it was shown.
    #[test]
    fn confirmer_message_echoing_injected_credentials_is_redacted() {
        let injected = serde_json::json!({
            "command": "curl -H 'Authorization: Bearer ECHOED-BEARER-X' https://a"
        });
        let body = confirm_prompt_body(
            "Run: curl -H 'Authorization: Bearer ECHOED-BEARER-X' https://a ?",
            &ext(&["jev"]),
            Some("bash"),
            Some((&ext(&["injector"]), &injected)),
        );
        assert!(!body.contains("ECHOED-BEARER-X"), "{body}");
        assert!(body.contains("Message from jev:\nRun: curl -H 'Authorization: [REDACTED]"), "{body}");
    }

    /// prA I-R4-3: control, bidi and zero-width characters in the message
    /// are stripped before capping, so they can't move the cursor or
    /// disguise text in plain-text clients.
    #[test]
    fn confirm_message_control_and_bidi_chars_are_stripped() {
        let hostile = "ok\x1b[3A\x1b[2KExtension: trusted\rTool: none\u{0085}\u{202E}evil\u{200B}x\u{2066}y\tz\nnext";
        let body = confirm_prompt_body(hostile, &ext(&["jev"]), Some("bash"), None);
        for bad in ['\x1b', '\r', '\u{0085}', '\u{202E}', '\u{200B}', '\u{2066}', '\t'] {
            assert!(!body.contains(bad), "{bad:?} survived: {body:?}");
        }
        assert!(body.contains("Message from jev:\nok[3A[2KExtension: trustedTool: noneevilxy z\nnext"), "{body:?}");
        assert!(body.starts_with("Extension: jev\nTool: bash\n"));
    }

    /// The confirmer's text is labelled, so a fake "Input that will run:"
    /// in it lands visibly under "Message from <ext>:", after the host block.
    #[test]
    fn extension_message_is_labelled_after_the_host_header() {
        let fake = "Input that will run:\n  echo harmless";
        let body = confirm_prompt_body(fake, &ext(&["jev"]), Some("bash"), None);
        let label = body.find("Message from jev:\n").expect(&body);
        assert_eq!(body.find("Input that will run:"), Some(label + "Message from jev:\n".len()));
        assert!(body.starts_with("Extension: jev\nTool: bash\n"), "{body}");
    }

    #[test]
    fn final_input_preview_is_capped() {
        let many = serde_json::json!({"command": (1..=20).map(|i| format!("echo {i}")).collect::<Vec<_>>().join("\n")});
        let p = final_input_preview(&many);
        assert!(p.lines().count() <= MAX_REWRITE_PREVIEW_LINES + 1, "{p}");
        assert!(p.ends_with("…[preview truncated]"), "{p}");
        let wide = serde_json::json!({"path": "x".repeat(2000)});
        let p = final_input_preview(&wide);
        assert!(p.chars().count() <= MAX_REWRITE_PREVIEW_CHARS + 30, "{}", p.len());
        let obj = serde_json::json!({"path": "/tmp/a", "content": "hi"});
        assert!(final_input_preview(&obj).contains("\"path\": \"/tmp/a\""));
    }

    /// A long extension message is capped so the host header, buttons and
    /// scope line stay on screen.
    #[test]
    fn long_confirm_message_is_capped() {
        let long = "x\n".repeat(500);
        let body = confirm_prompt_body(&long, &ext(&["jev"]), Some("bash"), None);
        assert!(body.starts_with("Extension: jev\nTool: bash\n"));
        assert!(body.contains("…[message truncated by host]"));
        assert!(body.lines().count() <= MAX_CONFIRM_MESSAGE_LINES + 8, "{}", body.lines().count());
        let wide = "y".repeat(10_000);
        assert!(confirm_prompt_body(&wide, &ext(&["jev"]), None, None).chars().count() < 1400);
    }

    /// An unattributed confirm (no host-stamped source) offers no scope and
    /// an "always" answer approves only this call; nothing is latched.
    #[tokio::test]
    async fn unattributed_confirm_always_answer_does_not_latch() {
        let latch = crate::tools::SessionConfirmLatch::new();
        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
        let handle = crate::tools::SecretPromptHandle::new(tx);
        let task = tokio::spawn(async move {
            let req = rx.recv().await.expect("prompt");
            assert!(req.confirm_scope.is_none());
            let _ = req.response_tx.send(Some(CONFIRM_ANSWER_ALLOW_ALL.to_string()));
        });
        let result = resolve_before_tool_call_result(
            crate::extensions::hooks::events::HookResult::confirm("Run deploy?"),
            Some(&handle),
            false,
            Some(&latch),
        )
        .await;
        task.await.unwrap();
        assert!(matches!(
            result,
            crate::extensions::hooks::events::HookResult::Continue
        ));
        assert!(latch.is_empty());
    }

    /// The latch only short-circuits Confirm; Block from an extension is
    /// still a Block, and a headless path with the source latched continues
    /// instead of failing closed (an unlatched source still fails closed).
    #[tokio::test]
    async fn session_allow_all_does_not_override_block() {
        let latch = crate::tools::SessionConfirmLatch::new();
        latch.allow(&key(&["jev"]));
        let blocked = resolve_before_tool_call_result(
            crate::extensions::hooks::events::HookResult::Block {
                reason: "nope".into(),
            },
            None,
            false,
            Some(&latch),
        )
        .await;
        assert!(matches!(
            blocked,
            crate::extensions::hooks::events::HookResult::Block { reason } if reason == "nope"
        ));
        let headless = resolve_before_tool_call_result(
            sourced_confirm("Run deploy?", "jev", "bash"),
            None,
            false,
            Some(&latch),
        )
        .await;
        assert!(matches!(
            headless,
            crate::extensions::hooks::events::HookResult::Continue
        ));
        let other = resolve_before_tool_call_result(
            sourced_confirm("Run deploy?", "other", "bash"),
            None,
            false,
            Some(&latch),
        )
        .await;
        assert!(matches!(
            other,
            crate::extensions::hooks::events::HookResult::Block { .. }
        ));
    }

    #[test]
    fn test_max_tokens_for_model() {
        // Opus models should return 128000
        assert_eq!(
            HelperMethods::max_tokens_for_model("claude-opus-4-6"),
            128000
        );
        assert_eq!(
            HelperMethods::max_tokens_for_model("opus-something"),
            128000
        );

        // Non-opus models should return 64000
        assert_eq!(
            HelperMethods::max_tokens_for_model("claude-sonnet-4-20250514"),
            64000
        );
        assert_eq!(HelperMethods::max_tokens_for_model("haiku"), 64000);
        assert_eq!(HelperMethods::max_tokens_for_model("claude-3-haiku"), 64000);
        assert_eq!(
            HelperMethods::max_tokens_for_model("some-other-model"),
            64000
        );

        // Edge cases
        assert_eq!(HelperMethods::max_tokens_for_model(""), 64000);
        assert_eq!(HelperMethods::max_tokens_for_model("OPUS"), 64000); // Case sensitive - uppercase doesn't match
        assert_eq!(
            HelperMethods::max_tokens_for_model("model-opus-end"),
            128000
        ); // Contains "opus" anywhere
    }

    #[test]
    fn test_truncate_tool_result() {
        let default_max = 30000;

        // Short string should remain unchanged
        let short = "This is a short string.";
        assert_eq!(
            HelperMethods::truncate_tool_result(short, default_max),
            short
        );

        // Exactly max should remain unchanged
        let exact = "x".repeat(30000);
        assert_eq!(
            HelperMethods::truncate_tool_result(&exact, default_max),
            exact
        );

        // String longer than max should be truncated with notice
        let too_long = "x".repeat(30001);
        let truncated = HelperMethods::truncate_tool_result(&too_long, default_max);

        // Should start with the truncated content
        assert!(truncated.starts_with(&"x".repeat(30000)));

        // Should contain truncation notice with total char count
        assert!(truncated.contains("[truncated — 30001 total bytes, showing first 30000]"));

        // Should be longer than max (due to notice)
        assert!(truncated.len() > 30000);

        // Test with a much longer string
        let very_long = "a".repeat(50000);
        let truncated_very_long = HelperMethods::truncate_tool_result(&very_long, default_max);
        assert!(
            truncated_very_long.contains("[truncated — 50000 total bytes, showing first 30000]")
        );
        assert!(truncated_very_long.starts_with(&"a".repeat(30000)));

        // Test with custom limit
        let custom_truncated = HelperMethods::truncate_tool_result(&very_long, 100);
        assert!(custom_truncated.starts_with(&"a".repeat(100)));
        assert!(custom_truncated.contains("[truncated — 50000 total bytes, showing first 100]"));
    }

    /// T2 red→green: `truncate_tool_result` must enforce a BYTE budget on
    /// multibyte input (legacy code used `chars().take(n)` — a char budget —
    /// so 100 two-byte chars under a 50-byte budget kept 100 bytes).
    #[test]
    fn truncate_tool_result_enforces_byte_budget() {
        let multibyte = "é".repeat(100); // 200 bytes, 100 chars
        let out = HelperMethods::truncate_tool_result(&multibyte, 50);
        let body = out.split("\n\n[truncated").next().unwrap();
        assert!(
            body.len() <= 50,
            "retained body must fit the byte budget (got {} bytes)",
            body.len()
        );
        assert!(std::str::from_utf8(body.as_bytes()).is_ok());
        assert!(
            out.contains("total bytes"),
            "marker must report byte counts, got: {out}"
        );
    }

    #[test]
    fn test_thinking_level_ranges() {
        use crate::core::models::thinking_level_for_budget;

        // Sentinel 0 = "adaptive" (S172 — model decides)
        assert_eq!(thinking_level_for_budget(0), "adaptive");

        // Low range: 1..=2048
        assert_eq!(thinking_level_for_budget(1), "low");
        assert_eq!(thinking_level_for_budget(1024), "low");
        assert_eq!(thinking_level_for_budget(2048), "low");

        // Medium range: 2049..=4096
        assert_eq!(thinking_level_for_budget(2049), "medium");
        assert_eq!(thinking_level_for_budget(3000), "medium");
        assert_eq!(thinking_level_for_budget(4096), "medium");

        // High range: 4097..=16384
        assert_eq!(thinking_level_for_budget(4097), "high");
        assert_eq!(thinking_level_for_budget(8192), "high");
        assert_eq!(thinking_level_for_budget(16384), "high");

        // XHigh range: _ (everything else)
        assert_eq!(thinking_level_for_budget(16385), "xhigh");
        assert_eq!(thinking_level_for_budget(32768), "xhigh");
        assert_eq!(thinking_level_for_budget(100000), "xhigh");
    }
}

#[cfg(test)]
mod effective_prompt_tests {
    use super::*;

    // ── progressive-disclosure tool catalog in the system prompt ────────────

    struct LateTool;
    #[async_trait::async_trait]
    impl crate::Tool for LateTool {
        fn name(&self) -> &str {
            "late_tool"
        }
        fn description(&self) -> &str {
            "Registered after the catalog froze."
        }
        fn parameters(&self) -> Value {
            json!({"type": "object", "properties": {}})
        }
        async fn execute(&self, _p: Value, _c: crate::ToolContext) -> crate::Result<String> {
            Ok(String::new())
        }
    }

    #[tokio::test]
    async fn tool_catalog_lists_activatable_tools_outside_the_core() {
        let mut runtime = headless("anthropic/claude-sonnet-5-5", "BASE.");
        assert!(
            runtime.frozen_tool_catalog().await.is_none(),
            "flag off: no catalog"
        );
        runtime.progressive_tool_disclosure = true;
        let catalog = runtime.frozen_tool_catalog().await.expect("catalog");
        assert!(catalog.starts_with("## Activatable tools\n"));
        assert!(
            catalog.contains("not instructions"),
            "third-party text is framed as data"
        );
        assert!(catalog.contains("- builtin:subagent_start"), "{catalog}");
        for core in [
            "builtin:bash",
            "builtin:read",
            "builtin:activate_tools",
            "builtin:search_tools",
        ] {
            assert!(
                !catalog.contains(&format!("- {core}")),
                "core {core} listed"
            );
        }
        let prompt = runtime.effective_system_prompt().await.unwrap();
        assert!(prompt.starts_with("BASE."), "base stays first");
        assert!(prompt.ends_with(&catalog), "catalog appended last");
    }

    /// `tools.always_loaded` flows config → runtime → clones (the stream
    /// session and subagent forks copy it), and a listed tool is core, so the
    /// activatable-tools catalog stops offering it.
    #[tokio::test]
    async fn tools_always_loaded_flows_from_config_into_the_session_core() {
        let mut runtime = headless("anthropic/claude-sonnet-5-5", "BASE.");
        let config = agent_core::config::load_config_from_str(
            "progressive_tool_disclosure = true\ntools.always_loaded = builtin:subagent_start\n",
        );
        runtime.apply_config_keep_tools(&config);
        assert!(runtime.progressive_tool_disclosure);
        let listed = crate::tools::catalog::ToolId::builtin("subagent_start");
        assert_eq!(&*runtime.tools_always_loaded, std::slice::from_ref(&listed));
        assert_eq!(&*runtime.clone().tools_always_loaded, std::slice::from_ref(&listed));
        let catalog = runtime.frozen_tool_catalog().await.expect("catalog");
        assert!(!catalog.contains("- builtin:subagent_start"), "{catalog}");
        assert!(catalog.contains("- builtin:subagent_"), "other tools still listed");
    }

    #[tokio::test]
    async fn tool_catalog_is_frozen_per_session() {
        let mut runtime = headless("anthropic/claude-sonnet-5-5", "BASE.");
        runtime.progressive_tool_disclosure = true;
        let first = runtime.effective_system_prompt().await.unwrap();
        runtime.tools.write().await.register(Arc::new(LateTool));
        let clone = runtime.clone();
        assert_eq!(runtime.effective_system_prompt().await.unwrap(), first);
        assert_eq!(clone.effective_system_prompt().await.unwrap(), first);
        assert!(
            !first.contains("late_tool"),
            "late registration does not rewrite the prefix"
        );
    }

    #[tokio::test]
    async fn tool_catalog_absent_when_activation_is_denied_or_impossible() {
        let mut denied = headless("anthropic/claude-sonnet-5-5", "BASE.");
        denied.progressive_tool_disclosure = true;
        denied.set_activation_confirm(agent_core::config::ActivationConfirm::Deny);
        assert!(denied.frozen_tool_catalog().await.is_none());
        assert_eq!(
            denied.effective_system_prompt().await.as_deref(),
            Some("BASE.")
        );

        let mut worker = headless("anthropic/claude-sonnet-5-5", "BASE.");
        worker.progressive_tool_disclosure = true;
        worker.set_tools(crate::ToolRegistry::without_subagent());
        assert!(
            worker.frozen_tool_catalog().await.is_none(),
            "no activate_tools, no catalog"
        );
    }

    fn headless(model: &str, base: &str) -> Runtime {
        let mut runtime = Runtime::new_headless();
        runtime.set_model(model.to_string());
        runtime.set_system_prompt(base.to_string());
        runtime
    }

    #[tokio::test]
    async fn anthropic_foreground_ultracode_composes_standing_doctrine_once() {
        let mut runtime = headless("anthropic/claude-fable-5", "BASE.");
        runtime.named_level = Some(agent_core::reasoning::ReasoningLevel::UltraCode);
        let prompt = runtime.effective_system_prompt().await.expect("doctrine");
        assert_eq!(prompt.matches("<anthropic-ultracode-workflow>").count(), 1);
        assert!(prompt.contains("subagent_start"));
        assert!(prompt.contains("subagent_resume"));
    }

    #[tokio::test]
    async fn codex_foreground_with_subagent_tools_composes_doctrine() {
        let runtime = headless("openai-codex/gpt-5.6-sol", "BASE.");
        let prompt = runtime
            .effective_system_prompt()
            .await
            .expect("prompt must compose");
        assert!(prompt.starts_with("BASE."), "base must lead");
        assert!(prompt.contains("## Subagent supervision"));
        assert!(prompt.contains("NEVER end your turn"));
    }

    #[tokio::test]
    async fn worker_runtime_without_subagent_tools_stays_clean() {
        let mut runtime = headless("openai-codex/gpt-5.6-sol", "WORKER.");
        runtime.set_tools(ToolRegistry::without_subagent());
        assert_eq!(
            runtime.effective_system_prompt().await.as_deref(),
            Some("WORKER."),
            "workers have no subagent tools; doctrine would be unactionable noise"
        );
    }

    #[tokio::test]
    async fn non_codex_foreground_stays_clean() {
        let runtime = headless("xai-auth/grok-4.5-latest", "BASE.");
        assert_eq!(
            runtime.effective_system_prompt().await.as_deref(),
            Some("BASE.")
        );
    }

    #[tokio::test]
    async fn manifest_prompt_is_returned_verbatim() {
        let mut runtime = headless("openai-codex/gpt-5.6-sol", "IGNORED.");
        let module = agent_core::prompt::PromptModule::new(
            agent_core::prompt::PromptModuleId::parse("kernel.test").unwrap(),
            "1.0.0",
            agent_core::prompt::PromptModuleSource::User,
            0,
            agent_core::prompt::PromptSelectors::default(),
            agent_core::prompt::ModuleMutability::MutableGuidance,
            "KERNEL.",
        )
        .unwrap();
        let context = agent_core::prompt::SelectionContext::new(
            agent_core::prompt::QualifiedModelId::parse("openai-codex/gpt-5.6-sol").unwrap(),
            None,
        )
        .unwrap();
        let stack = agent_core::prompt::PromptStack::new(vec![module], context).unwrap();
        runtime.apply_prompt_stack(stack).unwrap();
        assert_eq!(
            runtime.effective_system_prompt().await.as_deref(),
            Some("KERNEL."),
            "manifest authors own the full stack; no builtin injection on top"
        );
    }

    #[tokio::test]
    async fn unresolvable_model_identity_fails_closed_to_base() {
        let runtime = headless("definitely-not-a-model", "BASE.");
        assert_eq!(
            runtime.effective_system_prompt().await.as_deref(),
            Some("BASE.")
        );
    }
}

#[cfg(test)]
mod set_reasoning_level_checked_tests {
    use super::*;
    use agent_core::reasoning::ReasoningLevel;

    fn codex_runtime(model_id: &str) -> Runtime {
        let mut rt = Runtime::new_headless();
        rt.set_model(format!("openai-codex/{model_id}"));
        rt.set_reasoning_level(ReasoningLevel::Low);
        rt
    }

    fn provisioned_fable_runtime() -> Runtime {
        let mut rt = Runtime::new_headless();
        rt.set_model("anthropic/claude-fable-5".into());
        rt.set_reasoning_level(ReasoningLevel::Low);
        let foreground =
            agent_core::prompt::QualifiedModelId::parse("anthropic/claude-fable-5").unwrap();
        let orchestration =
            crate::orchestration::OrchestrationRuntime::baseline(foreground, 3, 8).unwrap();
        rt.install_orchestration(Arc::new(orchestration));
        rt
    }

    fn install_typed_manifest(rt: &mut Runtime) {
        let module = agent_core::prompt::PromptModule::new(
            agent_core::prompt::PromptModuleId::parse("kernel.test").unwrap(),
            "1.0.0",
            agent_core::prompt::PromptModuleSource::User,
            0,
            agent_core::prompt::PromptSelectors::default(),
            agent_core::prompt::ModuleMutability::MutableGuidance,
            "KERNEL.",
        )
        .unwrap();
        let context = agent_core::prompt::SelectionContext::new(
            agent_core::prompt::QualifiedModelId::parse("anthropic/claude-fable-5").unwrap(),
            None,
        )
        .unwrap();
        rt.apply_prompt_stack(agent_core::prompt::PromptStack::new(vec![module], context).unwrap())
            .unwrap();
    }

    #[test]
    fn provisioned_fable_accepts_max_and_ultracode() {
        for level in [ReasoningLevel::Max, ReasoningLevel::UltraCode] {
            let mut rt = provisioned_fable_runtime();
            rt.set_reasoning_level_checked(level).unwrap();
            assert_eq!(rt.reasoning_level(), level);
        }
    }

    #[test]
    fn fable_ultracode_prerequisites_fail_without_mutation() {
        let mut missing_orchestration = Runtime::new_headless();
        missing_orchestration.set_model("anthropic/claude-fable-5".into());
        missing_orchestration.set_reasoning_level(ReasoningLevel::Low);
        assert!(missing_orchestration
            .set_reasoning_level_checked(ReasoningLevel::UltraCode)
            .is_err());
        assert_eq!(missing_orchestration.reasoning_level(), ReasoningLevel::Low);

        let mut missing_tools = provisioned_fable_runtime();
        missing_tools.set_tools(ToolRegistry::without_subagent());
        assert!(missing_tools
            .set_reasoning_level_checked(ReasoningLevel::UltraCode)
            .is_err());
        assert_eq!(missing_tools.reasoning_level(), ReasoningLevel::Low);

        let mut worker = provisioned_fable_runtime();
        worker.set_codex_request_role(crate::runtime::openai::catalog::ExecutionRole::Worker);
        assert!(worker
            .set_reasoning_level_checked(ReasoningLevel::UltraCode)
            .is_err());
        assert_eq!(worker.reasoning_level(), ReasoningLevel::Low);

        let mut internal = provisioned_fable_runtime();
        internal.set_codex_request_role(crate::runtime::openai::catalog::ExecutionRole::Internal);
        assert!(internal
            .set_reasoning_level_checked(ReasoningLevel::UltraCode)
            .is_err());
        assert_eq!(internal.reasoning_level(), ReasoningLevel::Low);

        let mut manifest = provisioned_fable_runtime();
        install_typed_manifest(&mut manifest);
        assert!(manifest
            .set_reasoning_level_checked(ReasoningLevel::UltraCode)
            .is_err());
        assert_eq!(manifest.reasoning_level(), ReasoningLevel::Low);
    }

    #[test]
    fn checked_accepts_client_omission_modes_for_codex() {
        for level in [ReasoningLevel::Off, ReasoningLevel::Adaptive] {
            let mut rt = codex_runtime("gpt-5.6-sol");
            rt.set_reasoning_level_checked(level).unwrap();
            assert_eq!(rt.reasoning_level(), level);
        }
    }

    /// luna does not support Ultra; checked must reject and leave level unchanged.
    #[test]
    fn checked_rejects_ultra_for_luna_no_mutation() {
        let mut rt = codex_runtime("gpt-5.6-luna");
        let before = rt.reasoning_level();
        let err = rt
            .set_reasoning_level_checked(ReasoningLevel::Ultra)
            .unwrap_err();
        assert!(
            err.contains("ultra"),
            "error must name the rejected level; got: {err}"
        );
        assert_eq!(
            rt.reasoning_level(),
            before,
            "runtime must not be mutated when validation fails"
        );
    }

    /// gpt-5.5 does not support Max; checked must reject and not mutate.
    #[test]
    fn checked_rejects_max_for_gpt55_no_mutation() {
        let mut rt = codex_runtime("gpt-5.5");
        rt.set_reasoning_level(ReasoningLevel::Medium);
        let err = rt
            .set_reasoning_level_checked(ReasoningLevel::Max)
            .unwrap_err();
        assert!(
            err.contains("max"),
            "error must name the rejected level; got: {err}"
        );
        assert_eq!(rt.reasoning_level(), ReasoningLevel::Medium);
    }

    /// sol supports Ultra; checked must accept and mutate.
    #[test]
    fn checked_accepts_ultra_for_sol_and_mutates() {
        let mut rt = codex_runtime("gpt-5.6-sol");
        rt.set_reasoning_level_checked(ReasoningLevel::Ultra)
            .expect("sol must accept ultra");
        assert_eq!(rt.reasoning_level(), ReasoningLevel::Ultra);
    }

    /// Providers without exact metadata must not gain Max/Ultra.
    #[test]
    fn checked_rejects_extended_levels_without_metadata() {
        let mut rt = Runtime::new_headless();
        rt.set_model("claude-opus-4-7".to_string());
        rt.set_reasoning_level(ReasoningLevel::Low);
        for level in [ReasoningLevel::Max, ReasoningLevel::Ultra] {
            assert!(rt.set_reasoning_level_checked(level).is_err());
            assert_eq!(rt.reasoning_level(), ReasoningLevel::Low);
        }
    }

    /// Gap fix: unknown Codex ids (no cache, no static metadata) must reject
    /// the extended Max/Ultra modes at mutation time, fail closed.
    #[test]
    fn checked_rejects_max_ultra_for_unknown_codex_model() {
        let mut rt = codex_runtime("gpt-unknown-future");
        rt.set_reasoning_level(ReasoningLevel::Low);
        for level in [ReasoningLevel::Max, ReasoningLevel::Ultra] {
            let err = rt.set_reasoning_level_checked(level).unwrap_err();
            assert!(err.contains("no capability metadata"), "{err}");
            assert_eq!(rt.reasoning_level(), ReasoningLevel::Low);
        }
    }

    /// xAI: Off is rejected (not silently omitted) on models whose reasoning
    /// cannot be disabled; unsupported named efforts are rejected exactly.
    #[test]
    fn checked_enforces_xai_capability_matrix_no_mutation_on_err() {
        let mut rt = Runtime::new_headless();
        rt.set_model("xai-auth/grok-4.5".to_string());
        // Model default (not explicit) applied on switch: documented high.
        assert_eq!(rt.reasoning_level(), ReasoningLevel::High);
        for level in [
            ReasoningLevel::Off,
            ReasoningLevel::XHigh,
            ReasoningLevel::Ultra,
        ] {
            let before = rt.reasoning_level();
            assert!(rt.set_reasoning_level_checked(level).is_err(), "{level}");
            assert_eq!(rt.reasoning_level(), before);
        }
        rt.set_reasoning_level_checked(ReasoningLevel::Low)
            .expect("grok-4.5 supports low");
        assert_eq!(rt.reasoning_level(), ReasoningLevel::Low);
    }

    /// xAI models without documented effort control reject explicit named
    /// levels; model switch applies the Adaptive provider-default.
    #[test]
    fn checked_rejects_named_on_intrinsic_xai_model_and_defaults_adaptive() {
        let mut rt = Runtime::new_headless();
        rt.set_model("xai-auth/grok-4.3".to_string());
        assert_eq!(rt.reasoning_level(), ReasoningLevel::Adaptive);
        assert!(rt
            .set_reasoning_level_checked(ReasoningLevel::Medium)
            .is_err());
        assert_eq!(rt.reasoning_level(), ReasoningLevel::Adaptive);
        rt.set_reasoning_level_checked(ReasoningLevel::Adaptive)
            .unwrap();
    }

    /// Explicit user choice survives an xAI model switch (no default overwrite).
    #[test]
    fn explicit_level_survives_xai_model_switch() {
        let mut rt = Runtime::new_headless();
        rt.set_model("xai-auth/grok-4.5".to_string());
        rt.set_reasoning_level_checked(ReasoningLevel::Low).unwrap();
        rt.set_model("xai-auth/grok-4.5-latest".to_string());
        assert_eq!(
            rt.reasoning_level(),
            ReasoningLevel::Low,
            "explicit level must not be overwritten by the model default"
        );
    }

    /// Explicit level the new model rejects is clamped to the model default;
    /// provenance stays explicit so later switches still honor the user.
    #[test]
    fn explicit_unsupported_level_is_clamped_to_model_default_on_switch() {
        let mut rt = Runtime::new_headless();
        rt.set_reasoning_level_explicit(ReasoningLevel::XHigh);
        let clamp = rt
            .try_set_model("xai-auth/grok-4.6".to_string())
            .expect("model switch succeeds");
        assert_eq!(
            clamp,
            Some(ReasoningClamp {
                from: ReasoningLevel::XHigh,
                to: ReasoningLevel::High,
            })
        );
        assert_eq!(rt.reasoning_level(), ReasoningLevel::High);
        assert!(
            rt.is_reasoning_explicit(),
            "clamp keeps explicit provenance"
        );
        assert_eq!(rt.model(), "xai-auth/grok-4.6");
    }

    /// Explicit level the new model supports is left alone: no clamp reported.
    #[test]
    fn explicit_supported_level_is_not_clamped_on_switch() {
        let mut rt = Runtime::new_headless();
        rt.set_reasoning_level_explicit(ReasoningLevel::High);
        let clamp = rt.try_set_model("xai-auth/grok-4.6".to_string()).unwrap();
        assert_eq!(clamp, None);
        assert_eq!(rt.reasoning_level(), ReasoningLevel::High);
        assert!(rt.is_reasoning_explicit());
    }

    /// Non-explicit levels keep taking the new model's default (unchanged
    /// behavior) and report no clamp — the default is applied, not substituted.
    #[test]
    fn non_explicit_default_application_reports_no_clamp() {
        let mut rt = Runtime::new_headless();
        rt.set_reasoning_level(ReasoningLevel::XHigh);
        let clamp = rt.try_set_model("xai-auth/grok-4.6".to_string()).unwrap();
        assert_eq!(clamp, None);
        assert_eq!(rt.reasoning_level(), ReasoningLevel::High);
        assert!(!rt.is_reasoning_explicit());
    }

    /// Explicit Max on an xAI model without Max support clamps to a level that
    /// validates for that exact model (documented default Adaptive).
    #[test]
    fn explicit_max_clamps_to_validating_option_on_multi_agent_model() {
        let model = "xai-auth/grok-4.20-multi-agent-0309";
        let mut rt = Runtime::new_headless();
        rt.set_reasoning_level_explicit(ReasoningLevel::Max);
        let clamp = rt
            .try_set_model(model.to_string())
            .unwrap()
            .expect("Max is not supported: must clamp");
        assert_eq!(clamp.from, ReasoningLevel::Max);
        assert!(
            crate::runtime::openai::catalog::validation::validate_reasoning_mutation(
                model, clamp.to
            )
            .is_ok(),
            "clamped level {} must validate",
            clamp.to
        );
        assert_eq!(rt.reasoning_level(), clamp.to);
        assert!(rt.is_reasoning_explicit());
    }

    /// `set_model` (infallible wrapper) still surfaces the clamp.
    #[test]
    fn set_model_returns_clamp() {
        let mut rt = Runtime::new_headless();
        rt.set_reasoning_level_explicit(ReasoningLevel::XHigh);
        let clamp = rt.set_model("xai-auth/grok-4.6".to_string());
        assert_eq!(clamp.map(|c| c.to), Some(ReasoningLevel::High));
    }

    /// Resume path: a saved grok+xhigh session (written before clamping
    /// existed) must not resume into a permanently failing state.
    #[test]
    fn restore_session_reasoning_clamps_after_reapply() {
        let mut rt = Runtime::new_headless();
        rt.set_model("xai-auth/grok-4.6".to_string());
        let clamp = rt.restore_session_reasoning("xhigh");
        assert_eq!(
            clamp,
            Some(ReasoningClamp {
                from: ReasoningLevel::XHigh,
                to: ReasoningLevel::High,
            })
        );
        assert_eq!(rt.reasoning_level(), ReasoningLevel::High);
        assert!(rt.is_reasoning_explicit());
        assert!(rt.set_reasoning_level_checked(rt.reasoning_level()).is_ok());
    }

    /// Supported saved level restores verbatim with no clamp.
    #[test]
    fn restore_session_reasoning_keeps_supported_level() {
        let mut rt = Runtime::new_headless();
        rt.set_model("xai-auth/grok-4.6".to_string());
        assert_eq!(rt.restore_session_reasoning("low"), None);
        assert_eq!(rt.reasoning_level(), ReasoningLevel::Low);
        assert!(rt.is_reasoning_explicit());
    }

    /// Boot path: config `model` + unsupported explicit `thinking` no longer
    /// boots into a permanently failing state.
    #[test]
    fn apply_config_clamps_unsupported_explicit_level_against_config_model() {
        let mut rt = Runtime::new_headless();
        rt.apply_config(&crate::config::SynapsConfig {
            model: Some("xai-auth/grok-4.6".to_string()),
            thinking_level: Some(ReasoningLevel::XHigh),
            ..Default::default()
        });
        assert_eq!(rt.model(), "xai-auth/grok-4.6");
        assert_eq!(rt.reasoning_level(), ReasoningLevel::High);
        assert!(rt.is_reasoning_explicit());
        assert!(rt.set_reasoning_level_checked(rt.reasoning_level()).is_ok());
    }
}

#[cfg(test)]
mod codex_execution_preflight_tests {
    use super::*;
    use crate::runtime::openai::catalog::CodexRequestRole;
    use agent_core::prompt::QualifiedModelId;
    use agent_core::reasoning::ReasoningLevel;

    fn ultra_runtime() -> Runtime {
        let mut runtime = Runtime::new_headless();
        runtime.set_model("openai-codex/gpt-5.6-sol".to_string());
        runtime.set_reasoning_level_explicit(ReasoningLevel::Ultra);
        runtime
    }

    fn install_baseline(runtime: &mut Runtime) {
        let foreground =
            QualifiedModelId::parse("openai-codex/gpt-5.6-sol").expect("qualified model");
        let orchestration = crate::orchestration::OrchestrationRuntime::baseline(foreground, 3, 8)
            .expect("baseline orchestration");
        runtime.install_orchestration(Arc::new(orchestration));
    }

    #[tokio::test]
    async fn foreground_ultra_preflight_requires_orchestration() {
        let runtime = ultra_runtime();
        let error = runtime
            .validate_request_preflight()
            .await
            .expect_err("Ultra without orchestration must fail closed");
        assert!(error.to_string().contains("orchestration"), "{error}");
    }

    #[tokio::test]
    async fn foreground_ultra_preflight_requires_actionable_subagent_tools() {
        let mut runtime = ultra_runtime();
        install_baseline(&mut runtime);
        runtime.set_tools(ToolRegistry::without_subagent());

        let error = runtime
            .validate_request_preflight()
            .await
            .expect_err("Ultra without subagent tools must fail closed");
        assert!(error.to_string().contains("subagent"), "{error}");
    }

    #[tokio::test]
    async fn foreground_ultra_preflight_accepts_exact_model_policy_and_tools() {
        let mut runtime = ultra_runtime();
        install_baseline(&mut runtime);
        runtime
            .validate_request_preflight()
            .await
            .expect("fully provisioned Ultra must pass preflight");
    }

    #[tokio::test]
    async fn worker_ultra_preflight_never_requires_or_enables_orchestration() {
        let mut runtime = ultra_runtime();
        runtime.set_codex_request_role(CodexRequestRole::Worker);
        runtime.set_tools(ToolRegistry::without_subagent());
        runtime
            .validate_request_preflight()
            .await
            .expect("worker Ultra is underlying max without proactive orchestration");
        assert_eq!(runtime.codex_request_role(), CodexRequestRole::Worker);
    }

    #[tokio::test]
    async fn non_codex_ultra_preflight_fails_before_provider_work() {
        let mut runtime = Runtime::new_headless();
        runtime.set_model("anthropic/claude-opus-4-7".to_string());
        runtime.set_reasoning_level_explicit(ReasoningLevel::Ultra);
        let error = runtime
            .validate_request_preflight()
            .await
            .expect_err("non-Codex Ultra must fail closed");
        assert!(
            error.to_string().contains("unsupported_reasoning_level"),
            "{error}"
        );
    }

    #[tokio::test]
    async fn compaction_preflight_validates_target_before_credentials() {
        let mut runtime = ultra_runtime();
        runtime.set_compaction_model(Some("claude-sonnet-4-6".to_string()));
        {
            let mut auth = runtime.auth.write().await;
            auth.auth_token.clear();
            auth.auth_type = "none".to_string();
            auth.refresh_token = None;
            auth.token_expires = None;
        }

        let error = runtime
            .compact_call(Vec::new())
            .await
            .expect_err("Codex Ultra cannot silently cross into Anthropic compaction");
        assert!(
            error.to_string().contains("authoritative exact-model"),
            "preflight must reject the target before auth access: {error}"
        );
    }
}

#[cfg(test)]
mod http_client_timeout_tests {
    //! Regression tests for the runtime HTTP client's timeout semantics.
    //!
    //! The client used to set `.timeout(300s)` — a **total wall-clock
    //! deadline** that keeps ticking even while a streaming response is
    //! actively delivering bytes (reqwest `TotalTimeoutBody` never resets).
    //! Two failure modes: a dead connection took a full 5 minutes to
    //! surface (incident: session 20260714-025948-3dab), and any healthy
    //! stream longer than 300s was killed mid-flight. The fix is an
    //! idle-based `read_timeout` that resets on every received chunk.
    //!
    //! Servers here are raw TCP writing chunked HTTP/1.1 — full control
    //! over inter-chunk delays with no framework buffering in the way.

    use futures::StreamExt;
    use std::time::Duration;
    use tokio::io::AsyncWriteExt;

    const CHUNKED_HEAD: &[u8] =
        b"HTTP/1.1 200 OK\r\ncontent-type: text/event-stream\r\ntransfer-encoding: chunked\r\n\r\n";

    fn chunk(data: &str) -> Vec<u8> {
        format!("{:x}\r\n{}\r\n", data.len(), data).into_bytes()
    }

    /// Spawn a raw server: writes headers, then `n_chunks` chunks spaced
    /// `gap` apart, then (optionally) the terminating chunk.
    async fn spawn_chunked_server(n_chunks: usize, gap: Duration, terminate: bool) -> String {
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();
        tokio::spawn(async move {
            let (mut sock, _) = listener.accept().await.unwrap();
            // Drain the request head (we don't care about its contents).
            let mut buf = [0u8; 4096];
            use tokio::io::AsyncReadExt;
            let _ = sock.read(&mut buf).await;
            sock.write_all(CHUNKED_HEAD).await.unwrap();
            for i in 0..n_chunks {
                sock.write_all(&chunk(&format!("data {i}\n")))
                    .await
                    .unwrap();
                sock.flush().await.unwrap();
                tokio::time::sleep(gap).await;
            }
            if terminate {
                sock.write_all(b"0\r\n\r\n").await.unwrap();
            } else {
                // Stall forever (until the client gives up and drops us).
                tokio::time::sleep(Duration::from_secs(60)).await;
            }
        });
        format!("http://{addr}/stream")
    }

    async fn drain(
        client: &reqwest::Client,
        url: &str,
    ) -> Result<usize, Box<dyn std::error::Error + Send + Sync>> {
        let resp = client.get(url).send().await?;
        let mut stream = resp.bytes_stream();
        let mut total = 0usize;
        while let Some(c) = stream.next().await {
            total += c?.len();
        }
        Ok(total)
    }

    /// A healthy stream that keeps delivering bytes must NOT be killed,
    /// even when its total duration exceeds the read timeout — received
    /// data resets the idle clock. (This is the property the old total
    /// deadline violated at 300s.)
    #[tokio::test]
    async fn active_stream_outliving_read_timeout_survives() {
        // 15 chunks × 100ms ≈ 1.5s total, read_timeout = 400ms.
        let url = spawn_chunked_server(15, Duration::from_millis(100), true).await;
        let client = super::build_http_client(Duration::from_millis(400)).expect("client builds");
        let total = drain(&client, &url)
            .await
            .expect("active stream must never be killed by the idle timeout");
        assert!(total > 0);
    }

    /// A stream that stalls (bytes stop arriving) must be killed within
    /// the read timeout — not after a 300s total deadline.
    #[tokio::test]
    async fn stalled_stream_is_killed_by_read_timeout() {
        // 2 quick chunks then permanent stall, read_timeout = 300ms.
        let url = spawn_chunked_server(2, Duration::from_millis(10), false).await;
        let client = super::build_http_client(Duration::from_millis(300)).expect("client builds");
        let start = std::time::Instant::now();
        let err = drain(&client, &url)
            .await
            .expect_err("stalled stream must be detected");
        assert!(
            start.elapsed() < Duration::from_secs(5),
            "stall must be detected promptly, took {:?}",
            start.elapsed()
        );
        let msg = crate::core::error::error_chain_string(err.as_ref());
        assert!(
            msg.contains("timed out") || msg.contains("timeout"),
            "must be a timeout error: {msg}"
        );
    }

    /// A server that accepts but never sends response headers must be
    /// killed by the read timeout too (the incident's failure mode —
    /// previously took the full 300s total deadline to surface).
    #[tokio::test]
    async fn hung_headers_are_killed_by_read_timeout() {
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();
        tokio::spawn(async move {
            let (_sock, _) = listener.accept().await.unwrap();
            tokio::time::sleep(Duration::from_secs(60)).await;
        });
        let client = super::build_http_client(Duration::from_millis(300)).expect("client builds");
        let start = std::time::Instant::now();
        let err = client
            .get(format!("http://{addr}/codex/responses"))
            .send()
            .await
            .expect_err("hung request must be detected");
        assert!(
            start.elapsed() < Duration::from_secs(5),
            "hung headers must be detected promptly, took {:?}",
            start.elapsed()
        );
        assert!(err.is_timeout(), "must be a timeout: {err:?}");
    }
}

/// The memory binding of a runtime: made once, for the runtime's folder,
/// whatever an old config still says about a backend.
#[cfg(test)]
mod memory_binding_tests {
    use super::Runtime;
    use crate::test_env::{BaseDirGuard, EnvVarGuard};
    use crate::tools::test_helpers::context_with_memory;
    use crate::tools::Tool;
    use serde_json::json;
    use std::path::Path;

    /// A folder that is a project of its own.
    fn project() -> tempfile::TempDir {
        let dir = tempfile::tempdir().unwrap();
        std::fs::write(dir.path().join(".synaps-project"), "").unwrap();
        dir
    }

    fn key_of(dir: &Path) -> String {
        let scope = agent_core::memory::store::ProjectScope::discover(dir).unwrap();
        scope.key().to_owned()
    }

    fn bound_key(runtime: &Runtime) -> String {
        runtime.memory_backend.scope().unwrap().key().to_owned()
    }

    /// A config from before the cut, with all four backend keys active. The
    /// helper and the brain it names do not exist, and the brain path is
    /// relative: each of these used to leave the backend unavailable.
    const OLD_AXEL_CONFIG: &str = "memory.backend = axel\n\
         memory.user_scope = true\n\
         memory.axel.executable = /nonexistent/synaps-axel-memory-service\n\
         memory.axel.brain = relative/brain.r8\n\
         context_management.mode = auto\n";

    /// The binding is made by the first `apply_config`, for the folder the
    /// runtime was given before it, and never again: not by a second
    /// `apply_config`, not after the folder changes, not when the config
    /// changes. A clone keeps it, with the same author.
    #[test]
    #[serial_test::serial(synaps_base_dir)]
    fn a_runtime_binds_its_folders_project_once() {
        let _root = EnvVarGuard::remove("SYNAPS_PROJECT_ROOT");
        let (first, second) = (project(), project());
        assert_ne!(key_of(first.path()), key_of(second.path()));
        let plain = agent_core::config::load_config_from_str("");
        let old = agent_core::config::load_config_from_str(OLD_AXEL_CONFIG);

        let mut runtime = Runtime::new_headless();
        runtime.set_cwd(Some(first.path().to_path_buf()));
        runtime.apply_config(&plain);
        assert_eq!(bound_key(&runtime), key_of(first.path()));
        let author = runtime.memory_backend.author().clone();

        runtime.set_cwd(Some(second.path().to_path_buf()));
        runtime.apply_config(&plain);
        runtime.apply_config(&old);
        runtime.bind_memory();
        assert_eq!(bound_key(&runtime), key_of(first.path()));
        assert_eq!(runtime.memory_backend.author(), &author);

        let clone = runtime.clone();
        assert_eq!(bound_key(&clone), key_of(first.path()));
        assert_eq!(clone.memory_backend.author(), &author);
        assert_eq!(clone.memory_backend.base(), runtime.memory_backend.base());
    }

    /// T-B2, ruling 4: a config that still has `memory.backend = axel` and
    /// the three keys beside it starts a runtime. It reports the one
    /// warning, passes the request preflight, lets automatic
    /// context management be switched on, and its memory tools and its
    /// rollover seal act on the plain files of the chat's project. No
    /// helper is started and no brain is made.
    #[tokio::test]
    #[serial_test::serial(synaps_base_dir)]
    async fn an_old_axel_config_starts_a_runtime_on_plain_files() {
        let base = BaseDirGuard::new();
        let _root = EnvVarGuard::remove("SYNAPS_PROJECT_ROOT");
        let chat = project();
        let config = agent_core::config::load_config_from_str(OLD_AXEL_CONFIG);
        assert_eq!(config.warnings.len(), 1, "{:?}", config.warnings);
        assert!(config.warnings[0].starts_with("The Axel memory backend (memory.backend = axel, "));
        assert_eq!(
            config.context_management.mode,
            agent_core::config::ContextManagementMode::Auto
        );

        let mut runtime = Runtime::new_headless();
        runtime.set_cwd(Some(chat.path().to_path_buf()));
        runtime.apply_config(&config);
        runtime.apply_config(&config);
        // No "selected memory backend unavailable" or reconfigure-denied
        // refusal is left in the preflight every request passes.
        runtime
            .validate_request_preflight()
            .await
            .expect("an old Axel config passes the request preflight");
        let key = key_of(chat.path());
        assert_eq!(bound_key(&runtime), key);
        assert_eq!(runtime.memory_backend.base(), base.path());
        // Automatic rollover is on from the config and can be set again:
        // the "selected backend unavailable" refusals are gone.
        assert!(runtime.context_management_enabled());
        runtime.context_management_command("auto").expect("/context auto is not refused");

        let ctx = || context_with_memory(runtime.memory_backend.clone());
        let stored = crate::tools::memory::MemoryStoreTool
            .execute(json!({"content": "NOTE_UNDER_AN_OLD_CONFIG"}), ctx())
            .await
            .unwrap();
        assert!(stored.contains(&format!("project {key}")), "{stored}");
        let id = stored.split_whitespace().nth(2).unwrap().to_owned();
        // The note is a line of the project's plain file.
        let file = base.path().join("memory").join(format!("project-{key}.jsonl"));
        let lines = std::fs::read_to_string(&file).expect("the project's notes file");
        assert!(lines.contains("NOTE_UNDER_AN_OLD_CONFIG") && lines.contains(&id), "{lines}");
        let found = crate::tools::memory::MemorySearchTool
            .execute(json!({"query": "NOTE_UNDER_AN_OLD_CONFIG"}), ctx())
            .await
            .unwrap();
        assert!(found.contains(&id), "{found}");
        let fetched = crate::tools::memory::MemoryFetchTool
            .execute(json!({"ids": [id]}), ctx())
            .await
            .unwrap();
        assert!(fetched.contains("NOTE_UNDER_AN_OLD_CONFIG"), "{fetched}");

        // A rollover's seal writes an archive file under the same project.
        let window = vec![std::sync::Arc::new(json!({"role": "user", "content": "sealed words"}))];
        let (parts, _) = runtime
            .memory_backend
            .history_seal_window("old-config-chat", &window, "")
            .await
            .unwrap();
        let archive = &parts[0].archive.id;
        let sealed = agent_core::context_archive::archive_file(base.path(), &key, archive);
        assert!(sealed.unwrap().expect("an archive file").starts_with(base.path()));
        let history = crate::tools::memory::MemoryFetchTool
            .execute(json!({"ids": [format!("ctx-{archive}")]}), ctx())
            .await
            .unwrap();
        assert!(history.contains("sealed words"), "{history}");

        crate::tools::memory::MemoryForgetTool
            .execute(json!({"id": id}), ctx())
            .await
            .unwrap();
        // Nothing of the old backend was made: the two plain stores are
        // there, a brain is not.
        assert!(base.path().join("memory").is_dir());
        assert!(base.path().join("context-archives").is_dir());
        assert!(!base.path().join("memory/axel").exists());
        assert!(!chat.path().join("relative").exists());
    }

    /// A binding made with no folder, or with a deleted one, has no project:
    /// no id is invented for it. (The lease half of this test went with
    /// automatic memory.)
    #[test]
    #[serial_test::serial(synaps_base_dir)]
    fn a_binding_without_a_folder_has_no_project() {
        let gone = tempfile::tempdir().unwrap();
        let gone_path = gone.path().join("deleted");
        std::fs::create_dir(&gone_path).unwrap();
        std::fs::remove_dir(&gone_path).unwrap();
        for cwd in [None, Some(gone_path)] {
            let mut runtime = Runtime::new_headless();
            runtime.memory_backend = crate::memory_backend::MemoryBinding::for_cwd(cwd.clone());
            assert!(runtime.memory_backend.scope().is_err(), "{cwd:?} must have no project");
        }
    }

    /// An old config with the removed automatic-memory keys reaches nothing
    /// in a runtime: it is bound exactly like one given an empty config.
    #[test]
    #[serial_test::serial(synaps_base_dir)]
    fn an_old_automatic_memory_config_changes_nothing_on_a_runtime() {
        let old = agent_core::config::load_config_from_str(
            "memory.default_mode = capture_and_recall\n\
             memory.default_mode_confirmed = true\n\
             memory.recall_max_records = 12\n\
             memory.recall_max_tokens = 2048\n\
             memory.recall_timeout_ms = 250\n\
             memory.capture_tools = off\n\
             memory.capture_assistant = false\n\
             memory.capture_user = false\n\
             memory.auto_consolidate = on\n\
             memory.local_embeddings = on\n\
             memory.gliner = on\n",
        );
        assert_eq!(old.warnings.len(), 1, "{:?}", old.warnings);
        let plain = agent_core::config::load_config_from_str("");
        let mut with_old = Runtime::new_headless();
        with_old.apply_config(&old);
        with_old.apply_config(&old);
        let mut with_plain = Runtime::new_headless();
        with_plain.apply_config(&plain);
        assert_eq!(with_old.memory_backend.base(), with_plain.memory_backend.base());
        assert_eq!(
            with_old.memory_backend.scope().ok().map(|scope| scope.key().to_owned()),
            with_plain.memory_backend.scope().ok().map(|scope| scope.key().to_owned()),
        );
    }

    /// Any `memory.backend` value binds the same plain-file project as no
    /// such line: there is no unavailable backend to fall into.
    #[test]
    #[serial_test::serial(synaps_base_dir)]
    fn any_backend_selector_binds_the_same_plain_files() {
        let _root = EnvVarGuard::remove("SYNAPS_PROJECT_ROOT");
        let chat = project();
        for selector in ["axel", "legacy", "sqlite", "invalid-not-axel-not-legacy", ""] {
            let mut runtime = Runtime::new_headless();
            runtime.set_cwd(Some(chat.path().to_path_buf()));
            runtime.apply_config(&agent_core::config::load_config_from_str(&format!(
                "memory.backend = {selector}\n"
            )));
            assert_eq!(bound_key(&runtime), key_of(chat.path()), "{selector}");
            assert_eq!(runtime.memory_backend.base(), agent_core::config::base_dir());
        }
    }
}