agent_engine/extensions/

manager.rs

//! Extension manager — discovers, starts, and manages extension lifecycles.

use std::collections::HashMap;
use std::path::PathBuf;
use std::sync::Arc;

use super::capability::{
    ExtensionCapabilitySnapshot, FutureCapabilityEntry, HookCapabilityEntry, ToolCapabilityEntry,
};
use super::config::{diagnose_extension_config, ExtensionConfigDiagnostics};
use super::hooks::HookBus;
use super::info::PluginInfo;
use super::manifest::{ExtensionConfigEntry, ExtensionManifest};
use super::providers::{ProviderRegistry, RegisteredProvider, RegisteredProviderSummary};
use super::permissions::{Permission, PermissionSet};
use super::runtime::process::ProcessExtension;
use super::runtime::{ExtensionHandler, ExtensionHealth};
use serde_json::{Map, Value};

fn project_plugins_disabled() -> bool {
    std::env::var("SYNAPS_DISABLE_PROJECT_PLUGINS")
        .map(|value| {
            let normalized = value.trim().to_ascii_lowercase();
            matches!(normalized.as_str(), "1" | "true" | "yes" | "on")
        })
        .unwrap_or(false)
}

/// Whether a `plugins/` directory entry name denotes an INERT copy of a
/// plugin — a backup, disabled, or temporary directory — rather than an
/// installed plugin. Inert directories are excluded from discovery so a
/// stale copy can never register tools or execute alongside the real one
/// (user incident: `axel-memory-manager.backup.<TS>` dirs were discovered
/// as live plugins and their binaries spawned next to the real v0.3).
///
/// Name-based and conservative: only well-known suffix conventions (plus
/// the timestamped `.backup.<...>` infix) are excluded, so legitimate
/// dotted/hyphenated plugin names (`com.example.notes`, `backup-manager`)
/// stay discoverable. Exact plugin IDs of surviving plugins are untouched.
fn is_inert_plugin_dir_name(name: &str) -> bool {
    let lower = name.to_ascii_lowercase();
    // Timestamped backups: `<plugin>.backup.20260720101416`.
    if lower.contains(".backup.") {
        return true;
    }
    const INERT_SUFFIXES: &[&str] = &[
        ".backup",
        ".update-backup", // written by the plugin updater
        ".bak",
        ".disabled",
        ".old",
        ".orig",
        ".tmp",
        ".temp",
    ];
    INERT_SUFFIXES.iter().any(|s| lower.ends_with(s))
}

fn installed_plugin_setup_failure_in(
    state: &crate::skills::state::PluginsState,
    plugin_name: &str,
) -> Option<String> {
    let plugin = state.installed.iter().find(|p| p.name == plugin_name)?;
    match &plugin.setup_status {
        crate::skills::state::SetupStatus::Failed { message, .. } => Some(message.clone()),
        _ => None,
    }
}

/// The directories discovery walks, in override order: the user's
/// `plugins/`, then the project's `.synaps/plugins/` (unless disabled). A
/// plugin directory in a later root replaces one of the same name earlier.
fn plugin_roots() -> Vec<PathBuf> {
    let mut roots = vec![crate::config::base_dir().join("plugins")];
    if !project_plugins_disabled() {
        if let Ok(cwd) = std::env::current_dir() {
            let project_plugins = cwd.join(".synaps").join("plugins");
            if project_plugins != roots[0] {
                roots.push(project_plugins);
            }
        }
    }
    roots
}

/// The registry entries (name, current tool) of `names` that exist.
fn kept_entries(registry: &crate::ToolRegistry, names: &[String]) -> Vec<(String, Arc<dyn crate::Tool>)> {
    names
        .iter()
        .filter_map(|n| registry.get(n).map(|t| (n.clone(), Arc::clone(t))))
        .collect()
}

/// The same tool object (data pointer; vtables may differ per codegen unit).
fn same_tool(a: &Arc<dyn crate::Tool>, b: &Arc<dyn crate::Tool>) -> bool {
    std::ptr::eq(Arc::as_ptr(a) as *const (), Arc::as_ptr(b) as *const ())
}

/// The directory discovery would load plugin `name` from: the last root
/// (see [`plugin_roots`]) that has it. `None` when no root has it.
fn locate_plugin_dir(name: &str) -> Option<PathBuf> {
    plugin_roots()
        .into_iter()
        .rev()
        .map(|root| root.join(name))
        .find(|dir| dir.is_dir())
}

/// Read `plugin_dir`'s `.synaps-plugin/plugin.json` and resolve its
/// `extension` block the way discovery loads it (command and args made
/// absolute against the plugin directory). `Ok(None)`: no manifest, or a
/// manifest without an `extension` block (not an extension). Returns the
/// manifest path and the raw JSON too, for the load-failure hint.
fn read_extension_manifest(
    plugin_name: &str,
    plugin_dir: &std::path::Path,
) -> Result<Option<(ExtensionManifest, PathBuf, Value)>, ExtensionLoadFailure> {
    let manifest_path = plugin_dir.join(".synaps-plugin").join("plugin.json");
    if !manifest_path.exists() {
        return Ok(None);
    }

    let content = match std::fs::read_to_string(&manifest_path) {
        Ok(c) => c,
        Err(e) => {
            let reason = format!("Failed to read plugin manifest: {e}");
            tracing::warn!(plugin = %plugin_name, manifest = %manifest_path.display(), error = %e, "Failed to read plugin manifest");
            return Err(ExtensionLoadFailure::new(
                plugin_name,
                Some(manifest_path),
                reason,
                "Check manifest file permissions",
            ));
        }
    };

    let json: serde_json::Value = match serde_json::from_str(&content) {
        Ok(v) => v,
        Err(e) => {
            let reason = format!("Invalid plugin manifest JSON: {e}");
            tracing::warn!(plugin = %plugin_name, manifest = %manifest_path.display(), error = %e, "Invalid plugin manifest JSON");
            return Err(ExtensionLoadFailure::new(
                plugin_name,
                Some(manifest_path),
                reason,
                "Fix the JSON syntax in .synaps-plugin/plugin.json",
            ));
        }
    };

    let ext_value = match json.get("extension") {
        Some(v) => v.clone(),
        None => return Ok(None),
    };

    let ext_manifest: ExtensionManifest = match serde_json::from_value(ext_value) {
        Ok(m) => m,
        Err(e) => {
            let reason = format!("Failed to parse extension manifest: {e}");
            tracing::warn!(plugin = %plugin_name, manifest = %manifest_path.display(), error = %e, "Failed to parse extension manifest");
            return Err(ExtensionLoadFailure::new(
                plugin_name,
                Some(manifest_path),
                reason,
                "Check the `extension` block shape against docs/extensions/contract.json",
            ));
        }
    };

    #[allow(clippy::if_same_then_else)]
    let command = if std::path::Path::new(&ext_manifest.command).is_absolute() {
        ext_manifest.command.clone()
    } else if !ext_manifest.command.contains(std::path::MAIN_SEPARATOR)
        && !ext_manifest.command.contains('/')
    {
        ext_manifest.command.clone()
    } else {
        plugin_dir
            .join(&ext_manifest.command)
            .to_string_lossy()
            .to_string()
    };

    let args: Vec<String> = ext_manifest
        .args
        .iter()
        .map(|arg| {
            let arg_path = plugin_dir.join(arg);
            if arg_path.exists() {
                if let (Ok(canonical), Ok(plugin_canonical)) =
                    (arg_path.canonicalize(), plugin_dir.canonicalize())
                {
                    if canonical.starts_with(&plugin_canonical) {
                        return canonical.to_string_lossy().to_string();
                    }
                }
            }
            arg.clone()
        })
        .collect();

    let resolved = ExtensionManifest {
        theme_tokens: Default::default(),
        // Deferred declarations flow through from the installed
        // manifest (legacy `tools`/`activation` aliases are folded
        // into `deferred` at deserialization) so installed plugins
        // keep their zero-spawn deferred lifecycle instead of
        // silently loading eager.
        command,
        args,
        ..ext_manifest
    };
    Ok(Some((resolved, manifest_path, json)))
}

/// What [`ExtensionManager::reload_from_disk`] did, as the daemon reports
/// it to `synaps daemon reload-extension`.
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct ExtensionReloadReport {
    /// The plugin directory name the extension is keyed by.
    pub extension: String,
    /// It was loaded (live or deferred) before; `false` = first load, e.g. a
    /// plugin installed after the daemon started or one that failed then.
    pub was_loaded: bool,
    /// Loaded deferred (progressive disclosure): no process until first use.
    pub deferred: bool,
    /// Runtime names of the tools it has now (`<extension>:<tool>`), sorted.
    pub tools: Vec<String>,
    /// Tools it had before the reload that the new version does not have.
    #[serde(default)]
    pub removed_tools: Vec<String>,
    /// The directory it was loaded from.
    pub plugin_dir: PathBuf,
}

fn sanitize_hint_fragment(input: &str) -> String {
    input
        .chars()
        .map(|ch| if ch.is_control() { '?' } else { ch })
        .collect::<String>()
}

/// Actionable discovery/load failure for an installed plugin extension.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ExtensionLoadFailure {
    pub plugin: String,
    pub manifest_path: Option<PathBuf>,
    pub reason: String,
    pub hint: String,
}

impl ExtensionLoadFailure {
    fn new(
        plugin: impl Into<String>,
        manifest_path: Option<PathBuf>,
        reason: impl Into<String>,
        hint: impl Into<String>,
    ) -> Self {
        Self {
            plugin: plugin.into(),
            manifest_path,
            reason: reason.into(),
            hint: hint.into(),
        }
    }

    pub fn concise_message(&self) -> String {
        match &self.manifest_path {
            Some(path) => format!(
                "{} (manifest: {}; hint: {})",
                self.reason,
                path.display(),
                self.hint
            ),
            None => format!("{} (hint: {})", self.reason, self.hint),
        }
    }
}

/// Snapshot of a loaded extension's runtime status.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ExtensionStatus {
    pub id: String,
    pub health: ExtensionHealth,
    pub restart_count: usize,
}

/// Compute the hint for an extension load failure.
///
/// Two cases:
/// 1. **Missing extension binary AND plugin declares
///    `extension.setup`** — the plugin ships source only (the
///    binary is typically gitignored) and the setup script needs to
///    be run. The hint points the user at the exact command. This is
///    the common case for fresh marketplace installs of plugins that
///    build their extension binary from source.
/// 2. **Anything else** — a generic hint pointing at the manifest contract.
///
/// Pure function for unit-testability. Lives here (not in
/// `ExtensionLoadFailure`) because the setup convention is a
/// plugin-layer concern.
pub fn compute_extension_load_hint(
    error: &str,
    plugin_dir: &std::path::Path,
    declared_setup: Option<&str>,
) -> String {
    let missing_binary =
        error.contains("No such file or directory") || error.contains("os error 2");
    match (missing_binary, declared_setup) {
        (true, Some(setup)) => format!(
            "Extension binary missing — this plugin ships source only. Run the setup script from the plugin directory, then reload. plugin_dir={}, setup={}",
            sanitize_hint_fragment(&plugin_dir.display().to_string()),
            sanitize_hint_fragment(setup),
        ),
        _ => "Check the `extension` block in .synaps-plugin/plugin.json against docs/extensions/contract.json and confirm the extension command is installed"
            .to_string(),
    }
}

/// Process-level discovery state (daemon-mode phase 2, C2). Discovery walks
/// the plugin roots and spawns sidecars ONCE per manager; a second
/// `discover_and_load` call replays the recorded result instead of
/// re-walking (which would report every extension as "already loaded" —
/// N false failures on the second host in one process).
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub enum DiscoveryState {
    #[default]
    NotStarted,
    Running,
    Done {
        loaded: Vec<String>,
        failed: Vec<ExtensionLoadFailure>,
    },
}

/// Manages the lifecycle of all loaded extensions.
pub struct ExtensionManager {
    /// Once-per-process discovery record; see [`DiscoveryState`].
    discovery: DiscoveryState,
    /// The shared hook bus.
    hook_bus: Arc<HookBus>,
    /// Optional shared tool registry for extension-provided tools.
    tools: Option<Arc<tokio::sync::RwLock<crate::ToolRegistry>>>,
    /// Provider metadata registered by loaded extensions. Routing is not wired yet.
    providers: ProviderRegistry,
    /// Running extensions keyed by ID.
    extensions: HashMap<String, Arc<dyn ExtensionHandler>>,
    /// Validated permission metadata for successfully loaded EAGER handlers only.
    /// No resolved config or secrets; removed before any unload I/O.
    eager_permissions: HashMap<String, PermissionSet>,
    /// Declared manifest config entries per loaded extension, kept so we can
    /// produce diagnostics without re-reading the manifest.
    manifest_configs: HashMap<String, Vec<ExtensionConfigEntry>>,
    /// Capability declarations per loaded extension. Each plugin may
    /// declare zero or more capabilities (kind is plugin-defined; core
    /// does not enumerate). Populated on load.
    capabilities: HashMap<String, Vec<crate::extensions::runtime::process::CapabilityDeclaration>>,
    /// Optional plugin-reported info from the `info.get` RPC.
    plugin_info: HashMap<String, PluginInfo>,
    /// Theme tokens declared in each loaded extension's manifest (P19.2),
    /// kept so the TUI can merge them into the theme under
    /// `ext.<plugin-id>.<token>` without re-reading manifests. Extensions
    /// that declare none simply have no entry.
    manifest_theme_tokens: HashMap<String, std::collections::BTreeMap<String, String>>,
    /// Task 20: when true (progressive disclosure), tool-only manifests
    /// with validated passive declarations are NOT spawned at load — their
    /// dormant deferred tools are cataloged instead and the launch record
    /// is retained for exact-activation lease acquisition.
    progressive_deferral: bool,
    /// Launch records for deferred tool-only extensions (manifest + cwd +
    /// resolved config), keyed by plugin id, behind a SHARED handle: the
    /// [`crate::extensions::lease::ExtensionRuntimeManager`] reads the
    /// CURRENT record at every lease acquisition (sync lock, map ops only,
    /// never held across I/O). Consumed by the lease lifecycle only.
    deferred_tool_only: SharedDeferredRecords,
    /// Set by [`Self::reload`] for its load only: the dormant tool names of
    /// the deferred instance being replaced, which the new instance's
    /// catalog batch swaps out in the SAME registry write (see
    /// [`Self::reload`] on why the registry must never show a gap).
    reload_replaces: Vec<String>,
    /// Lazily-created shared extension runtime lease manager (Task 20
    /// Commit B). One instance shared by this manager (unload revocation)
    /// and the `Runtime` (per-session capabilities + session-end scope).
    extension_runtime: std::sync::OnceLock<Arc<crate::extensions::lease::ExtensionRuntimeManager>>,
    /// HOST-OWNED trusted project root, recorded ONCE per manager on first
    /// use (canonical `ProjectScope::discover` walk from the host's own
    /// working directory). Injected into resolved config for manifest
    /// entries declaring `host_context: "project_root"`; never sourced
    /// from env overrides, user config, or model input, and never
    /// forwarded as an environment variable.
    host_project_root: std::sync::OnceLock<Option<String>>,
}

/// Shared handle to the retained deferred launch records. Crate-private:
/// records hold RESOLVED (potentially secret) config.
pub(crate) type SharedDeferredRecords =
    Arc<std::sync::Mutex<HashMap<String, DeferredExtensionRecord>>>;

/// Retained launch expectation for a spawn-deferred tool-only extension.
///
/// INTERNAL AND OPAQUE (review fix A3): `config` holds RESOLVED values —
/// including `secret_env` material — so this record is crate-private,
/// exposes no public accessor, and deliberately implements neither
/// `Debug` nor `Serialize`. Public diagnostics may reveal only booleans/
/// counts (see [`ExtensionManager::is_deferred_tool_only`]); runtime
/// acquisition stays inside the manager/lease lifecycle API.
#[derive(Clone)]
pub(crate) struct DeferredExtensionRecord {
    pub(crate) manifest: ExtensionManifest,
    pub(crate) cwd: Option<std::path::PathBuf>,
    pub(crate) config: Value,
}

