trustee_api/

state.rs

//! Shared server state: per-user multi-session registry, broadcast channels, and auth state.
//!
//! ## Multi-Session Per User (MSU)
//!
//! Each authenticated user gets their own [`UserSessions`] containing N independent
//! [`UserSessionEntry`] instances (default max 4). Each entry has:
//! - An independent `Session` (workflow state, output, etc.)
//! - A dedicated broadcast channel for WebSocket fan-out
//! - Creation and last-active timestamps
//!
//! Sessions are keyed by user identity (`sub` claim from JWT, or `dev:email` for
//! dev mode). Unauthenticated deployments use a single `"default"` key, preserving
//! backward compatibility with single-user CLI operation.

use std::sync::Arc;

use dashmap::DashMap;
use tokio::sync::{broadcast, mpsc, Mutex};
use trustee_core::session::Session;
use trustee_core::types::TuiMessage;

use crate::auth::AuthState;

// ---------------------------------------------------------------------------
// 16F console dispatch: user-key pinning (nghr ec3e0622 companion)
// ---------------------------------------------------------------------------

tokio::task_local! {
    /// When set (by the `/xagent/{agent}` dispatch prelude), ALL
    /// user-key-derived resolution — home dir, session bucket, project id,
    /// secrets merge, MCP loader cache — pins to THIS key: the THQ profile
    /// binding id (the stable identity anchor), NOT the token's claims.
    /// Old-world 16E agents carry their Kanidm sub here, making the override
    /// a bit-identical no-op for them. The ec3e0622 guarantee: token claims
    /// can never re-home a principal.
    static USER_KEY_OVERRIDE: String;
}

/// The effective namespace key for `user_key` under the active dispatch
/// scope (see [`USER_KEY_OVERRIDE`]).
pub(crate) fn effective_user_key(user_key: &str) -> String {
    USER_KEY_OVERRIDE
        .try_with(|v| v.clone())
        .unwrap_or_else(|_| user_key.to_string())
}

/// Wrap a dispatched handler so every user-key-derived resolution inside it
/// pins to the binding's key (see [`USER_KEY_OVERRIDE`]).
pub(crate) async fn in_dispatch_scope<F: std::future::Future>(user_key: &str, fut: F) -> F::Output {
    USER_KEY_OVERRIDE.scope(user_key.to_string(), fut).await
}

/// Deterministic re-link of pre-rotation homes (nghr ec3e0622 acceptance 3).
///
/// `users/home_links.json`: `{ "links": [ { "from_hash": "<data dir>",
/// "to_hash": "<current claims-derived dir>" } ] }` — the data lives under
/// `from_hash`; the claims-derived `to_hash` home is the (empty) one the
/// rotation created. Applied at startup, idempotent:
/// - `to` missing → rename `from` → `to` (config + sessions restored);
/// - `to` exists but has ZERO entries → remove it, then rename;
/// - `to` exists with content → LOUD skip (manual reconciliation required).
pub fn apply_home_links() -> Vec<String> {
    let Some(home) = dirs::home_dir() else {
        return Vec::new();
    };
    let users = home.join(".trustee").join("users");
    let map_path = users.join("home_links.json");
    let Ok(text) = std::fs::read_to_string(&map_path) else {
        return Vec::new();
    };
    let Ok(doc) = serde_json::from_str::<serde_json::Value>(&text) else {
        tracing::warn!(
            "home_links.json is not valid JSON — ignored: {}",
            map_path.display()
        );
        return Vec::new();
    };
    let mut report = Vec::new();
    if let Some(links) = doc.get("links").and_then(|v| v.as_array()) {
        for l in links {
            let (Some(from), Some(to)) = (
                l.get("from_hash").and_then(|v| v.as_str()),
                l.get("to_hash").and_then(|v| v.as_str()),
            ) else {
                continue;
            };
            let from_dir = users.join(from);
            let to_dir = users.join(to);
            if !from_dir.exists() {
                continue; // already applied (or never existed)
            }
            if to_dir.exists() {
                let empty = std::fs::read_dir(&to_dir)
                    .map(|it| it.count() == 0)
                    .unwrap_or(false);
                if !empty {
                    let msg = format!(
                        "home link {from} → {to}: target exists with content — NOT touched, reconcile manually"
                    );
                    tracing::warn!("home_links: {msg}");
                    report.push(msg);
                    continue;
                }
                let _ = std::fs::remove_dir(&to_dir);
            }
            match std::fs::rename(&from_dir, &to_dir) {
                Ok(()) => {
                    let msg =
                        format!("home link applied: {from} → {to} (config/sessions restored)");
                    tracing::info!("home_links: {msg}");
                    report.push(msg);
                }
                Err(e) => {
                    let msg = format!("home link {from} → {to} FAILED: {e}");
                    tracing::warn!("home_links: {msg}");
                    report.push(msg);
                }
            }
        }
    }
    report
}

// ---------------------------------------------------------------------------
// Multi-session types
// ---------------------------------------------------------------------------

/// A single session with its own broadcast channel.
pub struct UserSessionEntry {
    /// The agent session, protected by a mutex.
    pub session: Arc<Mutex<Session>>,
    /// Broadcast sender for this session's WebSocket fan-out.
    pub ws_tx: broadcast::Sender<String>,
    /// When this session was created.
    pub created_at: chrono::DateTime<chrono::Utc>,
    /// Last time a command was submitted or state changed.
    /// Updated on every /sessions/{id}/command and /sessions/{id}/cancel call.
    pub last_active: Arc<Mutex<chrono::DateTime<chrono::Utc>>>,
}

/// All sessions belonging to one authenticated user.
pub struct UserSessions {
    /// session_id → session entry
    pub sessions: DashMap<String, UserSessionEntry>,
    /// Shared token store for all this user's sessions (MCP credential isolation).
    pub token_store: Arc<pep::MemoryTokenStore>,
}

/// Summary of an active session for listing (serializable for API responses).
#[derive(Debug, serde::Serialize)]
pub struct SessionListItem {
    pub session_id: String,
    pub session_name: Option<String>,
    pub workflow_state: String,
    pub created_at: String,
    pub last_active: String,
    /// Number of handoff rotations this session has performed (Bug 5). The
    /// in-memory registry key never changes across a rotation — only the
    /// checkpoint-chain name does — so a card whose name changed without this
    /// hint looks like it silently renamed. 0 = never rotated.
    pub handoff_count: u32,
}

/// Errors from multi-session operations.
#[derive(Debug)]
pub enum SessionError {
    /// User has reached max_sessions_per_user limit.
    MaxSessionsReached(usize),
    /// Session ID not found for this user.
    NotFound(String),
    /// Session is not Idle (cannot destroy/overwrite a running session).
    NotIdle(String),
}

impl std::fmt::Display for SessionError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            SessionError::MaxSessionsReached(n) => {
                write!(f, "Maximum {} sessions per user reached", n)
            }
            SessionError::NotFound(id) => write!(f, "Session {} not found", id),
            SessionError::NotIdle(state) => write!(f, "Session is not idle (state: {})", state),
        }
    }
}

impl std::error::Error for SessionError {}

/// Top-level registry: user_key → user's session collection.
pub type SessionRegistry = Arc<DashMap<String, UserSessions>>;

// ---------------------------------------------------------------------------
// ServerState
// ---------------------------------------------------------------------------

/// Cached per-user MCP tool loader (16C).
///
/// One entry per user hash. `loader: None` means the user's effective
/// config has MCP disabled — agents run MCP-less via abk's
/// `McpSource::Prebuilt(None)` (semantically identical to
/// `[mcp] enabled = false` today, but with zero config re-parsing).
pub struct McpLoaderEntry {
    /// Built loader; `None` = MCP disabled for this user.
    pub loader: Option<std::sync::Arc<abk::agent::McpToolLoader>>,
    /// Content fingerprint of the effective `[mcp]` config (SHA-256, first
    /// 8 bytes as u64). Content, never mtime — overlays are rewritten in place.
    pub fingerprint: u64,
    pub built_at: chrono::DateTime<chrono::Utc>,
    /// Set when the last build failed; the message surfaces to the user's
    /// next dispatch (fail loud). Other users are never affected.
    pub degraded: Option<String>,
    /// When `degraded` was set. Rebuilds are held back for
    /// [`MCP_BUILD_RETRY_BACKOFF`] — a poison entry never sticks forever.
    pub failed_at: Option<chrono::DateTime<chrono::Utc>>,
}

/// Minimum delay before retrying a failed MCP loader build (16C).
pub const MCP_BUILD_RETRY_BACKOFF: std::time::Duration = std::time::Duration::from_secs(30);

/// Maximum age of a cached MCP toolset manifest (42131deb).
///
/// The 16C fingerprint covers the LOCAL config only — a served-surface
/// change behind a byte-identical config (bridge/spec deploy) leaves the
/// fingerprint matching while the cached tool list goes stale: phantom
/// tools for every fresh session until a restart. Past this age the next
/// fresh-session dispatch REBUILDS the loader (fresh connect re-lists the
/// server's tools), bounding the phantom window to ≤ 10 minutes with no
/// operator action.
pub const MCP_TOOLSET_MAX_AGE: std::time::Duration = std::time::Duration::from_secs(600);

/// Why a cached MCP loader entry was (or wasn't) reused (42131deb).
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) enum LoaderFreshness {
    /// Fingerprint matches and the entry is young — reuse.
    Fresh,
    /// Effective config content changed — rebuild (existing 16C behavior).
    ConfigChanged,
    /// Entry older than [`MCP_TOOLSET_MAX_AGE`] — the SERVED surface may
    /// have changed under an unchanged local config; rebuild so the next
    /// fresh session re-lists the server's tools.
    AgedOut,
    /// Degraded entry still inside the retry backoff — fail loud.
    Backoff(std::time::Duration),
}

/// Pure staleness decision for a cache entry — unit-testable; the async
/// cache path consumes it verbatim (b597c9e9 decide-fn pattern).
pub(crate) fn loader_freshness(
    entry_fingerprint: u64,
    entry_built_at: chrono::DateTime<chrono::Utc>,
    entry_degraded: Option<&str>,
    entry_failed_at: Option<chrono::DateTime<chrono::Utc>>,
    wanted_fingerprint: u64,
    now: chrono::DateTime<chrono::Utc>,
) -> LoaderFreshness {
    // Degraded entries first: a poison entry never sticks forever, but it
    // is held back for the retry window regardless of age/changes.
    if entry_degraded.is_some() {
        if let Some(failed_at) = entry_failed_at {
            let backoff = chrono::Duration::from_std(MCP_BUILD_RETRY_BACKOFF)
                .unwrap_or_else(|_| chrono::Duration::seconds(30));
            if now < failed_at + backoff {
                return LoaderFreshness::Backoff(
                    (failed_at + backoff - now).to_std().unwrap_or_default(),
                );
            }
        }
    }
    if wanted_fingerprint != entry_fingerprint {
        return LoaderFreshness::ConfigChanged;
    }
    let max_age = chrono::Duration::from_std(MCP_TOOLSET_MAX_AGE)
        .unwrap_or_else(|_| chrono::Duration::seconds(600));
    if now > entry_built_at + max_age {
        LoaderFreshness::AgedOut
    } else {
        LoaderFreshness::Fresh
    }
}

/// 16F: per-agent dispatch target — THQ agent_name → the agent-user that
/// THQ-dispatched sessions must run AS.
///
/// Populated once at boot by the 16E discovery scan
/// (`thq_dispatch::populate`); restart-only lifecycle, exactly like the
/// enrollment lane itself.
#[derive(Debug, Clone)]
pub struct ThqDispatchEntry {
    /// The agent-user's stable key = its Kanidm `sub` (16E sub-pin). This is
    /// the identity impersonated on the inner dispatch: session bucket,
    /// per-user home, MCP loader cache, and Cedar principal all resolve
    /// through it.
    pub user_key: String,
    /// The agent's Kanidm service token from its per-user `.env` — exchanged
    /// for a short-lived `role=agent` Bearer at dispatch time. `None` = the
    /// agent has no provisioned credential (dispatch fails loud, 502).
    pub service_token: Option<String>,
    /// The issuer the token's own credential declares (`[mcp.credentials.*]
    /// .issuer_url` from the agent's overlay). Kanidm accepts a token
    /// exchange only on the origin the token was minted for — the issuer
    /// MUST travel with the credential that owns the token (verified live
    /// 2026-08-30: same token, 200 on its own vhost, 400 elsewhere).
    pub issuer_url: Option<String>,
}

/// 16F: the service-account issuer declared by the shared config's first
/// `[mcp.credentials.*]` entry with `type = "service-account"` (see
/// `ServerState::service_issuer` for why this exists).
pub fn service_issuers_from_config(config_toml: &str) -> Vec<String> {
    let Ok(v) = config_toml.parse::<toml::Value>() else {
        return Vec::new();
    };
    let Some(creds) = v
        .get("mcp")
        .and_then(|m| m.get("credentials"))
        .and_then(|c| c.as_table())
    else {
        return Vec::new();
    };
    let mut out = Vec::new();
    for (_name, cred) in creds {
        if cred.get("type").and_then(|t| t.as_str()) == Some("service-account") {
            if let Some(issuer) = cred.get("issuer_url").and_then(|i| i.as_str()) {
                if !out.iter().any(|e: &String| e == issuer) {
                    out.push(issuer.to_string());
                }
            }
        }
    }
    out
}

/// Shared state accessible by all axum handlers.
#[derive(Clone)]
pub struct ServerState {
    /// Per-user multi-session registry (MSU).
    pub sessions: SessionRegistry,
    /// Broadcast sender for backward compat — delegates to the default user's channel.
    pub ws_tx: broadcast::Sender<String>,
    /// Auth state (None = auth disabled, all endpoints open).
    pub auth: Option<Arc<AuthState>>,
    /// Shared config TOML (all users share the same agent config).
    pub config_toml: Option<String>,
    /// Shared secrets (injected into every per-user session).
    pub secrets: Option<std::collections::HashMap<String, String>>,
    /// Shared build info (injected into every per-user session).
    pub build_info: Option<trustee_core::types::BuildInfo>,
    /// Global concurrency limiter — limits the number of simultaneous workflows
    /// across all users. Default: 8 concurrent workflows.
    pub workflow_semaphore: Arc<tokio::sync::Semaphore>,
    /// Maximum number of concurrent sessions per user. Default: 4.
    pub max_sessions_per_user: usize,
    /// Whether per-user config overlays may override the `[llm]` section.
    /// Default: `false` — overlays are limited to `[mcp]`.
    pub allow_llm_overlay: bool,
    /// Per-user McpToolLoader cache (16C), keyed by user hash (never the raw key).
    pub mcp_loaders: Arc<DashMap<String, McpLoaderEntry>>,
    /// Single-flight build locks per user hash (16C).
    mcp_build_locks: Arc<DashMap<String, Arc<tokio::sync::Mutex<()>>>>,
    /// 16F: THQ per-agent dispatch table (agent_name → target), boot-populated.
    pub thq_dispatch: Arc<DashMap<String, ThqDispatchEntry>>,
    /// 16F: cached impersonation Bearers per agent user_key
    /// `(token, expires_at)` — keyed on the exchanged JWT's REAL `exp` claim
    /// (payload-parsed, never the `expires_in` estimate), proactively
    /// refreshed at T-120s before expiry (±15s per-process jitter), and
    /// invalidated + re-minted by the budget-of-one 401 retry in the
    /// xagent dispatch wrappers (199c4801/9618b41c; elimination spec
    /// fb82b866 response c35042bb leg L2).
    pub agent_dispatch_tokens: Arc<DashMap<String, (String, std::time::Instant)>>,
}

impl ServerState {
    /// Create new shared state from a default session, broadcast sender, and optional auth.
    pub fn new(
        session: Session,
        ws_tx: broadcast::Sender<String>,
        auth: Option<Arc<AuthState>>,
    ) -> Self {
        // nghr ec3e0622 acceptance 3: deterministic re-link of pre-rotation
        // homes, applied once at startup (idempotent; every action logged).
        apply_home_links();

        let sessions = Arc::new(DashMap::new());

        // Store the default user's UserSessions with an initial session
        let token_store = Arc::new(pep::MemoryTokenStore::new());
        let (ws_tx_entry, _) = broadcast::channel::<String>(256);

        let now = chrono::Utc::now();
        let initial_entry = UserSessionEntry {
            session: Arc::new(Mutex::new(session)),
            ws_tx: ws_tx_entry,
            created_at: now,
            last_active: Arc::new(Mutex::new(now)),
        };

        let user_sessions = UserSessions {
            sessions: DashMap::new(),
            token_store,
        };
        user_sessions
            .sessions
            .insert("default".to_string(), initial_entry);

        sessions.insert("default".to_string(), user_sessions);

        Self {
            sessions,
            ws_tx,
            auth,
            config_toml: None,
            secrets: None,
            build_info: None,
            workflow_semaphore: Arc::new(tokio::sync::Semaphore::new(8)),
            max_sessions_per_user: 4,
            allow_llm_overlay: false,
            mcp_loaders: Arc::new(DashMap::new()),
            mcp_build_locks: Arc::new(DashMap::new()),
            thq_dispatch: Arc::new(DashMap::new()),
            agent_dispatch_tokens: Arc::new(DashMap::new()),
        }
    }

    pub fn with_config_toml(mut self, config_toml: String) -> Self {
        // fffe9141 F3: loud boot WARN when [llm.provider].model is
        // missing/empty — the silent fictional fallback is gone (abk 0.22.8
        // F5), so a missing model explodes at the first LLM call with the
        // honest 'no model configured' error. THQ-runtime instances
        // legitimately boot without serving LLM sessions — warn, never
        // refuse.
        let model = config_toml
            .parse::<toml::Value>()
            .ok()
            .and_then(|v| {
                v.get("llm")
                    .and_then(|l| l.get("provider"))
                    .and_then(|p| p.get("model"))
                    .and_then(|m| m.as_str())
                    .map(|s| s.trim().to_string())
            })
            .unwrap_or_default();
        if model.is_empty() {
            tracing::warn!(
                "[llm.provider].model is missing or empty — LLM calls will fail with 'no model configured: set [llm.provider].model or OPENAI_DEFAULT_MODEL' until set (THQ-runtime instances that never serve LLM sessions may ignore this)"
            );
        }
        self.config_toml = Some(config_toml);
        self
    }

    /// 16F: the issuer URL Kanidm accepts SERVICE-ACCOUNT token exchange on.
    ///
    /// Kanidm binds exchange to the OAuth2 client's configured origin: the
    /// `[oidc]` issuer host serves logins/validation fine but REJECTS token
    /// exchange with `invalid_request`, while the `idp` vhost — the issuer
    /// every working `[mcp.credentials.*]` service-account entry already
    /// uses — accepts it (verified live 2026-08-30: same token, 200 vs 400).
    /// The exchange therefore takes its issuer from the shared config's
    /// first `service-account` credential; callers fall back to the auth
    /// issuer when absent.
    pub fn service_issuers(&self) -> Vec<String> {
        service_issuers_from_config(self.config_toml.as_deref().unwrap_or(""))
    }

    pub fn with_secrets(mut self, secrets: std::collections::HashMap<String, String>) -> Self {
        self.secrets = Some(secrets);
        self
    }

    pub fn with_build_info(mut self, build_info: trustee_core::types::BuildInfo) -> Self {
        self.build_info = Some(build_info);
        self
    }

    pub fn with_max_concurrent_workflows(mut self, max: usize) -> Self {
        self.workflow_semaphore = Arc::new(tokio::sync::Semaphore::new(max));
        self
    }

    /// Set the max sessions per user.
    pub fn with_max_sessions_per_user(mut self, max: usize) -> Self {
        self.max_sessions_per_user = max;
        self
    }

    /// Allow per-user config overlays to override the `[llm]` section.
    /// Default: `false` (overlays limited to `[mcp]`).
    pub fn with_allow_llm_overlay(mut self, allow: bool) -> Self {
        self.allow_llm_overlay = allow;
        self
    }

    // -----------------------------------------------------------------------
    // MSU: Multi-session methods
    // -----------------------------------------------------------------------

    /// Create a new session for a user. Returns the session_id.
    ///
    /// Creates a fresh `Session::new()`, copies shared config, sets per-user
    /// isolation, creates a broadcast channel, spawns a drain task, and inserts
    /// into the user's session DashMap.
    ///
    /// The session identity is PINNED here (c2f9508, restore-reattach fix):
    /// `session.session_id` = the registry key, so the disk checkpoint chain id
    /// and the live session id coincide and `get_or_restore_session_by_any_id`
    /// can re-attach a client by the id it already holds after an engine
    /// restart. The abk first-command auto-derive still fires for TUI/desktop
    /// sessions (no registry key there). Resumed sessions overwrite the pin
    /// with the resumed chain id (routes::resume_session) — chain semantics
    /// preserved.
    ///
    /// There is NO "active session" concept anymore — the legacy
    /// active-pointer surface was retired (2026-10-27, owner order); every
    /// consumer targets a session id via the session-scoped routes.
    pub async fn create_session(
        &self,
        user_key: &str,
        session_name: Option<String>,
        identity: Option<String>,
    ) -> Result<String, SessionError> {
        // Get or create the user's UserSessions entry
        let user_sessions = self
            .sessions
            .entry(user_key.to_string())
            .or_insert_with(|| UserSessions {
                sessions: DashMap::new(),
                token_store: Arc::new(pep::MemoryTokenStore::new()),
            });

        // Check session limit
        if user_sessions.sessions.len() >= self.max_sessions_per_user {
            return Err(SessionError::MaxSessionsReached(self.max_sessions_per_user));
        }

        // Create new Session
        let (mut session, workflow_rx) = Session::new();

        // Copy shared config
        if let Some(ref config_toml) = self.config_toml {
            session.config_toml = Some(config_toml.clone());
            session.parse_auto_handoff_config();
            if let Ok(table) = config_toml.parse::<toml::Value>() {
                if let Some(name) = table
                    .get("agent")
                    .and_then(|a| a.get("name"))
                    .and_then(|n| n.as_str())
                {
                    session.agent_name = name.to_string();
                }
            }
        }

        session.secrets = self.secrets.clone();
        session.build_info = self.build_info.clone();

        // Per-user isolation
        self.apply_user_isolation(&mut session, user_key);

        // Apply session_name if provided
        session.session_name = session_name;

        // Apply agent identity if provided
        session.identity = identity;

        // Create broadcast channel
        let (ws_tx_entry, _) = broadcast::channel::<String>(256);

        // Generate session_id
        let session_id = format!(
            "session_{}_{}",
            chrono::Utc::now().format("%Y_%m_%d_%H_%M"),
            &uuid::Uuid::new_v4().to_string()[..8]
        );

        // PIN the session identity (Hamrah 2026-10-26, restore-reattach fix):
        // the registry key IS the disk chain id. Without this, the first
        // command's abk run auto-generates a DIFFERENT session_id for the
        // on-disk checkpoint chain, and get_or_restore_session_by_any_id can
        // never find the chain by the live id after an engine restart.
        session.session_id = Some(session_id.clone());

        let now = chrono::Utc::now();

        // Insert into user's sessions DashMap
        user_sessions.sessions.insert(
            session_id.clone(),
            UserSessionEntry {
                session: Arc::new(Mutex::new(session)),
                ws_tx: ws_tx_entry.clone(),
                created_at: now,
                last_active: Arc::new(Mutex::new(now)),
            },
        );

        // Spawn drain task
        let session_arc = user_sessions
            .sessions
            .get(&session_id)
            .map(|e| e.session.clone());
        if let Some(session_arc) = session_arc {
            self.spawn_user_drain_task(session_id.clone(), session_arc, ws_tx_entry, workflow_rx);
        }

        Ok(session_id)
    }

    /// Get a specific session by user_key + session_id.
    /// Updates last_active on the session entry.
    pub async fn get_session(
        &self,
        user_key: &str,
        session_id: &str,
    ) -> Option<(Arc<Mutex<Session>>, broadcast::Sender<String>)> {
        let user_sessions = self.sessions.get(user_key)?;
        let entry = user_sessions.sessions.get(session_id)?;

        // Update last_active
        let now = chrono::Utc::now();
        *entry.last_active.lock().await = now;

        Some((entry.session.clone(), entry.ws_tx.clone()))
    }