impl ExtensionManager {
    /// Create a new manager with a shared hook bus.
    pub fn new(hook_bus: Arc<HookBus>) -> Self {
        Self {
            discovery: DiscoveryState::NotStarted,
            hook_bus,
            tools: None,
            providers: ProviderRegistry::new(),
            extensions: HashMap::new(),
            eager_permissions: HashMap::new(),
            manifest_configs: HashMap::new(),
            capabilities: HashMap::new(),
            plugin_info: HashMap::new(),
            manifest_theme_tokens: HashMap::new(),
            progressive_deferral: false,
            deferred_tool_only: Arc::new(std::sync::Mutex::new(HashMap::new())),
            reload_replaces: Vec::new(),
            extension_runtime: std::sync::OnceLock::new(),
            host_project_root: std::sync::OnceLock::new(),
        }
    }

    /// Create a new manager with shared hook bus and tool registry.
    pub fn new_with_tools(
        hook_bus: Arc<HookBus>,
        tools: Arc<tokio::sync::RwLock<crate::ToolRegistry>>,
    ) -> Self {
        Self {
            discovery: DiscoveryState::NotStarted,
            hook_bus,
            tools: Some(tools),
            providers: ProviderRegistry::new(),
            extensions: HashMap::new(),
            eager_permissions: HashMap::new(),
            manifest_configs: HashMap::new(),
            capabilities: HashMap::new(),
            plugin_info: HashMap::new(),
            manifest_theme_tokens: HashMap::new(),
            progressive_deferral: false,
            deferred_tool_only: Arc::new(std::sync::Mutex::new(HashMap::new())),
            reload_replaces: Vec::new(),
            extension_runtime: std::sync::OnceLock::new(),
            host_project_root: std::sync::OnceLock::new(),
        }
    }

    /// Load and start an extension from its manifest.
    pub async fn load(&mut self, id: &str, manifest: &ExtensionManifest) -> Result<(), String> {
        self.load_with_cwd(id, manifest, None).await
    }

    /// Load and start an extension from its manifest with a process cwd.
    pub async fn load_with_cwd(
        &mut self,
        id: &str,
        manifest: &ExtensionManifest,
        cwd: Option<std::path::PathBuf>,
    ) -> Result<(), String> {
        let deferred: Vec<String> = self
            .deferred_records()
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .keys()
            .cloned()
            .collect();
        let loaded = self
            .extensions
            .keys()
            .map(String::as_str)
            .chain(deferred.iter().map(String::as_str));
        if let Some(other) = Self::folded_id_collision(id, loaded) {
            return Err(format!(
                "Extension '{id}' collides with loaded extension '{other}': ids that differ only in '-'/'_' or case share SYNAPS_EXTENSION_* override variables"
            ));
        }
        let config = Self::resolve_config(id, &manifest.config, self.host_project_root())?;
        self.load_with_cwd_and_config(id, manifest, cwd, config)
            .await
    }

    /// Enable Task 20 progressive spawn deferral (set at engine boot from
    /// the progressive-disclosure flag; default false keeps every legacy
    /// eager path byte-for-byte).
    pub fn set_progressive_deferral(&mut self, enabled: bool) {
        self.progressive_deferral = enabled;
    }

    /// The HOST-OWNED trusted project root, discovered ONCE per manager
    /// from the host's own working directory via the canonical
    /// `agent_core` project-scope walk (nearest `.git`/`.synaps-project`
    /// marker, canonicalized). Recorded here so every extension load in
    /// this manager's lifetime sees one stable value. `None` when the
    /// working directory is unavailable or cannot be canonicalized.
    pub fn host_project_root(&self) -> Option<&str> {
        self.host_project_root
            .get_or_init(|| {
                let cwd = std::env::current_dir().ok()?;
                agent_core::memory::store::ProjectScope::discover(&cwd)
                    .ok()
                    .map(|scope| scope.root().display().to_string())
            })
            .as_deref()
    }

    /// Whether one plugin id is currently deferred (any Task 20 class:
    /// tool-only, provider, hook/lifecycle, UI/sidecar, mixed — zero
    /// spawn). Boolean-only diagnostic: the underlying launch record
    /// (which holds resolved — potentially secret — config) is never
    /// exposed.
    pub fn is_deferred(&self, id: &str) -> bool {
        self.deferred_records()
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .contains_key(id)
    }

    /// Compatibility alias for [`Self::is_deferred`] (Commit B name).
    pub fn is_deferred_tool_only(&self, id: &str) -> bool {
        self.is_deferred(id)
    }

    /// Number of retained deferred tool-only launch records (count-only
    /// diagnostic; never the records).
    pub fn deferred_tool_only_count(&self) -> usize {
        self.deferred_records()
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .len()
    }

    fn deferred_records(&self) -> &SharedDeferredRecords {
        &self.deferred_tool_only
    }

    /// The SHARED extension runtime lease manager (Task 20 Commit B),
    /// created on first use against this manager's retained launch
    /// records. The engine installs the same instance on the `Runtime`
    /// (session capabilities + durable session-end scope); this manager
    /// uses it for unload/reload lease revocation. Runtime acquisition —
    /// spawn, re-validation, exact declaration matching — happens ONLY
    /// inside that manager's lease lifecycle.
    pub fn extension_runtime(&self) -> Arc<crate::extensions::lease::ExtensionRuntimeManager> {
        self.extension_runtime_with_idle(crate::extensions::lease::DEFAULT_IDLE_MAX)
    }

    /// As [`Self::extension_runtime`], but the FIRST caller fixes the idle
    /// reap bound (tests use short bounds; production uses the default).
    pub fn extension_runtime_with_idle(
        &self,
        idle_max: std::time::Duration,
    ) -> Arc<crate::extensions::lease::ExtensionRuntimeManager> {
        Arc::clone(self.extension_runtime.get_or_init(|| {
            Arc::new(crate::extensions::lease::ExtensionRuntimeManager::new(
                Arc::clone(&self.deferred_tool_only),
                idle_max,
            ))
        }))
    }

    /// Test-only internal introspection: clone one retained launch record
    /// (the lease manager reads the shared records handle directly).
    #[cfg(test)]
    pub(crate) fn deferred_launch_record(&self, id: &str) -> Option<DeferredExtensionRecord> {
        self.deferred_records()
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .get(id)
            .cloned()
    }

    /// Test seam: tamper the retained manifest permissions of one deferred
    /// record IN PLACE (simulates internal-state drift the public API can
    /// never produce, to prove acquisition re-validation fails closed).
    /// Reveals nothing.
    #[doc(hidden)]
    pub fn tamper_deferred_manifest_permissions_for_tests(
        &self,
        id: &str,
        permissions: Vec<String>,
    ) -> bool {
        let mut map = self
            .deferred_tool_only
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        match map.get_mut(id) {
            Some(record) => {
                record.manifest.permissions = permissions;
                true
            }
            None => false,
        }
    }

    /// Test seam: tamper the retained RESOLVED config of one deferred
    /// record in place (simulates launch-record/config drift under a live
    /// lease). Reveals nothing.
    #[doc(hidden)]
    pub fn tamper_deferred_config_for_tests(&self, id: &str, config: Value) -> bool {
        let mut map = self
            .deferred_tool_only
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        match map.get_mut(id) {
            Some(record) => {
                record.config = config;
                true
            }
            None => false,
        }
    }

    async fn load_with_cwd_and_config(
        &mut self,
        id: &str,
        manifest: &ExtensionManifest,
        cwd: Option<std::path::PathBuf>,
        config: Value,
    ) -> Result<(), String> {
        // Don't load duplicates (live OR deferred).
        if self.extensions.contains_key(id) || self.is_deferred_tool_only(id) {
            return Err(format!("Extension '{}' is already loaded", id));
        }

        // Validate permissions and hook subscriptions before spawning the
        // extension process. This keeps malformed manifests from leaking child
        // processes when a later subscription step fails.
        let validated = manifest.validate(id)?;
        if let Some(notice) =
            crate::extensions::lifecycle::ignored_context_provider_notice(id, manifest)
        {
            tracing::warn!("{notice}");
        }
        let permissions = validated.permissions;
        let subscriptions = validated.subscriptions;

        // Task 20: under progressive disclosure, EVERY manifest with a
        // validated `deferred` block defers its spawn entirely (spec §7.5):
        //  - tool-only: dormant descriptor tools cataloged; exact
        //    activation acquires the lease;
        //  - provider: declared provider METADATA registered with a lazy
        //    handler; first selected complete/stream acquires;
        //  - hook/lifecycle: permission-validated subscriptions registered
        //    with a lazy handler; the first AUTHORIZED matching event
        //    acquires;
        //  - UI/sidecar: record retained only; explicit user APIs acquire;
        //  - mixed: all of the above against ONE shared per-plugin lease
        //    (earliest legitimate capability starts it; tool SEARCH alone
        //    never does).
        // Legacy manifests (no `deferred` block) and flag-off stay on the
        // documented eager path byte-for-byte.
        let class = crate::extensions::lifecycle::classify(manifest);
        if self.progressive_deferral
            && class != crate::extensions::lifecycle::ExtensionClass::LegacyEager
        {
            let declared_tool_names: Vec<String> = manifest
                .deferred
                .as_ref()
                .map(|d| d.tools.iter().map(|t| format!("{id}:{}", t.name)).collect())
                .unwrap_or_default();
            let dormant = crate::extensions::lifecycle::dormant_extension_tools(id, manifest);
            let tool_count = dormant.len();
            if !dormant.is_empty() {
                let Some(tools) = &self.tools else {
                    return Err(format!(
                        "Extension '{}' declares deferred tools but no tool registry is available",
                        id
                    ));
                };
                // A reload swaps the replaced instance's dormant entries in
                // this same write: no reader ever sees them missing.
                let replacing = std::mem::take(&mut self.reload_replaces);
                let mut registry = tools.write().await;
                registry
                    .try_disable(&replacing)
                    .and_then(|()| registry.try_register_batch(dormant))
                    .map_err(|e| {
                        format!("Extension '{}' deferred tool catalog rejected: {}", id, e)
                    })?;
            }
            // Retain the launch record BEFORE registering lazy capability
            // surfaces so a capability firing immediately can acquire.
            self.deferred_tool_only
                .lock()
                .unwrap_or_else(std::sync::PoisonError::into_inner)
                .insert(
                    id.to_string(),
                    DeferredExtensionRecord {
                        manifest: manifest.clone(),
                        cwd,
                        config: config.clone(),
                    },
                );
            // Bounded unwind shared by the failure paths below: remove the
            // record, the dormant batch, and any provider metadata.
            macro_rules! unwind_deferred {
                () => {{
                    self.deferred_tool_only
                        .lock()
                        .unwrap_or_else(std::sync::PoisonError::into_inner)
                        .remove(id);
                    self.providers.unregister_plugin(id);
                    if let Some(tools) = &self.tools {
                        let _ = tools.write().await.try_disable(&declared_tool_names);
                    }
                    self.hook_bus.unsubscribe_all(id).await;
                }};
            }
            let lazy: Option<Arc<dyn ExtensionHandler>> = {
                // Created only when a lazy capability surface exists, so a
                // tool-only load never pre-initializes the runtime manager
                // (its idle policy stays first-caller-configurable).
                let needs_lazy = manifest
                    .deferred
                    .as_ref()
                    .map(|d| !d.providers.is_empty())
                    .unwrap_or(false)
                    || !manifest.hooks.is_empty();
                needs_lazy.then(|| {
                    Arc::new(crate::extensions::lease::LazyExtensionHandler::new(
                        id,
                        self.extension_runtime(),
                    )) as Arc<dyn ExtensionHandler>
                })
            };
            // Declared provider METADATA (validated shape) registered with
            // the lazy handler; routing re-validates the live runtime's
            // declarations against the manifest before any call.
            let declared_providers: Vec<_> = manifest
                .deferred
                .as_ref()
                .map(|d| d.providers.clone())
                .unwrap_or_default();
            let mut registered_provider_ids = Vec::new();
            for declared in &declared_providers {
                let spec = declared.to_registered_spec();
                if let Err(error) = Self::validate_provider_config_requirements(id, &spec, &config)
                {
                    unwind_deferred!();
                    return Err(error);
                }
                let handler = lazy.clone().expect("providers imply a lazy handler");
                match self
                    .providers
                    .register_with_handler(id, spec, Some(handler))
                {
                    Ok(runtime_id) => registered_provider_ids.push(runtime_id),
                    Err(error) => {
                        unwind_deferred!();
                        return Err(error);
                    }
                }
            }
            // Tool-use audit parity with the eager path.
            for runtime_id in &registered_provider_ids {
                if let Some(provider) = self.providers.get(runtime_id) {
                    let tool_use = provider.spec.models.iter().any(|m| {
                        m.capabilities
                            .get("tool_use")
                            .and_then(|v| v.as_bool())
                            .unwrap_or(false)
                    });
                    if tool_use {
                        tracing::warn!(
                            "Provider '{}' is tool-use capable: it can request Synaps tools through provider mediation. Use `/extensions trust disable {}` to block routing.",
                            runtime_id,
                            runtime_id,
                        );
                    }
                }
            }
            // Permission-validated hook subscriptions with the lazy
            // handler: the first AUTHORIZED matching event acquires.
            for (kind, tool_filter, matcher) in subscriptions {
                let handler = lazy.clone().expect("subscriptions imply a lazy handler");
                if let Err(error) = self
                    .hook_bus
                    .subscribe(kind, handler, tool_filter, matcher, permissions.clone())
                    .await
                {
                    unwind_deferred!();
                    return Err(error);
                }
            }
            self.manifest_configs
                .insert(id.to_string(), manifest.config.clone());
            if !manifest.theme_tokens.is_empty() {
                self.manifest_theme_tokens
                    .insert(id.to_string(), manifest.theme_tokens.clone());
            }
            tracing::info!(extension = %id, class = ?class, tools = tool_count,
                providers = registered_provider_ids.len(), hooks = manifest.hooks.len(),
                "Extension deferred: passive surfaces registered, no process started");
            return Ok(());
        }

        // Spawn the extension process only after the manifest is known-good.
        let process =
            ProcessExtension::spawn_with_cwd(id, &manifest.command, &manifest.args, cwd.clone())
                .await?;
        // Publish permissions to the inbound-request dispatcher so memory.*
        // calls during initialize can be authorized correctly.
        process.set_permissions(permissions.clone()).await;
        let capabilities = match process.initialize(cwd.clone(), config.clone()).await {
            Ok(capabilities) => capabilities,
            Err(error) => {
                process.shutdown().await;
                return Err(error);
            }
        };
        let registered_tools = capabilities.tools;
        let registered_providers = capabilities.providers;
        let capability_declarations = capabilities.capabilities;
        let should_probe_info = !registered_tools.is_empty()
            || !registered_providers.is_empty()
            || !capability_declarations.is_empty();
        let handler: Arc<dyn ExtensionHandler> = Arc::new(process);
        if !registered_tools.is_empty()
            && !permissions.has(crate::extensions::permissions::Permission::ToolsRegister)
        {
            handler.shutdown().await;
            return Err(format!(
                "Extension '{}' registered tools but lacks permission 'tools.register'",
                id
            ));
        }
        if !registered_providers.is_empty()
            && !permissions.has(crate::extensions::permissions::Permission::ProvidersRegister)
        {
            handler.shutdown().await;
            return Err(format!(
                "Extension '{}' registered providers but lacks permission 'providers.register'",
                id
            ));
        }
        for decl in &capability_declarations {
            if let Err(err) =
                crate::extensions::runtime::process::validate_capability(decl, &permissions)
            {
                handler.shutdown().await;
                return Err(format!(
                    "Extension '{}' capability '{}' invalid: {}",
                    id, decl.kind, err
                ));
            }
        }
        if !registered_providers.is_empty() {
            let mut registered_ids = Vec::new();
            for provider in registered_providers {
                if let Err(error) =
                    Self::validate_provider_config_requirements(id, &provider, &config)
                {
                    self.providers.unregister_plugin(id);
                    handler.shutdown().await;
                    return Err(error);
                }
                match self
                    .providers
                    .register_with_handler(id, provider, Some(handler.clone()))
                {
                    Ok(runtime_id) => registered_ids.push(runtime_id),
                    Err(error) => {
                        self.providers.unregister_plugin(id);
                        handler.shutdown().await;
                        return Err(error);
                    }
                }
            }
            tracing::info!(extension = %id, providers = ?registered_ids, "Extension provider metadata registered");
            // Warn for tool-use-capable providers so authors and users can audit them.
            for runtime_id in &registered_ids {
                if let Some(provider) = self.providers.get(runtime_id) {
                    let tool_use = provider.spec.models.iter().any(|m| {
                        m.capabilities
                            .get("tool_use")
                            .and_then(|v| v.as_bool())
                            .unwrap_or(false)
                    });
                    if tool_use {
                        tracing::warn!(
                            "Provider '{}' is tool-use capable: it can request Synaps tools through provider mediation. Use `/extensions trust disable {}` to block routing.",
                            runtime_id,
                            runtime_id,
                        );
                    }
                }
            }
        }
        if !registered_tools.is_empty() {
            let Some(tools) = &self.tools else {
                handler.shutdown().await;
                return Err(format!(
                    "Extension '{}' registered tools but no tool registry is available",
                    id
                ));
            };
            let mut registry = tools.write().await;
            for spec in registered_tools {
                let tool_name = spec.name.clone();
                if let Err(e) = registry.try_register(Arc::new(crate::tools::ExtensionTool::new(
                    id,
                    spec,
                    handler.clone(),
                ))) {
                    tracing::warn!(
                        extension = %id,
                        tool = %tool_name,
                        error = %e,
                        "Refusing to expose extension tool the capability catalog could not record"
                    );
                }
            }
        }

        // Do not probe optional info.get for legacy hook-only extensions. The
        // best-effort call can race with simple fixtures that exit after
        // shutdown/EOF and is only needed for richer extension-capability
        // surfaces (providers/tools/plugin-defined capabilities).
        let info = if should_probe_info {
            match handler.get_info().await {
                Ok(info) => Some(info),
                Err(error) => {
                    if error.contains("method not found") || error.contains("unknown method") {
                        tracing::debug!(
                            extension = %id,
                            error = %error,
                            "Extension did not provide optional info.get metadata",
                        );
                        None
                    } else {
                        tracing::warn!(
                            extension = %id,
                            error = %error,
                            "Ignoring invalid optional info.get metadata",
                        );
                        None
                    }
                }
            }
        } else {
            None
        };

        // Register hook subscriptions
        for (kind, tool_filter, matcher) in subscriptions {
            self.hook_bus
                .subscribe(
                    kind,
                    handler.clone(),
                    tool_filter,
                    matcher,
                    permissions.clone(),
                )
                .await?;
        }

        self.extensions.insert(id.to_string(), handler);
        self.eager_permissions.insert(id.to_string(), permissions);
        self.manifest_configs
            .insert(id.to_string(), manifest.config.clone());
        if !manifest.theme_tokens.is_empty() {
            self.manifest_theme_tokens
                .insert(id.to_string(), manifest.theme_tokens.clone());
        }
        if !capability_declarations.is_empty() {
            self.capabilities
                .insert(id.to_string(), capability_declarations);
        }
        if let Some(info) = info {
            self.plugin_info.insert(id.to_string(), info);
        }
        tracing::info!(extension = %id, hooks = manifest.hooks.len(), "Extension loaded");
        Ok(())
    }