    /// Get a session by EITHER its live MSU registry key OR its
    /// checkpoint/session identity (`session.session_id`).
    ///
    /// The web frontend tracks `currentSessionId` from the `ResumeInfo` WS
    /// message, which carries the auto-derived checkpoint id
    /// (`session_YYYY_MM_DD_HH_MM_uuid8`) — NOT the live MSU registry key
    /// (`"default"` or the key from `create_session()`). External clients
    /// like Torpi/THQ pass the live registry key. This resolver accepts both:
    ///
    /// 1. Try registry-key lookup first (precise, used by Torpi/THQ).
    /// 2. Fall back to scanning the user's live sessions for one whose
    ///    `session.session_id` matches the requested id (used by the
    ///    embedded web UI after a command or resume).
    ///
    /// Returns `(live_registry_key, session_arc, ws_tx)`, or `None` if not
    /// found. The live key is returned so callers that need to set it as
    /// active (or otherwise reference the registry) use the real key.
    pub async fn get_session_by_any_id(
        &self,
        user_key: &str,
        id: &str,
    ) -> Option<(String, Arc<Mutex<Session>>, broadcast::Sender<String>)> {
        // Fast path: registry key match.
        let user_sessions = self.sessions.get(user_key)?;
        if let Some(entry) = user_sessions.sessions.get(id) {
            // Update last_active
            let now = chrono::Utc::now();
            *entry.last_active.lock().await = now;
            return Some((id.to_string(), entry.session.clone(), entry.ws_tx.clone()));
        }

        // Slow path: scan live sessions for a matching session.session_id.
        for entry in user_sessions.sessions.iter() {
            let session = entry.session.lock().await;
            if session.session_id.as_deref() == Some(id) {
                let key = entry.key().clone();
                let ws_tx = entry.ws_tx.clone();
                drop(session);
                // Update last_active
                let now = chrono::Utc::now();
                *entry.last_active.lock().await = now;
                return Some((key, entry.session.clone(), ws_tx));
            }
        }

        None
    }

    /// Tombstone directory for destroyed-but-on-disk session chains.
    ///
    /// Destroy removes the live session but deliberately keeps checkpoints
    /// ("checkpoint data is preserved"). Restoring must NOT resurrect a
    /// chain the user explicitly terminated, so destroy writes a marker
    /// here and [`Self::restore_session_from_checkpoint`] refuses to
    /// hydrate tombstoned chains. Markers are tiny and intentionally
    /// permanent.
    fn tombstone_path(&self, user_key: &str, chain_id: &str) -> Option<std::path::PathBuf> {
        if chain_id.is_empty() || !chain_id.starts_with("session_") {
            return None;
        }
        let user_key = effective_user_key(user_key);
        let hash = trustee_core::user_hash(&user_key);
        dirs::home_dir().map(|home| {
            home.join(".trustee")
                .join("users")
                .join(&hash)
                .join("session_tombstones")
                .join(chain_id)
        })
    }

    /// Write a destroy tombstone for a checkpoint chain (best effort).
    ///
    /// Called from [`Self::destroy_session`] when the destroyed live
    /// session was attached to an on-disk chain.
    fn write_tombstone(&self, user_key: &str, chain_id: &str) {
        if let Some(path) = self.tombstone_path(user_key, chain_id) {
            if let Some(parent) = path.parent() {
                let _ = std::fs::create_dir_all(parent);
            }
            let _ = std::fs::write(&path, chrono::Utc::now().to_rfc3339());
        }
    }

    /// Read the destroy-tombstone stamp for a chain, if one exists.
    ///
    /// 7df3cae2 (tombstone-shadow class): a destroyed chain whose
    /// checkpoints sit intact on disk must be DISCLOSED, never shadowed by
    /// a lying generic absence — readers use this stamp to answer
    /// tombstoned-but-non-empty loudly. `None` = no marker (chain either
    /// alive/resumable or genuinely absent).
    pub fn chain_tombstone_stamp(&self, user_key: &str, chain_id: &str) -> Option<String> {
        let path = self.tombstone_path(user_key, chain_id)?;
        let text = std::fs::read_to_string(&path).ok()?;
        let trimmed = text.trim();
        if trimmed.is_empty() {
            None
        } else {
            Some(trimmed.to_string())
        }
    }

    /// Look up a live session by any id; on a miss, transparently restore
    /// the checkpoint chain from disk.
    ///
    /// Restart resilience: trustee-api holds live sessions in memory only,
    /// so a process restart orphans every open console. When a client asks
    /// for a chain id that is not in memory but EXISTS on disk
    /// (checkpoints present, not tombstoned by destroy), a live session is
    /// rebuilt with `resume_info` pointing at the latest checkpoint and
    /// registered under the chain id. The next command continues the
    /// conversation; the console's alive-check and WS reconnect succeed.
    ///
    /// Returns `(live_registry_key, session_arc, ws_tx)` — the key is the
    /// chain id for restored sessions.
    pub async fn get_or_restore_session_by_any_id(
        &self,
        user_key: &str,
        id: &str,
    ) -> Option<(String, Arc<Mutex<Session>>, broadcast::Sender<String>)> {
        if let Some(found) = self.get_session_by_any_id(user_key, id).await {
            return Some(found);
        }
        self.restore_session_from_checkpoint(user_key, id).await
    }

    /// Rebuild a live session from an on-disk checkpoint chain.
    ///
    /// Persona note: dispatched (THQ/xagent) sessions carry `identity:
    /// None` by design — their persona is the per-user overlay config
    /// ([lifecycle].system_template), which `apply_user_isolation` +
    /// `get_user_config_and_home` re-apply here exactly as at creation.
    /// Explicit per-request identity overrides (rare) are not persisted
    /// and therefore not restored.
    async fn restore_session_from_checkpoint(
        &self,
        user_key: &str,
        chain_id: &str,
    ) -> Option<(String, Arc<Mutex<Session>>, broadcast::Sender<String>)> {
        // Never resurrect an explicitly destroyed chain.
        if let Some(path) = self.tombstone_path(user_key, chain_id) {
            if path.exists() {
                return None;
            }
        }

        // Restore needs the caller's resolved config (persona) and home.
        let (config_toml, home_dir) = self.get_user_config_and_home(user_key);
        let config_toml = config_toml?;

        // The chain must exist on disk with at least one checkpoint;
        // otherwise this id is genuinely unknown → 404 semantics.
        //
        // Resume-handle resolution (Ravand 2026-10-27): a resumed session's
        // live id is a fresh registry key — its disk dir carries a
        // `resumed_from` constant naming the SOURCE chain (recorded at
        // resume time by write_resume_pointer). Follow it so
        // restore-by-held-id works after an engine restart: the SOURCE
        // chain is restored under the HELD id, and the next command
        // continues the source conversation.
        let info = match trustee_core::sessions::create_resume_info(
            &config_toml,
            chain_id,
            home_dir.as_deref(),
        )
        .await
        .ok()
        .flatten()
        {
            Some(info) => info,
            None => {
                let source = trustee_core::sessions::resolve_resume_handle(
                    &config_toml,
                    chain_id,
                    home_dir.as_deref(),
                )
                .await
                .ok()
                .flatten()?;
                trustee_core::sessions::create_resume_info(
                    &config_toml,
                    &source,
                    home_dir.as_deref(),
                )
                .await
                .ok()
                .flatten()?
            }
        };

        // Single-flight per chain id: racing lookups dedupe on the DashMap
        // entry — the first insert wins, the loser adopts the winner.
        let user_sessions = self
            .sessions
            .entry(user_key.to_string())
            .or_insert_with(|| UserSessions {
                sessions: DashMap::new(),
                token_store: Arc::new(pep::MemoryTokenStore::new()),
            });

        if let Some(existing) = user_sessions.sessions.get(chain_id) {
            let now = chrono::Utc::now();
            *existing.last_active.lock().await = now;
            return Some((
                chain_id.to_string(),
                existing.session.clone(),
                existing.ws_tx.clone(),
            ));
        }

        // Respect the per-user cap: evict the least-recently-active Idle
        // live session to make room (restores are user-initiated
        // reattaches, so refusing them would recreate the lost-session
        // bug for heavy users). If nothing is evictable, refuse.
        if user_sessions.sessions.len() >= self.max_sessions_per_user {
            let mut evict: Option<(String, chrono::DateTime<chrono::Utc>)> = None;
            for entry in user_sessions.sessions.iter() {
                let is_idle = entry.session.lock().await.workflow_state
                    == trustee_core::types::WorkflowState::Idle;
                if !is_idle {
                    continue;
                }
                let la = entry.last_active.lock().await;
                if evict.as_ref().is_none_or(|(_, t)| *la < *t) {
                    evict = Some((entry.key().clone(), *la));
                }
            }
            if let Some((victim, _)) = evict {
                user_sessions.sessions.remove(&victim);
            } else {
                return None;
            }
        }

        // Build the session exactly like create_session, then attach the
        // on-disk chain identity and resume pointer.
        let (mut session, workflow_rx) = Session::new();
        session.config_toml = Some(config_toml.clone());
        session.parse_auto_handoff_config();
        if let Ok(table) = config_toml.parse::<toml::Value>() {
            if let Some(name) = table
                .get("agent")
                .and_then(|a| a.get("name"))
                .and_then(|n| n.as_str())
            {
                session.agent_name = name.to_string();
            }
        }
        session.secrets = self.secrets.clone();
        session.build_info = self.build_info.clone();
        self.apply_user_isolation(&mut session, user_key);
        session.session_id = Some(chain_id.to_string());
        session.resume_info = Some(info);
        session.session_name = None; // display name re-derives from metadata

        let (ws_tx_entry, _) = broadcast::channel::<String>(256);
        let now = chrono::Utc::now();

        user_sessions.sessions.insert(
            chain_id.to_string(),
            UserSessionEntry {
                session: Arc::new(Mutex::new(session)),
                ws_tx: ws_tx_entry.clone(),
                created_at: now,
                last_active: Arc::new(Mutex::new(now)),
            },
        );

        // Drain task: forward workflow output into the WS broadcast channel
        // (same contract as create_session — without this the restored
        // session would run silently).
        let session_arc = user_sessions
            .sessions
            .get(chain_id)
            .map(|e| e.session.clone())
            .ok_or_else(|| SessionError::NotFound(chain_id.to_string()))
            .ok()?;
        self.spawn_user_drain_task(
            chain_id.to_string(),
            session_arc.clone(),
            ws_tx_entry.clone(),
            workflow_rx,
        );

        Some((chain_id.to_string(), session_arc, ws_tx_entry))
    }

    /// List all active sessions for a user, sorted by last_active desc.
    pub async fn list_sessions(&self, user_key: &str) -> Vec<SessionListItem> {
        let Some(user_sessions) = self.sessions.get(user_key) else {
            return Vec::new();
        };

        let mut items = Vec::new();
        for entry in user_sessions.sessions.iter() {
            let session = entry.session.lock().await;
            let workflow_state = match session.workflow_state {
                trustee_core::types::WorkflowState::Idle => "Idle",
                trustee_core::types::WorkflowState::Running => "Running",
                trustee_core::types::WorkflowState::Cancelling => "Cancelling",
            };
            let last_active = entry.last_active.lock().await;
            items.push(SessionListItem {
                session_id: entry.key().clone(),
                session_name: session.session_name.clone(),
                workflow_state: workflow_state.to_string(),
                created_at: entry.created_at.to_rfc3339(),
                last_active: last_active.to_rfc3339(),
                handoff_count: session.handoff_count,
            });
        }
        drop(user_sessions);

        // Sort by last_active descending
        items.sort_by(|a, b| b.last_active.cmp(&a.last_active));
        items
    }

    /// Destroy a session. The session must be Idle.
    pub async fn destroy_session(
        &self,
        user_key: &str,
        session_id: &str,
    ) -> Result<(), SessionError> {
        let user_sessions = self
            .sessions
            .get(user_key)
            .ok_or_else(|| SessionError::NotFound(session_id.to_string()))?;

        // Check workflow state before removing; capture the on-disk chain
        // id (if any) so destroy can tombstone it against later restores.
        let chain_id_for_tombstone = {
            let entry = user_sessions
                .sessions
                .get(session_id)
                .ok_or_else(|| SessionError::NotFound(session_id.to_string()))?;
            let session = entry.session.lock().await;
            if session.workflow_state != trustee_core::types::WorkflowState::Idle {
                let state_str = match session.workflow_state {
                    trustee_core::types::WorkflowState::Running => "Running",
                    trustee_core::types::WorkflowState::Cancelling => "Cancelling",
                    _ => "Unknown",
                };
                return Err(SessionError::NotIdle(state_str.to_string()));
            }
            session.session_id.clone()
        };

        // Remove from DashMap
        user_sessions.sessions.remove(session_id);

        // If the destroyed live session was attached to an on-disk chain,
        // tombstone it so a later get_or_restore lookup does not resurrect
        // a session the user explicitly terminated.
        if let Some(chain_id) = chain_id_for_tombstone {
            self.write_tombstone(user_key, &chain_id);
        }

        Ok(())
    }

    // -----------------------------------------------------------------------
    // Read-only helpers (no session creation side effects)
    // -----------------------------------------------------------------------

    /// Resolve a user's home_dir without creating an in-memory session.
    ///
    /// This is the read-only equivalent of the isolation logic in
    /// `apply_user_isolation`. Used by endpoints that only need to read
    /// checkpoint data from disk (history, session list, session detail)
    /// and must NOT create ghost sessions as a side effect.
    pub fn get_user_home_dir(&self, user_key: &str) -> Option<std::path::PathBuf> {
        // TRUSTEE_CLASSIC_HOME=1 (Android single-user loopback, Hamrah
        // 2026-10-26): use the CLASSIC home — the same one the command path
        // and the CLI write to — so the session list/resume/history see the
        // real sessions instead of an empty users/{hash}/ namespace.
        //
        // LEAK GUARD (Ravand 2026-10-27 review): under this gate EVERY user
        // shares the classic home — per-user checkpoint isolation is OFF.
        // That is correct ONLY for the single-user loopback. If the env is
        // set on a multi-user server by accident, all users would see each
        // other's checkpoint chains — say so LOUD, once, at first use.
        if std::env::var("TRUSTEE_CLASSIC_HOME").as_deref() == Ok("1") {
            static GATE_WARNED: std::sync::Once = std::sync::Once::new();
            GATE_WARNED.call_once(|| {
                tracing::warn!(
                    "TRUSTEE_CLASSIC_HOME=1 — ALL users share the classic home \
                     (~/.trustee); per-user checkpoint isolation is OFF. \
                     Single-user loopback ONLY — unset this env on multi-user servers."
                );
            });
            return dirs::home_dir().map(|home| home.join(".trustee"));
        }
        let user_key = effective_user_key(user_key);
        let hash = trustee_core::user_hash(&user_key);
        dirs::home_dir().map(|home| home.join(".trustee").join("users").join(&hash))
    }

    /// Resolve config_toml and home_dir without creating an in-memory session.
    ///
    /// Returns `(config_toml, home_dir)`. If config is not loaded,
    /// config_toml will be None.
    pub fn get_user_config_and_home(
        &self,
        user_key: &str,
    ) -> (Option<String>, Option<std::path::PathBuf>) {
        (self.config_toml.clone(), self.get_user_home_dir(user_key))
    }

    // -----------------------------------------------------------------------
    // Private helpers
    // -----------------------------------------------------------------------

    /// Apply per-user isolation: SHA-256 hash → home_dir + project_id.
    ///
    /// Hashing goes through the single consolidated [`trustee_core::user_hash`]
    /// so the web path and the CLI path can never drift apart.
    fn apply_user_isolation(&self, session: &mut Session, user_key: &str) {
        // TRUSTEE_CLASSIC_HOME=1 (Android single-user loopback): the classic
        // home — the per-user isolation only applies in multi-user web mode.
        if std::env::var("TRUSTEE_CLASSIC_HOME").as_deref() == Ok("1") {
            if let Some(home) = dirs::home_dir() {
                session.home_dir = Some(home.join(".trustee"));
            }
            return;
        }
        let user_key = effective_user_key(user_key);
        let user_hash = trustee_core::user_hash(&user_key);

        // Set per-user home directory for checkpoint isolation
        let user_home = if let Some(home) = dirs::home_dir() {
            let user_home = home.join(".trustee").join("users").join(&user_hash);
            session.home_dir = Some(user_home.clone());
            Some(user_home)
        } else {
            None
        };

        // 92a36073 boot-surface: a MATERIALIZED namespace without per-user
        // credentials is the shadowing scenario — say so NOW, not in
        // forensics after a global token answered for it.
        if let Some(ref user_home) = user_home {
            if materialized_user_missing_env(user_home) {
                tracing::warn!(
                    "user namespace {} is materialized but has NO per-user .env — \
                     ${{…}} references will fall through to process env \
                     (global ~/.trustee/.env) and execute under the GLOBAL principal",
                    &user_hash[..8.min(user_hash.len())]
                );
            }
        }

        session.project_id = Some(format!("web{}", &user_hash[..16]));

        // ── Per-user .env (Task 2) ──────────────────────────────────────
        //
        // Load per-user secrets from ~/.trustee/users/{hash}/.env
        // These are merged on top of shared secrets (per-user wins).
        // They are NEVER set as process env vars — used only for ${VAR}
        // substitution in the config TOML below.
        let shared_secrets = session.secrets.clone().unwrap_or_default();
        let mut merged_secrets = shared_secrets.clone();

        if let Some(ref user_home) = user_home {
            if let Ok(merged) = self.load_user_secrets(user_home, &merged_secrets) {
                merged_secrets = merged;
            }
        }

        // ── Per-user config overlay (Task 3) ────────────────────────────
        if let Some(ref user_home) = user_home {
            if let Some(merged) = self.merge_user_config(user_home, session.config_toml.as_deref())
            {
                session.config_toml = Some(merged);
                tracing::debug!("Merged per-user config into session");
            }
        }

        // ── ${VAR} substitution (Task 4) ────────────────────────────────
        if let Some(ref mut config_toml) = session.config_toml {
            substitute_env_vars(
                config_toml,
                &merged_secrets,
                &format!("user {}", &user_hash[..8.min(user_hash.len())]),
            );
        }

        // ── Strip per-user secrets (Task 5) ────────────────────────────
        session.secrets = Some(shared_secrets);
    }

    /// Load secrets from a user's ~/.trustee/users/{hash}/.env and merge
    /// on top of `base` (user wins). Returns the merged map.
    fn load_user_secrets(
        &self,
        user_home: &std::path::Path,
        base: &std::collections::HashMap<String, String>,
    ) -> std::io::Result<std::collections::HashMap<String, String>> {
        let user_env_path = user_home.join(".env");
        if !user_env_path.exists() {
            return Ok(base.clone());
        }
        let content = std::fs::read_to_string(&user_env_path)?;
        let mut merged = base.clone();
        for line in content.lines() {
            let line = line.trim();
            if line.is_empty() || line.starts_with('#') {
                continue;
            }
            if let Some((key, value)) = line.split_once('=') {
                let key = key.trim().to_string();
                let value = value
                    .trim()
                    .trim_matches('"')
                    .trim_matches('\'')
                    .to_string();
                merged.insert(key, value);
            }
        }
        tracing::debug!("Loaded per-user secrets from {}", user_env_path.display());
        Ok(merged)
    }

    /// Deep-merge a user's per-user config overlay on top of the shared
    /// config. Returns the merged TOML string, or None if no overlay exists.
    ///
    /// Overlay allowlist (task 16B): only allowlisted top-level sections of
    /// the user overlay are merged; anything else is dropped loudly.
    /// - `[mcp]` is always allowed — per-user MCP tool sets are the point of
    ///   the overlay convention.
    /// - `[llm]` is allowed only when the instance opted in via the
    ///   `[users].allow_llm_overlay` knob (default **false**).
    ///
    /// Everything else (`[server]`, `[auth]`, `[storage]`, `[web]`, …) is
    /// boot-time, instance-level config and must not be rewritable through a
    /// per-user file. This is predictability hardening, not a security
    /// boundary: those sections were never re-read from session config.
    fn merge_user_config(
        &self,
        user_home: &std::path::Path,
        shared_config: Option<&str>,
    ) -> Option<String> {
        let user_config_path = user_home.join("config").join("trustee.toml");
        if !user_config_path.exists() {
            return None;
        }
        let user_config_toml = std::fs::read_to_string(&user_config_path).ok()?;
        let shared = shared_config.unwrap_or("").parse::<toml::Value>().ok()?;
        let overlay = user_config_toml.parse::<toml::Value>().ok()?;

        let allowed = |section: &str| {
            section == "mcp"
                // 16E: [thq] is per-user identity config (read from the overlay
                // file at boot); keeping it in the merged session config is
                // inert but avoids a false "dropping" warn on every dispatch.
                || section == "thq"
                // xagent personas: an agent's [lifecycle].system_template IS
                // her identity — user-owned by design. The WASM extension
                // switch ([lifecycle].enabled) is stripped below: persona is
                // user-owned, extension control is not.
                || section == "lifecycle"
                || (self.allow_llm_overlay && section == "llm")
        };
        // Mask the user in logs: user_home's dir name IS the hash — never
        // log the raw key (keys may be emails). Log a short hash prefix.
        let dir_name = user_home
            .file_name()
            .and_then(|n| n.to_str())
            .unwrap_or("<unknown>");
        let masked_user = dir_name.get(..8).unwrap_or(dir_name);

        let mut filtered_overlay = toml::map::Map::new();
        if let Some(table) = overlay.as_table() {
            for (section, value) in table {
                if section == "lifecycle" {
                    let mut v = value.clone();
                    if let Some(t) = v.as_table_mut() {
                        if t.remove("enabled").is_some() {
                            tracing::info!(
                                "user config overlay: [lifecycle].enabled is not user-settable — stripped for user {}",
                                masked_user
                            );
                        }
                    }
                    filtered_overlay.insert(section.clone(), v);
                } else if allowed(section) {
                    filtered_overlay.insert(section.clone(), value.clone());
                } else {
                    tracing::warn!(
                        "user config overlay: dropping non-allowlisted section [{}] for user {}",
                        section,
                        masked_user
                    );
                }
            }
        }

        if filtered_overlay.is_empty() {
            // Nothing survived the allowlist — no-op, keep shared config as-is.
            return None;
        }
        let overlay = toml::Value::Table(filtered_overlay);

        let mut shared = shared;
        deep_merge_toml(&mut shared, &overlay);
        let merged = toml::to_string(&shared).ok()?;
        tracing::debug!("Merged per-user config from {}", user_config_path.display());
        Some(merged)
    }

    /// Resolve the fully-merged config TOML for a user WITHOUT creating a session.
    ///
    /// This is the read-only equivalent of the config resolution in
    /// `apply_user_isolation`: shared config + per-user overlay + ${VAR}
    /// substitution. Used by endpoints that need to inspect config (e.g.
    /// listing available LLM models) without spawning a ghost session.
    ///
    /// Returns None if no shared config is loaded.
    pub fn resolve_user_config(&self, user_key: &str) -> Option<String> {
        let config_toml = self.config_toml.clone()?;

        // Resolve user home dir (same hash scheme as apply_user_isolation)
        let user_home = self.get_user_home_dir(user_key)?;

        // Start from shared secrets; merge per-user .env on top
        let mut merged_secrets = self.secrets.clone().unwrap_or_default();
        if let Ok(merged) = self.load_user_secrets(&user_home, &merged_secrets) {
            merged_secrets = merged;
        }

        // Merge per-user config overlay
        let mut resolved = config_toml;
        if let Some(merged) = self.merge_user_config(&user_home, Some(&resolved)) {
            resolved = merged;
        }

        // Substitute ${VAR} from merged secrets
        substitute_env_vars(
            &mut resolved,
            &merged_secrets,
            &format!(
                "resolve_user_config user {}",
                &trustee_core::user_hash(user_key)[..8]
            ),
        );

        Some(resolved)
    }