    fn validate_provider_config_requirements(
        id: &str,
        provider: &crate::extensions::runtime::process::RegisteredProviderSpec,
        config: &Value,
    ) -> Result<(), String> {
        let Some(required) = provider
            .config_schema
            .as_ref()
            .and_then(|schema| schema.get("required"))
            .and_then(Value::as_array)
        else {
            return Ok(());
        };
        for key in required {
            let Some(key) = key.as_str() else {
                return Err(format!(
                    "Extension '{}' provider '{}' config_schema.required must contain only strings",
                    id, provider.id,
                ));
            };
            let present = config
                .as_object()
                .map(|map| map.contains_key(key))
                .unwrap_or(false);
            if !present {
                return Err(format!(
                    "Extension '{}' provider '{}' missing required provider config '{}'",
                    id, provider.id, key,
                ));
            }
        }
        Ok(())
    }

    /// Env vars an extension may never pull in through `secret_env`.
    ///
    /// Security (#207 MED-2): `secret_env` names an arbitrary process
    /// environment variable, and [`Self::resolve_config`] reads it verbatim
    /// into the config handed to the extension at `initialize`. Nothing
    /// constrained that name, so a manifest could declare the HOST's own
    /// provider credential — `secret_env: "ANTHROPIC_API_KEY"` — and a plugin
    /// installed for something unrelated would be handed the user's API key.
    ///
    /// Default-deny the credentials the host itself holds, plus the whole
    /// `SYNAPS_*` host namespace except this extension's own scoped override.
    /// Third-party secrets an extension legitimately needs (`GITHUB_TOKEN`,
    /// and friends) are unaffected.
    ///
    /// The credential set is derived from the provider registries (every
    /// `env_vars` entry of the broker's static table and the engine catalog)
    /// plus the credential sources read outside those tables, with the
    /// original hand-written list kept as a floor. Drift tests below fail if a
    /// registered or hand-read credential env var is not covered.
    ///
    /// `own_keys` are this manifest's declared config keys: inside `SYNAPS_*`
    /// only their exact override names are allowed, so extension `foo`
    /// cannot read `foo-bar`'s `SYNAPS_EXTENSION_FOO_BAR_TOKEN`.
    ///
    /// `var` must already have passed [`Self::secret_env_name`].
    pub(crate) fn denied_secret_env(
        var: &str,
        id: &str,
        own_keys: &[&str],
    ) -> Option<&'static str> {
        if var.starts_with("SYNAPS_")
            && !own_keys
                .iter()
                .any(|k| Self::extension_override_var(id, k) == var)
        {
            return Some(
                "SYNAPS_* is the host namespace; only this extension's own SYNAPS_EXTENSION_<ID>_<KEY> for a declared key is allowed",
            );
        }
        if Self::is_host_credential_env(var) {
            return Some(
                "that variable holds a host credential; extensions are never handed the host's own provider keys",
            );
        }
        None
    }

    /// The usable `secret_env` name: `raw` trimmed, if it matches
    /// `[A-Z_][A-Z0-9_]*`. `None` means a malformed name; the caller warns and
    /// ignores the `secret_env` (it is never read) instead of failing the load.
    pub(crate) fn secret_env_name(raw: &str) -> Option<&str> {
        let name = raw.trim();
        let mut chars = name.chars();
        let valid = chars
            .next()
            .is_some_and(|c| c.is_ascii_uppercase() || c == '_')
            && chars.all(|c| c.is_ascii_uppercase() || c.is_ascii_digit() || c == '_');
        valid.then_some(name)
    }

    /// The per-extension env override name for config `key`
    /// (`SYNAPS_EXTENSION_<ID>_<KEY>`, upper-cased, `-` folded to `_`).
    fn extension_override_var(id: &str, key: &str) -> String {
        format!(
            "SYNAPS_EXTENSION_{}_{}",
            id.replace('-', "_").to_ascii_uppercase(),
            key.replace('-', "_").to_ascii_uppercase()
        )
    }

    /// Another loaded id that differs from `id` only in `-`/`_` or case. The
    /// two would share every `SYNAPS_EXTENSION_<ID>_*` override variable.
    fn folded_id_collision<'a>(id: &str, loaded: impl Iterator<Item = &'a str>) -> Option<&'a str> {
        let fold = |s: &str| s.replace('-', "_").to_ascii_uppercase();
        let mine = fold(id);
        loaded
            .filter(|other| *other != id)
            .find(|other| fold(other) == mine)
    }

    /// `up` must already be upper-cased.
    fn is_host_credential_env(up: &str) -> bool {
        // Floor: the original #207 list. Never shrink it.
        const FLOOR: &[&str] = &[
            "ANTHROPIC_API_KEY",
            "OPENAI_API_KEY",
            "GEMINI_API_KEY",
            "GOOGLE_API_KEY",
            "XAI_API_KEY",
            "GROQ_API_KEY",
            "OPENROUTER_API_KEY",
            "DEEPSEEK_API_KEY",
            "MISTRAL_API_KEY",
            "AWS_SECRET_ACCESS_KEY",
            "AWS_SESSION_TOKEN",
        ];
        // Credential sources that are not rows in a provider registry.
        // (`SYNAPS_*` entries are also caught by the namespace rule; they are
        // listed so relaxing that rule can never expose them.)
        const OTHER_SOURCES: &[&str] = &[
            agent_core::auth::muse::API_KEY_ENV,
            "LOCAL_API_KEY",
            "OPENAI_CODEX_ACCESS_TOKEN",
            "AWS_ACCESS_KEY_ID",
            "SYNAPS_MACHINE_TOKEN",
            "SYNAPS_BROKER_TOKEN",
            // reqwest reads these by default; a proxy URL can embed
            // `user:pass@`.
            "HTTPS_PROXY",
            "HTTP_PROXY",
            "ALL_PROXY",
            "NO_PROXY",
        ];
        FLOOR.contains(&up)
            || OTHER_SOURCES.contains(&up)
            || agent_core::auth::static_providers::STATIC_PROVIDERS
                .iter()
                .any(|s| s.env_vars.contains(&up))
            || crate::runtime::openai::registry::providers()
                .iter()
                .any(|s| s.env_vars.contains(&up))
    }

    fn resolve_config(
        id: &str,
        entries: &[ExtensionConfigEntry],
        host_project_root: Option<&str>,
    ) -> Result<Value, String> {
        Self::resolve_config_with_env(id, entries, host_project_root, &|name| {
            std::env::var(name).ok()
        })
    }

    /// [`Self::resolve_config`] with the environment lookup injected, so
    /// tests can prove which variable names are (and are never) read.
    fn resolve_config_with_env(
        id: &str,
        entries: &[ExtensionConfigEntry],
        host_project_root: Option<&str>,
        env: &dyn Fn(&str) -> Option<String>,
    ) -> Result<Value, String> {
        let mut out = Map::new();
        let own_keys: Vec<&str> = entries.iter().map(|e| e.key.trim()).collect();
        for entry in entries {
            let key = entry.key.trim();
            if key.is_empty() {
                return Err(format!("Extension '{}' declares config with empty key", id));
            }
            if key.contains('.') || key.contains('/') || key.contains(' ') {
                return Err(format!(
                    "Extension '{}' config key '{}' must not contain dots, slashes, or spaces",
                    id, key,
                ));
            }
            // RESERVED host-context entries: the value comes EXCLUSIVELY
            // from trusted host state recorded once by this manager. Env
            // overrides, secret_env, plugin config, and legacy config keys
            // are all deliberately skipped — user or model input can never
            // supply or widen a host-context value.
            if let Some(host_key) = entry.host_context {
                if entry.secret_env.is_some() {
                    return Err(format!(
                        "Extension '{}' config key '{}' declares both host_context and secret_env; host-context values never carry secrets",
                        id, key,
                    ));
                }
                match host_key {
                    crate::extensions::manifest::HostContextKey::ProjectRoot => {
                        if let Some(root) = host_project_root {
                            out.insert(key.to_string(), Value::String(root.to_string()));
                        } else if let Some(default) = &entry.default {
                            out.insert(key.to_string(), default.clone());
                        } else if entry.required {
                            return Err(format!(
                                "Extension '{}' requires host context 'project_root' but the host could not resolve a trusted project root",
                                id,
                            ));
                        }
                    }
                }
                continue;
            }
            let config_key = format!("extension.{}.{}", id, key);
            // Security (#207 MED-2): validate the declared env var name BEFORE
            // any resolution path can read it. A denied name (host credential,
            // SYNAPS_*) fails the load loudly. A malformed name is ignored with
            // a warning: it is never read, and the value resolves as if no
            // secret_env had been declared.
            let secret_env: Option<&str> = match entry.secret_env.as_deref() {
                None => None,
                Some(raw) => match Self::secret_env_name(raw) {
                    None => {
                        let shown: String = raw.chars().take(64).collect();
                        tracing::warn!(
                            extension = %id,
                            key = %key,
                            secret_env = ?shown,
                            "ignoring secret_env: not an upper-case variable name matching [A-Z_][A-Z0-9_]*"
                        );
                        None
                    }
                    Some(name) => {
                        if let Some(reason) = Self::denied_secret_env(name, id, &own_keys) {
                            return Err(format!(
                                "Extension '{}' config key '{}' declares secret_env '{}': {}",
                                id, key, name, reason
                            ));
                        }
                        Some(name)
                    }
                },
            };
            if let Some(value) = env(&Self::extension_override_var(id, key)) {
                out.insert(key.to_string(), Value::String(value));
                continue;
            }
            if let Some(secret_env) = secret_env {
                if let Some(value) = env(secret_env) {
                    out.insert(key.to_string(), Value::String(value));
                    continue;
                }
            }
            if let Some(value) = crate::extensions::config_store::read_plugin_config(id, key) {
                out.insert(key.to_string(), Value::String(value));
                continue;
            }
            if let Some(value) = crate::config::read_config_value(&config_key) {
                out.insert(key.to_string(), Value::String(value));
                continue;
            }
            if let Some(default) = &entry.default {
                out.insert(key.to_string(), default.clone());
                continue;
            }
            if entry.required {
                let hint = if let Some(secret_env) = secret_env {
                    format!(
                        "set environment variable '{}' or config key '{}'",
                        secret_env, config_key
                    )
                } else {
                    format!("set config key '{}'", config_key)
                };
                return Err(format!(
                    "Extension '{}' missing required config '{}': {}",
                    id, key, hint
                ));
            }
        }
        Ok(Value::Object(out))
    }

    /// Test-only seeder: synthetically insert capability declarations
    /// for an extension id. Used to exercise capability snapshot
    /// rendering without spinning up a real plugin process.
    #[cfg(test)]
    pub(crate) fn test_seed_capabilities(
        &mut self,
        id: &str,
        decls: Vec<crate::extensions::runtime::process::CapabilityDeclaration>,
    ) {
        self.capabilities.insert(id.to_string(), decls);
    }

    /// Unload an extension — unsubscribe hooks and shut down the process.
    ///
    /// A DEFERRED tool-only extension (Task 20) has no live handler:
    /// unloading it removes the retained launch record, deregisters its
    /// dormant catalog batch (advancing the catalog generation so stale
    /// grants cannot survive), and terminates every session's runtime
    /// lease for the plugin.
    pub async fn unload(&mut self, id: &str) -> Result<(), String> {
        self.unload_inner(id, false).await.map(|_| ())
    }

    /// [`Self::unload`]; with `keep_tools` (a reload only) the extension's
    /// registry entries stay where they are and are returned, for
    /// [`Self::settle_replaced_tools`] to remove whatever the new instance
    /// did not replace. Until then a call to one fails against the stopped
    /// instance, as it would against a missing tool.
    async fn unload_inner(
        &mut self,
        id: &str,
        keep_tools: bool,
    ) -> Result<Vec<(String, Arc<dyn crate::Tool>)>, String> {
        let deferred = self
            .deferred_tool_only
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .remove(id);
        if let Some(record) = deferred {
            // Record removed FIRST: any concurrent lease acquisition now
            // fails closed (NotDeferred) regardless of the steps below.
            if let Some(runtime) = self.extension_runtime.get() {
                let revoked = runtime.revoke_plugin_all_sessions(id);
                if revoked > 0 {
                    tracing::info!(extension = %id, leases = revoked,
                        "Terminated live runtime leases of unloaded deferred extension");
                }
            }
            self.hook_bus.unsubscribe_all(id).await;
            self.providers.unregister_plugin(id);
            let mut kept = Vec::new();
            if let Some(tools) = &self.tools {
                let names: Vec<String> = record
                    .manifest
                    .deferred
                    .as_ref()
                    .map(|d| {
                        d.tools
                            .iter()
                            .map(|t| format!("{}:{}", id, t.name))
                            .collect()
                    })
                    .unwrap_or_default();
                if keep_tools {
                    let registry = tools.read().await;
                    kept = kept_entries(&registry, &names);
                    self.reload_replaces = names.clone();
                } else if let Err(error) = tools.write().await.try_disable(&names) {
                    // Fail-closed either way: with the record gone, dormant
                    // descriptors can never execute; surface the partial
                    // cleanup instead of hiding it.
                    return Err(format!(
                        "Extension '{}' unloaded (record removed, leases revoked) but its dormant catalog batch could not be deregistered: {}",
                        id, error
                    ));
                }
            }
            self.manifest_configs.remove(id);
            self.manifest_theme_tokens.remove(id);
            tracing::info!(extension = %id, "Deferred extension unloaded");
            return Ok(kept);
        }
        let handler = self
            .extensions
            .remove(id)
            .ok_or_else(|| format!("Extension '{}' not found", id))?;

        self.hook_bus.unsubscribe_all(id).await;
        self.providers.unregister_plugin(id);
        // Its tools go with it: each holds the handler being shut down, so
        // a tool left behind would fail every call, and a reload whose new
        // version no longer declares it would never replace it.
        let mut kept = Vec::new();
        let tool_error = match &self.tools {
            Some(tools) if keep_tools => {
                let registry = tools.read().await;
                let names = registry.tool_names_for_extension(id);
                kept = kept_entries(&registry, &names);
                // A new version that defers swaps these in its batch write.
                self.reload_replaces = names;
                None
            }
            Some(tools) => {
                let mut registry = tools.write().await;
                let names = registry.tool_names_for_extension(id);
                registry.try_disable(&names).err()
            }
            None => None,
        };
        self.eager_permissions.remove(id);
        self.manifest_configs.remove(id);
        self.capabilities.remove(id);
        self.plugin_info.remove(id);
        self.manifest_theme_tokens.remove(id);
        handler.shutdown().await;

        if let Some(error) = tool_error {
            // Fail-closed like the deferred path: the extension is gone
            // either way; surface the partial cleanup.
            return Err(format!(
                "Extension '{}' unloaded but its tools could not be deregistered: {}",
                id, error
            ));
        }
        tracing::info!(extension = %id, "Extension unloaded");
        Ok(kept)
    }

    /// After a reload's load: remove each kept entry the new instance did
    /// not replace (it still holds the stopped instance), in ONE registry
    /// write. On success that is exactly the tools the new version no
    /// longer declares; after a failed load it is all of them (fail-closed:
    /// no tool outlives its process).
    async fn settle_replaced_tools(&mut self, id: &str, kept: Vec<(String, Arc<dyn crate::Tool>)>) -> Result<(), String> {
        self.reload_replaces.clear();
        let Some(tools) = &self.tools else { return Ok(()) };
        if kept.is_empty() {
            return Ok(());
        }
        let mut registry = tools.write().await;
        let stale: Vec<String> = kept
            .into_iter()
            .filter(|(name, old)| registry.get(name).is_some_and(|now| same_tool(now, old)))
            .map(|(name, _)| name)
            .collect();
        registry
            .try_disable(&stale)
            .map_err(|e| format!("Extension '{id}' reloaded but its stale tools could not be deregistered: {e}"))
    }

    /// Reload one extension by unloading any existing instance first, then loading
    /// the supplied manifest. If the new load fails, the previous instance remains
    /// unloaded so duplicate handlers cannot survive a broken reload.
    ///
    /// The extension's tools never leave the shared registry during the
    /// swap: the new instance replaces them by name (a deferred one in a
    /// single write), and only what it did not replace is removed after.
    /// A session that starts a request mid-reload therefore still finds its
    /// activated tools, so the round-top rebuild keeps them and the tools
    /// array it sends (part of the cached prompt prefix) is unchanged. A
    /// gap would drop those activations for good: a prompt-cache rewrite
    /// then, and another when the model re-activates.
    pub async fn reload(
        &mut self,
        id: &str,
        manifest: &ExtensionManifest,
        cwd: Option<std::path::PathBuf>,
    ) -> Result<(), String> {
        let kept = if self.extensions.contains_key(id) || self.is_deferred_tool_only(id) {
            self.unload_inner(id, true).await?
        } else {
            Vec::new()
        };
        let loaded = self.load_with_cwd(id, manifest, cwd).await;
        let settled = self.settle_replaced_tools(id, kept).await;
        loaded.and(settled)
    }

    /// Reload extension `id` from what is installed now, without touching
    /// any other extension: find its plugin directory the way discovery
    /// does, read and validate its manifest, then stop the running instance
    /// (if any) and start the new one. Its tools are re-registered in the
    /// shared registry (so every session sees the new set from its next
    /// request), its hooks re-subscribed, and a deferred extension's
    /// session leases revoked. A plugin that is installed but not loaded
    /// (added after discovery, or failed then) is loaded.
    ///
    /// Everything that can be checked without stopping the running
    /// instance is checked first; a failure there leaves it running. Only a
    /// failure to START the new version leaves the extension unloaded (as
    /// [`Self::reload`] documents), and the error says so. Discovery's
    /// recorded result is updated either way, so a later replay agrees.
    ///
    /// Needs discovery to have finished on this manager: before that, a
    /// load here would make discovery's own load of the same plugin fail.
    pub async fn reload_from_disk(
        &mut self,
        id: &str,
    ) -> Result<ExtensionReloadReport, ExtensionLoadFailure> {
        let refuse = |reason: String, hint: &str| ExtensionLoadFailure::new(id, None, reason, hint);
        if id.is_empty()
            || id == "."
            || id == ".."
            || id.contains('/')
            || id.contains('\\')
            || id.contains(':')
            || id.chars().any(char::is_control)
        {
            return Err(refuse(
                format!("'{id}' is not a plugin name"),
                "Give the plugin's directory name, as `/extensions` lists it",
            ));
        }
        if is_inert_plugin_dir_name(id) {
            return Err(refuse(
                format!("'{id}' is a backup, disabled or temp directory name; discovery never loads it"),
                "Give the plugin's directory name without the suffix",
            ));
        }
        if !matches!(self.discovery, DiscoveryState::Done { .. }) {
            return Err(refuse(
                "extensions have not finished loading yet".into(),
                "Retry in a moment",
            ));
        }
        let was_loaded = self.is_loaded(id);
        let Some(plugin_dir) = locate_plugin_dir(id) else {
            let roots = plugin_roots()
                .iter()
                .map(|r| r.display().to_string())
                .collect::<Vec<_>>()
                .join(", ");
            return Err(refuse(
                format!("no plugin directory named '{id}' in {roots}"),
                "Check the name against `ls ~/.synaps-cli/plugins` or `/extensions`",
            ));
        };
        if crate::config::load_config()
            .disabled_plugins
            .iter()
            .any(|d| d == id)
        {
            return Err(refuse(
                format!("'{id}' is disabled (disabled_plugins in config)"),
                "Remove it from disabled_plugins first",
            ));
        }
        let plugins_state = {
            let state_path = crate::skills::state::PluginsState::default_path();
            crate::skills::state::PluginsState::load_from(&state_path).unwrap_or_default()
        };
        if let Some(message) = installed_plugin_setup_failure_in(&plugins_state, id) {
            return Err(refuse(
                format!("Post-install setup failed: {message}"),
                "Open /plugins, reinstall or update the plugin after fixing setup",
            ));
        }
        let (resolved, manifest_path, json) = match read_extension_manifest(id, &plugin_dir)? {
            Some(read) => read,
            None => {
                return Err(ExtensionLoadFailure::new(
                    id,
                    Some(plugin_dir.join(".synaps-plugin").join("plugin.json")),
                    format!("'{id}' has no extension (no plugin.json, or no `extension` block in it)"),
                    "Only plugins with an `extension` block run as extensions",
                ));
            }
        };
        if let Err(e) = resolved.validate(id) {
            return Err(ExtensionLoadFailure::new(
                id,
                Some(manifest_path),
                e,
                "Fix the manifest; the running instance was not touched",
            ));
        }

        // Commit point: from here the running instance is stopped.
        let prefix = format!("{id}:");
        let owned_tools = |registry: &crate::ToolRegistry| -> Vec<String> {
            registry
                .iter_tools_sorted()
                .into_iter()
                .map(|t| t.name().to_string())
                .filter(|n| n.starts_with(&prefix))
                .collect()
        };
        let before = match &self.tools {
            Some(tools) => owned_tools(&*tools.read().await),
            None => Vec::new(),
        };
        let outcome = self.reload(id, &resolved, Some(plugin_dir.clone())).await;
        let failure = outcome.as_ref().err().map(|e| {
            let setup_script = json.pointer("/extension/setup").and_then(|v| v.as_str());
            let hint = compute_extension_load_hint(e, &plugin_dir, setup_script);
            let reason = if was_loaded {
                format!("{e} (the previous instance was stopped, so '{id}' is not loaded now)")
            } else {
                e.clone()
            };
            ExtensionLoadFailure::new(id, Some(manifest_path.clone()), reason, hint)
        });
        if let DiscoveryState::Done { loaded, failed } = &mut self.discovery {
            loaded.retain(|l| l != id);
            failed.retain(|f| f.plugin != id);
            match &failure {
                None => loaded.push(id.to_string()),
                Some(f) => failed.push(f.clone()),
            }
        }
        if let Some(failure) = failure {
            tracing::warn!(extension = %id, error = %failure.reason, "Extension reload failed");
            return Err(failure);
        }
        let tools = match &self.tools {
            Some(tools) => owned_tools(&*tools.read().await),
            None => Vec::new(),
        };
        let removed_tools = before.into_iter().filter(|n| !tools.contains(n)).collect();
        let deferred = self.is_deferred_tool_only(id);
        tracing::info!(extension = %id, was_loaded, deferred, tools = tools.len(), path = %plugin_dir.display(), "Extension reloaded");
        Ok(ExtensionReloadReport {
            extension: id.to_string(),
            was_loaded,
            deferred,
            tools,
            removed_tools,
            plugin_dir,
        })
    }

    /// `id` is loaded, live or deferred (no process yet).
    pub fn is_loaded(&self, id: &str) -> bool {
        self.extensions.contains_key(id) || self.is_deferred_tool_only(id)
    }

    /// Current process-level discovery state.
    pub fn discovery_state(&self) -> &DiscoveryState {
        &self.discovery
    }

    /// `Some((loaded, failed))` once discovery has completed on this
    /// manager; `None` while not started / running. Idempotence seam for
    /// [`crate::extensions::loader::spawn_discover_and_load`].
    pub fn discovery_done(&self) -> Option<(Vec<String>, Vec<ExtensionLoadFailure>)> {
        match &self.discovery {
            DiscoveryState::Done { loaded, failed } => Some((loaded.clone(), failed.clone())),
            _ => None,
        }
    }

    /// Shut down all extensions gracefully. Resets discovery so a later
    /// `discover_and_load` on this manager walks the roots again.
    pub async fn shutdown_all(&mut self) {
        self.discovery = DiscoveryState::NotStarted;
        let ids: Vec<String> = self.extensions.keys().cloned().collect();
        for id in ids {
            let _ = self.unload(&id).await;
        }
        // Deferred extensions (Task 20): unload each retained record so
        // dormant catalog batches and any live runtime leases go with it.
        let deferred_ids: Vec<String> = {
            let map = self
                .deferred_tool_only
                .lock()
                .unwrap_or_else(std::sync::PoisonError::into_inner);
            map.keys().cloned().collect()
        };
        for id in deferred_ids {
            let _ = self.unload(&id).await;
        }
        if let Some(runtime) = self.extension_runtime.get() {
            runtime.terminate_all();
        }
    }

    /// Start shutting down all extensions in the background.
    ///
    /// This is intended for process exit: the UI should not hang waiting for
    /// extension child processes to acknowledge shutdown. Dropping the join handle
    /// lets Tokio abort remaining work when the runtime exits.
    pub fn shutdown_all_detached(
        manager: Arc<tokio::sync::RwLock<Self>>,
    ) -> tokio::task::JoinHandle<()> {
        tokio::spawn(async move {
            manager.write().await.shutdown_all().await;
        })
    }

    /// List running extension IDs.
    pub fn list(&self) -> Vec<&str> {
        self.extensions.keys().map(|s| s.as_str()).collect()
    }

    /// Return Arc references to all running extension handlers, sorted by ID.
    /// Intended for background notification watchers that need to hold onto
    /// handlers beyond the lifetime of a manager lock.
    pub fn handlers(&self) -> Vec<(String, Arc<dyn super::runtime::ExtensionHandler>)> {
        let mut out: Vec<_> = self
            .extensions
            .iter()
            .map(|(id, h)| (id.clone(), Arc::clone(h)))
            .collect();
        out.sort_by(|a, b| a.0.cmp(&b.0));
        out
    }

    /// Number of running extensions.
    pub fn count(&self) -> usize {
        self.extensions.len()
    }

    /// Return health snapshots for all loaded extensions, sorted by ID.
    pub async fn statuses(&self) -> Vec<ExtensionStatus> {
        let mut handlers: Vec<(String, Arc<dyn ExtensionHandler>)> = self
            .extensions
            .iter()
            .map(|(id, handler)| (id.clone(), handler.clone()))
            .collect();
        handlers.sort_by(|a, b| a.0.cmp(&b.0));

        let mut statuses = Vec::with_capacity(handlers.len());
        for (id, handler) in handlers {
            statuses.push(ExtensionStatus {
                id,
                health: handler.health().await,
                restart_count: handler.restart_count().await,
            });
        }
        statuses
    }

    /// Return registered provider metadata sorted by runtime id.
    pub fn providers(&self) -> Vec<&RegisteredProvider> {
        self.providers.list()
    }

    /// Return registered provider metadata by runtime id.
    pub fn provider(&self, runtime_id: &str) -> Option<&RegisteredProvider> {
        self.providers.get(runtime_id)
    }

    /// Return optional cached plugin info reported by `info.get`.
    pub fn plugin_info(&self, id: &str) -> Option<&PluginInfo> {
        self.plugin_info.get(id)
    }

    /// Manifest-declared theme tokens for every loaded extension, sorted by
    /// extension ID (P19.2). Each entry is `(plugin-id, token -> hex color)`;
    /// the TUI namespaces these as `ext.<plugin-id>.<token>` when merging
    /// into the active theme. Extensions without `theme_tokens` don't appear.
    pub fn theme_tokens(&self) -> Vec<(String, std::collections::BTreeMap<String, String>)> {
        let mut out: Vec<_> = self
            .manifest_theme_tokens
            .iter()
            .map(|(id, tokens)| (id.clone(), tokens.clone()))
            .collect();
        out.sort_by(|a, b| a.0.cmp(&b.0));
        out
    }

    /// Handler for explicit USER-triggered plugin APIs (interactive
    /// commands, settings editors). Live extensions use their
    /// running handler; a DEFERRED extension (Task 20) gets a lazy handler
    /// whose first use acquires the shared per-plugin runtime lease — a
    /// user action is a legitimate activation trigger. Discovery/search/
    /// diagnostics never call this.
    pub fn user_action_handler(
        &self,
        id: &str,
    ) -> Result<Arc<dyn super::runtime::ExtensionHandler>, String> {
        if let Some(handler) = self.extensions.get(id) {
            return Ok(handler.clone());
        }
        if self.is_deferred(id) {
            return Ok(Arc::new(
                crate::extensions::lease::LazyExtensionHandler::new(id, self.extension_runtime()),
            ));
        }
        Err(format!("unknown extension '{}'", id))
    }

    /// Resolve authority for an explicitly user-armed session driver without I/O.
    /// Only successful eager loads retain validated permission metadata. Deferred,
    /// unknown, failed, and unloaded instances fail closed; never acquire a lease.
    /// Callers pin this Arc and compare it with `Arc::ptr_eq` on every timer and
    /// before applying an async result. Drop the manager guard before plugin RPC.
    pub fn session_driver_handler(&self, id: &str) -> Result<Arc<dyn ExtensionHandler>, String> {
        if self.is_deferred(id) {
            return Err("session drivers require an eagerly loaded extension; deferred drivers are unsupported".into());
        }
        let handler = self
            .extensions
            .get(id)
            .ok_or_else(|| "session driver extension is not loaded".to_string())?;
        if !self
            .eager_permissions
            .get(id)
            .is_some_and(|permissions| permissions.has(Permission::SessionDrive))
        {
            return Err("session driver extension lacks validated session.drive permission".into());
        }
        Ok(handler.clone())
    }

    /// (E-P8) Whether an eagerly-loaded extension holds the validated
    /// `session.drive` permission. The TUI uses this to route a driver
    /// plugin's slash command to `DriverStart` (the actor front door)
    /// instead of a plain interactive invoke. Fails closed for deferred,
    /// unknown, or permission-less extensions.
    pub fn has_session_drive(&self, id: &str) -> bool {
        self.eager_permissions
            .get(id)
            .is_some_and(|permissions| permissions.has(Permission::SessionDrive))
    }

    /// Invoke an interactive plugin command on extension `id`. Streams
    /// `command.output` (matching `request_id`) and `task.*` notifications
    /// to `sink`. Returns the final JSON-RPC response value.
    ///
    /// CP-11 fix-3: `sink` is the bounded, metered producer half of
    /// [`crate::extensions::invoke_output::invoke_event_channel`]; the
    /// caller MUST drive the matching collector concurrently with this
    /// call (an awaited bounded send with no consumer would stall the
    /// invocation loop until the 120 s timeout). Prefer
    /// [`Self::invoke_command_collected`], which wires that correctly.
    pub async fn invoke_command(
        &self,
        id: &str,
        command: &str,
        args: Vec<String>,
        request_id: &str,
        sink: crate::extensions::invoke_output::InvokeEventSink,
    ) -> Result<serde_json::Value, String> {
        let handler = self.user_action_handler(id)?;
        handler
            .invoke_command(command, args, request_id, sink)
            .await
    }

    /// Invoke an interactive plugin command and collect its budgeted
    /// output. This is THE production entry point for user-facing
    /// surfaces: it joins the invocation with an eagerly concurrent
    /// collector so hostile `command.output` floods are paced and capped
    /// at production time (never parked unbounded), then returns the
    /// final response alongside the retained events and exact
    /// produced/consumed/retained/truncated/dropped accounting.
    ///
    /// Dropping the returned future cancels the invocation, drops the
    /// sink, and releases the collector plus any blocked producers — no
    /// detached tasks are spawned.
    pub async fn invoke_command_collected(
        &self,
        id: &str,
        command: &str,
        args: Vec<String>,
        request_id: &str,
    ) -> (
        Result<serde_json::Value, String>,
        crate::extensions::invoke_output::InvokeOutputReport,
    ) {
        let (sink, collector) = crate::extensions::invoke_output::invoke_event_channel(
            crate::extensions::invoke_output::InvokeOutputBudget::default(),
        );
        let handler = match self.user_action_handler(id) {
            Ok(handler) => handler,
            Err(error) => {
                drop(sink);
                return (Err(error), collector.collect().await);
            }
        };
        let invoke = handler.invoke_command(command, args, request_id, sink);
        let (result, report) = tokio::join!(invoke, collector.collect());
        (result, report)
    }

    pub async fn settings_editor_open(
        &self,
        id: &str,
        category: &str,
        field: &str,
    ) -> Result<serde_json::Value, String> {
        let handler = self.user_action_handler(id)?;
        handler.settings_editor_open(category, field).await
    }

    pub async fn settings_editor_key(
        &self,
        id: &str,
        category: &str,
        field: &str,
        key: &str,
    ) -> Result<serde_json::Value, String> {
        let handler = self.user_action_handler(id)?;
        handler.settings_editor_key(category, field, key).await
    }

    pub async fn settings_editor_commit(
        &self,
        id: &str,
        category: &str,
        field: &str,
        value: serde_json::Value,
    ) -> Result<serde_json::Value, String> {
        let handler = self.user_action_handler(id)?;
        handler.settings_editor_commit(category, field, value).await
    }

    /// Return all cached plugin info sorted by extension id.
    pub fn plugin_infos(&self) -> Vec<(&str, &PluginInfo)> {
        let mut entries: Vec<_> = self
            .plugin_info
            .iter()
            .map(|(id, info)| (id.as_str(), info))
            .collect();
        entries.sort_by(|a, b| a.0.cmp(b.0));
        entries
    }

    /// Return provider status summaries sorted by provider runtime id.
    pub fn provider_summaries(&self) -> Vec<RegisteredProviderSummary> {
        self.providers.summaries()
    }

    /// Unified capability snapshot per loaded extension, sorted by id.
    ///
    /// Aggregates hook subscriptions, extension-provided tools, and registered
    /// providers. `future` carries plugin-defined capability kinds and
    /// capabilities land.
    pub async fn capability_snapshots(&self) -> Vec<ExtensionCapabilitySnapshot> {
        let mut handlers: Vec<(String, Arc<dyn ExtensionHandler>)> = self
            .extensions
            .iter()
            .map(|(id, handler)| (id.clone(), handler.clone()))
            .collect();
        handlers.sort_by(|a, b| a.0.cmp(&b.0));

        let provider_summaries = self.providers.summaries();
        let plugin_id_lookup: std::collections::HashMap<String, String> = self
            .providers
            .list()
            .into_iter()
            .map(|p| (p.runtime_id.clone(), p.plugin_id.clone()))
            .collect();

        let mut out = Vec::with_capacity(handlers.len());
        for (id, handler) in handlers {
            let health = handler.health().await;
            let restart_count = handler.restart_count().await;

            let hook_pairs = self.hook_bus.subscriptions_for(&id).await;
            let hooks: Vec<HookCapabilityEntry> = hook_pairs
                .into_iter()
                .map(|(kind, tool_filter)| HookCapabilityEntry {
                    kind: kind.as_str().to_string(),
                    tool_filter,
                })
                .collect();

            let tools: Vec<ToolCapabilityEntry> = if let Some(tools) = &self.tools {
                let registry = tools.read().await;
                registry
                    .tool_names_for_extension(&id)
                    .into_iter()
                    .map(|name| ToolCapabilityEntry { name })
                    .collect()
            } else {
                Vec::new()
            };

            let providers: Vec<RegisteredProviderSummary> = provider_summaries
                .iter()
                .filter(|summary| {
                    plugin_id_lookup
                        .get(&summary.runtime_id)
                        .map(|p| p == &id)
                        .unwrap_or(false)
                })
                .cloned()
                .collect();

            let future: Vec<FutureCapabilityEntry> = self
                .capabilities
                .get(&id)
                .map(|decls| {
                    decls
                        .iter()
                        .map(|d| FutureCapabilityEntry {
                            kind: d.kind.clone(),
                            name: d.name.clone(),
                        })
                        .collect()
                })
                .unwrap_or_default();

            out.push(ExtensionCapabilitySnapshot {
                id,
                health,
                restart_count,
                hooks,
                tools,
                providers,
                future,
            });
        }
        out
    }

    /// Return runtime ids of registered providers that declare at least one
    /// tool-use-capable model. Sorted by runtime id.
    pub fn provider_tool_use_runtime_ids(&self) -> Vec<String> {
        let mut ids: Vec<String> = self
            .providers
            .list()
            .into_iter()
            .filter(|p| {
                p.spec.models.iter().any(|m| {
                    m.capabilities
                        .get("tool_use")
                        .and_then(|v| v.as_bool())
                        .unwrap_or(false)
                })
            })
            .map(|p| p.runtime_id.clone())
            .collect();
        ids.sort();
        ids
    }

    /// Return a `runtime_id -> enabled` map for every registered provider, computed
    /// from the persisted trust state. Providers without an entry default to
    /// enabled. If the trust state file is missing, all providers are reported
    /// as enabled (default). If the file is corrupt, all providers are reported
    /// as **disabled** (fail-closed) and a warning is logged.
    pub fn provider_trust_view(&self) -> std::collections::BTreeMap<String, bool> {
        let trust = match crate::extensions::trust::load_trust_state() {
            Ok(t) => t,
            Err(e) => {
                tracing::warn!("trust.json corrupt or unreadable, failing closed (all providers disabled): {e}");
                // Return all providers as disabled
                return self
                    .providers
                    .list()
                    .into_iter()
                    .map(|p| (p.runtime_id.clone(), false))
                    .collect();
            }
        };
        self.providers
            .list()
            .into_iter()
            .map(|p| {
                let enabled = crate::extensions::trust::is_provider_enabled(&trust, &p.runtime_id);
                (p.runtime_id.clone(), enabled)
            })
            .collect()
    }

    /// Compute config diagnostics for a loaded extension by id.
    /// Returns `None` if the extension is not loaded.
    pub fn config_diagnostics(&self, id: &str) -> Option<ExtensionConfigDiagnostics> {
        let manifest_config = self.manifest_configs.get(id)?;

        // Collect provider required keys from registered providers' config_schema.
        let mut provider_required: Vec<(String, Vec<String>)> = Vec::new();
        for provider in self.providers.list() {
            if provider.plugin_id != id {
                continue;
            }
            let required: Vec<String> = provider
                .spec
                .config_schema
                .as_ref()
                .and_then(|schema| schema.get("required"))
                .and_then(Value::as_array)
                .map(|arr| {
                    arr.iter()
                        .filter_map(|v| v.as_str().map(|s| s.to_string()))
                        .collect()
                })
                .unwrap_or_default();
            provider_required.push((provider.provider_id.clone(), required));
        }
        provider_required.sort_by(|a, b| a.0.cmp(&b.0));

        let env_lookup = |name: &str| std::env::var(name).ok();
        let plugin_config_lookup =
            |key: &str| crate::extensions::config_store::read_plugin_config(id, key);
        let legacy_config_lookup = |key: &str| crate::config::read_config_value(key);

        Some(diagnose_extension_config(
            id,
            manifest_config,
            &provider_required,
            &env_lookup,
            &plugin_config_lookup,
            &legacy_config_lookup,
        ))
    }

    /// Diagnostics for all loaded extensions, sorted alphabetically by id.
    pub fn all_config_diagnostics(&self) -> Vec<ExtensionConfigDiagnostics> {
        let mut ids: Vec<&String> = self.manifest_configs.keys().collect();
        ids.sort();
        ids.into_iter()
            .filter_map(|id| self.config_diagnostics(id))
            .collect()
    }

    /// Get the shared hook bus.
    pub fn hook_bus(&self) -> &Arc<HookBus> {
        &self.hook_bus
    }

    /// Get the shared tool registry, when this manager was constructed with one.
    pub fn tools_shared(&self) -> Option<Arc<tokio::sync::RwLock<crate::ToolRegistry>>> {
        self.tools.clone()
    }

    /// Test-only: register a provider spec + in-process handler directly,
    /// without spawning an extension process. Used by trace wiring tests.
    #[cfg(test)]
    pub(crate) fn register_provider_handler_for_tests(
        &mut self,
        plugin_id: &str,
        spec: crate::extensions::runtime::process::RegisteredProviderSpec,
        handler: Arc<dyn ExtensionHandler>,
    ) -> Result<String, String> {
        self.providers
            .register_with_handler(plugin_id, spec, Some(handler))
    }

    /// Discover and load all extensions from the user and project plugin directories.
    ///
    /// Scans `~/.synaps-cli/plugins/*/.synaps-plugin/plugin.json` and
    /// `./.synaps/plugins/*/.synaps-plugin/plugin.json` for manifests that contain
    /// an `extension` field. Project-local plugins override user plugins with the
    /// same directory name.
    pub async fn discover_and_load(&mut self) -> (Vec<String>, Vec<ExtensionLoadFailure>) {
        self.discover_and_load_with_progress(|_| {}).await
    }

    /// Discover and load all extensions, invoking `progress` after each load
    /// attempt. Used by the async UI loader to update startup toasts without
    /// blocking first paint.
    pub async fn discover_and_load_with_progress<F>(
        &mut self,
        mut progress: F,
    ) -> (Vec<String>, Vec<ExtensionLoadFailure>)
    where
        F: FnMut(crate::extensions::loader::ExtensionLoaderEvent),
    {
        // Idempotent (daemon-mode C2): discovery runs once per manager.
        // A second call replays the recorded result and spawns nothing.
        if let Some(done) = self.discovery_done() {
            tracing::debug!(
                loaded = done.0.len(),
                failed = done.1.len(),
                "Extension discovery already done; replaying recorded result"
            );
            return done;
        }
        self.discovery = DiscoveryState::Running;
        let (loaded, failed) = self.discover_and_load_uncached(&mut progress).await;
        self.discovery = DiscoveryState::Done {
            loaded: loaded.clone(),
            failed: failed.clone(),
        };
        (loaded, failed)
    }

    async fn discover_and_load_uncached<F>(
        &mut self,
        progress: &mut F,
    ) -> (Vec<String>, Vec<ExtensionLoadFailure>)
    where
        F: FnMut(crate::extensions::loader::ExtensionLoaderEvent),
    {
        let mut plugin_dirs: HashMap<String, PathBuf> = HashMap::new();
        let mut failed: Vec<ExtensionLoadFailure> = Vec::new();

        for plugins_dir in plugin_roots() {
            if !plugins_dir.exists() {
                continue;
            }

            let entries = match std::fs::read_dir(&plugins_dir) {
                Ok(e) => e,
                Err(e) => {
                    tracing::warn!(path = %plugins_dir.display(), error = %e, "Failed to read plugins directory");
                    failed.push(ExtensionLoadFailure::new(
                        "plugins",
                        Some(plugins_dir.clone()),
                        format!("Failed to read plugins directory: {e}"),
                        "Check directory permissions and retry",
                    ));
                    continue;
                }
            };

            for entry in entries.flatten() {
                let plugin_name = entry.file_name().to_string_lossy().to_string();
                if is_inert_plugin_dir_name(&plugin_name) {
                    tracing::debug!(
                        plugin = %plugin_name,
                        path = %entry.path().display(),
                        "Skipping backup/disabled/temp plugin directory"
                    );
                    continue;
                }
                plugin_dirs.insert(plugin_name, entry.path());
            }
        }

        let mut plugin_dirs: Vec<(String, PathBuf)> = plugin_dirs.into_iter().collect();
        plugin_dirs.sort_by(|a, b| a.0.cmp(&b.0));

        let mut loaded = Vec::new();
        let disabled_plugins = crate::config::load_config().disabled_plugins;
        // Hoist plugins.json read out of the per-plugin loop — was N reads + parses
        // of the same file (one per discovered plugin). Read it once here.
        let plugins_state = {
            let state_path = crate::skills::state::PluginsState::default_path();
            crate::skills::state::PluginsState::load_from(&state_path).unwrap_or_default()
        };
        for (plugin_name, plugin_dir) in plugin_dirs {
            if disabled_plugins.iter().any(|d| d == &plugin_name) {
                tracing::debug!(plugin = %plugin_name, "Extension disabled via disabled_plugins config");
                continue;
            }
            if let Some(message) = installed_plugin_setup_failure_in(&plugins_state, &plugin_name) {
                tracing::warn!(plugin = %plugin_name, error = %message, "Skipping extension with failed post-install setup");
                failed.push(ExtensionLoadFailure::new(
                    plugin_name,
                    None,
                    format!("Post-install setup failed: {message}"),
                    "Open /plugins, reinstall or update the plugin after fixing setup; extension load is disabled until setup succeeds",
                ));
                continue;
            }
            let (resolved, manifest_path, json) =
                match read_extension_manifest(&plugin_name, &plugin_dir) {
                    Ok(Some(read)) => read,
                    Ok(None) => continue,
                    Err(failure) => {
                        failed.push(failure);
                        continue;
                    }
                };

            match self
                .load_with_cwd(&plugin_name, &resolved, Some(plugin_dir.clone()))
                .await
            {
                Ok(()) => {
                    tracing::info!(plugin = %plugin_name, path = %plugin_dir.display(), "Extension loaded from plugins/");
                    loaded.push(plugin_name.clone());
                    progress(crate::extensions::loader::ExtensionLoaderEvent::Loaded {
                        plugin: plugin_name,
                        loaded: loaded.len(),
                        failed: failed.len(),
                    });
                }
                Err(e) => {
                    tracing::warn!(plugin = %plugin_name, manifest = %manifest_path.display(), error = %e, "Failed to load extension");
                    let setup_script = json.pointer("/extension/setup").and_then(|v| v.as_str());
                    let hint = compute_extension_load_hint(&e, &plugin_dir, setup_script);
                    let failure =
                        ExtensionLoadFailure::new(plugin_name, Some(manifest_path), e, hint);
                    failed.push(failure.clone());
                    progress(crate::extensions::loader::ExtensionLoaderEvent::Failed {
                        failure,
                        loaded: loaded.len(),
                        failed: failed.len(),
                    });
                }
            }
        }

        (loaded, failed)
    }
}