    /// Get or build the per-user MCP tool loader (16C).
    ///
    /// Cache semantics:
    /// - fingerprint = content hash of the effective `[mcp]` section
    ///   (shared + allowlist-filtered overlay + ${VAR} substitution, i.e.
    ///   exactly what `resolve_user_config` produces) — stale on ANY change.
    /// - hit + match → `Arc` clone, zero network I/O.
    /// - miss/stale → single-flight build (one build per user at a time;
    ///   late arrivals re-check and reuse the winner's entry).
    /// - `Ok(None)` = MCP disabled for this user → agent runs MCP-less via
    ///   abk `McpSource::Prebuilt(None)` (same semantics as
    ///   `[mcp] enabled = false`, no per-task re-evaluation).
    /// - build failure → cached degraded entry with
    ///   [`MCP_BUILD_RETRY_BACKOFF`]; the error is returned so THIS user's
    ///   dispatch fails loud while other users are unaffected.
    pub async fn get_or_build_mcp_loader(
        &self,
        user_key: &str,
        token_store: &Arc<pep::MemoryTokenStore>,
    ) -> Result<Option<std::sync::Arc<abk::agent::McpToolLoader>>, String> {
        let user_key = effective_user_key(user_key);
        let user_hash = trustee_core::user_hash(&user_key);

        // Effective per-user config — the fingerprint source of truth.
        let resolved = self.resolve_user_config(&user_key);
        let fingerprint = fingerprint_mcp_section(resolved.as_deref());

        // Fast path: non-degraded entry that is fresh AND content-current.
        if let Some(entry) = self.mcp_loaders.get(&user_hash) {
            match loader_freshness(
                entry.fingerprint,
                entry.built_at,
                entry.degraded.as_deref(),
                entry.failed_at,
                fingerprint,
                chrono::Utc::now(),
            ) {
                LoaderFreshness::Fresh => return Ok(entry.loader.clone()),
                LoaderFreshness::ConfigChanged => {
                    // fall through to the rebuild path below
                }
                LoaderFreshness::AgedOut => {
                    // 42131deb: the served surface may have changed behind
                    // a byte-identical config — rebuild (re-lists tools).
                    tracing::info!(
                        "MCP toolset for user {} is {}s old (max {}s) — rebuilding \
                         to re-list the served surface (phantom-tool guard)",
                        &user_hash[..8.min(user_hash.len())],
                        (chrono::Utc::now() - entry.built_at).num_seconds(),
                        MCP_TOOLSET_MAX_AGE.as_secs()
                    );
                }
                LoaderFreshness::Backoff(remaining) => {
                    let err = entry.degraded.clone().unwrap_or_else(|| {
                        "previous MCP loader build failed; retry after backoff".to_string()
                    });
                    let _ = remaining;
                    return Err(err);
                }
            }
        }

        // Single-flight: one build per user at a time.
        let lock = self
            .mcp_build_locks
            .entry(user_hash.clone())
            .or_insert_with(|| Arc::new(tokio::sync::Mutex::new(())))
            .clone();
        let _guard = lock.lock().await;

        // Double-check: another task may have built while we waited.
        if let Some(entry) = self.mcp_loaders.get(&user_hash) {
            if loader_freshness(
                entry.fingerprint,
                entry.built_at,
                entry.degraded.as_deref(),
                entry.failed_at,
                fingerprint,
                chrono::Utc::now(),
            ) == LoaderFreshness::Fresh
            {
                return Ok(entry.loader.clone());
            }
        }

        match self
            .build_mcp_loader(&user_hash, resolved.as_deref(), fingerprint, token_store)
            .await
        {
            Ok(entry) => {
                self.mcp_loaders.insert(user_hash.clone(), entry);
                Ok(self.mcp_loaders.get(&user_hash).unwrap().loader.clone())
            }
            Err(err) => {
                tracing::warn!(
                    "MCP loader build FAILED for user {}; dispatch fails loud, retry after {:?}",
                    &user_hash[..8.min(user_hash.len())],
                    MCP_BUILD_RETRY_BACKOFF
                );
                self.mcp_loaders.insert(
                    user_hash,
                    McpLoaderEntry {
                        loader: None,
                        fingerprint,
                        built_at: chrono::Utc::now(),
                        degraded: Some(err.clone()),
                        failed_at: Some(chrono::Utc::now()),
                    },
                );
                Err(err)
            }
        }
    }

    /// Build a loader entry from the effective config. No caching here —
    /// the caller owns insertion and degraded handling.
    async fn build_mcp_loader(
        &self,
        user_hash: &str,
        resolved: Option<&str>,
        fingerprint: u64,
        token_store: &Arc<pep::MemoryTokenStore>,
    ) -> Result<McpLoaderEntry, String> {
        let mcp_config: Option<abk::config::McpConfig> = match resolved {
            Some(toml_str) => {
                let value = toml_str
                    .parse::<toml::Value>()
                    .map_err(|e| format!("config parse failed: {}", e))?;
                match value.get("mcp") {
                    Some(section) => {
                        // 0.19.6 (incident fc94aea9, T3): a leftover ${…} in
                        // the effective [mcp] section means the credential
                        // reference is UNRESOLVED — the literal placeholder
                        // must never reach an agent (it once rode to
                        // production auth as the Bearer). Fail the build
                        // loud; the degraded path handles retry/backoff.
                        let mut leftovers: Vec<String> = Vec::new();
                        collect_unresolved_placeholders(section, &mut leftovers);
                        if !leftovers.is_empty() {
                            return Err(format!(
                                "unresolved {} placeholder(s) in [mcp]: {} — supply the \
                                 exact-case key in the profile secrets",
                                leftovers.len(),
                                leftovers.join(", ")
                            ));
                        }
                        use serde::Deserialize as _;
                        Some(
                            abk::config::McpConfig::deserialize(section.clone())
                                .map_err(|e| format!("invalid [mcp] config: {}", e))?,
                        )
                    }
                    None => None,
                }
            }
            None => None,
        };

        let loader = match mcp_config {
            Some(cfg) if cfg.enabled => {
                let built = abk::agent::McpToolLoader::with_token_store(
                    &cfg,
                    Some(token_store.clone() as std::sync::Arc<dyn pep::token_store::TokenStore>),
                )
                .await
                .map_err(|e| format!("MCP loader build failed: {}", e))?;

                // THE parity-evidence log line (16C acceptance + migration task):
                // one INFO line per build; a task loop that reuses the cache
                // shows exactly one line per user per fingerprint.
                let servers: Vec<String> = built
                    .server_statuses
                    .iter()
                    .map(|s| {
                        if s.connected {
                            format!("{}(up,{}tools)", s.name, s.tool_count)
                        } else {
                            format!("{}(DOWN)", s.name)
                        }
                    })
                    .collect();
                tracing::info!(
                    "MCP loader built for user {}: servers=[{}] total_tools={}",
                    &user_hash[..8.min(user_hash.len())],
                    servers.join(", "),
                    built.tool_count
                );
                Some(std::sync::Arc::new(built))
            }
            _ => None,
        };

        Ok(McpLoaderEntry {
            loader,
            fingerprint,
            built_at: chrono::Utc::now(),
            degraded: None,
            failed_at: None,
        })
    }

    /// Spawn a background drain task for a specific session's workflow receiver.
    fn spawn_user_drain_task(
        &self,
        session_id: String,
        session: Arc<Mutex<Session>>,
        ws_tx: broadcast::Sender<String>,
        mut workflow_rx: mpsc::UnboundedReceiver<TuiMessage>,
    ) {
        // Bug 6: broadcast StateChanged only on actual transitions. Without
        // this, every WS message was accompanied by a duplicate StateChanged,
        // doubling traffic and re-triggering frontend updateState on hot
        // StreamDelta/ReasoningDelta bursts. New subscribers learn the current
        // state from the WS snapshot, so the initial None is safe.
        let mut last_broadcast_state: Option<String> = None;
        tokio::spawn(async move {
            while let Some(msg) = workflow_rx.recv().await {
                {
                    let mut session = session.lock().await;
                    session.handle_workflow_message(msg.clone());

                    let state_str = match session.workflow_state {
                        trustee_core::types::WorkflowState::Idle => "Idle",
                        trustee_core::types::WorkflowState::Running => "Running",
                        trustee_core::types::WorkflowState::Cancelling => "Cancelling",
                    };
                    if last_broadcast_state.as_deref() != Some(state_str) {
                        last_broadcast_state = Some(state_str.to_string());
                        let state_msg = serde_json::json!({
                            "type": "StateChanged",
                            "state": state_str
                        });
                        let _ = ws_tx.send(state_msg.to_string());
                    }
                }

                let json = serde_json::to_string(&SerializableMessage(&msg)).unwrap_or_default();
                let _ = ws_tx.send(json);
            }
            tracing::debug!("Drain task ended for session: {}", session_id);
        });
    }

    /// Spawn the default user's drain task (backward compatibility).
    /// Called during server startup for the initial session.
    pub fn spawn_drain_task(self, mut workflow_rx: mpsc::UnboundedReceiver<TuiMessage>) {
        // Get the default user's first session
        let default_user = self
            .sessions
            .get("default")
            .expect("default user must exist");
        let first_entry = default_user
            .sessions
            .iter()
            .next()
            .expect("default user must have at least one session");
        let session = first_entry.session.clone();
        let ws_tx = first_entry.ws_tx.clone();
        let session_id = first_entry.key().clone();
        drop(first_entry);
        drop(default_user);

        // Bug 6: transition-only StateChanged broadcasts (see
        // spawn_user_drain_task). The task starts in Running state because
        // spawn_drain_task is only called after a command began executing.
        let mut last_broadcast_state = Some("Running".to_string());
        tokio::spawn(async move {
            while let Some(msg) = workflow_rx.recv().await {
                {
                    let mut session = session.lock().await;
                    session.handle_workflow_message(msg.clone());

                    let state_str = match session.workflow_state {
                        trustee_core::types::WorkflowState::Idle => "Idle",
                        trustee_core::types::WorkflowState::Running => "Running",
                        trustee_core::types::WorkflowState::Cancelling => "Cancelling",
                    };
                    if last_broadcast_state.as_deref() != Some(state_str) {
                        last_broadcast_state = Some(state_str.to_string());
                        let state_msg = serde_json::json!({
                            "type": "StateChanged",
                            "state": state_str
                        });
                        let _ = ws_tx.send(state_msg.to_string());
                    }
                }

                let json = serde_json::to_string(&SerializableMessage(&msg)).unwrap_or_default();
                let _ = ws_tx.send(json);
            }
            tracing::debug!("Drain task ended for session: {}", session_id);
        });
    }

    /// Resolve the user key from request headers.
    pub async fn resolve_user_key(&self, headers: &axum::http::HeaderMap) -> String {
        let Some(ref auth) = self.auth else {
            return "default".to_string();
        };

        // Try Bearer header first
        if let Some(token) = headers
            .get(axum::http::header::AUTHORIZATION)
            .and_then(|v| v.to_str().ok())
            .and_then(|v| v.strip_prefix("Bearer "))
            .map(|s| s.to_string())
        {
            if token.starts_with("dev:") {
                let parts: Vec<&str> = token.splitn(4, ':').collect();
                if parts.len() >= 4 {
                    return format!("dev:{}", parts[1]);
                }
            }
            if let Ok(claims) = auth.validate_token(&token).await {
                return claims.sub;
            }
        }

        // Try cookie
        let cookie_session_id = headers
            .get(axum::http::header::COOKIE)
            .and_then(|v| v.to_str().ok())
            .and_then(|cookies| {
                cookies.split(';').map(|c| c.trim()).find_map(|c| {
                    c.strip_prefix(&format!("{}=", auth.config.cookie_name))
                        .map(|s| s.to_string())
                })
            });

        if let Some(session_id) = cookie_session_id {
            if session_id.starts_with("dev:") {
                let parts: Vec<&str> = session_id.splitn(4, ':').collect();
                if parts.len() >= 4 {
                    return format!("dev:{}", parts[1]);
                }
            }

            if let Ok(access_token) = auth.session_manager.get_token(&session_id).await {
                if let Ok(claims) = auth.validate_token(&access_token).await {
                    return claims.sub;
                }
            }
        }

        "default".to_string()
    }
}

// ---------------------------------------------------------------------------
// SerializableMessage (unchanged)
// ---------------------------------------------------------------------------

/// Wrapper to serialize `TuiMessage` as JSON with a `type` discriminator.
struct SerializableMessage<'a>(&'a TuiMessage);

impl<'a> serde::Serialize for SerializableMessage<'a> {
    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
    where
        S: serde::Serializer,
    {
        use serde::ser::SerializeStruct;

        match self.0 {
            TuiMessage::OutputLine(line) => {
                let mut s = serializer.serialize_struct("msg", 2)?;
                s.serialize_field("type", "OutputLine")?;
                s.serialize_field("line", line)?;
                s.end()
            }
            TuiMessage::StreamDelta(delta) => {
                let mut s = serializer.serialize_struct("msg", 2)?;
                s.serialize_field("type", "StreamDelta")?;
                s.serialize_field("delta", delta)?;
                s.end()
            }
            TuiMessage::ReasoningDelta(delta) => {
                let mut s = serializer.serialize_struct("msg", 2)?;
                s.serialize_field("type", "ReasoningDelta")?;
                s.serialize_field("delta", delta)?;
                s.end()
            }
            TuiMessage::WorkflowCompleted => {
                let mut s = serializer.serialize_struct("msg", 2)?;
                s.serialize_field("type", "WorkflowCompleted")?;
                s.serialize_field("state", "Idle")?;
                s.end()
            }
            TuiMessage::WorkflowError(err) => {
                let mut s = serializer.serialize_struct("msg", 2)?;
                s.serialize_field("type", "WorkflowError")?;
                s.serialize_field("error", err)?;
                s.end()
            }
            TuiMessage::ResumeInfo(info) => match info {
                Some(ri) => {
                    let mut s = serializer.serialize_struct("msg", 5)?;
                    s.serialize_field("type", "ResumeInfo")?;
                    s.serialize_field("state", "Idle")?;
                    s.serialize_field("session_id", &ri.session_id)?;
                    s.serialize_field("checkpoint_id", &ri.checkpoint_id)?;
                    s.serialize_field("iteration", &ri.iteration)?;
                    s.end()
                }
                None => {
                    let mut s = serializer.serialize_struct("msg", 2)?;
                    s.serialize_field("type", "ResumeInfo")?;
                    s.serialize_field("state", "Idle")?;
                    s.end()
                }
            },
            TuiMessage::TodoUpdate(content) => {
                let mut s = serializer.serialize_struct("msg", 2)?;
                s.serialize_field("type", "TodoUpdate")?;
                s.serialize_field("content", content)?;
                s.end()
            }
            TuiMessage::WorkflowCancelled => {
                let mut s = serializer.serialize_struct("msg", 2)?;
                s.serialize_field("type", "WorkflowCancelled")?;
                s.serialize_field("state", "Idle")?;
                s.end()
            }
            TuiMessage::HandoffReady(briefing) => {
                let mut s = serializer.serialize_struct("msg", 3)?;
                s.serialize_field("type", "HandoffReady")?;
                s.serialize_field("state", "Idle")?;
                s.serialize_field("briefing", briefing)?;
                s.end()
            }
            TuiMessage::SessionRotated { old, new } => {
                let mut s = serializer.serialize_struct("msg", 3)?;
                s.serialize_field("type", "SessionRotated")?;
                s.serialize_field("old", old)?;
                s.serialize_field("new", new)?;
                s.end()
            }
            TuiMessage::HandoffFailed => {
                let mut s = serializer.serialize_struct("msg", 2)?;
                s.serialize_field("type", "HandoffFailed")?;
                s.serialize_field("state", "Idle")?;
                s.end()
            }
            TuiMessage::ToolPending { tool_name, hint } => {
                let mut s = serializer.serialize_struct("msg", 3)?;
                s.serialize_field("type", "ToolPending")?;
                s.serialize_field("tool_name", tool_name)?;
                s.serialize_field("hint", hint)?;
                s.end()
            }
            TuiMessage::ToolDone {
                tool_name,
                success,
                hint,
            } => {
                let mut s = serializer.serialize_struct("msg", 4)?;
                s.serialize_field("type", "ToolDone")?;
                s.serialize_field("tool_name", tool_name)?;
                s.serialize_field("success", success)?;
                s.serialize_field("hint", hint)?;
                s.end()
            }
            TuiMessage::ContextTokensUpdated(count) => {
                let mut s = serializer.serialize_struct("msg", 2)?;
                s.serialize_field("type", "ContextTokensUpdated")?;
                s.serialize_field("count", count)?;
                s.end()
            }
            TuiMessage::McpServerStatus {
                name,
                connected,
                tool_count,
                error,
            } => {
                let mut s = serializer.serialize_struct("msg", 5)?;
                s.serialize_field("type", "McpServerStatus")?;
                s.serialize_field("name", name)?;
                s.serialize_field("connected", connected)?;
                s.serialize_field("tool_count", tool_count)?;
                s.serialize_field("error", error)?;
                s.end()
            }
            TuiMessage::SessionTitleUpdated(title) => {
                let mut s = serializer.serialize_struct("msg", 2)?;
                s.serialize_field("type", "SessionTitleUpdated")?;
                s.serialize_field("title", title)?;
                s.end()
            }
        }
    }
}

// ---------------------------------------------------------------------------
// Per-user config helpers
// ---------------------------------------------------------------------------

/// Deep-merge a TOML overlay on top of a base value (in-place).
///
/// - Tables: recursively merge key-by-key (overlay wins on conflict).
/// - Arrays: overlay replaces base entirely (no merging).
/// - Scalars: overlay replaces base.
/// - If a key exists in overlay but not base, it's added.
fn deep_merge_toml(base: &mut toml::Value, overlay: &toml::Value) {
    match (base, overlay) {
        (toml::Value::Table(base_table), toml::Value::Table(overlay_table)) => {
            for (key, overlay_val) in overlay_table {
                match base_table.get_mut(key) {
                    Some(base_val) => {
                        // Both exist — recurse if both are tables, else replace
                        deep_merge_toml(base_val, overlay_val);
                    }
                    None => {
                        // Key only in overlay — insert
                        base_table.insert(key.clone(), overlay_val.clone());
                    }
                }
            }
        }
        // Non-table: overlay replaces base
        (base, overlay) => {
            *base = overlay.clone();
        }
    }
}

/// Replace `${VAR_NAME}` references in a string with values from a secrets map.
///
/// Falls back to process environment if the variable is not in the map.
/// Variables not found in either are left as-is.
/// Walk a TOML value tree; collect every string that still contains a
/// `${…}` placeholder (post-substitution leftovers — unresolved references).
fn collect_unresolved_placeholders(value: &toml::Value, out: &mut Vec<String>) {
    match value {
        toml::Value::String(s) => {
            if let Some(start) = s.find("${") {
                if let Some(end_rel) = s[start + 2..].find('}') {
                    let name = &s[start + 2..start + 2 + end_rel];
                    let rendered = format!("${{{name}}}");
                    if !out.contains(&rendered) {
                        out.push(rendered);
                    }
                }
            }
        }
        toml::Value::Table(t) => {
            for v in t.values() {
                collect_unresolved_placeholders(v, out);
            }
        }
        toml::Value::Array(a) => {
            for v in a {
                collect_unresolved_placeholders(v, out);
            }
        }
        _ => {}
    }
}

/// Where a `${VAR}` answer came from (92a36073 credential provenance).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum SecretSource {
    /// The secrets map (shared secrets + per-user `.env`, user wins) — the
    /// attributed, per-identity answer.
    SecretsMap,
    /// The PROCESS environment — for the trustee service that is the global
    /// `~/.trustee/.env` loaded at boot. When this answers for a per-user
    /// namespace, the per-user credential did NOT answer and a global
    /// credential did: actions execute under the global principal. Never
    /// silent — every occurrence is logged at ERROR with the variable named.
    ProcessEnv,
    /// Found nowhere; the literal `${…}` remains (the fc94aea9 guard).
    Unresolved,
}

/// One resolved (or unresolved) variable from [`substitute_env_vars`].
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct EnvVarResolution {
    pub name: String,
    pub source: SecretSource,
}

fn substitute_env_vars(
    s: &mut String,
    secrets: &std::collections::HashMap<String, String>,
    ctx: &str,
) -> Vec<EnvVarResolution> {
    // Simple state machine: scan for ${, read until }, replace.
    let mut result = String::with_capacity(s.len());
    let bytes = s.as_bytes();
    let mut i = 0;
    // One entry per DISTINCT variable, first sighting wins (a var used
    // twice resolves from the same source both times).
    let mut report: Vec<EnvVarResolution> = Vec::new();

    while i < bytes.len() {
        if i + 1 < bytes.len() && bytes[i] == b'$' && bytes[i + 1] == b'{' {
            // Find closing }
            if let Some(end) = s[i + 2..].find('}') {
                let var_name = &s[i + 2..i + 2 + end];
                // Look up in per-user secrets first, then process env
                let (replacement, source) = if let Some(value) = secrets.get(var_name) {
                    (value.clone(), SecretSource::SecretsMap)
                } else if let Ok(value) = std::env::var(var_name) {
                    (value, SecretSource::ProcessEnv)
                } else {
                    // Not found — leave as-is BUT NOT SILENTLY (incident
                    // fc94aea9, T3): an unresolved ${VAR} once rode to
                    // production authentication as the literal credential.
                    // Every leftover is named, every time.
                    (s[i..i + 2 + end + 1].to_string(), SecretSource::Unresolved)
                };
                if !report.iter().any(|r| r.name == var_name) {
                    report.push(EnvVarResolution {
                        name: var_name.to_string(),
                        source,
                    });
                }
                result.push_str(&replacement);
                i = i + 2 + end + 1;
            } else {
                // No closing } — copy as-is
                result.push('$');
                i += 1;
            }
        } else {
            result.push(bytes[i] as char);
            i += 1;
        }
    }

    // 92a36073 — never-silent credential provenance. A process-env answer
    // for a per-user namespace is the global-token shadowing hazard: the
    // global principal is about to execute what the per-user identity
    // should have. ERROR level, EVERY occurrence (no dedup across calls).
    for r in &report {
        match r.source {
            SecretSource::ProcessEnv => tracing::error!(
                "CREDENTIAL PROVENANCE [{}]: ${{{}}} resolved from PROCESS ENV \
                 (global ~/.trustee/.env at boot) — no per-user/profile secret \
                 answered; calls using it execute under the GLOBAL principal. \
                 Supply the exact-case key in the per-user .env / profile \
                 secrets to restore attribution.",
                ctx,
                r.name
            ),
            SecretSource::Unresolved => {
                // Compose the leftover list for the existing loud line.
            }
            SecretSource::SecretsMap => tracing::debug!(
                "CREDENTIAL PROVENANCE [{}]: ${{{}}} resolved from profile/user secrets",
                ctx,
                r.name
            ),
        }
    }
    let unresolved: Vec<&str> = report
        .iter()
        .filter(|r| r.source == SecretSource::Unresolved)
        .map(|r| r.name.as_str())
        .collect();
    if !unresolved.is_empty() {
        tracing::warn!(
            "ENV SUBSTITUTION LEFT {} UNRESOLVED [{}]: {} — the literal ${{…}} remains in the \
             effective config; supply the key in the profile secrets (exact-case KEY=VALUE) \
             or the environment",
            unresolved.len(),
            ctx,
            unresolved
                .iter()
                .map(|v| format!("${{{v}}}"))
                .collect::<Vec<_>>()
                .join(", ")
        );
    }

    *s = result;
    report
}

/// 92a36073 boot-surface check: is a MATERIALIZED namespace missing its
/// per-user credentials? A directory without a `.env` is exactly the
/// shadowing scenario (rotated/empty home + global token downstream) —
/// surface it at boot, not in forensics. Pure so it stays unit-tested.
pub(crate) fn materialized_user_missing_env(user_home: &std::path::Path) -> bool {
    user_home.is_dir() && !user_home.join(".env").exists()
}

// ---------------------------------------------------------------------------
// Tests (task 16B: overlay allowlist + consolidated user hash)
// ---------------------------------------------------------------------------