#[cfg(test)]
mod invoke_command_collected_tests {
    //! CP-11 fix-3: prove the manager's collected entry point pairs the
    //! bounded sink with an eagerly concurrent collector — an in-process
    //! handler that floods far past the channel capacity AND the retained
    //! budget must neither deadlock nor grow retention, and accounting
    //! must be exact.
    use super::*;
    use crate::extensions::commands::CommandOutputEvent;
    use crate::extensions::invoke_output::INVOKE_RETAINED_BYTE_BUDGET;
    use crate::extensions::runtime::InvokeCommandEvent;

    struct FloodingHandler;

    #[async_trait::async_trait]
    impl super::ExtensionHandler for FloodingHandler {
        fn id(&self) -> &str {
            "flooding-fake"
        }
        async fn handle(
            &self,
            _event: &crate::extensions::hooks::events::HookEvent,
        ) -> crate::extensions::hooks::events::HookResult {
            Default::default()
        }
        async fn shutdown(&self) {}

        async fn invoke_command(
            &self,
            _command: &str,
            _args: Vec<String>,
            _request_id: &str,
            sink: crate::extensions::invoke_output::InvokeEventSink,
        ) -> Result<Value, String> {
            // 2048 x 4 KiB = 8 MiB >> 256 KiB budget, and 2048 events
            // >> channel capacity 8: every awaited send past capacity
            // requires the concurrent collector to make progress.
            for _ in 0..2048 {
                let event = InvokeCommandEvent::Output(CommandOutputEvent::Text {
                    content: "f".repeat(4096),
                });
                if sink.send(event).await.is_err() {
                    return Err("collector vanished".to_string());
                }
            }
            let _ = sink
                .send(InvokeCommandEvent::Output(CommandOutputEvent::Done))
                .await;
            Ok(serde_json::json!({"status": "flooded"}))
        }
    }