/// Fingerprint the effective `[mcp]` section of a resolved config (16C).
///
/// Content hash (SHA-256, first 8 bytes as u64) — never mtime: overlays are
/// rewritten in place. Configs without `[mcp]` fingerprint to a stable
/// constant, so the disabled state caches too.
fn fingerprint_mcp_section(resolved: Option<&str>) -> u64 {
    use sha2::{Digest, Sha256};
    let section = resolved
        .and_then(|s| s.parse::<toml::Value>().ok())
        .and_then(|v| v.get("mcp").cloned());
    let bytes = match section {
        Some(v) => v.to_string().into_bytes(),
        None => b"<no-mcp>".to_vec(),
    };
    let digest = Sha256::digest(&bytes);
    u64::from_be_bytes(digest[..8].try_into().expect("sha256 digest >= 8 bytes"))
}

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

    /// Default-knob ServerState for overlay tests.
    fn test_state() -> ServerState {
        let (session, _rx) = Session::new();
        let (ws_tx, _ws_rx) = tokio::sync::broadcast::channel::<String>(16);
        ServerState::new(session, ws_tx, None)
    }

    #[tokio::test]
    async fn dispatch_scope_pins_user_key_resolution() {
        // THE 16F pin (ec3e0622 companion): inside the dispatch scope, EVERY
        // user-key-derived resolution lands in the binding's home — the
        // caller's own key (and its claims) become irrelevant.
        let state = test_state();
        let caller = "some-human-sub";
        let plain_home = state.get_user_home_dir(caller).unwrap();
        let pinned_key = "prof-1234";
        let pinned_home = in_dispatch_scope(pinned_key, async {
            state.get_user_home_dir(caller).unwrap()
        })
        .await;
        let users_root = dirs::home_dir().unwrap().join(".trustee").join("users");
        assert_eq!(plain_home, users_root.join(trustee_core::user_hash(caller)));
        assert_eq!(
            pinned_home,
            users_root.join(trustee_core::user_hash(pinned_key)),
            "dispatch scope must pin resolution to the binding key"
        );
        assert_ne!(plain_home, pinned_home);
    }

    /// Unique temp dir with a `config/` subdir (std-only; no tempfile dep).
    fn temp_user_home(tag: &str) -> std::path::PathBuf {
        static COUNTER: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
        let n = COUNTER.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
        let dir = std::env::temp_dir().join(format!(
            "trustee-state-test-{}-{}-{}",
            tag,
            std::process::id(),
            n
        ));
        std::fs::create_dir_all(dir.join("config")).expect("create temp user home");
        dir
    }

    fn parse(toml_str: &str) -> toml::Value {
        toml_str.parse::<toml::Value>().expect("valid test TOML")
    }

    /// (a) Overlay with [mcp] + [server] + [auth]: [mcp] applied (user wins),
    /// shared [server] preserved untouched, overlay [auth] dropped entirely.
    #[test]
    fn overlay_allowlist_drops_non_allowlisted_sections() {
        let state = test_state();
        let home = temp_user_home("allowlist");
        std::fs::write(
            home.join("config").join("trustee.toml"),
            "[server]\nport = 1\n\n[auth]\nmode = \"kanidm\"\n\n[mcp]\nmode = \"user\"\n",
        )
        .expect("write overlay");

        let shared = "[server]\nport = 8080\n\n[mcp]\nmode = \"shared\"\n";
        let merged = state
            .merge_user_config(&home, Some(shared))
            .expect("overlay has allowlisted content");

        let merged_val = parse(&merged);
        let expected = parse("[server]\nport = 8080\n\n[mcp]\nmode = \"user\"\n");
        assert_eq!(
            merged_val, expected,
            "merged config must be shared + [mcp] overlay only"
        );
        assert!(
            merged_val.get("auth").is_none(),
            "overlay [auth] must be dropped"
        );
    }

    /// (b) No overlay file → merge is a no-op (None), shared config untouched.
    #[test]
    fn no_overlay_file_returns_none() {
        let state = test_state();
        let home = temp_user_home("empty");
        assert!(state
            .merge_user_config(&home, Some("[server]\nport = 8080\n"))
            .is_none());
    }

    /// Overlay whose sections are ALL non-allowlisted → no-op (None).
    #[test]
    fn overlay_with_no_allowlisted_sections_is_noop() {
        let state = test_state();
        let home = temp_user_home("all-dropped");
        std::fs::write(
            home.join("config").join("trustee.toml"),
            "[server]\nport = 1\n\n[storage]\npath = \"/tmp/x\"\n",
        )
        .expect("write overlay");
        assert!(state
            .merge_user_config(&home, Some("[server]\nport = 8080\n"))
            .is_none());
    }

    /// (c) [llm] dropped when allow_llm_overlay=false (default), applied when
    /// the instance opted in via with_allow_llm_overlay(true).
    #[test]
    fn llm_overlay_dropped_by_default_and_kept_when_enabled() {
        let shared = "[llm]\nprovider = \"openai\"\n\n[mcp]\nmode = \"shared\"\n";
        let overlay = "[llm]\nprovider = \"anthropic\"\n";

        // Default: knob false → [llm]-only overlay is a no-op.
        let state = test_state();
        assert!(!state.allow_llm_overlay);
        let home = temp_user_home("llm-off");
        std::fs::write(home.join("config").join("trustee.toml"), overlay).expect("write overlay");
        assert!(state.merge_user_config(&home, Some(shared)).is_none());

        // Opted in: [llm] kept, user value wins; shared [mcp] untouched.
        let state = test_state().with_allow_llm_overlay(true);
        assert!(state.allow_llm_overlay);
        let home = temp_user_home("llm-on");
        std::fs::write(home.join("config").join("trustee.toml"), overlay).expect("write overlay");
        let merged = state
            .merge_user_config(&home, Some(shared))
            .expect("[llm] overlay applies when opted in");
        let merged_val = parse(&merged);
        assert_eq!(merged_val["llm"]["provider"].as_str(), Some("anthropic"));
        assert_eq!(merged_val["mcp"]["mode"].as_str(), Some("shared"));
    }

    /// (d) Web-path home dir resolves through the consolidated
    /// trustee_core::user_hash (determinism/known-vectors covered in
    /// trustee-core's own tests).
    #[test]
    fn user_home_dir_uses_consolidated_hash() {
        let state = test_state();
        if let Some(home) = state.get_user_home_dir("farzan@example.com") {
            assert_eq!(
                home.file_name().and_then(|n| n.to_str()),
                Some(trustee_core::user_hash("farzan@example.com")).as_deref()
            );
            let users_root = dirs::home_dir().unwrap().join(".trustee").join("users");
            assert_eq!(home.parent(), Some(&users_root).map(|p| p.as_path()));
        }
        // dirs::home_dir() unavailable in sandbox → covered by core tests.
    }

    /// xagent personas: [lifecycle] survives the overlay allowlist (persona
    /// is user-owned) but the WASM extension switch does not.
    #[test]
    fn lifecycle_overlay_survives_allowlist_enabled_stripped() {
        let state = test_state();
        let home = temp_user_home("lifecycle-persona");
        std::fs::write(
            home.join("config").join("trustee.toml"),
            "[lifecycle]\nsystem_template = \"You are Saman.\"\nenabled = true\n",
        )
        .expect("write overlay");

        let merged = state
            .merge_user_config(
                &home,
                Some("[lifecycle]\nsystem_template = \"shared default\"\n"),
            )
            .expect("lifecycle overlay is allowlisted");

        let v = merged.parse::<toml::Value>().expect("valid TOML");
        assert_eq!(
            v["lifecycle"]["system_template"].as_str(),
            Some("You are Saman."),
            "agent persona wins over shared default"
        );
        assert_eq!(
            v["lifecycle"].get("enabled"),
            None,
            "WASM extension switch is not user-settable"
        );
    }

    /// 16D integration guard: dev-agent principals (user_key `agent-<name>`)
    /// get their OWN session bucket and home-dir namespace — one agent can
    /// never see another's sessions, and both go through the same
    /// user_hash isolation path as humans.
    #[tokio::test]
    async fn agent_principals_get_isolated_session_buckets() {
        let state = test_state();
        let key_a = "agent-farzan";
        let key_b = "agent-paydar";

        let sid_a = state
            .create_session(key_a, None, None)
            .await
            .expect("create session");
        let _sid_b = state
            .create_session(key_b, None, None)
            .await
            .expect("create session");

        // Each agent resolves its own session; B cannot see A's by id.
        assert!(
            state.get_session_by_any_id(key_a, &sid_a).await.is_some(),
            "owner bucket resolves its own session"
        );
        assert!(
            state.get_session_by_any_id(key_b, &sid_a).await.is_none(),
            "cross-agent session access must be 404/None"
        );

        // Same isolation path as humans: home dir = users_root/user_hash(key).
        if let Some(home) = state.get_user_home_dir(key_a) {
            assert_eq!(
                home.file_name().and_then(|n| n.to_str()),
                Some(trustee_core::user_hash(key_a)).as_deref()
            );
        }
    }
}

// ---------------------------------------------------------------------------
// 16C — per-user McpToolLoader cache
// ---------------------------------------------------------------------------

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

    fn state_with_shared(shared: &str) -> ServerState {
        let (session, _rx) = Session::new();
        let (ws_tx, _ws_rx) = tokio::sync::broadcast::channel::<String>(16);
        let mut state = ServerState::new(session, ws_tx, None);
        state.config_toml = Some(shared.to_string());
        state
    }

    /// Deterministic throwaway user: real user-home path under
    /// ~/.trustee/users/{hash} (that IS the resolution path under test),
    /// with config/ subdir; caller cleans up.
    struct TempUser {
        key: String,
        home: std::path::PathBuf,
    }

    impl TempUser {
        fn new(tag: &str) -> Self {
            let key = format!("16c-{tag}-{}@test.invalid", std::process::id());
            let home = dirs::home_dir()
                .expect("HOME available in test env")
                .join(".trustee")
                .join("users")
                .join(trustee_core::user_hash(&key));
            std::fs::create_dir_all(home.join("config")).expect("create user home");
            Self { key, home }
        }

        fn write_overlay(&self, toml_str: &str) {
            std::fs::write(self.home.join("config").join("trustee.toml"), toml_str)
                .expect("write overlay");
        }
    }

    impl Drop for TempUser {
        fn drop(&mut self) {
            let _ = std::fs::remove_dir_all(&self.home);
        }
    }

    #[tokio::test]
    async fn no_mcp_config_caches_disabled_marker() {
        let state = state_with_shared("[server]\nport = 8080\n");
        let user = TempUser::new("nomcp");
        let ts = Arc::new(pep::MemoryTokenStore::new());

        let first = state.get_or_build_mcp_loader(&user.key, &ts).await.unwrap();
        assert!(first.is_none(), "no [mcp] anywhere → disabled marker");
        assert_eq!(state.mcp_loaders.len(), 1, "exactly one cache entry");

        // Second call is a fingerprint hit — same disabled marker, still one entry.
        let second = state.get_or_build_mcp_loader(&user.key, &ts).await.unwrap();
        assert!(second.is_none());
        assert_eq!(state.mcp_loaders.len(), 1);
    }

    #[tokio::test]
    async fn concurrent_cold_builds_single_flight() {
        let state = Arc::new(state_with_shared("[server]\nport = 8080\n"));
        let user = TempUser::new("singleflight");
        let ts = Arc::new(pep::MemoryTokenStore::new());

        let mut handles = Vec::new();
        for _ in 0..5 {
            let state = state.clone();
            let key = user.key.clone();
            let ts = ts.clone();
            handles.push(tokio::spawn(async move {
                state.get_or_build_mcp_loader(&key, &ts).await
            }));
        }
        for h in handles {
            h.await.unwrap().expect("all five succeed");
        }
        assert_eq!(state.mcp_loaders.len(), 1, "single-flight → one entry");
    }

    #[tokio::test]
    async fn fingerprint_change_triggers_rebuild() {
        let state = state_with_shared("[server]\nport = 8080\n");
        let user = TempUser::new("fpchange");
        let ts = Arc::new(pep::MemoryTokenStore::new());

        // v1: an enabled MCP server pointing at a blackhole port — abk keeps
        // the loader with a DOWN status (connect refused is fast on loopback).
        user.write_overlay("[mcp]\nenabled = true\n\n[[mcp.servers]]\nname = \"v1\"\nurl = \"http://127.0.0.1:9/sse\"\n");
        let v1 = state.get_or_build_mcp_loader(&user.key, &ts).await.unwrap();
        assert!(v1.is_some(), "enabled [mcp] → real loader");
        let fp1 = state
            .mcp_loaders
            .get(&trustee_core::user_hash(&user.key))
            .unwrap()
            .fingerprint;

        // v2: different server URL → different content fingerprint → rebuild.
        std::thread::sleep(std::time::Duration::from_millis(5));
        user.write_overlay("[mcp]\nenabled = true\n\n[[mcp.servers]]\nname = \"v2\"\nurl = \"http://127.0.0.1:9/other\"\n");
        let v2 = state.get_or_build_mcp_loader(&user.key, &ts).await.unwrap();
        assert!(v2.is_some());
        let entry = state
            .mcp_loaders
            .get(&trustee_core::user_hash(&user.key))
            .unwrap();
        assert_ne!(
            entry.fingerprint, fp1,
            "fingerprint must change with content"
        );
        assert!(entry.degraded.is_none());

        // The old Arc stays valid for in-flight sessions (no panic, no revoke).
        let _still_usable = v1.as_ref().unwrap().tool_count;
    }

    #[tokio::test]
    async fn degraded_entry_fails_loud_within_backoff_and_isolates_users() {
        let state = state_with_shared("[server]\nport = 8080\n");
        let bad = TempUser::new("degraded-bad");
        let good = TempUser::new("degraded-good");
        let ts = Arc::new(pep::MemoryTokenStore::new());

        // [mcp] present but not a table → McpConfig deserialize fails → Err.
        bad.write_overlay("[mcp]\nenabled = \"not-a-bool\"\n");
        let err = match state.get_or_build_mcp_loader(&bad.key, &ts).await {
            Ok(_) => panic!("invalid [mcp] must fail loud"),
            Err(e) => e,
        };
        assert!(
            err.contains("invalid [mcp]"),
            "surfaces the parse error: {err}"
        );

        let entry = state
            .mcp_loaders
            .get(&trustee_core::user_hash(&bad.key))
            .unwrap();
        assert!(entry.degraded.is_some(), "poison entry recorded");

        // Within backoff: fast-fail again (cached error).
        let err2 = match state.get_or_build_mcp_loader(&bad.key, &ts).await {
            Ok(_) => panic!("still within backoff"),
            Err(e) => e,
        };
        assert_eq!(err, err2, "same cached error");

        // Other users are completely unaffected.
        let good_loader = state
            .get_or_build_mcp_loader(&good.key, &ts)
            .await
            .expect("other user unaffected");
        assert!(
            good_loader.is_none(),
            "good user has no [mcp] → disabled marker"
        );
    }

    #[tokio::test]
    async fn aged_out_entry_rebuilds_without_restart() {
        // THE 42131deb guard: a served-surface change behind a byte-identical
        // config must be reflected in the NEXT fresh session's manifest —
        // simulated by rewinding built_at past MCP_TOOLSET_MAX_AGE; the next
        // dispatch must NOT reuse the stale Arc (it rebuilds + re-lists).
        let state = state_with_shared("[server]\nport = 8080\n");
        let user = TempUser::new("agedout");
        let ts = Arc::new(pep::MemoryTokenStore::new());

        user.write_overlay("[mcp]\nenabled = true\n\n[[mcp.servers]]\nname = \"v1\"\nurl = \"http://127.0.0.1:9/sse\"\n");
        let v1 = state
            .get_or_build_mcp_loader(&user.key, &ts)
            .await
            .expect("initial build");
        let v1 = v1.expect("enabled [mcp] → loader");

        // Simulate time: the entry is now older than the max age, config
        // byte-identical (fingerprint UNCHANGED — that's the trap).
        {
            let mut entry = state
                .mcp_loaders
                .get_mut(&trustee_core::user_hash(&user.key))
                .unwrap();
            entry.built_at = chrono::Utc::now() - chrono::Duration::seconds(601);
        }

        let v2 = state
            .get_or_build_mcp_loader(&user.key, &ts)
            .await
            .expect("aged-out dispatch rebuilds");
        let v2 = v2.expect("rebuilt loader present");
        assert!(
            !Arc::ptr_eq(&v1, &v2),
            "aged-out entry must NOT be reused — fresh connect re-lists the served surface"
        );
        let entry = state
            .mcp_loaders
            .get(&trustee_core::user_hash(&user.key))
            .unwrap();
        assert!(
            chrono::Utc::now() - entry.built_at < chrono::Duration::seconds(5),
            "built_at refreshed by the rebuild"
        );

        // And a young, unchanged entry still hits (no rebuild churn).
        let v3 = state
            .get_or_build_mcp_loader(&user.key, &ts)
            .await
            .expect("fresh dispatch");
        assert!(Arc::ptr_eq(&v2, &v3.unwrap()), "young entry hits the cache");
    }
}

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

    fn fresh_state() -> ServerState {
        let (session, _rx) = Session::new();
        let (ws_tx, _ws_rx) = tokio::sync::broadcast::channel::<String>(16);
        let mut state = ServerState::new(session, ws_tx, None);
        state.config_toml = Some("[agent]\nname = \"trustee\"\n".to_string());
        state
    }

    /// Deterministic throwaway user (same pattern as mcp_loader_cache_tests).
    struct TempUser {
        key: String,
    }

    impl TempUser {
        fn new(tag: &str) -> Self {
            let key = format!("restore-{tag}-{}@test.invalid", std::process::id());
            let home = dirs::home_dir()
                .expect("HOME available in test env")
                .join(".trustee")
                .join("users")
                .join(trustee_core::user_hash(&key));
            std::fs::create_dir_all(home).expect("create user home");
            Self { key }
        }
    }

    #[tokio::test]
    async fn restore_unknown_chain_returns_none() {
        let state = fresh_state();
        let user = TempUser::new("unknown");
        let out = state
            .get_or_restore_session_by_any_id(&user.key, "session_2099_01_01_00_00_deadbeef")
            .await;
        assert!(out.is_none(), "unknown chain must 404, not restore");
    }

    #[tokio::test]
    async fn destroy_tombstones_chain_id() {
        let state = fresh_state();
        let user = TempUser::new("tombstone");
        let sid = state
            .create_session(&user.key, None, None)
            .await
            .expect("create session");

        // Attach the live session to an on-disk-style chain id.
        let (session, _ws) = state.get_session(&user.key, &sid).await.unwrap();
        {
            let mut s = session.lock().await;
            s.session_id = Some("session_2026_09_04_10_00_aaaabbbb".to_string());
        }

        state
            .destroy_session(&user.key, &sid)
            .await
            .expect("destroy");

        let path = state
            .tombstone_path(&user.key, "session_2026_09_04_10_00_aaaabbbb")
            .expect("tombstone path");
        assert!(path.exists(), "destroy must write a chain tombstone");

        // And a fresh live session with NO chain id must not tombstone anything.
        let sid2 = state
            .create_session(&user.key, None, None)
            .await
            .expect("create second session");
        state
            .destroy_session(&user.key, &sid2)
            .await
            .expect("destroy 2");
        // (No assertion beyond "no panic": session_id was None → no new markers.)
    }

    #[tokio::test]
    async fn restore_refuses_registry_key_shapes() {
        // Registry keys ("default", "web…") are not checkpoint chains; even
        // if create_resume_info could not find them anyway, the tombstone
        // guard short-circuits non-`session_` ids without touching disk.
        let state = fresh_state();
        let user = TempUser::new("shapes");
        assert!(state.tombstone_path(&user.key, "default").is_none());
        let out = state
            .get_or_restore_session_by_any_id(&user.key, "default")
            .await;
        assert!(out.is_none());
    }
}

#[test]
fn collect_unresolved_placeholders_walks_the_tree() {
    let section: toml::Value = toml::from_str(
        r#"
[credentials.farzan_service_account]
service_token = "${FARZAN_SERVICE_ACCOUNT}"
issuer_url = "https://ok.example"

[[servers]]
name = "fame"
url = "https://fame.example"
credentials = "farzan_service_account"

[[servers]]
name = "other"
url = "https://other.example"
token = "${OTHER_TOKEN}"
"#,
    )
    .unwrap();
    let mut out = Vec::new();
    collect_unresolved_placeholders(&section, &mut out);
    out.sort();
    assert_eq!(
        out,
        vec![
            "${FARZAN_SERVICE_ACCOUNT}".to_string(),
            "${OTHER_TOKEN}".to_string()
        ],
        "every leftover placeholder in the [mcp] tree must be collected, deduped"
    );
    // clean section → nothing
    let clean: toml::Value = toml::from_str("[x]\ny = \"z\"\n").unwrap();
    let mut out2 = Vec::new();
    collect_unresolved_placeholders(&clean, &mut out2);
    assert!(out2.is_empty());
}

#[test]
fn substitute_env_vars_resolves_and_keeps_leftovers_deterministically() {
    let mut secrets = std::collections::HashMap::new();
    secrets.insert("A_TOKEN".to_string(), "resolved-value".to_string());
    let mut s = "x = \"${A_TOKEN}\" y = \"${MISSING_TOKEN}\"".to_string();
    let report = substitute_env_vars(&mut s, &secrets, "test");
    assert!(s.contains("resolved-value"), "declared key resolves: {s}");
    assert!(
        s.contains("${MISSING_TOKEN}"),
        "undeclared key keeps the literal (the loader guard turns this into a loud failure): {s}"
    );
    // 92a36073: the report names each distinct variable and its source.
    assert_eq!(report.len(), 2, "one entry per distinct var: {report:?}");
    assert!(report.contains(&EnvVarResolution {
        name: "A_TOKEN".to_string(),
        source: SecretSource::SecretsMap
    }));
    assert!(report.contains(&EnvVarResolution {
        name: "MISSING_TOKEN".to_string(),
        source: SecretSource::Unresolved
    }));
}

#[test]
fn substitute_env_vars_reports_process_env_shadow() {
    // 92a36073: a var ABSENT from the per-user/profile secrets but set
    // in the process environment is the global-token shadowing shape —
    // it must be REPORTED as ProcessEnv, and the secrets map must still
    // win when it DOES carry the key.
    let var = format!("RAVAND_PROVENANCE_TEST_{}", std::process::id());
    std::env::set_var(&var, "global-env-answer");

    // Absent from secrets → process env answers → ProcessEnv reported.
    let mut secrets = std::collections::HashMap::new();
    let mut s = format!("x = \"${{{var}}}\"");
    let report = substitute_env_vars(&mut s, &secrets, "test-shadow");
    assert!(s.contains("global-env-answer"), "process env answered: {s}");
    assert_eq!(
        report,
        vec![EnvVarResolution {
            name: var.clone(),
            source: SecretSource::ProcessEnv
        }]
    );

    // Same var present in secrets → secrets WIN (per-user precedence
    // unchanged), reported as SecretsMap.
    secrets.insert(var.clone(), "per-user-answer".to_string());
    let mut s2 = format!("x = \"${{{var}}}\"");
    let report2 = substitute_env_vars(&mut s2, &secrets, "test-wins");
    assert!(s2.contains("per-user-answer"), "user secrets win: {s2}");
    assert_eq!(
        report2,
        vec![EnvVarResolution {
            name: var.clone(),
            source: SecretSource::SecretsMap
        }]
    );

    std::env::remove_var(&var);
}

#[test]
fn materialized_user_missing_env_only_fires_on_dir_without_env() {
    // 92a36073 boot-surface: warn shape is dir-present + .env-absent.
    let base = std::env::temp_dir().join(format!("trustee-envcheck-{}", std::process::id()));
    let _ = std::fs::remove_dir_all(&base);

    // Not materialized at all → NOT the warning shape.
    assert!(!materialized_user_missing_env(&base.join("ghost")));

    // Materialized WITHOUT .env → the warning shape.
    std::fs::create_dir_all(&base).unwrap();
    assert!(materialized_user_missing_env(&base));

    // Materialized WITH .env → not the warning shape.
    std::fs::write(base.join(".env"), "K=V\n").unwrap();
    assert!(!materialized_user_missing_env(&base));

    let _ = std::fs::remove_dir_all(&base);
}

#[test]
fn loader_freshness_decides_all_four_states() {
    let now = chrono::Utc::now();
    let fp = 0xABCD_u64;
    let young = now - chrono::Duration::seconds(60);
    let old = now - chrono::Duration::seconds(601);

    // Fresh: same fingerprint, young entry.
    assert_eq!(
        loader_freshness(fp, young, None, None, fp, now),
        LoaderFreshness::Fresh
    );
    // ConfigChanged: fingerprint differs wins over age.
    assert_eq!(
        loader_freshness(fp, old, None, None, 0x1234, now),
        LoaderFreshness::ConfigChanged
    );
    // AgedOut: same fingerprint past the max age — the 42131deb guard.
    assert_eq!(
        loader_freshness(fp, old, None, None, fp, now),
        LoaderFreshness::AgedOut
    );
    // Backoff: degraded inside the retry window, even old + unchanged.
    let failed = now - chrono::Duration::seconds(10);
    match loader_freshness(fp, old, Some("boom"), Some(failed), fp, now) {
        LoaderFreshness::Backoff(remaining) => {
            assert!(remaining.as_secs() >= 15, "~20s left: {remaining:?}");
        }
        other => panic!("expected Backoff, got {other:?}"),
    }
    // Degraded but PAST backoff → age rules apply (rebuild proceeds).
    let failed_long = now - chrono::Duration::seconds(120);
    assert_eq!(
        loader_freshness(fp, old, Some("boom"), Some(failed_long), fp, now),
        LoaderFreshness::AgedOut
    );
}