    #[tokio::test]
    async fn collected_invoke_bounds_in_process_flood_without_deadlock() {
        let bus = Arc::new(HookBus::new());
        let mut mgr = ExtensionManager::new(bus);
        mgr.extensions
            .insert("flooding-fake".to_string(), Arc::new(FloodingHandler));

        let (result, report) = tokio::time::timeout(
            std::time::Duration::from_secs(30),
            mgr.invoke_command_collected("flooding-fake", "any", vec![], "req-1"),
        )
        .await
        .expect("bounded sink + concurrent collector must not deadlock");

        assert_eq!(result.expect("invoke ok")["status"], "flooded");
        let c = &report.counters;
        assert_eq!(c.produced_events, 2049);
        assert_eq!(c.produced_payload_bytes, 2048 * 4096);
        assert_eq!(c.consumed_events, c.produced_events);
        assert_eq!(c.consumed_payload_bytes, c.produced_payload_bytes);
        assert_eq!(c.retained_payload_bytes, INVOKE_RETAINED_BYTE_BUDGET as u64);
        assert!(c.saw_done);
        assert_eq!(
            c.consumed_payload_bytes,
            c.retained_payload_bytes + c.truncated_payload_bytes + c.dropped_payload_bytes,
        );
        assert_eq!(c.consumed_events, c.retained_events + c.dropped_events);
    }

    #[tokio::test]
    async fn collected_invoke_reports_unknown_extension_with_empty_report() {
        let bus = Arc::new(HookBus::new());
        let mgr = ExtensionManager::new(bus);
        let (result, report) = mgr
            .invoke_command_collected("nope", "any", vec![], "req-1")
            .await;
        assert!(result
            .expect_err("unknown id")
            .contains("unknown extension"));
        assert!(report.events.is_empty());
        assert_eq!(report.counters.produced_events, 0);
    }
}

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

    #[tokio::test]
    async fn capability_snapshots_empty_when_no_extensions() {
        let bus = Arc::new(HookBus::new());
        let mgr = ExtensionManager::new(bus);
        assert!(mgr.capability_snapshots().await.is_empty());
    }

    /// Review fix A3: the retained deferred launch record resolves
    /// `secret_env` config values, so NOTHING public/debug/diagnostic may
    /// ever reveal them — only booleans/counts. The record type itself is
    /// crate-private, field-private, and implements neither `Debug` nor
    /// `Serialize` (compiler-enforced); this test additionally proves the
    /// live diagnostic surfaces never carry the resolved secret.
    #[tokio::test]
    async fn deferred_launch_record_never_leaks_secret_config_through_diagnostics() {
        const SECRET: &str = "task20-A3-sentinel-5f2c9d";
        // NOTE: deliberately NOT in the SYNAPS_* host namespace — that is denied by
        // `denied_secret_env` (#207 MED-2). This test is about diagnostic leakage,
        // not namespace policy, so it uses a realistic third-party var name.
        const VAR: &str = "T20_A3_THIRD_PARTY_SECRET";
        std::env::set_var(VAR, SECRET);

        let manifest: ExtensionManifest = serde_json::from_value(serde_json::json!({
            "runtime": "process",
            "command": "/bin/false",
            "permissions": ["tools.register"],
            "config": [{
                "key": "api_key",
                "type": "string",
                "required": true,
                "secret_env": VAR
            }],
            "deferred": {
                "tools": [{
                    "name": "search",
                    "description": "deferred search",
                    "input_schema": {"type": "object"}
                }]
            }
        }))
        .unwrap();

        let registry = Arc::new(tokio::sync::RwLock::new(crate::ToolRegistry::new()));
        let mut mgr = ExtensionManager::new_with_tools(Arc::new(HookBus::new()), registry);
        mgr.set_progressive_deferral(true);
        mgr.load("secretive", &manifest).await.unwrap();

        // Boolean/count-only public surface for the deferral state.
        assert!(mgr.is_deferred_tool_only("secretive"));
        assert_eq!(mgr.deferred_tool_only_count(), 1);
        // The internal record DID resolve the secret (lease path needs it).
        let record = mgr.deferred_launch_record("secretive").expect("retained");
        assert_eq!(record.config["api_key"], serde_json::json!(SECRET));

        // Every public/debug/diagnostic surface stays free of the value.
        let mut rendered = String::new();
        rendered.push_str(&format!("{:?}", mgr.statuses().await));
        rendered.push_str(&format!("{:?}", mgr.config_diagnostics("secretive")));
        rendered.push_str(&format!("{:?}", mgr.all_config_diagnostics()));
        rendered.push_str(&format!("{:?}", mgr.capability_snapshots().await));
        rendered.push_str(&format!("{:?}", mgr.plugin_infos()));
        rendered.push_str(&format!("{:?}", mgr.list()));
        assert!(
            !rendered.contains(SECRET),
            "resolved secret config leaked through a public diagnostic surface"
        );

        std::env::remove_var(VAR);
    }

    #[tokio::test]
    async fn capability_snapshot_lists_hooks_for_loaded_extension() {
        let bus = Arc::new(HookBus::new());
        let mut mgr = ExtensionManager::new(bus.clone());
        let manifest = ExtensionManifest {
            theme_tokens: Default::default(),
            deferred: None,
            protocol_version: 1,
            runtime: crate::extensions::manifest::ExtensionRuntime::Process,
            command: "python3".to_string(),
            setup: None,
            prebuilt: ::std::collections::HashMap::new(),
            args: vec![
                "tests/fixtures/process_extension.py".to_string(),
                "normal".to_string(),
                "/tmp/synaps-capability-test.log".to_string(),
            ],
            permissions: vec!["tools.intercept".to_string()],
            hooks: vec![crate::extensions::manifest::HookSubscription {
                hook: "before_tool_call".to_string(),
                tool: Some("bash".to_string()),
                matcher: None,
            }],
            config: vec![],
        };

        mgr.load("cap-snap", &manifest).await.unwrap();

        let snaps = mgr.capability_snapshots().await;
        assert_eq!(snaps.len(), 1);
        let snap = &snaps[0];
        assert_eq!(snap.id, "cap-snap");
        assert_eq!(snap.hooks.len(), 1);
        assert_eq!(snap.hooks[0].kind, "before_tool_call");
        assert_eq!(snap.hooks[0].tool_filter.as_deref(), Some("bash"));
        assert!(snap.tools.is_empty());
        assert!(snap.providers.is_empty());
        assert!(snap.future.is_empty());

        mgr.shutdown_all().await;
    }

    #[tokio::test]
    async fn capability_snapshot_surfaces_seeded_capabilities() {
        let bus = Arc::new(HookBus::new());
        let mut mgr = ExtensionManager::new(bus.clone());
        let manifest = ExtensionManifest {
            theme_tokens: Default::default(),
            deferred: None,
            protocol_version: 1,
            runtime: crate::extensions::manifest::ExtensionRuntime::Process,
            command: "python3".to_string(),
            setup: None,
            prebuilt: ::std::collections::HashMap::new(),
            args: vec![
                "tests/fixtures/process_extension.py".to_string(),
                "normal".to_string(),
                "/tmp/synaps-capability-snapshot-test.log".to_string(),
            ],
            permissions: vec!["tools.intercept".to_string()],
            hooks: vec![crate::extensions::manifest::HookSubscription {
                hook: "before_tool_call".to_string(),
                tool: Some("bash".to_string()),
                matcher: None,
            }],
            config: vec![],
        };

        mgr.load("multi-cap", &manifest).await.unwrap();

        // Seed two capabilities of *different* kinds — proves the
        // snapshot rendering iterates a generic list and uses the
        // plugin-supplied `kind` rather than hardcoding any modality.
        mgr.test_seed_capabilities(
            "multi-cap",
            vec![
                crate::extensions::runtime::process::CapabilityDeclaration {
                    kind: "capture".to_string(),
                    name: "Local Sample STT".to_string(),
                    permissions: vec!["audio.input".to_string()],
                    params: serde_json::Value::Null,
                },
                crate::extensions::runtime::process::CapabilityDeclaration {
                    kind: "ocr".to_string(),
                    name: "Tesseract".to_string(),
                    permissions: vec![],
                    params: serde_json::Value::Null,
                },
            ],
        );

        let snaps = mgr.capability_snapshots().await;
        let snap = snaps
            .iter()
            .find(|s| s.id == "multi-cap")
            .expect("multi-cap snapshot");
        assert_eq!(snap.future.len(), 2);
        let kinds: Vec<&str> = snap.future.iter().map(|e| e.kind.as_str()).collect();
        assert!(kinds.contains(&"capture"), "got kinds {:?}", kinds);
        assert!(kinds.contains(&"ocr"), "got kinds {:?}", kinds);
        let names: Vec<&str> = snap.future.iter().map(|e| e.name.as_str()).collect();
        assert!(names.contains(&"Local Sample STT"), "got {:?}", names);
        assert!(names.contains(&"Tesseract"), "got {:?}", names);

        mgr.unload("multi-cap").await.unwrap();
        let snaps = mgr.capability_snapshots().await;
        assert!(snaps.iter().all(|s| s.id != "multi-cap"));

        mgr.shutdown_all().await;
    }

    #[tokio::test]
    async fn new_manager_has_no_extensions() {
        let bus = Arc::new(HookBus::new());
        let mgr = ExtensionManager::new(bus);
        assert_eq!(mgr.count(), 0);
        assert!(mgr.list().is_empty());
    }

    #[tokio::test]
    async fn unload_nonexistent_returns_error() {
        let bus = Arc::new(HookBus::new());
        let mut mgr = ExtensionManager::new(bus);
        let result = mgr.unload("nope").await;
        assert!(result.is_err());
    }

    #[tokio::test]
    async fn reload_unsubscribes_old_handler_before_loading_new_one() {
        let bus = Arc::new(HookBus::new());
        let mut mgr = ExtensionManager::new(bus.clone());
        let manifest = ExtensionManifest {
            theme_tokens: Default::default(),
            deferred: None,
            protocol_version: 1,
            runtime: crate::extensions::manifest::ExtensionRuntime::Process,
            command: "python3".to_string(),
            setup: None,
            prebuilt: ::std::collections::HashMap::new(),
            args: vec![
                "tests/fixtures/process_extension.py".to_string(),
                "normal".to_string(),
                "/tmp/synaps-reload-test.log".to_string(),
            ],
            permissions: vec!["tools.intercept".to_string()],
            hooks: vec![crate::extensions::manifest::HookSubscription {
                hook: "before_tool_call".to_string(),
                tool: Some("bash".to_string()),
                matcher: None,
            }],
            config: vec![],
        };

        mgr.load("reload-test", &manifest).await.unwrap();
        assert_eq!(bus.handler_count().await, 1);

        mgr.reload("reload-test", &manifest, None).await.unwrap();

        assert_eq!(mgr.count(), 1);
        assert_eq!(bus.handler_count().await, 1);
        mgr.shutdown_all().await;
    }

    #[tokio::test]
    async fn reload_failure_leaves_previous_instance_unloaded() {
        let bus = Arc::new(HookBus::new());
        let mut mgr = ExtensionManager::new(bus.clone());
        let good = ExtensionManifest {
            theme_tokens: Default::default(),
            deferred: None,
            protocol_version: 1,
            runtime: crate::extensions::manifest::ExtensionRuntime::Process,
            command: "python3".to_string(),
            setup: None,
            prebuilt: ::std::collections::HashMap::new(),
            args: vec![
                "tests/fixtures/process_extension.py".to_string(),
                "normal".to_string(),
                "/tmp/synaps-reload-failure-test.log".to_string(),
            ],
            permissions: vec!["tools.intercept".to_string()],
            hooks: vec![crate::extensions::manifest::HookSubscription {
                hook: "before_tool_call".to_string(),
                tool: Some("bash".to_string()),
                matcher: None,
            }],
            config: vec![],
        };
        let bad = ExtensionManifest {
            theme_tokens: Default::default(),
            deferred: None,
            command: "/definitely/not/a/real/extension-binary".to_string(),
            setup: None,
            prebuilt: ::std::collections::HashMap::new(),
            ..good.clone()
        };

        mgr.load("reload-failure-test", &good).await.unwrap();
        let err = mgr
            .reload("reload-failure-test", &bad, None)
            .await
            .unwrap_err();

        assert!(err.contains("Failed to spawn extension"), "{err}");
        assert_eq!(mgr.count(), 0);
        assert_eq!(bus.handler_count().await, 0);
    }

    #[test]
    fn project_plugins_disable_env_parser_accepts_truthy_values() {
        for value in ["1", "true", "TRUE", "yes", "on"] {
            std::env::set_var("SYNAPS_DISABLE_PROJECT_PLUGINS", value);
            assert!(project_plugins_disabled());
        }
        for value in ["", "0", "false", "off", "no"] {
            std::env::set_var("SYNAPS_DISABLE_PROJECT_PLUGINS", value);
            assert!(!project_plugins_disabled());
        }
        std::env::remove_var("SYNAPS_DISABLE_PROJECT_PLUGINS");
    }

    fn with_temp_base_dir<T>(path: &std::path::Path, f: impl FnOnce() -> T) -> T {
        let old_base_dir = std::env::var("SYNAPS_BASE_DIR").ok();
        crate::config::set_base_dir_for_tests(path.to_path_buf());
        let out = f();
        match old_base_dir {
            Some(old) => std::env::set_var("SYNAPS_BASE_DIR", old),
            None => std::env::remove_var("SYNAPS_BASE_DIR"),
        }
        out
    }

    #[test]
    #[serial_test::serial(synaps_base_dir)]
    fn resolve_config_prefers_plugin_namespaced_config_before_legacy_global_key() {
        let dir = tempfile::tempdir().unwrap();
        with_temp_base_dir(dir.path(), || {
            crate::extensions::config_store::write_plugin_config(
                "sample-plugin",
                "backend",
                "cpu",
            )
            .unwrap();
            crate::config::write_config_value("extension.sample-plugin.backend", "auto").unwrap();

            let resolved = ExtensionManager::resolve_config(
                "sample-plugin",
                &[ExtensionConfigEntry {
                    key: "backend".to_string(),
                    value_type: None,
                    description: None,
                    required: true,
                    default: None,
                    secret_env: None,
                    host_context: None,
                }],
                None,
            )
            .unwrap();

            assert_eq!(
                resolved["backend"],
                serde_json::Value::String("cpu".to_string())
            );
        });
    }

    #[test]
    #[serial_test::serial(synaps_base_dir)]
    fn resolve_config_keeps_legacy_global_extension_key_as_fallback() {
        let dir = tempfile::tempdir().unwrap();
        with_temp_base_dir(dir.path(), || {
            crate::config::write_config_value("extension.sample-plugin.backend", "auto").unwrap();

            let resolved = ExtensionManager::resolve_config(
                "sample-plugin",
                &[ExtensionConfigEntry {
                    key: "backend".to_string(),
                    value_type: None,
                    description: None,
                    required: true,
                    default: None,
                    secret_env: None,
                    host_context: None,
                }],
                None,
            )
            .unwrap();

            assert_eq!(
                resolved["backend"],
                serde_json::Value::String("auto".to_string())
            );
        });
    }

    #[tokio::test]
    async fn config_diagnostics_returns_none_for_unknown_extension() {
        let bus = Arc::new(HookBus::new());
        let mgr = ExtensionManager::new(bus);
        assert!(mgr.config_diagnostics("nope").is_none());
        assert!(mgr.all_config_diagnostics().is_empty());
    }

    #[tokio::test]
    async fn config_diagnostics_reports_loaded_manifest_entries() {
        let bus = Arc::new(HookBus::new());
        let mut mgr = ExtensionManager::new(bus);
        let manifest = ExtensionManifest {
            theme_tokens: Default::default(),
            deferred: None,
            protocol_version: 1,
            runtime: crate::extensions::manifest::ExtensionRuntime::Process,
            command: "python3".to_string(),
            setup: None,
            prebuilt: ::std::collections::HashMap::new(),
            args: vec![
                "tests/fixtures/process_extension.py".to_string(),
                "normal".to_string(),
                "/tmp/synaps-config-diag-test.log".to_string(),
            ],
            permissions: vec!["tools.intercept".to_string()],
            hooks: vec![crate::extensions::manifest::HookSubscription {
                hook: "before_tool_call".to_string(),
                tool: Some("bash".to_string()),
                matcher: None,
            }],
            config: vec![crate::extensions::manifest::ExtensionConfigEntry {
                key: "region".to_string(),
                value_type: None,
                description: Some("AWS region".to_string()),
                required: false,
                default: Some(serde_json::Value::String("us-east-1".to_string())),
                secret_env: None,
                host_context: None,
            }],
        };

        mgr.load("config-diag-test", &manifest).await.unwrap();

        let diag = mgr
            .config_diagnostics("config-diag-test")
            .expect("diagnostics should be available for loaded extension");
        assert_eq!(diag.extension_id, "config-diag-test");
        assert_eq!(diag.entries.len(), 1);
        assert_eq!(diag.entries[0].key, "region");
        assert!(diag.entries[0].has_value);
        assert!(diag.provider_missing.is_empty());

        let all = mgr.all_config_diagnostics();
        assert_eq!(all.len(), 1);
        assert_eq!(all[0].extension_id, "config-diag-test");

        mgr.shutdown_all().await;
        // After shutdown, manifest config storage is cleared.
        assert!(mgr.config_diagnostics("config-diag-test").is_none());
    }

    #[tokio::test]
    async fn provider_trust_view_is_empty_for_no_providers() {
        let bus = Arc::new(HookBus::new());
        let mgr = ExtensionManager::new(bus);
        let view = mgr.provider_trust_view();
        assert!(view.is_empty());
    }

    #[tokio::test]
    async fn provider_tool_use_runtime_ids_lists_only_tool_use_capable() {
        use crate::extensions::runtime::process::{
            RegisteredProviderModelSpec, RegisteredProviderSpec,
        };
        let bus = Arc::new(HookBus::new());
        let mut mgr = ExtensionManager::new(bus);
        // Tool-use capable provider.
        let tool_spec = RegisteredProviderSpec {
            id: "alpha".into(),
            display_name: "Alpha".into(),
            description: "tool-use".into(),
            models: vec![RegisteredProviderModelSpec {
                id: "m1".into(),
                display_name: None,
                capabilities: serde_json::json!({"tool_use": true}),
                context_window: None,
            }],
            config_schema: None,
        };
        // Plain provider, no tool_use.
        let plain_spec = RegisteredProviderSpec {
            id: "beta".into(),
            display_name: "Beta".into(),
            description: "plain".into(),
            models: vec![RegisteredProviderModelSpec {
                id: "m1".into(),
                display_name: None,
                capabilities: serde_json::json!({"streaming": true}),
                context_window: None,
            }],
            config_schema: None,
        };
        mgr.providers.register("plug", tool_spec).unwrap();
        mgr.providers.register("plug", plain_spec).unwrap();
        let ids = mgr.provider_tool_use_runtime_ids();
        assert_eq!(ids, vec!["plug:alpha".to_string()]);
    }

    // ---- compute_extension_load_hint --------------------------------

    #[test]
    fn hint_missing_binary_with_declared_setup_points_at_script() {
        let hint = compute_extension_load_hint(
            "Failed to spawn extension 'sample-plugin': No such file or directory (os error 2)",
            std::path::Path::new("/home/u/.synaps-cli/plugins/sample-plugin"),
            Some("scripts/setup.sh"),
        );
        assert!(
            hint.contains("Extension binary missing"),
            "missing-binary case should be flagged: {hint}"
        );
        assert!(
            hint.contains("/home/u/.synaps-cli/plugins/sample-plugin"),
            "hint should include the plugin dir: {hint}"
        );
        assert!(
            hint.contains("setup=scripts/setup.sh"),
            "hint should show sanitized setup path without copy-paste shell command: {hint}"
        );
    }

    #[test]
    fn hint_missing_binary_without_declared_setup_falls_back_to_generic() {
        let hint = compute_extension_load_hint(
            "Failed to spawn extension 'foo': No such file or directory (os error 2)",
            std::path::Path::new("/x/y"),
            None,
        );
        assert!(
            hint.contains("contract.json"),
            "no setup declared → generic hint: {hint}"
        );
        assert!(
            !hint.contains("Extension binary missing"),
            "should not falsely promise a setup script: {hint}"
        );
    }

    #[test]
    fn hint_other_error_with_declared_setup_falls_back_to_generic() {
        let hint = compute_extension_load_hint(
            "Extension 'foo' must subscribe to at least one hook or request a registration permission",
            std::path::Path::new("/x/y"),
            Some("scripts/setup.sh"),
        );
        // Setup script is declared, but the error is *not* a missing
        // binary — running the script wouldn't help. Fall back to the
        // generic hint so we don't mislead the user.
        assert!(hint.contains("contract.json"), "got {hint}");
        assert!(!hint.contains("Extension binary missing"), "got {hint}");
    }

    #[test]
    fn hint_recognises_os_error_2_format() {
        // Older / cross-platform error formats may include the kernel
        // errno but not the "No such file or directory" English text.
        let hint = compute_extension_load_hint(
            "spawn failed (os error 2)",
            std::path::Path::new("/p"),
            Some("setup.sh"),
        );
        assert!(hint.contains("Extension binary missing"), "got {hint}");
    }

    // ── bugfix: installed-plugin backup discovery ───────────────────────────

    /// Minimal deferred tool-only plugin.json (Axel-shaped: `memory_fetch`
    /// takes a single `{id}`), written under `<dir>/.synaps-plugin/`.
    fn write_installed_plugin(plugins_root: &std::path::Path, dir_name: &str) {
        let plugin_dir = plugins_root.join(dir_name).join(".synaps-plugin");
        std::fs::create_dir_all(&plugin_dir).unwrap();
        let manifest = serde_json::json!({
            "extension": {
                "runtime": "process",
                "command": "/bin/false",
                "permissions": ["tools.register"],
                "deferred": {
                    "tools": [{
                        "name": "memory_fetch",
                        "description": "Fetch one memory by exact ID.",
                        "input_schema": {
                            "type": "object",
                            "properties": {"id": {"type": "string"}},
                            "required": ["id"]
                        }
                    }]
                }
            }
        });
        std::fs::write(
            plugin_dir.join("plugin.json"),
            serde_json::to_string_pretty(&manifest).unwrap(),
        )
        .unwrap();
    }

    /// Regression (user incident, session 20260720-141844-fb5a): `plugins/`
    /// held the installed `axel-memory-manager` AND timestamped copies
    /// (`axel-memory-manager.backup.<TS>`). Discovery treated every
    /// directory with a valid manifest as a live plugin, so the backups
    /// were loaded and their binaries executed alongside the real one.
    /// Backup/disabled/temp directory names must never be discovered: only
    /// active plugins load, backup tools are never cataloged, and (since a
    /// backup that is never loaded owns no lease) a backup process can
    /// never be spawned.
    #[tokio::test]
    #[serial_test::serial(synaps_base_dir)]
    async fn discover_and_load_skips_backup_and_disabled_plugin_directories() {
        let base = crate::test_env::BaseDirGuard::new();
        let _no_project_plugins =
            crate::test_env::EnvVarGuard::set("SYNAPS_DISABLE_PROJECT_PLUGINS", "1");
        let plugins_root = base.path().join("plugins");

        write_installed_plugin(&plugins_root, "axel-memory-manager");
        // The two timestamped backups from the incident, the updater's own
        // `.update-backup` convention, and common manual conventions.
        for inert in [
            "axel-memory-manager.backup.20260719210053",
            "axel-memory-manager.backup.20260720101416",
            "axel-memory-manager.update-backup",
            "axel-memory-manager.backup",
            "axel-memory-manager.bak",
            "axel-memory-manager.disabled",
            "axel-memory-manager.old",
            "axel-memory-manager.orig",
            "axel-memory-manager.tmp",
        ] {
            write_installed_plugin(&plugins_root, inert);
        }
        // Dotted names that are NOT backups stay discoverable.
        write_installed_plugin(&plugins_root, "com.example.notes");

        let registry = Arc::new(tokio::sync::RwLock::new(
            crate::ToolRegistry::without_subagent(),
        ));
        let mut mgr = ExtensionManager::new_with_tools(Arc::new(HookBus::new()), registry.clone());
        mgr.set_progressive_deferral(true);

        let (mut loaded, failed) = mgr.discover_and_load().await;
        loaded.sort();

        assert!(failed.is_empty(), "no failures expected: {failed:?}");
        assert_eq!(
            loaded,
            vec![
                "axel-memory-manager".to_string(),
                "com.example.notes".to_string()
            ],
            "backup/disabled/temp plugin directories must not be discovered"
        );

        let reg = registry.read().await;
        assert!(reg.get("axel-memory-manager:memory_fetch").is_some());
        assert!(reg.get("com.example.notes:memory_fetch").is_some());
        assert!(
            reg.get("axel-memory-manager.backup.20260719210053:memory_fetch")
                .is_none(),
            "backup plugin tools must never be cataloged"
        );
        assert!(reg
            .get("axel-memory-manager.backup.20260720101416:memory_fetch")
            .is_none());
        drop(reg);

        // Deferred lifecycle + no backup load ⇒ zero live extension
        // processes were started by discovery.
        assert_eq!(mgr.count(), 0, "discovery must spawn nothing");
    }

    // ── a manifest that declares only a context provider (B-4) ──────────────

    /// A `plugin.json` whose only `deferred` content is a context provider,
    /// with the permission that used to go with it.
    fn write_provider_only_plugin(
        plugins_root: &std::path::Path,
        dir_name: &str,
        command: &str,
        args: Vec<String>,
    ) {
        let plugin_dir = plugins_root.join(dir_name).join(".synaps-plugin");
        std::fs::create_dir_all(&plugin_dir).unwrap();
        let manifest = serde_json::json!({
            "extension": {
                "runtime": "process",
                "command": command,
                "args": args,
                "permissions": ["context_providers.register"],
                "deferred": {"context_providers": [{"id": "recall"}]}
            }
        });
        std::fs::write(
            plugin_dir.join("plugin.json"),
            serde_json::to_string_pretty(&manifest).unwrap(),
        )
        .unwrap();
    }

    #[derive(Clone, Default)]
    struct LogCapture(Arc<std::sync::Mutex<Vec<u8>>>);
    struct LogWriter(Arc<std::sync::Mutex<Vec<u8>>>);
    impl std::io::Write for LogWriter {
        fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
            self.0.lock().unwrap().extend_from_slice(buf);
            Ok(buf.len())
        }
        fn flush(&mut self) -> std::io::Result<()> {
            Ok(())
        }
    }
    impl LogCapture {
        fn subscriber(&self) -> impl tracing::Subscriber + Send + Sync {
            let sink = self.0.clone();
            tracing_subscriber::fmt()
                .with_max_level(tracing::Level::WARN)
                .with_writer(move || LogWriter(sink.clone()))
                .with_ansi(false)
                .finish()
        }
        fn text(&self) -> String {
            String::from_utf8(self.0.lock().unwrap().clone()).unwrap()
        }
    }

    /// Through discovery (the real load path), with progressive disclosure
    /// on: a provider-only manifest starts like one with an empty `deferred`
    /// block (eager, its process runs `initialize`), and the load writes the
    /// "declaration is ignored" notice to the log.
    #[tokio::test]
    #[serial_test::serial(synaps_base_dir)]
    async fn a_provider_only_manifest_loads_eagerly_and_logs_the_ignored_declaration() {
        let base = crate::test_env::BaseDirGuard::new();
        let _no_project_plugins =
            crate::test_env::EnvVarGuard::set("SYNAPS_DISABLE_PROJECT_PLUGINS", "1");
        let plugins_root = base.path().join("plugins");
        let spy = base.path().join("spy.log");
        let tools_json = base.path().join("tools.json");
        std::fs::write(&tools_json, "[]").unwrap();
        let fixture = std::path::Path::new(env!("CARGO_MANIFEST_DIR"))
            .join("tests/fixtures/extension_fixture.py");
        write_provider_only_plugin(
            &plugins_root,
            "provider-only",
            "python3",
            vec![
                fixture.display().to_string(),
                spy.display().to_string(),
                tools_json.display().to_string(),
                "ok".to_string(),
            ],
        );

        let registry = Arc::new(tokio::sync::RwLock::new(
            crate::ToolRegistry::without_subagent(),
        ));
        let mut mgr = ExtensionManager::new_with_tools(Arc::new(HookBus::new()), registry);
        mgr.set_progressive_deferral(true);
        let log = LogCapture::default();
        let (loaded, failed) = {
            let _guard = tracing::subscriber::set_default(log.subscriber());
            mgr.discover_and_load().await
        };

        assert!(failed.is_empty(), "no failures expected: {failed:?}");
        assert_eq!(loaded, vec!["provider-only".to_string()]);
        assert!(!mgr.is_deferred("provider-only"), "not dormant");
        assert!(mgr.is_loaded("provider-only"));
        assert_eq!(mgr.count(), 1, "its process is running");
        let spy_log = std::fs::read_to_string(&spy).unwrap_or_default();
        assert!(
            spy_log.lines().any(|l| l == "request:initialize"),
            "the process was started and initialized at load: {spy_log:?}"
        );
        let text = log.text();
        assert_eq!(
            text.matches(&format!(
                "extension provider-only declares a context provider; built-in automatic \
                 memory was removed in {}, the declaration is ignored",
                agent_core::AXEL_BACKEND_REMOVED_IN
            ))
            .count(),
            1,
            "one notice in the log: {text}"
        );
        mgr.shutdown_all().await;
    }

    /// With progressive disclosure on, a provider-only manifest whose
    /// command does not start fails to load like any eager manifest: the
    /// failure is collected and the other extensions still load.
    #[tokio::test]
    #[serial_test::serial(synaps_base_dir)]
    async fn a_provider_only_manifest_with_a_dead_command_fails_alone() {
        let base = crate::test_env::BaseDirGuard::new();
        let _no_project_plugins =
            crate::test_env::EnvVarGuard::set("SYNAPS_DISABLE_PROJECT_PLUGINS", "1");
        let plugins_root = base.path().join("plugins");
        write_provider_only_plugin(&plugins_root, "provider-only", "/bin/false", Vec::new());
        write_installed_plugin(&plugins_root, "com.example.notes");

        let registry = Arc::new(tokio::sync::RwLock::new(
            crate::ToolRegistry::without_subagent(),
        ));
        let mut mgr = ExtensionManager::new_with_tools(Arc::new(HookBus::new()), registry.clone());
        mgr.set_progressive_deferral(true);
        let (loaded, failed) = mgr.discover_and_load().await;

        assert_eq!(loaded, vec!["com.example.notes".to_string()]);
        assert_eq!(failed.len(), 1, "{failed:?}");
        assert_eq!(failed[0].plugin, "provider-only");
        assert!(!mgr.is_loaded("provider-only"));
        assert!(!mgr.is_deferred("provider-only"));
        assert!(registry
            .read()
            .await
            .get("com.example.notes:memory_fetch")
            .is_some());
    }

    /// The exclusion helper is name-based and conservative: it must catch
    /// known backup/disabled/temp conventions without swallowing
    /// legitimate dotted or hyphenated plugin names.
    #[test]
    fn inert_plugin_dir_name_filter_is_exact() {
        for inert in [
            "axel-memory-manager.backup.20260720101416",
            "axel-memory-manager.BACKUP.1",
            "axel-memory-manager.backup",
            "axel-memory-manager.update-backup",
            "axel-memory-manager.bak",
            "axel-memory-manager.disabled",
            "axel-memory-manager.old",
            "axel-memory-manager.orig",
            "axel-memory-manager.tmp",
            "axel-memory-manager.temp",
        ] {
            assert!(
                is_inert_plugin_dir_name(inert),
                "{inert} should be excluded from discovery"
            );
        }
        for live in [
            "axel-memory-manager",
            "com.example.notes",
            "backup-manager",
            "old-school-tools",
            "tmpfs-inspector",
            "my.backups.viewer",
        ] {
            assert!(
                !is_inert_plugin_dir_name(live),
                "{live} is a legitimate plugin name and must stay discoverable"
            );
        }
    }
}

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

    fn entry(key: &str, secret_env: &str) -> ExtensionConfigEntry {
        serde_json::from_value(serde_json::json!({
            "key": key,
            "type": "string",
            "secret_env": secret_env,
        }))
        .unwrap()
    }

    /// #207 MED-2: a manifest must not be able to name the host's own
    /// provider credential and have it handed over at initialize.
    #[test]
    fn host_provider_credentials_are_rejected() {
        for var in [
            "ANTHROPIC_API_KEY",
            "OPENAI_API_KEY",
            "AWS_SECRET_ACCESS_KEY",
        ] {
            let err = ExtensionManager::resolve_config("evil", &[entry("k", var)], None)
                .expect_err(&format!("{var} must be rejected"));
            assert!(err.contains("host credential"), "unexpected error: {err}");
        }
        // Padding is trimmed before the denylist check, as before.
        let entries = [entry("k", "  ANTHROPIC_API_KEY ")];
        let err = ExtensionManager::resolve_config("evil", &entries, None)
            .expect_err("padded host credential must be rejected");
        assert!(err.contains("host credential"), "unexpected error: {err}");
        // A lower-case name is malformed: ignored (never read), not fatal.
        let entries = [entry("k", "anthropic_api_key")];
        let out = ExtensionManager::resolve_config("evil", &entries, None)
            .expect("malformed name is dropped, not fatal");
        assert!(out.get("k").is_none(), "{out}");
    }

    /// The whole SYNAPS_* namespace is the host's, except this extension's
    /// own scoped override.
    #[test]
    fn host_namespace_is_rejected_but_own_scope_is_allowed() {
        let err =
            ExtensionManager::resolve_config("evil", &[entry("k", "SYNAPS_BROKER_TOKEN")], None)
                .expect_err("host namespace must be rejected");
        assert!(err.contains("host namespace"), "unexpected error: {err}");

        // an extension's own scoped var is fine
        assert!(ExtensionManager::denied_secret_env(
            "SYNAPS_EXTENSION_MY_PLUGIN_TOKEN",
            "my-plugin",
            &["token"]
        )
        .is_none());
    }

    /// Third-party secrets an extension legitimately needs still work.
    #[test]
    fn third_party_secrets_are_still_allowed() {
        for var in ["GITHUB_TOKEN", "STRIPE_KEY", "MY_SERVICE_TOKEN"] {
            assert!(
                ExtensionManager::denied_secret_env(var, "some-ext", &[]).is_none(),
                "{var} should be allowed"
            );
        }
    }

    /// 01 I1: the denylist is built from the provider registries, so a new
    /// provider (or credential source) cannot drift out of it.
    #[test]
    fn every_registered_provider_env_var_is_denied() {
        let mut vars: Vec<&str> = Vec::new();
        for spec in agent_core::auth::static_providers::STATIC_PROVIDERS {
            vars.extend(spec.env_vars.iter().copied());
        }
        for spec in crate::runtime::openai::registry::providers() {
            vars.extend(spec.env_vars.iter().copied());
        }
        vars.extend([
            agent_core::auth::muse::API_KEY_ENV,
            "META_API_KEY",
            "KIMI_API_KEY",
            "MOONSHOT_API_KEY",
            "OPENAI_CODEX_ACCESS_TOKEN",
            "LOCAL_API_KEY",
            "TOGETHER_API_KEY",
            "CEREBRAS_API_KEY",
            "AWS_ACCESS_KEY_ID",
        ]);
        assert!(vars.len() > 20, "registry walk found too few vars");
        for var in vars {
            let reason = ExtensionManager::denied_secret_env(var, "some-ext", &[])
                .unwrap_or_else(|| panic!("{var} must be denied as secret_env"));
            assert!(reason.contains("host credential"), "{var}: {reason}");
        }
    }

    /// Literal names passed to `pat` (e.g. `env::var(`), skipping any
    /// whitespace between the open paren and the opening quote.
    fn literal_args(src: &str, pat: &str) -> Vec<String> {
        let mut out = Vec::new();
        let mut rest = src;
        while let Some(i) = rest.find(pat) {
            rest = &rest[i + pat.len()..];
            let arg = rest.trim_start();
            if let Some(lit) = arg.strip_prefix('"') {
                if let Some(end) = lit.find('"') {
                    out.push(lit[..end].to_string());
                }
            }
        }
        out
    }

    /// Drift guard for credential env vars read by hand rather than through
    /// a registry: walk every workspace source file (recursively: all crates
    /// plus the root `src/`) for `env::var(…"NAME"…)` / `env::var_os(…)`
    /// literals (whitespace and rustfmt line breaks allowed inside the call)
    /// and `*_ENV: &str = "…"` consts whose names look like secrets, and
    /// require every one to be refused.
    #[test]
    fn credential_env_reads_in_workspace_sources_are_denied() {
        fn walk(dir: &std::path::Path, out: &mut Vec<std::path::PathBuf>) {
            for e in std::fs::read_dir(dir).unwrap() {
                let p = e.unwrap().path();
                if p.is_dir() {
                    walk(&p, out);
                } else if p.extension().is_some_and(|x| x == "rs") {
                    out.push(p);
                }
            }
        }
        let root = std::path::Path::new(env!("CARGO_MANIFEST_DIR")).join("../..");
        let mut files = Vec::new();
        walk(&root.join("src"), &mut files);
        for e in std::fs::read_dir(root.join("crates")).unwrap() {
            let src = e.unwrap().path().join("src");
            if src.is_dir() {
                walk(&src, &mut files);
            }
        }
        // The scan must reach the places the review listed, or it rotted.
        for must in [
            "crates/agent-core/src/core/auth/muse.rs",
            "crates/agent-core/src/core/config.rs",
            "crates/agent-engine/src/runtime/google_gemini/runtime.rs",
            "crates/agent-engine/src/session/types.rs",
            "crates/agent-engine/src/daemon/mod.rs",
            "crates/agent-tui/src/lib.rs",
            "src/cmd/auth_broker.rs",
            "src/cmd/login.rs",
        ] {
            assert!(
                files.iter().any(|f| f.ends_with(must)),
                "scan does not cover {must}"
            );
        }

        let mut seen = 0;
        for f in &files {
            let src = std::fs::read_to_string(f).unwrap();
            let mut names = literal_args(&src, "env::var(");
            names.extend(literal_args(&src, "env::var_os("));
            names.extend(literal_args(&src, "_ENV: &str ="));
            for name in names {
                let up = name.to_ascii_uppercase();
                let looks_secret = ["KEY", "TOKEN", "SECRET", "PASSWORD"]
                    .iter()
                    .any(|w| up.contains(w));
                if !looks_secret {
                    continue;
                }
                seen += 1;
                assert!(
                    ExtensionManager::secret_env_name(&name).is_none()
                        || ExtensionManager::denied_secret_env(&name, "some-ext", &[]).is_some(),
                    "{} reads credential env var {name}, which extensions may still request via secret_env",
                    f.display()
                );
            }
        }
        assert!(
            seen > 3,
            "source scan matched almost nothing; the patterns rotted"
        );
    }

    #[test]
    fn literal_scan_handles_rustfmt_split_calls() {
        // Guard the scanner itself against the rustfmt shape that the old
        // version missed: `env::var(\n    "X",\n)`.
        let src = "let a = std::env::var(\n        \"SPLIT_TOKEN\",\n    );\nenv::var(\"ONE\")";
        assert_eq!(literal_args(src, "env::var("), ["SPLIT_TOKEN", "ONE"]);
    }

    /// reqwest honours these and a proxy URL can carry `user:pass@`.
    #[test]
    fn proxy_variables_are_denied() {
        for var in ["HTTPS_PROXY", "HTTP_PROXY", "ALL_PROXY", "NO_PROXY"] {
            assert!(
                ExtensionManager::denied_secret_env(var, "some-ext", &[]).is_some(),
                "{var} must be denied"
            );
            // Lower case (which reqwest also reads) is malformed, so it is
            // ignored and never read. If the name rule is ever relaxed, this
            // check must become case-insensitive.
            let lower = var.to_ascii_lowercase();
            assert!(
                ExtensionManager::secret_env_name(&lower).is_none(),
                "{lower} must never be read"
            );
        }
    }

    /// `foo` must not read `foo-bar`'s override var; only its own declared
    /// keys' override names are allowed in the SYNAPS_ namespace.
    #[test]
    fn own_scope_is_limited_to_declared_keys() {
        assert!(ExtensionManager::denied_secret_env(
            "SYNAPS_EXTENSION_FOO_BAR_TOKEN",
            "foo",
            &["api_key"]
        )
        .is_some());
        assert!(ExtensionManager::denied_secret_env(
            "SYNAPS_EXTENSION_FOO_API_KEY",
            "foo",
            &["api_key"]
        )
        .is_none());
        assert!(ExtensionManager::denied_secret_env(
            "SYNAPS_EXTENSION_FOO_BAR_API_KEY",
            "foo-bar",
            &["api-key"]
        )
        .is_none());
        // through resolve_config, with the manifest's own keys
        let err = ExtensionManager::resolve_config(
            "foo",
            &[entry("k", "SYNAPS_EXTENSION_FOO_BAR_TOKEN")],
            None,
        )
        .expect_err("cross-extension override must be rejected");
        assert!(err.contains("host namespace"), "{err}");
    }

    #[test]
    fn ids_differing_only_in_separators_collide() {
        assert_eq!(
            ExtensionManager::folded_id_collision("foo_bar", ["foo-bar", "x"].into_iter()),
            Some("foo-bar")
        );
        assert_eq!(
            ExtensionManager::folded_id_collision("Foo-Bar", ["foo-bar"].into_iter()),
            Some("foo-bar")
        );
        // the same id (a reload) and unrelated ids are not collisions
        assert_eq!(
            ExtensionManager::folded_id_collision("foo-bar", ["foo-bar", "foo"].into_iter()),
            None
        );
    }

    #[test]
    fn secret_env_name_format() {
        for bad in [
            "",
            "   ",
            "lower_case",
            "HAS SPACE",
            "A=B",
            "A-B",
            "1ABC",
            "A\0B",
            "ÄPI_KEY",
        ] {
            assert_eq!(ExtensionManager::secret_env_name(bad), None, "{bad:?}");
        }
        for good in ["GITHUB_TOKEN", "_X", "A1"] {
            assert_eq!(ExtensionManager::secret_env_name(good), Some(good));
            assert!(ExtensionManager::denied_secret_env(good, "some-ext", &[]).is_none());
        }
        // Padding is trimmed, as before.
        assert_eq!(
            ExtensionManager::secret_env_name("  PADDED "),
            Some("PADDED")
        );
    }

    /// A malformed name must never be READ, not merely tolerated: with
    /// `lower_token` set, the resolved value is still the default. Every name
    /// the resolver asks for is recorded, and `lower_token` is not among them.
    #[test]
    fn malformed_secret_env_is_never_read_even_when_set() {
        let entry: ExtensionConfigEntry = serde_json::from_value(serde_json::json!({
            "key": "k",
            "secret_env": "lower_token",
            "default": "fallback",
        }))
        .unwrap();
        let asked = std::sync::Mutex::new(Vec::<String>::new());
        let env = |name: &str| {
            asked.lock().unwrap().push(name.to_string());
            (name == "lower_token").then(|| "LEAK".to_string())
        };
        let out = ExtensionManager::resolve_config_with_env("ext", &[entry], None, &env)
            .expect("malformed secret_env must not fail the load");
        assert_eq!(out["k"], "fallback", "{out}");
        assert!(
            !asked.lock().unwrap().iter().any(|n| n == "lower_token"),
            "malformed name was looked up: {:?}",
            asked.lock().unwrap()
        );

        // Control: a well-formed name IS read through the same seam.
        let good: ExtensionConfigEntry = serde_json::from_value(serde_json::json!({
            "key": "k",
            "secret_env": "  GOOD_TOKEN ",
            "default": "fallback",
        }))
        .unwrap();
        let env = |name: &str| (name == "GOOD_TOKEN").then(|| "value".to_string());
        let out = ExtensionManager::resolve_config_with_env("ext", &[good], None, &env).unwrap();
        assert_eq!(out["k"], "value");
    }

    /// A malformed `secret_env` is warned about and dropped: the plugin still
    /// loads and the value resolves from the other sources as usual. A
    /// `required` entry with nothing else set fails with the normal error.
    #[test]
    fn malformed_secret_env_is_dropped_not_fatal() {
        let with_default: ExtensionConfigEntry = serde_json::from_value(serde_json::json!({
            "key": "k",
            "secret_env": "my-token",
            "default": "fallback",
        }))
        .unwrap();
        let out = ExtensionManager::resolve_config("ext", &[with_default], None)
            .expect("malformed secret_env must not fail the load");
        assert_eq!(out["k"], "fallback");

        let required: ExtensionConfigEntry = serde_json::from_value(serde_json::json!({
            "key": "k",
            "secret_env": "lower_case",
            "required": true,
        }))
        .unwrap();
        let err = ExtensionManager::resolve_config("ext", &[required], None)
            .expect_err("required entry with no value still fails");
        assert!(err.contains("missing required config"), "{err}");
        assert!(
            !err.contains("lower_case"),
            "dropped name is not suggested: {err}"
        );
    }
}