car-server-core 0.52.1

//! Server-side session state — shared across all connections.

use car_engine::{Runtime, ToolExecutor};
use car_eventlog::EventLog;
use car_proto::{ToolCancelRequest, ToolExecuteRequest, ToolExecuteResponse};
use futures::Sink;
use serde::{Deserialize, Serialize};
use serde_json::Value;
use sha2::{Digest, Sha256};
use std::collections::HashMap;
use std::path::{Path, PathBuf};
use std::pin::Pin;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::Arc;
use tokio::sync::{mpsc, oneshot, Mutex};
use tokio_tungstenite::tungstenite::{Error as WsError, Message};

/// Type-erased WebSocket sink. The dispatch loop accepts either a
/// `WebSocketStream<TcpStream>` (the legacy car-server TCP listener)
/// or a `WebSocketStream<UnixStream>` (the daemon-as-default UDS
/// listener) — both implement `Sink<Message, Error = WsError>` after
/// the tungstenite handshake. Erasing the type here avoids cascading
/// a generic parameter through every WsChannel / Session / ServerState
/// touchpoint in the dispatcher.
pub type WsSink = Pin<Box<dyn Sink<Message, Error = WsError> + Send + Unpin + 'static>>;

/// Grace window applied on disconnect before a still-open run is marked
/// `Incomplete` (agent run tracing, U1 — R5). Short on purpose: it only
/// has to cover the gap between a healthy `runs.complete` being
/// dispatched on a spawned task and its terminal record landing, not any
/// real work. Long enough to absorb that scheduling jitter, short enough
/// that a genuinely abandoned run is reported `Incomplete` promptly.
pub const RUN_COMPLETE_GRACE: std::time::Duration = std::time::Duration::from_millis(250);

/// Bounded in-process lease for a disconnected run whose producer negotiated
/// `runs.resume.v1`. Ten seconds leaves practical headroom for a loaded Mac's
/// Python controller to establish a fresh local socket, authenticate, and
/// negotiate capabilities without turning this into restart recovery. Runs
/// without the capability keep the original 250 ms behavior; an unclaimed
/// resumable run still becomes `Incomplete`.
pub const RUN_RESUME_LEASE: std::time::Duration = std::time::Duration::from_secs(10);

/// Maximum time an async run-lifecycle mutation waits for the journal writer's
/// durability acknowledgement. Five seconds matches the server's existing
/// acknowledgement/grace budget and remains well inside the default mutating
/// handler deadline. If an operator override cancels earlier, the async append
/// still marks the exact row pending before its first yield.
const CRITICAL_JOURNAL_ACKNOWLEDGEMENT_TIMEOUT: std::time::Duration =
    std::time::Duration::from_secs(5);

/// Startup uses the same durability budget as live lifecycle mutations. A
/// stalled journal must fail the constructor before the daemon listens rather
/// than occupying the startup thread forever.
const STARTUP_RECONCILIATION_ACKNOWLEDGEMENT_TIMEOUT: std::time::Duration =
    CRITICAL_JOURNAL_ACKNOWLEDGEMENT_TIMEOUT;

/// Per-run observation bound for disconnect cleanup: the ten-second resume
/// lease, two bounded five-second journal acknowledgement attempts (the second
/// is an exact retry after durability-unknown), and scheduler headroom. Cleanup
/// never clears the run or binding unless an acknowledgement arrives.
pub const RUN_DISCONNECT_CLEANUP_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(22);

const RUN_SUBSCRIBE_SUMMARY_STATE_RETRY_LIMIT: usize = 3;

/// Server-side credentials for continuing an A2A-owned A2UI surface.
///
/// This intentionally lives outside `car_a2ui::A2uiSurfaceOwner` so
/// renderers can inspect surface ownership without receiving secrets.
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase", tag = "type")]
pub enum A2aRouteAuth {
    None,
    Bearer { token: String },
    Header { name: String, value: String },
}

#[cfg(test)]
mod run_reservation_liveness_tests {
    use super::*;
    use crate::run_store::RunStoreLookupGate;
    use std::time::Duration;

    fn pristine_run(run_id: &str, client_id: &str) -> RunMeta {
        RunMeta {
            run_id: run_id.to_string(),
            agent_id: "agent-release".to_string(),
            client_id: client_id.to_string(),
            active_client_id: client_id.to_string(),
            resume_predecessor_client_id: None,
            resume_lease: None,
            intent: "release without blocking the registry".to_string(),
            outcome_description: None,
            started_at: chrono::Utc::now(),
            termination: None,
            ended_at: None,
            turns: Vec::new(),
            start_committed: false,
            pending_terminal: None,
            cancellation_pending: None,
            cancellation_receipt: None,
            trace_corruption: None,
            durability_generation: 0,
        }
    }

    #[tokio::test]
    async fn blocked_release_lookup_does_not_hold_global_runs_lock() {
        let tmp = tempfile::TempDir::new().unwrap();
        let gate = RunStoreLookupGate::default();
        let state = Arc::new(ServerState::with_config(
            ServerStateConfig::new(tmp.path().join("journals"))
                .with_run_store_lookup_gate(gate.clone()),
        ));
        let session = state
            .create_session("release-client", Arc::new(WsChannel::test_stub()))
            .await
            .unwrap();
        let expected = pristine_run("release-candidate", &session.client_id);
        assert!(matches!(
            state.reserve_run(expected.clone()).await,
            Ok(RunReservation::New)
        ));

        gate.block_next();
        let release_state = state.clone();
        let release_session = session.clone();
        let release_expected = expected.clone();
        let release = tokio::spawn(async move {
            release_state
                .release_unpersisted_run_reservation(&release_session, &release_expected)
                .await
        });
        let wait_gate = gate.clone();
        assert!(
            tokio::task::spawn_blocking(move || {
                wait_gate.wait_until_entered(Duration::from_secs(3))
            })
            .await
            .unwrap(),
            "release must reach the blocked durable lookup"
        );

        let registry = tokio::time::timeout(Duration::from_secs(1), state.runs.lock())
            .await
            .expect("blocked release lookup must not hold the global runs lock");
        drop(registry);
        gate.release();
        assert_eq!(release.await.unwrap(), Ok(()));
        assert!(state.run_meta(&expected.run_id).await.is_none());
    }
}

/// Shared write half of the WebSocket, plus pending callback channels.
/// `write` is type-erased via [`WsSink`] so the dispatcher can run
/// against any transport-specific WebSocketStream (TCP or UDS today;
/// axum-bridged in future) without templatizing every consumer.
pub struct WsChannel {
    pub write: Mutex<WsSink>,
    /// Pending tool execution callbacks: request_id → oneshot sender
    pub pending: Mutex<HashMap<String, oneshot::Sender<ToolExecuteResponse>>>,
    /// Body-free correlation for active callbacks. A session owns at most one
    /// active run, so every entry belongs to that run while its lifecycle
    /// guard is held by proposal execution.
    pub active_actions: Mutex<HashMap<String, String>>,
    pub next_id: AtomicU64,
}

/// A WS connection is a host event sink (#418): `HostState` broadcasts
/// `host.event` frames to subscribers through this trait, so it never has to
/// know about the concrete `WsChannel` / tungstenite types.
#[async_trait::async_trait]
impl crate::host::EventSubscriber for WsChannel {
    async fn send_text(&self, json: String) {
        use futures::SinkExt;
        // Bound both lock acquisition and the socket write. Host events are
        // emitted from reducer transitions; a suspended client must not hold
        // that reducer's announcement lock (and browser controls behind it)
        // indefinitely through TCP backpressure.
        let _ = tokio::time::timeout(std::time::Duration::from_secs(10), async {
            self.write
                .lock()
                .await
                .send(Message::Text(json.into()))
                .await
        })
        .await;
    }
}

impl WsChannel {
    pub fn next_request_id(&self) -> String {
        let id = self.next_id.fetch_add(1, Ordering::SeqCst);
        format!("cb-{}", id)
    }

    /// Test-only stub that returns a WsChannel whose write sink drains
    /// to nowhere. Used by `host.rs` tests that need a real
    /// `Arc<WsChannel>` in the subscribers map (to exercise membership
    /// checks like the cross-session resolve fan-out) without
    /// constructing a tungstenite handshake. Never writes are
    /// performed against this stub; if anything tries, the drain sink
    /// quietly absorbs.
    #[cfg(test)]
    pub fn test_stub() -> Self {
        use futures::sink::SinkExt;
        let sink: WsSink = Box::pin(
            futures::sink::drain()
                .sink_map_err(|_| tokio_tungstenite::tungstenite::Error::ConnectionClosed),
        );
        WsChannel {
            write: Mutex::new(sink),
            pending: Mutex::new(HashMap::new()),
            active_actions: Mutex::new(HashMap::new()),
            next_id: AtomicU64::new(0),
        }
    }

    /// Test-only channel whose write sink is *observable*: every
    /// `Message::Text` payload written to it is appended to the returned
    /// shared Vec. `test_stub` drains to nowhere, which makes it useless
    /// for asserting that a specific frame — e.g. the `tools.cancel`
    /// notification — actually reached the wire.
    ///
    /// The Vec is behind a `std::sync::Mutex` (not tokio's) because the
    /// sink closure is synchronous; the lock is taken and released inside
    /// the closure before the returned future is handed back, so a test
    /// holding the lock to assert can never deadlock a concurrent write.
    #[cfg(test)]
    pub fn test_capture() -> (Self, std::sync::Arc<std::sync::Mutex<Vec<String>>>) {
        let frames = std::sync::Arc::new(std::sync::Mutex::new(Vec::<String>::new()));
        let sink: WsSink = Box::pin(futures::sink::unfold(
            frames.clone(),
            |frames: std::sync::Arc<std::sync::Mutex<Vec<String>>>, msg: Message| {
                if let Message::Text(text) = &msg {
                    // Scoped so the lock is released before we yield the
                    // (already-ready) future back to the sink machinery.
                    if let Ok(mut held) = frames.lock() {
                        held.push(text.as_str().to_string());
                    }
                }
                futures::future::ready(Ok::<_, WsError>(frames))
            },
        ));
        (
            WsChannel {
                write: Mutex::new(sink),
                pending: Mutex::new(HashMap::new()),
                active_actions: Mutex::new(HashMap::new()),
                next_id: AtomicU64::new(0),
            },
            frames,
        )
    }
}

/// In-flight `agents.chat` session bookkeeping. Created when a host
/// client calls `agents.chat`, removed when the agent emits a terminal
/// `agent.chat.event` (`kind: "done"` or `"error"`), when either side
/// disconnects, or when the host cancels via `agents.chat.cancel`.
///
/// The session_id is host-supplied (or server-generated when omitted)
/// and threads through every `agent.chat.event` notification so the
/// server can route streamed deltas back to the originating host
/// without needing per-session subscriptions. See
/// `docs/proposals/agent-chat-surface.md` for the wire contract.
#[derive(Debug, Clone)]
pub struct ChatSession {
    /// Agent that owns this chat — populated from
    /// `attached_agents` at `agents.chat` dispatch time.
    pub agent_id: String,
    /// Client id of the host that issued `agents.chat`. The server
    /// forwards `agent.chat.event` notifications back to *this* host
    /// only, so two CarHost windows chatting with the same agent are
    /// independent streams.
    pub host_client_id: String,
    /// Unix-seconds creation time — used by the future stale-session
    /// sweeper to drop sessions whose agent died without emitting a
    /// terminal event.
    pub created_at: u64,
    /// In-process cancellation flag for daemon-owned chat runs. External
    /// attached agents receive `agent.chat.cancel` instead.
    pub local_cancel: Option<Arc<AtomicBool>>,
}

/// Host-visible standing goal for a chat session. `goal.set` stores this keyed
/// by `session_id`; `agents.chat` reuses it for that session when the turn does
/// not pass an inline `goal`.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ChatGoalState {
    pub session_id: String,
    pub check: String,
    pub max_iterations: u32,
    pub status: String,
    pub last_iteration: Option<u32>,
    pub last_met: Option<bool>,
    pub last_grounded: Option<bool>,
    pub last_reason: Option<String>,
    pub terminal_kind: Option<String>,
    pub terminal_message: Option<String>,
    pub updated_at: u64,
}

fn chat_goals_path_from_journal_dir(journal_dir: &Path) -> PathBuf {
    let car_dir = journal_dir
        .parent()
        .map(PathBuf::from)
        .unwrap_or_else(|| PathBuf::from("."));
    car_dir.join("chat-goals.json")
}

fn load_chat_goals_from_disk(journal_dir: &Path) -> HashMap<String, ChatGoalState> {
    let path = chat_goals_path_from_journal_dir(journal_dir);
    let Ok(text) = std::fs::read_to_string(&path) else {
        return HashMap::new();
    };
    match serde_json::from_str::<HashMap<String, ChatGoalState>>(&text) {
        Ok(mut goals) => {
            for goal in goals.values_mut() {
                if goal.status == "running" {
                    goal.status = "active".to_string();
                }
            }
            goals
        }
        Err(e) => {
            tracing::warn!(
                path = %path.display(),
                error = %e,
                "chat-goals store was unreadable; starting with an empty in-memory goal registry"
            );
            HashMap::new()
        }
    }
}

fn write_chat_goals_atomic(
    path: &std::path::Path,
    goals: &HashMap<String, ChatGoalState>,
) -> std::io::Result<()> {
    use std::sync::atomic::{AtomicU64, Ordering};
    static SEQ: AtomicU64 = AtomicU64::new(0);
    if let Some(parent) = path.parent() {
        std::fs::create_dir_all(parent)?;
    }
    let json = serde_json::to_vec_pretty(goals)?;
    let seq = SEQ.fetch_add(1, Ordering::Relaxed);
    let mut tmp_os = path.as_os_str().to_owned();
    tmp_os.push(format!(".tmp.{}.{}", std::process::id(), seq));
    let tmp = PathBuf::from(tmp_os);
    std::fs::write(&tmp, json)?;
    std::fs::rename(&tmp, path)
}

/// One normalized chunk of an `agent.chat` stream, delivered to an in-process
/// [`chat_collectors`](ServerState::chat_collectors) sink (the A2A bridge). A
/// `token` chunk carries a `delta`; a terminal chunk has `kind` `"done"`
/// (success) or `"error"` (with `error` text).
#[derive(Debug, Clone)]
pub struct ChatStreamChunk {
    pub kind: String,
    pub delta: Option<String>,
    pub error: Option<String>,
}

/// An in-process `agent.chat` collector entry: the sink that
/// [`try_forward_agent_chat_event`](crate::handler::try_forward_agent_chat_event)
/// feeds, plus the host client whose disconnect should abandon it. Holding the
/// sole sender here means dropping the entry closes the stream, so the
/// collecting task wakes immediately instead of waiting out its timeout.
pub struct ChatCollector {
    pub tx: tokio::sync::mpsc::UnboundedSender<ChatStreamChunk>,
    pub host_client_id: String,
}

#[derive(Debug, Clone)]
#[doc(hidden)]
pub struct RunResumeLease {
    pub disconnected_client_id: String,
    pub expires_at: tokio::time::Instant,
}

/// Deterministic test seam at the authenticated completion boundary. It is
/// inert unless explicitly armed through [`ServerStateConfig`].
#[derive(Debug, Clone, Default)]
#[doc(hidden)]
pub struct RunCompletionFenceGate {
    armed: Arc<std::sync::atomic::AtomicBool>,
    entered: Arc<tokio::sync::Notify>,
    release: Arc<tokio::sync::Notify>,
}

impl RunCompletionFenceGate {
    pub fn block_next(&self) {
        self.armed.store(true, std::sync::atomic::Ordering::Release);
    }

    pub async fn wait_until_entered(&self, timeout: std::time::Duration) -> bool {
        tokio::time::timeout(timeout, self.entered.notified())
            .await
            .is_ok()
    }

    pub fn release(&self) {
        self.armed
            .store(false, std::sync::atomic::Ordering::Release);
        self.release.notify_one();
    }

    async fn wait_if_armed(&self) {
        if !self.armed.load(std::sync::atomic::Ordering::Acquire) {
            return;
        }
        self.entered.notify_one();
        if self.armed.load(std::sync::atomic::Ordering::Acquire) {
            self.release.notified().await;
        }
    }
}

/// Daemon-side record of a single agent run (agent run tracing, U1).
///
/// Keyed by `run_id` in the process-wide [`ServerState::runs`] registry
/// so the record survives the WS connection that produced it — needed
/// both for the disconnect grace window (R5) and so the disk store (U3)
/// can flush a run even after its session is gone. U1 keeps this purely
/// in memory; persistence is U3.
#[derive(Debug, Clone)]
pub struct RunMeta {
    pub run_id: String,
    /// Owning agent. Set at `runs.start` from the resolved id; used by
    /// the read RPCs' ownership check (U5/KTD10) and the disk key (U3).
    pub agent_id: String,
    /// Immutable WS `client_id` that durably called `runs.start`. This remains
    /// the provenance identity written to `RunStarted`, proposal receipts, and
    /// `RunEnded`, even if an authenticated replacement socket resumes the
    /// in-memory run.
    pub client_id: String,
    /// The one live socket currently authorized to continue the run. Equal to
    /// `client_id` for a fresh run; changed only by authenticated orphan
    /// reclaim after the previous owner has disappeared from `sessions`.
    pub active_client_id: String,
    /// Stable predecessor returned by an idempotent `runs.resume` retry on the
    /// winning replacement socket.
    pub resume_predecessor_client_id: Option<String>,
    /// Ephemeral, in-process authority window installed atomically when the
    /// active socket leaves the session registry. It is intentionally absent
    /// from RunStore/journals and therefore never survives daemon restart.
    #[doc(hidden)]
    pub resume_lease: Option<RunResumeLease>,
    pub intent: String,
    pub outcome_description: Option<String>,
    pub started_at: chrono::DateTime<chrono::Utc>,
    /// `None` while the run is in progress; `Some` once a terminal
    /// record (a reported outcome or an `Incomplete` marker) is
    /// written. The presence of this field is the "is this run still
    /// open?" signal the grace window checks.
    pub termination: Option<car_proto::RunTermination>,
    /// When the terminal record was written, if any.
    pub ended_at: Option<chrono::DateTime<chrono::Utc>>,
    /// The ordered per-turn trace recorded for this run (agent run
    /// tracing, U2). Each entry is a [`car_proto::RunRecord::Turn`]
    /// produced by the recorder from a submitted proposal + its
    /// `ActionResult`s. U1 leaves this empty; the U2 recorder appends to
    /// it via [`ServerState::record_run_turns`]. U3 flushes this buffer
    /// to disk and U4 broadcasts it — both read it through
    /// [`ServerState::run_turns`]. The `turn.index` field is monotonic
    /// across the run's proposals, so this Vec is the clean ordered
    /// stream those units consume.
    pub turns: Vec<car_proto::RunRecord>,
    /// True only after the exact RunStarted row reached flush+fsync and the
    /// authenticated run_started journal event received the same durability
    /// acknowledgement. A reserved-but-uncommitted run blocks all proposal
    /// writes and can only be resumed by its owning client.
    pub start_committed: bool,
    /// Stable terminal transaction preimage retained across a failed
    /// RunStore or journal durability boundary. Retries must present the same
    /// RunTermination and finish this exact RunEnded; a conflicting terminal
    /// is rejected.
    pub pending_terminal: Option<car_proto::RunEnded>,
    /// Durable cancellation request. Presence quarantines new proposals and
    /// `runs.complete` until CAR either confirms a Cancelled terminal or an
    /// operator independently resolves the unconfirmed stop.
    pub cancellation_pending: Option<car_proto::RunCancellationRequested>,
    /// Stable response for same-key idempotent retries.
    pub cancellation_receipt: Option<car_proto::RunCancelResponse>,
    /// Stable fail-closed error once the durable JSONL is known corrupt.
    /// A quarantined run accepts no more proposals/turns, cannot complete,
    /// and cannot install or retain live subscribers.
    pub trace_corruption: Option<String>,
    /// Monotonic in-memory commit generation for durability work which drops
    /// the global run-registry lock while filesystem I/O executes.
    #[doc(hidden)]
    pub durability_generation: u64,
}

#[derive(Debug, Clone)]
pub enum RunReservation {
    New,
    Existing(RunMeta),
}

/// In-memory ownership result for the authenticated WebSocket resume path.
/// `active_client_id` is the fenced socket allowed to use the original
/// durable run provenance now.
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct RunResumeBinding {
    pub run_id: String,
    pub agent_id: String,
    pub active_client_id: String,
    pub resumed_from_client_id: String,
}

/// Canonical terminal identity shared by the durable `RunEnded` row, terminal
/// journal event, and `runs.complete` response. The input is exactly the RFC
/// 8785/JCS serialization of `RunTermination`; it deliberately does not include
/// timestamps, run ids, client ids, or Daily artifact bytes.
pub(crate) fn run_completion_digest(
    termination: &car_proto::RunTermination,
) -> Result<String, String> {
    let canonical = car_inference::catalog_identity::canonical_json(termination)?;
    Ok(format!("{:x}", Sha256::digest(canonical.as_bytes())))
}

fn run_trace_corruption_message(run_id: &str, detail: impl std::fmt::Display) -> String {
    format!(
        "{} run `{run_id}`: {detail}",
        car_proto::RUN_TRACE_CORRUPTION_MESSAGE_PREFIX
    )
}

fn same_run_termination(
    left: &car_proto::RunTermination,
    right: &car_proto::RunTermination,
) -> bool {
    match (run_completion_digest(left), run_completion_digest(right)) {
        (Ok(left), Ok(right)) => left == right,
        _ => false,
    }
}

fn same_run_ended(left: &car_proto::RunEnded, right: &car_proto::RunEnded) -> bool {
    left.run_id == right.run_id
        && left.client_id == right.client_id
        && left.agent_id == right.agent_id
        && left.completion_digest == right.completion_digest
        && left.ended_at == right.ended_at
        && same_run_termination(&left.termination, &right.termination)
}

struct StartupJournalCache {
    journal_dir: PathBuf,
    failures: Option<car_eventlog::JournalFailureInjector>,
    logs: HashMap<String, EventLog>,
}

impl StartupJournalCache {
    fn new(journal_dir: &Path, failures: Option<&car_eventlog::JournalFailureInjector>) -> Self {
        Self {
            journal_dir: journal_dir.to_path_buf(),
            failures: failures.cloned(),
            logs: HashMap::new(),
        }
    }

    fn get(&mut self, client_id: &str) -> Result<&mut EventLog, String> {
        if !self.logs.contains_key(client_id) {
            let path = self.journal_dir.join(format!("{client_id}.jsonl"));
            let log = match (path.exists(), self.failures.as_ref()) {
                (true, Some(failures)) => {
                    EventLog::load_with_journal_failure_injector(&path, failures.clone())
                        .map_err(|error| error.to_string())
                }
                (true, None) => EventLog::load(&path).map_err(|error| error.to_string()),
                (false, Some(failures)) => Ok(EventLog::with_journal_failure_injector(
                    path,
                    failures.clone(),
                )),
                (false, None) => Ok(EventLog::with_journal(path)),
            }?;
            self.logs.insert(client_id.to_string(), log);
        }
        self.logs
            .get_mut(client_id)
            .ok_or_else(|| "startup journal cache insertion failed".to_string())
    }
}

fn append_recovered_cancellation_result_journal(
    journal_dir: PathBuf,
    failures: Option<car_eventlog::JournalFailureInjector>,
    client_id: String,
    result: car_proto::RunCancelResponse,
) -> Result<bool, String> {
    let mut journals = StartupJournalCache::new(&journal_dir, failures.as_ref());
    let log = journals.get(&client_id)?;
    log.bind_run(&result.run_id, &client_id)?;
    let Value::Object(data) = serde_json::to_value(&result).map_err(|error| error.to_string())?
    else {
        return Err("recovered cancellation result did not serialize as an object".into());
    };
    let data: HashMap<String, Value> = data.into_iter().collect();
    let already_durable = log.events().iter().any(|event| {
        event.kind == car_eventlog::EventKind::RunCancellationResult
            && event.run_id.as_deref() == Some(result.run_id.as_str())
            && event.client_id.as_deref() == Some(client_id.as_str())
            && event.action_id.as_deref() == result.action_id.as_deref()
            && event.proposal_id.is_none()
            && event.data == data
    });
    log.append_critical_bounded(
        car_eventlog::EventKind::RunCancellationResult,
        result.action_id.as_deref(),
        None,
        data,
        CRITICAL_JOURNAL_ACKNOWLEDGEMENT_TIMEOUT,
    )
    .map_err(|error| format!("recovered cancellation journal append failed: {error}"))?;
    Ok(!already_durable)
}

fn reconcile_durable_run_journals(
    journals: &mut StartupJournalCache,
    store: &crate::run_store::RunStore,
    acknowledgement_timeout: std::time::Duration,
) -> Result<(), String> {
    type DurableRunBoundaries = (
        car_proto::RunStarted,
        Option<car_proto::RunEnded>,
        Option<car_proto::RunCancellationRequested>,
        Option<car_proto::RunCancelResponse>,
    );
    let mut by_client: HashMap<String, Vec<DurableRunBoundaries>> = HashMap::new();
    store.visit_run_boundaries(|started, ended, requested, result| {
        let Some(client_id) = started.client_id.as_deref() else {
            return;
        };
        if client_id.is_empty()
            || client_id.contains('/')
            || client_id.contains('\\')
            || client_id == "."
            || client_id == ".."
        {
            tracing::error!(run_id = %started.run_id, "run-journal outbox has unsafe client_id");
            return;
        }
        by_client
            .entry(client_id.to_string())
            .or_default()
            .push((started, ended, requested, result));
    });

    for (client_id, mut runs) in by_client {
        runs.sort_by_key(|(started, _, _, _)| started.started_at);
        let log = match journals.get(&client_id) {
            Ok(log) => log,
            Err(error) => {
                tracing::error!(%client_id, %error, "cannot load run journal for outbox reconciliation");
                continue;
            }
        };
        for (started, ended, requested, result) in runs {
            if let Err(error) = log.bind_run(&started.run_id, &client_id) {
                tracing::error!(run_id = %started.run_id, %error, "cannot bind run journal during outbox reconciliation");
                continue;
            }
            let mut start_data = HashMap::from([
                (
                    "agent_id".to_string(),
                    Value::from(started.agent_id.clone()),
                ),
                ("intent".to_string(), Value::from(started.intent.clone())),
                (
                    "started_at".to_string(),
                    serde_json::to_value(started.started_at).unwrap_or(Value::Null),
                ),
            ]);
            if let Some(description) = &started.outcome_description {
                start_data.insert(
                    "outcome_description".to_string(),
                    Value::from(description.clone()),
                );
            }
            log.append_critical_bounded(
                car_eventlog::EventKind::RunStarted,
                None,
                None,
                start_data,
                acknowledgement_timeout,
            )
            .map_err(|error| {
                format!(
                    "startup reconciliation failed for run {} run_started: {error}",
                    started.run_id
                )
            })?;

            if let Some(requested) = requested {
                let Value::Object(data) = serde_json::to_value(&requested).unwrap_or(Value::Null)
                else {
                    continue;
                };
                log.append_critical_bounded(
                    car_eventlog::EventKind::RunCancellationRequested,
                    requested.action_id.as_deref(),
                    None,
                    data.into_iter().collect(),
                    acknowledgement_timeout,
                )
                .map_err(|error| {
                    format!(
                        "startup reconciliation failed for run {} cancellation request: {error}",
                        started.run_id
                    )
                })?;
            }

            if ended.is_none() {
                if let Some(result) = result {
                    let Value::Object(data) = serde_json::to_value(&result).unwrap_or(Value::Null)
                    else {
                        continue;
                    };
                    log.append_critical_bounded(
                        car_eventlog::EventKind::RunCancellationResult,
                        result.action_id.as_deref(),
                        None,
                        data.into_iter().collect(),
                        acknowledgement_timeout,
                    )
                    .map_err(|error| {
                        format!(
                            "startup reconciliation failed for run {} cancellation result: {error}",
                            started.run_id
                        )
                    })?;
                }
            }

            let Some(ended) = ended else {
                continue;
            };
            let Some(completion_digest) = ended.completion_digest.as_deref() else {
                log.clear_run_binding(&started.run_id, &client_id)?;
                continue;
            };
            if ended.client_id.as_deref() != Some(client_id.as_str()) {
                tracing::error!(run_id = %started.run_id, "RunEnded owner does not match RunStarted during outbox reconciliation");
                log.clear_run_binding(&started.run_id, &client_id)?;
                continue;
            }
            let termination_kind = match &ended.termination {
                car_proto::RunTermination::Outcome { .. } => "outcome",
                car_proto::RunTermination::Incomplete => "incomplete",
                car_proto::RunTermination::Cancelled { .. } => "cancelled",
            };
            let data = HashMap::from([
                (
                    "termination_kind".to_string(),
                    Value::from(termination_kind),
                ),
                (
                    "completion_digest".to_string(),
                    Value::from(completion_digest),
                ),
                (
                    "termination".to_string(),
                    serde_json::to_value(&ended.termination).unwrap_or(Value::Null),
                ),
            ]);
            log.append_critical_bounded(
                car_eventlog::EventKind::RunCompleted,
                None,
                None,
                data,
                acknowledgement_timeout,
            )
            .map_err(|error| {
                format!(
                    "startup reconciliation failed for run {} run_completed: {error}",
                    started.run_id
                )
            })?;
            log.clear_run_binding(&started.run_id, &client_id)?;
        }
    }
    Ok(())
}

/// Finish proposal terminals whose actions already ran before the daemon lost
/// the journal acknowledgement. The outbox owns the exact normal-serde result
/// and JCS digest, so startup only replays that terminal row; it never invokes
/// the runtime or a tool executor.
fn reconcile_pending_proposal_journals(
    journals: &mut StartupJournalCache,
    store: &crate::run_store::RunStore,
    acknowledgement_timeout: std::time::Duration,
) -> Result<(), String> {
    let pending_proposals = match store.all_pending_proposals() {
        Ok(pending) => pending,
        Err(error) => {
            tracing::error!(%error, "cannot enumerate proposal-finalization outbox; all unresolved runs remain quarantined");
            return Ok(());
        }
    };
    for pending in pending_proposals {
        match store.completed_proposal(
            &pending.run_id,
            &pending.client_id,
            pending.requested_policy_session_id.as_deref(),
            &pending.original_submission,
        ) {
            Ok(Some(receipt)) => {
                if let Err(error) = store.cleanup_completed_proposal_guards(&receipt) {
                    tracing::error!(run_id = %pending.run_id, %error, "completed proposal response is durable; startup guard cleanup remains pending");
                }
                continue;
            }
            Ok(None) => {}
            Err(error) => {
                tracing::error!(run_id = %pending.run_id, %error, "cannot validate exact completed proposal response during startup cleanup");
                continue;
            }
        }
        let (started, marker) = match store.pending_provenance(&pending) {
            Ok(provenance) => provenance,
            Err(error) => {
                tracing::error!(run_id = %pending.run_id, %error, "proposal-finalization provenance is invalid; preserving outcome-unknown quarantine");
                continue;
            }
        };
        let Some(client_id) = started.client_id.as_deref() else {
            tracing::error!(run_id = %pending.run_id, "durable RunStarted is missing client_id");
            continue;
        };
        if client_id.is_empty()
            || client_id.contains('/')
            || client_id.contains('\\')
            || client_id == "."
            || client_id == ".."
        {
            tracing::error!(run_id = %pending.run_id, "proposal-finalization outbox has unsafe client_id");
            continue;
        }
        let canonical = match car_inference::catalog_identity::canonical_json(
            &pending.proposal_result,
        ) {
            Ok(canonical) => canonical,
            Err(error) => {
                tracing::error!(run_id = %pending.run_id, %error, "proposal-finalization result is not RFC 8785 canonicalizable");
                continue;
            }
        };
        let digest = format!("{:x}", Sha256::digest(canonical.as_bytes()));
        if digest != pending.result_digest {
            tracing::error!(run_id = %pending.run_id, "proposal-finalization digest does not match its durable result preimage");
            continue;
        }

        let log = match journals.get(client_id) {
            Ok(log) => log,
            Err(error) => {
                tracing::error!(run_id = %pending.run_id, %error, "cannot load proposal journal for outbox reconciliation");
                continue;
            }
        };
        if let Err(error) = log.bind_run(&started.run_id, client_id) {
            tracing::error!(run_id = %pending.run_id, %error, "cannot bind proposal journal during outbox reconciliation");
            continue;
        }
        if let Some(policy_session_id) = marker.policy_session_id.as_deref() {
            if let Err(error) = log.bind_policy_session(policy_session_id) {
                tracing::error!(run_id = %pending.run_id, %error, "cannot restore proposal policy binding during outbox reconciliation");
                continue;
            }
        }
        store.ensure_proposal_turns(&pending).map_err(|error| {
            format!(
                "startup reconciliation failed for run {} proposal trace: {error}",
                pending.run_id
            )
        })?;
        log.append_critical_bounded(
            car_eventlog::EventKind::ProposalCompleted,
            None,
            Some(&pending.final_proposal_id),
            pending.event_data(),
            acknowledgement_timeout,
        )
        .map_err(|error| {
            format!(
                "startup reconciliation failed for run {} proposal_completed: {error}",
                pending.run_id
            )
        })?;
        if let Some(policy_session_id) = marker.policy_session_id.as_deref() {
            log.clear_policy_session(policy_session_id)?;
        }
        let receipt = match store.write_completed_proposal(&pending) {
            Ok(receipt) => receipt,
            Err(error) => {
                tracing::error!(run_id = %pending.run_id, %error, "cannot persist reconciled completed proposal response");
                continue;
            }
        };
        if let Err(error) = store.cleanup_completed_proposal_guards(&receipt) {
            tracing::error!(run_id = %pending.run_id, %error, "completed proposal response is durable; startup guard cleanup remains pending");
            continue;
        }
    }
    Ok(())
}

/// A completed response can outlive either cleanup guard when the daemon dies
/// between independently-fsynced unlink boundaries. Reconcile only under the
/// receipt's exact typed authority; never dispatch or append another event.
fn reconcile_completed_proposal_guards(store: &crate::run_store::RunStore) {
    if let Err(error) = store.reconcile_completed_proposal_migration() {
        tracing::error!(%error, "cannot migrate completed proposal response index; guards remain quarantined");
    }
}

/// Per-run turn ceiling — a runaway-loop backstop sized well above any
/// healthy main-agent-only cycle (tens of turns), never a trimmer. This is
/// a TRUE hard cap, and it is enforced HERE — inside
/// [`ServerState::record_run_turns`], under the `runs` lock — not only in
/// the WS handler's pre-check. The handler keeps a fast-path pre-check off a
/// lock-free snapshot, but the dispatcher spawns a task per frame, so
/// pipelined `runs.record_turns` batches can all read a sub-ceiling snapshot
/// and pass that pre-check before any of them appends. Only the under-lock
/// check below is atomic with the append, so it is the one that actually
/// bounds the run (ADV-1). A batch that would take the run PAST this many
/// recorded turns is refused WHOLE.
pub const RECORD_TURNS_RUN_CEILING: usize = 2000;

/// Outcome of [`ServerState::record_run_turns`] — distinguishes the three
/// reasons a batch can fail to land so the caller maps each to the right
/// `runs.record_turns` drop reason (ADV-1). Before this enum the function
/// returned a bare `usize` (the new total, or `0` for "nothing appended"),
/// which collapsed an under-lock CEILING refusal and an unknown/terminal run
/// into the same `0` — the handler then mislabeled a ceiling refusal as
/// `run_terminal`.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum RecordRunTurnsOutcome {
    /// The batch was appended; the run's new total turn count.
    Appended { new_total: usize },
    /// The batch was refused WHOLE because it would take the run past
    /// [`RECORD_TURNS_RUN_CEILING`] (a runaway backstop). Maps to
    /// `dropped: "run_turn_limit"`.
    RefusedCeiling,
    /// The run id is unknown to this process OR already terminal — nothing
    /// was appended. Maps to `dropped: "run_terminal"` (the benign TOCTOU:
    /// the run went terminal between the handler's pre-check and the
    /// append), matching the prior silent-zero semantics.
    UnknownOrTerminal,
    /// The batch cannot advance canonical memory/subscribers. Ordinary I/O
    /// failures mean the durable append boundary was not acknowledged and an
    /// exact retry is safe. A `run trace corruption:` error instead means the
    /// bytes exposed committed middle corruption; the run is quarantined and
    /// must not be retried or completed.
    PersistenceFailed(String),
}

pub(crate) enum RunSubscribePageResult {
    Ready(car_proto::RunSubscribeResponse),
    Durable {
        agent_id: String,
        status: car_proto::RunLiveStatus,
    },
}

impl RunMeta {
    /// True once a terminal record (outcome or `Incomplete`) is set.
    pub fn is_terminal(&self) -> bool {
        self.termination.is_some()
    }

    pub fn accepts_proposals(&self) -> bool {
        self.start_committed
            && self.pending_terminal.is_none()
            && self.cancellation_pending.is_none()
            && self.trace_corruption.is_none()
            && !self.is_terminal()
    }

    /// The coarse live status of this run for the U4 subscribe snapshot /
    /// `runs.trace.event` (agent run tracing). Mirrors `RunStore`'s
    /// `RunStatus` but reads off the in-memory `RunMeta` so the live path
    /// never touches disk.
    pub fn live_status(&self) -> car_proto::RunLiveStatus {
        match &self.termination {
            None if self.cancellation_pending.is_some() => {
                car_proto::RunLiveStatus::CancellationPending
            }
            None => car_proto::RunLiveStatus::InProgress,
            Some(car_proto::RunTermination::Outcome { .. }) => car_proto::RunLiveStatus::Completed,
            Some(car_proto::RunTermination::Incomplete) => car_proto::RunLiveStatus::Incomplete,
            Some(car_proto::RunTermination::Cancelled { .. }) => {
                car_proto::RunLiveStatus::Cancelled
            }
        }
    }

    /// Count of `RunRecord::Turn` entries in this run's buffer — the live
    /// turn cursor. Equals `turns.len()` because the buffer holds only
    /// `Turn` records (Started/Ended are not buffered here).
    pub fn turn_cursor(&self) -> usize {
        self.turns.len()
    }
}

/// Default ceiling for the daemon→host `tools.execute` callback wait when
/// an action carries no explicit `timeout_ms`. Overridable via
/// `CAR_TOOL_TIMEOUT` (seconds). Raised from the old hardcoded 60s — real
/// tools (build steps, CLI drivers, slow APIs) routinely run longer, and a
/// 60s ceiling reaped them regardless of the agent's budget (Parslee-ai/car#259).
pub const DEFAULT_TOOL_TIMEOUT_MS: u64 = 300_000;

/// Grace added to the callback wait above an action's own `timeout_ms`. The
/// executor (`car-engine`) already wraps each attempt in
/// `timeout(action.timeout_ms, dispatch)`; if this wait used the *same*
/// value it would race that outer deadline with zero slack. By waiting a
/// little longer, the executor's deadline fires first and reaps+rolls back
/// cleanly with its structured "action timed out" error — this wait is only
/// a transport backstop for the case the executor deadline doesn't apply.
const TOOL_TIMEOUT_GRACE_MS: u64 = 5_000;

/// The callback-wait budget for one tool call.
///
/// - **`Some(X)`**: the executor is the authority (it bounds the attempt at
///   `X`); this wait is `X + grace`, a backstop that lets the executor reap
///   first. So the agent's per-action budget takes effect — the bug was that
///   this wait used a hardcoded 60s and ignored `X` entirely (car#259).
/// - **`None`**: the executor applies **no** deadline, so this wait is the
///   *sole* bound on the call — the `CAR_TOOL_TIMEOUT`-overridable
///   [`DEFAULT_TOOL_TIMEOUT_MS`].
fn tool_callback_timeout(action_timeout_ms: Option<u64>) -> std::time::Duration {
    let ms = match action_timeout_ms {
        Some(budget) => budget.saturating_add(TOOL_TIMEOUT_GRACE_MS),
        None => std::env::var("CAR_TOOL_TIMEOUT")
            .ok()
            .and_then(|s| s.trim().parse::<u64>().ok())
            .map(|secs| secs.saturating_mul(1000))
            .unwrap_or(DEFAULT_TOOL_TIMEOUT_MS),
    };
    std::time::Duration::from_millis(ms)
}

/// Build and write the fire-and-forget `tools.cancel` notification for a
/// cancelled `tools.execute` callback (Parslee-ai/car#264).
///
/// This is the ONE place that constructs a [`ToolCancelRequest`] frame and
/// puts it on the wire; every cancellation path routes through
/// [`PendingToolCall`]'s drop, which calls this. Best-effort by design — the
/// daemon has already given up on the call, so a failed write changes nothing
/// it can act on.
pub(crate) async fn write_tool_cancel(
    channel: &WsChannel,
    request_id: String,
    action_id: String,
    reason: String,
) {
    use futures::SinkExt;
    let cancel = ToolCancelRequest {
        request_id,
        action_id,
        reason,
    };
    let notification = serde_json::json!({
        "jsonrpc": "2.0",
        "method": "tools.cancel",
        "params": cancel,
    });
    let Ok(text) = serde_json::to_string(&notification) else {
        return;
    };
    let _ = channel
        .write
        .lock()
        .await
        .send(Message::Text(text.into()))
        .await;
}

/// RAII guard covering one in-flight daemon→host `tools.execute` callback
/// (Parslee-ai/car#264).
///
/// Armed from the moment the oneshot sender lands in
/// [`WsChannel::pending`]; disarmed only when a `ToolExecuteResponse` is
/// actually received. If it is still armed when it drops, the call was
/// cancelled and the guard both releases the `pending` entry and tells the
/// host to abort the in-flight child.
///
/// Why RAII rather than an explicit cleanup branch: `tokio::time::timeout`
/// **drops** the inner future on expiry, and the daemon's callback wait is
/// deliberately the *outermost* of three deadlines
/// (`TOOL_TIMEOUT_GRACE_MS` above the executor's per-action budget). So on
/// every real reap the whole dispatch future is dropped while parked on
/// `rx`, and any code written after the `await` — including the explicit
/// timeout arm — never runs. A `Drop` impl is the only cleanup that survives
/// being dropped.
struct PendingToolCall {
    channel: Arc<WsChannel>,
    request_id: String,
    action_id: String,
    reason: String,
    armed: bool,
}

impl PendingToolCall {
    fn new(channel: Arc<WsChannel>, request_id: String, action_id: String, reason: String) -> Self {
        Self {
            channel,
            request_id,
            action_id,
            reason,
            armed: true,
        }
    }

    /// A response was delivered — this call completed, so no cancel is owed.
    fn disarm(&mut self) {
        self.armed = false;
    }
}

impl Drop for PendingToolCall {
    fn drop(&mut self) {
        if !self.armed {
            return;
        }
        // `pending` is a `tokio::sync::Mutex` and the write is async, but
        // `Drop` is sync — so the cleanup has to be a detached task. Outside
        // a runtime (a plain `drop` in sync code) there is nothing to spawn
        // onto; skip quietly rather than panicking out of a destructor.
        let Ok(handle) = tokio::runtime::Handle::try_current() else {
            return;
        };
        let channel = self.channel.clone();
        let request_id = std::mem::take(&mut self.request_id);
        let action_id = std::mem::take(&mut self.action_id);
        let reason = std::mem::take(&mut self.reason);
        handle.spawn(async move {
            // Only cancel if WE claimed the pending entry. The response
            // demux in `handler.rs` removes it on the success path, and
            // connection teardown clears the whole map — in either case the
            // host is not waiting on anything we should abort.
            let claimed = channel.pending.lock().await.remove(&request_id).is_some();
            channel.active_actions.lock().await.remove(&request_id);
            if claimed {
                write_tool_cancel(&channel, request_id, action_id, reason).await;
            }
        });
    }
}

/// Tool executor that sends callbacks to the client over WebSocket.
pub struct WsToolExecutor {
    pub channel: Arc<WsChannel>,
    negotiated_capabilities: Arc<std::sync::RwLock<std::collections::BTreeSet<String>>>,
}

#[async_trait::async_trait]
impl ToolExecutor for WsToolExecutor {
    async fn execute(&self, tool: &str, params: &Value) -> Result<Value, String> {
        // Legacy callers that don't have a proposal-level Action.id
        // (e.g. internal `executor.execute` chains in tests) — emit an
        // empty action_id so the client-side handler can still see the
        // payload shape and decide whether to fail loudly.
        self.execute_with_action(tool, params, "", None).await
    }

    async fn execute_with_action(
        &self,
        tool: &str,
        params: &Value,
        action_id: &str,
        timeout_ms: Option<u64>,
    ) -> Result<Value, String> {
        self.execute_with_action_in_session(tool, params, action_id, timeout_ms, None, 1)
            .await
    }

    async fn execute_with_action_in_session(
        &self,
        tool: &str,
        params: &Value,
        action_id: &str,
        timeout_ms: Option<u64>,
        session_id: Option<&str>,
        attempt: u32,
    ) -> Result<Value, String> {
        self.execute_callback(
            tool,
            params,
            action_id,
            timeout_ms,
            session_id,
            attempt,
            &HashMap::new(),
            None,
            false,
        )
        .await
        .map(|execution| execution.output)
    }

    async fn execute_with_action_state_in_session(
        &self,
        tool: &str,
        params: &Value,
        action_id: &str,
        timeout_ms: Option<u64>,
        session_id: Option<&str>,
        attempt: u32,
        expected_effects: &HashMap<String, Value>,
        return_schema: Option<&Value>,
    ) -> Result<car_engine::ToolExecution, String> {
        self.execute_callback(
            tool,
            params,
            action_id,
            timeout_ms,
            session_id,
            attempt,
            expected_effects,
            return_schema,
            self.callback_state_negotiated(),
        )
        .await
    }
}

impl WsToolExecutor {
    pub fn new(
        channel: Arc<WsChannel>,
        negotiated_capabilities: Arc<std::sync::RwLock<std::collections::BTreeSet<String>>>,
    ) -> Self {
        Self {
            channel,
            negotiated_capabilities,
        }
    }

    #[cfg(test)]
    fn legacy(channel: Arc<WsChannel>) -> Self {
        Self::new(
            channel,
            Arc::new(std::sync::RwLock::new(std::collections::BTreeSet::new())),
        )
    }

    fn callback_state_negotiated(&self) -> bool {
        self.negotiated_capabilities
            .read()
            .map(|capabilities| capabilities.contains(car_proto::TOOLS_CALLBACK_STATE_CAPABILITY))
            .unwrap_or(false)
    }

    #[allow(clippy::too_many_arguments)]
    async fn execute_callback(
        &self,
        tool: &str,
        params: &Value,
        action_id: &str,
        timeout_ms: Option<u64>,
        session_id: Option<&str>,
        attempt: u32,
        expected_effects: &HashMap<String, Value>,
        return_schema: Option<&Value>,
        callback_state_negotiated: bool,
    ) -> Result<car_engine::ToolExecution, String> {
        use futures::SinkExt;

        // The JSON-RPC request id is the daemon's callback-routing key
        // (used by the pending-response map below). The `action_id`
        // FIELD on the payload is the originating proposal Action.id
        // surfaced to the host so process-wide handlers can route
        // concurrent callbacks back to per-call dispatchers
        // (Parslee-ai/car-releases#43 follow-up). They serve different
        // purposes and must stay distinct: routing id is daemon-side,
        // action id is host-side.
        let request_id = self.channel.next_request_id();

        let callback = ToolExecuteRequest {
            action_id: action_id.to_string(),
            tool: tool.to_string(),
            parameters: params.clone(),
            // Surface the action's budget to the host so its own tool runner
            // can bound the work too (Parslee-ai/car#259) — was always None.
            timeout_ms,
            // The engine's real retry counter, 1-based. This was hardcoded to
            // `1`, which made the field a constant on the wire: a host building
            // the retry-disambiguating join the field exists for got a key that
            // never varied, and would only find out under the concurrent-retry
            // conditions the join is for (Parslee-ai/car#928).
            attempt,
            // Surface the routing id so the host can key a per-call abort
            // registry on it; a reap emits `tools.cancel` carrying the same id
            // (Parslee-ai/car#264).
            request_id: request_id.clone(),
            // Server-stamped correlation. The executor has always passed this
            // in; the parameter was `_session_id` and the value was discarded,
            // so every host that needed attribution had to rebuild it from
            // client-side convention (Parslee-ai/car#904).
            session_id: session_id.map(str::to_string),
        };

        // Create a oneshot channel for the response
        let (tx, rx) = oneshot::channel();
        self.channel
            .pending
            .lock()
            .await
            .insert(request_id.clone(), tx);
        self.channel
            .active_actions
            .lock()
            .await
            .insert(request_id.clone(), action_id.to_string());

        // From here on the call is in flight. The guard makes cleanup +
        // host-abort survive ANY cancellation of this future — the
        // executor's per-action `timeout_ms` deadline (the common case, it
        // fires `TOOL_TIMEOUT_GRACE_MS` before our own wait below), the
        // server's per-method deadline in `handler.rs`, or the callback wait
        // here — not just the one arm we can write code in
        // (Parslee-ai/car#264).
        let mut pending_call = PendingToolCall::new(
            self.channel.clone(),
            request_id.clone(),
            action_id.to_string(),
            format!("tool '{tool}' call cancelled before completion (request {request_id})"),
        );

        // Send the callback to the client as a JSON-RPC request
        let rpc_request = serde_json::json!({
            "jsonrpc": "2.0",
            "method": "tools.execute",
            "params": callback,
            "id": request_id,
        });

        let msg = Message::Text(
            serde_json::to_string(&rpc_request)
                .map_err(|e| e.to_string())?
                .into(),
        );
        self.channel
            .write
            .lock()
            .await
            .send(msg)
            .await
            .map_err(|e| format!("failed to send tool callback: {}", e))?;

        // Wait for the client to respond, bounded by the action's budget
        // (or the configurable default) — NOT a hardcoded 60s. This is the
        // ceiling that previously ignored `Action.timeout_ms` and reaped
        // legitimately-long tool calls (Parslee-ai/car#259).
        let wait = tool_callback_timeout(timeout_ms);
        let response = match tokio::time::timeout(wait, rx).await {
            Ok(inner) => inner.map_err(|_| format!("tool '{}' callback channel closed", tool))?,
            Err(_) => {
                // Reaped by our own callback wait. Sharpen the guard's reason
                // to name the timeout, then let it do the cleanup + host abort
                // on the way out — the same path a dropped future takes
                // (Parslee-ai/car#264).
                let reason = format!("tool '{}' callback timed out ({}s)", tool, wait.as_secs());
                pending_call.reason = reason.clone();
                return Err(reason);
            }
        };

        // A response was delivered (error field set or not): the call is
        // finished, so no cancel is owed to the host.
        self.channel.active_actions.lock().await.remove(&request_id);
        pending_call.disarm();

        if let Some(err) = response.error {
            return Err(err);
        }
        decode_callback_execution(
            tool,
            action_id,
            response.output.unwrap_or(Value::Null),
            expected_effects,
            return_schema,
            callback_state_negotiated,
        )
    }
}

fn decode_callback_execution(
    tool: &str,
    action_id: &str,
    result: Value,
    expected_effects: &HashMap<String, Value>,
    return_schema: Option<&Value>,
    callback_state_negotiated: bool,
) -> Result<car_engine::ToolExecution, String> {
    if !callback_state_negotiated {
        return Ok(car_engine::ToolExecution::output_only(result));
    }
    let envelope = result.as_object().filter(|object| {
        object.len() == 2 && object.contains_key("output") && object.contains_key("state_changes")
    });

    if expected_effects.is_empty() && envelope.is_none() {
        return Ok(car_engine::ToolExecution::output_only(result));
    }
    let envelope = envelope.ok_or_else(|| {
        format!(
            "tool '{tool}' callback for action '{action_id}' must return exact envelope {{output,state_changes}}"
        )
    })?;
    let output = envelope
        .get("output")
        .cloned()
        .expect("exact envelope contains output");
    let state_changes: HashMap<String, Value> = serde_json::from_value(
        envelope
            .get("state_changes")
            .cloned()
            .expect("exact envelope contains state_changes"),
    )
    .map_err(|_| {
        format!("tool '{tool}' callback for action '{action_id}' state_changes must be an object")
    })?;
    let expected_keys: std::collections::BTreeSet<&str> =
        expected_effects.keys().map(String::as_str).collect();
    let actual_keys: std::collections::BTreeSet<&str> =
        state_changes.keys().map(String::as_str).collect();
    if expected_keys != actual_keys {
        return Err(format!(
            "tool '{tool}' callback state_changes keys do not match action '{action_id}': expected {:?}, got {:?}",
            expected_keys, actual_keys
        ));
    }
    if let Some(schema) = return_schema {
        car_engine::validate_tool_output(tool, schema, &output)?;
    }
    let changes_value = serde_json::to_value(&state_changes).map_err(|error| {
        format!("tool '{tool}' callback state_changes serialization failed: {error}")
    })?;
    car_inference::catalog_identity::canonical_json(&changes_value).map_err(|error| {
        format!("tool '{tool}' callback state_changes failed JCS/I-JSON validation: {error}")
    })?;
    Ok(car_engine::ToolExecution {
        output,
        state_changes,
    })
}

#[cfg(test)]
mod tool_cancel_tests {
    use super::*;
    use serde_json::json;
    use std::time::Duration;

    /// Every `Message::Text` the executor wrote, parsed as JSON.
    fn written(frames: &std::sync::Arc<std::sync::Mutex<Vec<String>>>) -> Vec<Value> {
        frames
            .lock()
            .unwrap()
            .iter()
            .map(|t| serde_json::from_str::<Value>(t).expect("frame is JSON"))
            .collect()
    }

    fn frame_with_method<'a>(frames: &'a [Value], method: &str) -> Option<&'a Value> {
        frames.iter().find(|f| f["method"] == json!(method))
    }

    /// Let detached cleanup tasks (the `PendingToolCall` drop spawn) run.
    async fn settle() {
        for _ in 0..8 {
            tokio::task::yield_now().await;
        }
        tokio::time::sleep(Duration::from_millis(100)).await;
        for _ in 0..8 {
            tokio::task::yield_now().await;
        }
    }

    /// The daemon's own callback wait expiring still cancels — the arm that
    /// used to hold the inline cleanup now delegates to the drop guard, and
    /// the observable behaviour (Err text, cancel frame, drained `pending`)
    /// must be unchanged.
    #[tokio::test(start_paused = true)]
    async fn callback_wait_expiry_cancels() {
        let (channel, frames) = WsChannel::test_capture();
        let channel = Arc::new(channel);
        let executor = WsToolExecutor::legacy(channel.clone());

        // 1ms budget → a ~5s callback wait; paused time makes that instant.
        let err = executor
            .execute_with_action_in_session("drive_cli", &json!({}), "a0", Some(1), None, 1)
            .await
            .expect_err("the callback wait must expire");
        assert!(
            err.contains("tool 'drive_cli' callback timed out"),
            "unexpected error text: {err}"
        );

        settle().await;

        let frames = written(&frames);
        let execute = frame_with_method(&frames, "tools.execute").expect("tools.execute was sent");
        let cancel = frame_with_method(&frames, "tools.cancel")
            .expect("a tools.cancel notification must reach the wire");
        assert_eq!(
            cancel["params"]["request_id"], execute["params"]["request_id"],
            "cancel must correlate to the originating request_id"
        );
        assert_eq!(cancel["params"]["action_id"], json!("a0"));
        assert!(
            cancel["id"].is_null(),
            "cancel is a notification, not a request"
        );
        assert!(
            channel.pending.lock().await.is_empty(),
            "the pending entry must be released"
        );
    }

    /// Parslee-ai/car#904 — the execution session must reach the wire.
    ///
    /// The executor has always passed a session id in; the parameter was named
    /// `_session_id` and the value was dropped on the floor, so every host that
    /// needed to know which mission a callback belonged to had to rebuild it
    /// from client-side convention. Asserted on the actual `tools.execute`
    /// frame rather than on the struct, because the struct field existing is
    /// not the property that was missing.
    #[tokio::test(start_paused = true)]
    async fn tools_execute_carries_the_execution_session() {
        let (channel, frames) = WsChannel::test_capture();
        let channel = Arc::new(channel);
        let executor = WsToolExecutor::legacy(channel.clone());

        let _ = tokio::time::timeout(
            Duration::from_millis(50),
            executor.execute_with_action_in_session(
                "search",
                &json!({}),
                "a0",
                Some(10_000),
                Some("sess-42"),
                1,
            ),
        )
        .await;
        settle().await;

        let frames = written(&frames);
        let execute = frame_with_method(&frames, "tools.execute").expect("tools.execute was sent");
        assert_eq!(
            execute["params"]["session_id"],
            json!("sess-42"),
            "the daemon must stamp the session it already knows, not leave \
             attribution to the host"
        );
    }

    /// Parslee-ai/car#928 — the retry counter on the wire must be the engine's
    /// real one.
    ///
    /// It was hardcoded to `1` here, so the field a host would build its
    /// retry-disambiguating join on never varied. That is worse than the field
    /// being absent: the join looks correct and degenerates silently, and only
    /// under the concurrent-retry conditions it exists to handle.
    #[tokio::test(start_paused = true)]
    async fn tools_execute_carries_the_real_attempt_number() {
        for attempt in [1u32, 2, 7] {
            let (channel, frames) = WsChannel::test_capture();
            let channel = Arc::new(channel);
            let executor = WsToolExecutor::legacy(channel.clone());

            let _ = tokio::time::timeout(
                Duration::from_millis(50),
                executor.execute_with_action_in_session(
                    "search",
                    &json!({}),
                    "a0",
                    Some(10_000),
                    Some("sess-1"),
                    attempt,
                ),
            )
            .await;
            settle().await;

            let frames = written(&frames);
            let execute =
                frame_with_method(&frames, "tools.execute").expect("tools.execute was sent");
            assert_eq!(
                execute["params"]["attempt"],
                json!(attempt),
                "the payload must carry the engine's attempt, not a constant"
            );
        }
    }

    /// The other half: a caller with no session must omit the key rather than
    /// send a null. `execute()` and in-process executors legitimately have
    /// none, and a null would make every such payload differ from the pre-#904
    /// shape for no gain.
    #[tokio::test(start_paused = true)]
    async fn a_sessionless_call_omits_the_session_key() {
        let (channel, frames) = WsChannel::test_capture();
        let channel = Arc::new(channel);
        let executor = WsToolExecutor::legacy(channel.clone());

        let _ = tokio::time::timeout(
            Duration::from_millis(50),
            executor.execute_with_action_in_session(
                "search",
                &json!({}),
                "a0",
                Some(10_000),
                None,
                1,
            ),
        )
        .await;
        settle().await;

        let frames = written(&frames);
        let execute = frame_with_method(&frames, "tools.execute").expect("tools.execute was sent");
        assert!(
            execute["params"].get("session_id").is_none(),
            "a sessionless call must omit the key entirely, got: {}",
            execute["params"]
        );
    }

    /// REGRESSION (Parslee-ai/car#264): the executor's per-action deadline
    /// fires `TOOL_TIMEOUT_GRACE_MS` BEFORE the daemon's callback wait, so it
    /// drops this whole future while it is parked on `rx`. Before the drop
    /// guard, nothing ran on that path: no `tools.cancel` frame was ever
    /// written and the `pending` entry leaked until connection teardown,
    /// orphaning the host's in-flight child.
    #[tokio::test(start_paused = true)]
    async fn engine_deadline_drop_cancels() {
        let (channel, frames) = WsChannel::test_capture();
        let channel = Arc::new(channel);
        let executor = WsToolExecutor::legacy(channel.clone());

        // Outer deadline (standing in for the engine's per-action timeout)
        // at 50ms; the callback wait would be 10s + 5s grace.
        let outcome = tokio::time::timeout(
            Duration::from_millis(50),
            executor.execute_with_action_in_session(
                "drive_cli",
                &json!({}),
                "a0",
                Some(10_000),
                None,
                1,
            ),
        )
        .await;
        assert!(
            outcome.is_err(),
            "the outer deadline must fire first and drop the dispatch future"
        );

        settle().await;

        let frames = written(&frames);
        let execute = frame_with_method(&frames, "tools.execute").expect("tools.execute was sent");
        let cancel = frame_with_method(&frames, "tools.cancel").expect(
            "dropping the dispatch future must still emit tools.cancel so the host child isn't orphaned",
        );
        assert_eq!(
            cancel["params"]["request_id"], execute["params"]["request_id"],
            "cancel must correlate to the originating request_id"
        );
        assert!(
            channel.pending.lock().await.is_empty(),
            "the pending entry must not leak"
        );
    }

    /// The happy path must stay silent: a delivered response disarms the
    /// guard, so no host abort is emitted for a call that completed.
    #[tokio::test]
    async fn success_emits_no_cancel() {
        let (channel, frames) = WsChannel::test_capture();
        let channel = Arc::new(channel);
        let executor = WsToolExecutor::legacy(channel.clone());

        let responder = channel.clone();
        let host = tokio::spawn(async move {
            loop {
                let claimed = {
                    let mut pending = responder.pending.lock().await;
                    let key = pending.keys().next().cloned();
                    key.and_then(|k| pending.remove(&k).map(|tx| (k, tx)))
                };
                if let Some((request_id, tx)) = claimed {
                    let _ = tx.send(ToolExecuteResponse {
                        action_id: request_id,
                        output: Some(json!({"ok": true})),
                        error: None,
                    });
                    return;
                }
                tokio::task::yield_now().await;
            }
        });

        let out = executor
            .execute_with_action_in_session("drive_cli", &json!({}), "a0", Some(10_000), None, 1)
            .await
            .expect("the host responded, so the call succeeds");
        assert_eq!(out, json!({"ok": true}));
        host.await.expect("responder task finishes");

        settle().await;

        let frames = written(&frames);
        assert!(
            frame_with_method(&frames, "tools.execute").is_some(),
            "tools.execute was sent"
        );
        assert!(
            frame_with_method(&frames, "tools.cancel").is_none(),
            "a completed call must not emit tools.cancel"
        );
        assert!(channel.pending.lock().await.is_empty());
    }
}

/// The bare commodity-tool names owned by the engine's bound substrate.
/// When a session binds a non-Local substrate (`session.bindSubstrate`),
/// these names must reach the engine's substrate-after-fall_through path
/// (`executor.rs`), NOT the WS tool callback. Everything else — connector
/// `mcp_{slug}_*` routes and host-executed client tools — is untouched.
///
/// `calculate` is deliberately excluded: it is a pure, environment-free
/// tool that the engine already handles in-process via `agent_basics`, so
/// it never needs to be re-routed to a substrate.
pub(crate) const SUBSTRATE_OWNED_TOOLS: &[&str] = &[
    "read_file",
    "write_file",
    "edit_file",
    "list_dir",
    "find_files",
    "grep_files",
];

/// Executor wrapper used **only** by substrate-bound sessions
/// (`session.bindSubstrate`). It is the low-blast-radius "option (a)" from
/// `docs/execution-substrate.md` §3: keep the existing
/// `share_with_fallback(ws_executor)` composition intact for connector
/// routes and host tool callbacks, but for the bare substrate-owned
/// built-in names ([`SUBSTRATE_OWNED_TOOLS`]) return an `"unknown tool:
/// <name>"` error so the engine's dispatch (`executor.rs`) takes the
/// `fall_through` branch and routes those names to the runtime's bound
/// substrate (the connector/VM `McpSubstrate`) instead of the WS client.
///
/// This preserves every existing contract — GUI/connector/coder sessions
/// never construct one of these — and is what makes a connector-driven
/// session *coherent*: its file/exec built-ins land on the same machine as
/// its `mcp_{slug}_*` tools, not split between the WS client and a remote
/// MCP server.
pub struct SubstrateShadowExecutor {
    inner: Arc<dyn ToolExecutor>,
}

impl SubstrateShadowExecutor {
    pub fn new(inner: Arc<dyn ToolExecutor>) -> Self {
        Self { inner }
    }
}

#[async_trait::async_trait]
impl ToolExecutor for SubstrateShadowExecutor {
    async fn execute(&self, tool: &str, params: &Value) -> Result<Value, String> {
        self.execute_with_action(tool, params, "", None).await
    }

    async fn execute_with_action(
        &self,
        tool: &str,
        params: &Value,
        action_id: &str,
        timeout_ms: Option<u64>,
    ) -> Result<Value, String> {
        self.execute_with_action_in_session(tool, params, action_id, timeout_ms, None, 1)
            .await
    }

    async fn execute_with_action_in_session(
        &self,
        tool: &str,
        params: &Value,
        action_id: &str,
        timeout_ms: Option<u64>,
        session_id: Option<&str>,
        attempt: u32,
    ) -> Result<Value, String> {
        if SUBSTRATE_OWNED_TOOLS.contains(&tool) {
            // Signal the engine to fall through to the bound substrate.
            // The exact `"unknown tool"` prefix is the contract the engine
            // dispatch matches on (`executor.rs`).
            return Err(format!("unknown tool: {tool}"));
        }
        self.inner
            .execute_with_action_in_session(
                tool, params, action_id, timeout_ms, session_id, attempt,
            )
            .await
    }

    async fn execute_with_action_state_in_session(
        &self,
        tool: &str,
        params: &Value,
        action_id: &str,
        timeout_ms: Option<u64>,
        session_id: Option<&str>,
        attempt: u32,
        expected_effects: &HashMap<String, Value>,
        return_schema: Option<&Value>,
    ) -> Result<car_engine::ToolExecution, String> {
        if SUBSTRATE_OWNED_TOOLS.contains(&tool) {
            return Err(format!("unknown tool: {tool}"));
        }
        self.inner
            .execute_with_action_state_in_session(
                tool,
                params,
                action_id,
                timeout_ms,
                session_id,
                attempt,
                expected_effects,
                return_schema,
            )
            .await
    }
}

/// Voice event sink that forwards events to a specific WebSocket client
/// as `voice.event` JSON-RPC notifications.
///
/// Each `voice.transcribe_stream.start` call constructs one of these
/// bound to the originating client's [`WsChannel`], so a client only
/// receives events for sessions it started.
pub struct WsVoiceEventSink {
    pub channel: Arc<WsChannel>,
}

impl car_voice::VoiceEventSink for WsVoiceEventSink {
    fn send(&self, session_id: &str, event_json: String) {
        use futures::SinkExt;
        let channel = self.channel.clone();
        let session_id = session_id.to_string();
        tokio::spawn(async move {
            let payload: Value = serde_json::from_str(&event_json)
                .unwrap_or_else(|_| Value::String(event_json.clone()));
            let notification = serde_json::json!({
                "jsonrpc": "2.0",
                "method": "voice.event",
                "params": {
                    "session_id": session_id,
                    "event": payload,
                },
            });
            let Ok(text) = serde_json::to_string(&notification) else {
                return;
            };
            let _ = channel
                .write
                .lock()
                .await
                .send(Message::Text(text.into()))
                .await;
        });
    }

    fn send_binary(&self, frame: Vec<u8>) {
        use futures::SinkExt;
        let channel = self.channel.clone();
        tokio::spawn(async move {
            let _ = channel
                .write
                .lock()
                .await
                .send(Message::Binary(frame.into()))
                .await;
        });
    }
}

/// Per-meeting fanout sink that ingests transcript text into a
/// session-scoped memgine using the `Arc<tokio::sync::Mutex<...>>`
/// wrapper, then forwards every event upstream untouched.
///
/// Lives here (not in `car-ffi-common`) because the engine handle uses
/// `tokio::sync::Mutex` per the "one-wrapper rule" — the FFI-common
/// `MeetingMemgineFanout` still uses `std::sync::Mutex` for the NAPI/
/// PyO3 bindings, which keep their sync wrappers. Each binding owns the
/// fanout that matches its lock primitive; the parsing/formatting logic
/// itself is shared via [`car_meeting::extract_transcript_for_ingest`].
///
/// `send` is called from the voice drain task and must be non-blocking,
/// so the lock acquisition is shipped to a `tokio::spawn`. Transcript
/// events are independent so reordering across spawned tasks is fine.
pub struct WsMemgineIngestSink {
    pub meeting_id: String,
    pub engine: Arc<Mutex<car_memgine::MemgineEngine>>,
    pub upstream: Arc<dyn car_voice::VoiceEventSink>,
}

impl car_voice::VoiceEventSink for WsMemgineIngestSink {
    fn send(&self, voice_session_id: &str, event_json: String) {
        if let Ok(value) = serde_json::from_str::<Value>(&event_json) {
            if let Some((speaker, text)) = car_meeting::extract_transcript_for_ingest(
                &value,
                &self.meeting_id,
                voice_session_id,
            ) {
                let engine = self.engine.clone();
                tokio::spawn(async move {
                    let mut guard = engine.lock().await;
                    guard.ingest_conversation(&speaker, &text, chrono::Utc::now());
                });
            }
        }
        self.upstream.send(voice_session_id, event_json);
    }
}

/// The last assistant turn a chat session produced, remembered so the *next*
/// user turn can be classified against it (the conversation-outcome signal).
/// The `trace_id`/`model_id` are carried straight from the `InferenceResult`
/// that produced the turn — never reconstructed from message order — so credit
/// always lands on the exact model/trace that generated the judged turn.
#[derive(Debug, Clone)]
pub struct LastChatTurn {
    /// The user request this assistant turn answered (for restatement similarity).
    pub user_text: String,
    pub assistant_text: String,
    pub trace_id: String,
    pub model_id: String,
}

/// Per-client session.
pub struct ClientSession {
    pub client_id: String,
    pub runtime: Arc<Runtime>,
    pub channel: Arc<WsChannel>,
    pub host: Arc<crate::host::HostState>,
    /// Memgine handle. Wrapped in `tokio::sync::Mutex` so dispatcher
    /// handlers can hold the lock across `.await` points without
    /// risking poisoning. Migrated from `std::sync::Mutex` in the
    /// car-server-core extraction (U1) per the "one-wrapper rule".
    pub memgine: Arc<Mutex<car_memgine::MemgineEngine>>,
    /// Lazy browser session — first `browser.run` call launches Chromium,
    /// subsequent calls reuse it so element IDs resolve across invocations
    /// within the same WebSocket connection.
    pub browser: car_ffi_common::browser::BrowserSessionSlot,
    /// Per-connection auth state. Starts `false`; flips to `true`
    /// after a successful `session.auth` handshake. Always considered
    /// authenticated when `ServerState::auth_token` is unset (auth
    /// disabled). Closes Parslee-ai/car-releases#32.
    pub authenticated: Arc<std::sync::atomic::AtomicBool>,
    /// Negotiated daemon JSON-RPC protocol for this WebSocket connection.
    /// Starts at `0` (unnegotiated) and is set to
    /// `car_proto::PROTOCOL_VERSION` only after an exact-version
    /// `server.handshake`. The state is connection-scoped by construction, so
    /// a reconnect always has to negotiate again before using protocol-v2-only
    /// host/auth methods.
    pub negotiated_protocol_version: std::sync::atomic::AtomicU32,
    /// Capability names negotiated by this connection's authenticated
    /// `server.handshake`. Connection-scoped for the same reason as the wire
    /// version: a reconnect cannot inherit another socket's proof.
    pub negotiated_capabilities: Arc<std::sync::RwLock<std::collections::BTreeSet<String>>>,
    /// Bounded lifecycle state for inferences started on this exact socket.
    /// Keeping it per-session prevents an authenticated peer from probing IDs
    /// owned by another connection.
    pub inference_control: Arc<crate::inference_control::InferenceRegistry>,
    /// Host-management role (Parslee-ai/car#254). Starts `false`; flips
    /// to `true` only when the connection presents the per-launch host
    /// token via `session.auth { host_token }` (validated against
    /// `ServerState::host_token`). `authorize_run_access` requires this
    /// for cross-agent run-trace reads — being merely `host.subscribe`d
    /// is no longer sufficient, which is what closes the self-elevation
    /// hole. Cleared implicitly when the connection drops.
    pub is_host: std::sync::atomic::AtomicBool,
    /// Bound agent identity (#169). `Some(id)` once a lifecycle-agent
    /// child has called `session.auth { token, agent_id }` and the
    /// supervisor confirmed `agent_id` is supervised + token matches.
    /// Used by `agents.list` to surface which managed agents have
    /// actually attached vs. just being marked `Running` at the
    /// process level. Cleared at disconnect by `remove_session`.
    pub agent_id: Arc<tokio::sync::Mutex<Option<String>>>,
    /// Canonical schema digests for the narrow reverse-callback tools this
    /// client registered. Server-owned registry entries never land here.
    pub callback_tool_schema_digests:
        Arc<tokio::sync::RwLock<std::collections::HashMap<String, String>>>,
    /// Bound memory namespace (`session.auth { memory_namespace }`, #79/#80).
    ///
    /// The namespace identity has to survive the handshake, not just the
    /// lookup that binds `bound_memgine`. Without it the daemon knows *which
    /// graph* a session is using but not *what to call it*, so nothing can
    /// write that graph back to
    /// `~/.car/memory/memory-namespaces/<encoded-ns>.json` — which is
    /// why namespace memory was durable only until the next restart
    /// (car-releases#82). Paired with `bound_memgine` exactly as `agent_id` is.
    pub memory_namespace: tokio::sync::Mutex<Option<String>>,
    /// Bound persistent memgine (#170). `Some` after `session.auth`
    /// successfully attaches the connection to a daemon-owned
    /// per-agent memgine (paired with `agent_id`). Memory handlers
    /// route through [`ClientSession::effective_memgine`] which
    /// returns this when set, falling back to the ephemeral
    /// `memgine` field for browser/host/CLI connections.
    pub bound_memgine: tokio::sync::Mutex<Option<Arc<Mutex<car_memgine::MemgineEngine>>>>,
    /// The run currently bracketed on this connection (agent run
    /// tracing, U1). Set by `runs.start` **before** that handler
    /// responds, so the per-turn recorder (U2) always reads the
    /// `run_id` the bracket established — no race across the
    /// concurrently-spawned dispatch tasks (KTD3). Cleared by
    /// `runs.complete`. On disconnect with a still-set current run and
    /// no recorded terminal, the daemon marks it `Incomplete` (R5).
    pub current_run_id: tokio::sync::Mutex<Option<String>>,
    /// Serializes the authenticated run bracket on one WebSocket: start,
    /// proposal lifecycle, complete, and disconnect terminalization cannot
    /// interleave and produce ambiguous journal ordering.
    pub run_lifecycle_guard: Arc<tokio::sync::Mutex<()>>,
    /// Per-session permission-tier gate (survey §3.4.3/§5.2.5). Backs the
    /// `permission.*` JSON-RPC methods: the session holds the granted standing
    /// tier + risk classifier. Defaults to `SandboxEdit`, the tier
    /// `docs/websocket-protocol.md` publishes as the session default.
    ///
    /// `Arc`, not a bare lock, because this is also the gate
    /// [`crate::permission_gate::PermissionAdmissionGate`] enforces at
    /// proposal admission (Parslee-ai/car#890). ONE gate object behind both:
    /// a second instance would let the advisory `permission.evaluate` and the
    /// enforcement point disagree about the same action, which is precisely
    /// the bug class the admission gate was added to close.
    ///
    /// NOTE (kernel review C1): approval *records* do NOT live here. The
    /// gate's embedded ledger is deliberately unused in the daemon — every
    /// fingerprint-keyed HITL read/write goes through the SHARED
    /// [`ServerState::approval_ledger`] (journal-backed), so an approval
    /// recorded on one connection is visible to every other and survives
    /// restart. Tier state stays per-session; the approval store is
    /// daemon-wide.
    pub permission_gate: Arc<tokio::sync::RwLock<car_policy::PermissionGate>>,
    /// Non-overlap guard for `evolution.run` on this session (kernel review
    /// S3): the dispatcher spawns requests concurrently even on one
    /// connection, and two interleaved cycles would double-dispatch the
    /// evolution mechanisms. A second `evolution.run` while one is in flight
    /// errs instead of overlapping.
    pub evolution_guard: crate::evolution::CycleGuard,
    /// Last assistant turn produced on this session's chat path, so the next
    /// user turn can be scored as a conversation outcome (Part B). `None` until
    /// the first chat reply; only set for declared-chat inference.
    pub last_chat_turn: tokio::sync::Mutex<Option<LastChatTurn>>,
    /// In-flight chat-infer count. Concurrent infers on one session have no
    /// well-defined turn adjacency (the dispatcher spawns a task per frame), so
    /// the outcome scorer only records a turn that began while the session was
    /// idle — otherwise it would score one infer's trace against an unrelated
    /// concurrent user turn (an active mislabel of routing stats). neo #1.
    pub chat_inflight: std::sync::atomic::AtomicUsize,
    /// Tenant identity bound at `session.auth { tenant_id }` (linus
    /// review C-4). Once bound, every tenant-scoped handler uses THIS
    /// identity: a per-request `tenant_id` param may restate it but a
    /// mismatch is rejected — an authenticated connection cannot hop
    /// into another tenant's namespace by editing request params.
    /// `None` (unbound) preserves the legacy per-request behavior.
    pub tenant: tokio::sync::Mutex<Option<String>>,
    /// Whether this connection has a live `tools.stream.event` forwarder
    /// task (C2). Set by `tools.stream.subscribe`; a concurrent second
    /// subscribe is a no-op instead of spawning a duplicate forwarder
    /// (which would double every notification). `Arc` so the forwarder
    /// task can RESET it on exit (write timeout on a stalled-but-
    /// recovering socket, broadcast closed) — a later re-subscribe then
    /// spawns a fresh forwarder rather than silently no-op'ing (Q2).
    pub tool_stream_subscribed: std::sync::Arc<std::sync::atomic::AtomicBool>,
}

impl ClientSession {
    /// Returns the memgine handle the memory.* handlers should use:
    /// the bound per-agent memgine when this session attached via
    /// `session.auth { agent_id }` (#169 + #170), otherwise the
    /// ephemeral per-WS memgine. Cheap (one async lock + Arc clone).
    pub async fn effective_memgine(&self) -> Arc<Mutex<car_memgine::MemgineEngine>> {
        if let Some(eng) = self.bound_memgine.lock().await.as_ref() {
            return eng.clone();
        }
        self.memgine.clone()
    }

    pub async fn bind_run_journal(&self, run_id: &str) -> Result<(), String> {
        self.runtime
            .event_log_handle()
            .lock()
            .await
            .bind_run(run_id, &self.client_id)
    }

    pub async fn require_run_journal_binding(&self, run_id: &str) -> Result<(), String> {
        let log = self.runtime.event_log_handle();
        let log = log.lock().await;
        match log.active_run_binding() {
            Some((bound_run, bound_client, _))
                if bound_run == run_id && bound_client == self.client_id =>
            {
                Ok(())
            }
            Some((bound_run, bound_client, _)) => Err(format!(
                "active journal binding mismatch: expected run_id `{run_id}` / client_id `{}`, got `{bound_run}` / `{bound_client}`",
                self.client_id
            )),
            None => Err(format!(
                "active run `{run_id}` has no authenticated journal binding"
            )),
        }
    }

    pub async fn clear_run_journal_binding(&self, run_id: &str) -> Result<(), String> {
        self.runtime
            .event_log_handle()
            .lock()
            .await
            .clear_run_binding(run_id, &self.client_id)
    }

    pub async fn append_run_terminal_event(
        &self,
        ended: &car_proto::RunEnded,
    ) -> Result<(), String> {
        self.append_run_terminal_event_once(ended, &self.client_id)
            .await
            .map_err(|error| error.to_string())
    }

    /// Append a terminal from an authenticated replacement socket while
    /// preserving the immutable client identity in the durable run trace.
    pub(crate) async fn append_resumed_run_terminal_event(
        &self,
        ended: &car_proto::RunEnded,
        durable_client_id: &str,
    ) -> Result<(), String> {
        self.append_run_terminal_event_once(ended, durable_client_id)
            .await
            .map_err(|error| error.to_string())
    }

    pub async fn append_run_cancellation_requested_event(
        &self,
        requested: &car_proto::RunCancellationRequested,
    ) -> Result<(), String> {
        self.require_run_journal_binding(&requested.run_id).await?;
        let value = serde_json::to_value(requested).map_err(|error| error.to_string())?;
        let Value::Object(data) = value else {
            return Err("cancellation request did not serialize as an object".into());
        };
        self.runtime
            .event_log_handle()
            .lock()
            .await
            .append_critical_async(
                car_eventlog::EventKind::RunCancellationRequested,
                requested.action_id.as_deref(),
                None,
                data.into_iter().collect(),
                CRITICAL_JOURNAL_ACKNOWLEDGEMENT_TIMEOUT,
            )
            .await
            .map_err(|error| error.to_string())?;
        Ok(())
    }

    pub async fn append_run_cancellation_result_event(
        &self,
        result: &car_proto::RunCancelResponse,
    ) -> Result<(), String> {
        self.require_run_journal_binding(&result.run_id).await?;
        let value = serde_json::to_value(result).map_err(|error| error.to_string())?;
        let Value::Object(data) = value else {
            return Err("cancellation result did not serialize as an object".into());
        };
        self.runtime
            .event_log_handle()
            .lock()
            .await
            .append_critical_async(
                car_eventlog::EventKind::RunCancellationResult,
                result.action_id.as_deref(),
                None,
                data.into_iter().collect(),
                CRITICAL_JOURNAL_ACKNOWLEDGEMENT_TIMEOUT,
            )
            .await
            .map_err(|error| error.to_string())?;
        Ok(())
    }

    async fn append_run_terminal_event_once(
        &self,
        ended: &car_proto::RunEnded,
        durable_client_id: &str,
    ) -> Result<(), car_eventlog::CriticalAppendError> {
        let rejected = |reason| car_eventlog::CriticalAppendError::Rejected { reason };
        self.require_run_journal_binding(&ended.run_id)
            .await
            .map_err(rejected)?;
        if ended.client_id.as_deref() != Some(durable_client_id) {
            return Err(rejected(
                "RunEnded client_id does not match the durable run owner".into(),
            ));
        }
        let completion_digest = ended
            .completion_digest
            .as_deref()
            .ok_or_else(|| rejected("RunEnded is missing completion_digest".into()))?;
        let termination_kind = match &ended.termination {
            car_proto::RunTermination::Outcome { .. } => "outcome",
            car_proto::RunTermination::Incomplete => "incomplete",
            car_proto::RunTermination::Cancelled { .. } => "cancelled",
        };
        let log = self.runtime.event_log_handle();
        let mut log = log.lock().await;
        log.append_critical_async(
            car_eventlog::EventKind::RunCompleted,
            None,
            None,
            [
                (
                    "termination_kind".to_string(),
                    Value::from(termination_kind),
                ),
                (
                    "completion_digest".to_string(),
                    Value::from(completion_digest),
                ),
                (
                    "termination".to_string(),
                    serde_json::to_value(&ended.termination)
                        .map_err(|error| rejected(error.to_string()))?,
                ),
            ]
            .into(),
            CRITICAL_JOURNAL_ACKNOWLEDGEMENT_TIMEOUT,
        )
        .await?;
        Ok(())
    }

    pub async fn append_run_started_event(
        &self,
        started: &car_proto::RunStarted,
    ) -> Result<(), String> {
        self.require_run_journal_binding(&started.run_id).await?;
        if started.client_id.as_deref() != Some(self.client_id.as_str()) {
            return Err("RunStarted client_id does not match the live ClientSession".into());
        }
        let mut data = HashMap::from([
            (
                "agent_id".to_string(),
                Value::from(started.agent_id.clone()),
            ),
            ("intent".to_string(), Value::from(started.intent.clone())),
            (
                "started_at".to_string(),
                serde_json::to_value(started.started_at).unwrap_or(Value::Null),
            ),
        ]);
        if let Some(description) = &started.outcome_description {
            data.insert(
                "outcome_description".to_string(),
                Value::from(description.clone()),
            );
        }
        self.runtime
            .event_log_handle()
            .lock()
            .await
            .append_critical_async(
                car_eventlog::EventKind::RunStarted,
                None,
                None,
                data,
                CRITICAL_JOURNAL_ACKNOWLEDGEMENT_TIMEOUT,
            )
            .await
            .map_err(|error| error.to_string())?;
        Ok(())
    }
}

/// Builder for constructing a [`ServerState`] with embedder-supplied
/// dependencies. Embedders (e.g. `tokhn-daemon`) use this to inject
/// their own memgine handle and other shared infrastructure; the
/// Approval-gate policy for high-risk WS methods.
///
/// Every method in `methods` must be acknowledged via
/// `host.resolve_approval` before the dispatcher will route the
/// request to its handler. The dispatcher waits up to `timeout` for
/// a resolution; on timeout (or any non-`approve` resolution) the
/// request fails with JSON-RPC error `-32003`.
///
/// Default: gate enabled, the host-automation surface
/// (`automation.run_applescript`, `automation.run_powershell`,
/// `automation.shortcuts.run`, `messages.send`, `mail.send`, `vision.ocr`),
/// 60-second timeout.
/// `car-server --no-approvals` (or embedders calling
/// [`ServerStateConfig::with_approval_gate`] with `enabled=false`)
/// turns it off — only appropriate when no untrusted caller can
/// reach the WS port.
#[derive(Debug, Clone)]
pub struct ApprovalGate {
    /// Master switch. When `false`, every method dispatches without
    /// raising an approval — the pre-2026-05 behaviour.
    pub enabled: bool,
    /// Methods that require approval. Match is by exact method-name
    /// string against the JSON-RPC `method` field.
    pub methods: std::collections::HashSet<String>,
    /// How long to wait for the user to resolve the approval before
    /// timing out and surfacing an error to the caller.
    pub timeout: std::time::Duration,
}

impl Default for ApprovalGate {
    fn default() -> Self {
        let methods = [
            "automation.run_applescript",
            "automation.run_powershell",
            "automation.shortcuts.run",
            "messages.send",
            "mail.send",
            "vision.ocr",
        ]
        .iter()
        .map(|s| s.to_string())
        .collect();
        Self {
            enabled: true,
            methods,
            timeout: std::time::Duration::from_secs(60),
        }
    }
}

impl ApprovalGate {
    /// Disable the gate entirely. Equivalent to passing
    /// `car-server --no-approvals`. Only appropriate when no
    /// untrusted caller can reach the WS port.
    pub fn disabled() -> Self {
        Self {
            enabled: false,
            methods: std::collections::HashSet::new(),
            timeout: std::time::Duration::from_secs(60),
        }
    }

    /// `true` if this method must be acknowledged before dispatch.
    pub fn requires_approval(&self, method: &str) -> bool {
        self.enabled && self.methods.contains(method)
    }
}

/// standalone `car-server` binary uses [`ServerState::standalone`]
/// which calls `with_config` under the hood.
pub struct ServerStateConfig {
    pub journal_dir: PathBuf,
    /// Optional pre-constructed memgine engine. When `None`, each
    /// `create_session` call builds a fresh engine; embedders that want
    /// to share a single engine across sessions can supply a clone of
    /// their `Arc<Mutex<MemgineEngine>>` here.
    pub shared_memgine: Option<Arc<Mutex<car_memgine::MemgineEngine>>>,
    /// Optional pre-constructed inference engine.
    pub inference: Option<Arc<car_inference::InferenceEngine>>,
    /// Optional embedder-supplied A2A runtime. Used by the in-core
    /// `A2aDispatcher` to execute peer-driven proposals. When `None`,
    /// the dispatcher uses a fresh `Runtime` with `register_agent_basics`
    /// — peer agents see CAR's built-in tools and nothing else,
    /// matching the behaviour of the standalone `start_a2a_listener`.
    pub a2a_runtime: Option<Arc<car_engine::Runtime>>,
    /// Optional embedder-supplied A2A task store. When `None`,
    /// defaults to `InMemoryTaskStore`. tokhn-style embedders that
    /// want a polling-friendly persistent store plug it in here.
    pub a2a_store: Option<Arc<dyn car_a2a::TaskStore>>,
    /// Optional embedder-supplied agent card factory. When `None`,
    /// the dispatcher serves a card built from the A2A runtime's
    /// tool schemas at construction time, advertising its public URL
    /// as `ws://127.0.0.1:9100/` (the WS surface the dispatcher itself
    /// is reachable on).
    pub a2a_card_source: Option<Arc<car_a2a::AgentCardSource>>,
    /// Approval-gate policy. When `None`, the dispatcher uses
    /// [`ApprovalGate::default`] (gate ON, the macOS-automation
    /// surface gated, 60s timeout). Pass
    /// [`ApprovalGate::disabled`] to opt out — only appropriate
    /// when no untrusted caller can reach the WS port.
    pub approval_gate: Option<ApprovalGate>,
    /// Journal path for the daemon-wide shared HITL approval ledger
    /// ([`ServerState::approval_ledger`]). `None` = the default
    /// `~/.car/approvals.jsonl`. Tests and embedders that must not touch the
    /// user's real home point this at their own path.
    pub approval_journal: Option<PathBuf>,
    /// Directory for the daemon-wide trajectory store
    /// ([`ServerState::trajectory_store`]). `None` = the default
    /// `~/.car/trajectories/`. Tests and embedders that must not touch the
    /// user's real home point this at their own path, same as
    /// `approval_journal`.
    pub trajectory_dir: Option<PathBuf>,
    /// Deterministic durability fault seam used by direct lifecycle tests.
    pub run_store_failures: Option<crate::run_store::RunStoreFailureInjector>,
    /// Deterministic blocking-read seam used by run-liveness tests.
    #[doc(hidden)]
    pub run_store_summary_read_gate: Option<crate::run_store::RunStoreSummaryReadGate>,
    /// Test-only scheduling seam around durable run-summary publication.
    #[doc(hidden)]
    pub run_store_summary_write_gate: Option<crate::run_store::RunStoreSummaryWriteGate>,
    /// Deterministic blocking append seam used by run-liveness tests.
    #[doc(hidden)]
    pub run_store_append_gate: Option<crate::run_store::RunStoreAppendGate>,
    #[doc(hidden)]
    pub run_completion_fence_gate: Option<RunCompletionFenceGate>,
    /// Deterministic blocking durable-lookup seam used by run-liveness tests.
    #[doc(hidden)]
    pub run_store_lookup_gate: Option<crate::run_store::RunStoreLookupGate>,
    /// Deterministic first-use directory-entry durability seam for RunStore.
    pub run_store_private_path_failures: Option<car_secrets::PrivatePathDurabilityFailureInjector>,
    /// Deterministic critical-journal fault seam used by direct lifecycle tests.
    pub journal_failures: Option<car_eventlog::JournalFailureInjector>,
    /// Watch-only self-heal ledger. The standalone daemon sets this to
    /// `<CAR_HOME>/selfheal/detections.jsonl`; embedders/tests can isolate it.
    pub selfheal_ledger: Option<PathBuf>,
    /// Detection cadence. This changes only how often the same watch-only tick
    /// runs; it never selects a different implementation or enables filing.
    pub selfheal_interval_secs: u64,
    /// Deterministic evidence seam for integration tests and embedders.
    #[doc(hidden)]
    pub selfheal_evidence: Option<crate::selfheal::SelfhealEvidence>,
    /// Bounded source-checkout probe. Embedders fail closed unless they supply
    /// candidates; the standalone daemon derives this from process-local paths.
    pub selfheal_source_probe: crate::selfheal::SelfhealSourceProbe,
    /// Test-only override for the startup reconciliation acknowledgement bound.
    #[doc(hidden)]
    pub startup_reconciliation_acknowledgement_timeout: std::time::Duration,
    /// Test-only override for the bounded in-process resume lease.
    #[doc(hidden)]
    pub run_resume_lease: std::time::Duration,
}

impl ServerStateConfig {
    /// Minimal config suitable for the standalone car-server binary:
    /// only the journal dir is required; everything else is lazily
    /// constructed at first use.
    pub fn new(journal_dir: PathBuf) -> Self {
        Self {
            journal_dir,
            shared_memgine: None,
            inference: None,
            a2a_runtime: None,
            a2a_store: None,
            a2a_card_source: None,
            approval_gate: None,
            approval_journal: None,
            trajectory_dir: None,
            run_store_failures: None,
            run_store_summary_read_gate: None,
            run_store_summary_write_gate: None,
            run_store_append_gate: None,
            run_completion_fence_gate: None,
            run_store_lookup_gate: None,
            run_store_private_path_failures: None,
            journal_failures: None,
            selfheal_ledger: None,
            selfheal_interval_secs: crate::selfheal::DEFAULT_SELFHEAL_INTERVAL_SECS,
            selfheal_evidence: None,
            selfheal_source_probe: crate::selfheal::SelfhealSourceProbe::empty(),
            startup_reconciliation_acknowledgement_timeout:
                STARTUP_RECONCILIATION_ACKNOWLEDGEMENT_TIMEOUT,
            run_resume_lease: RUN_RESUME_LEASE,
        }
    }

    pub fn with_shared_memgine(mut self, engine: Arc<Mutex<car_memgine::MemgineEngine>>) -> Self {
        self.shared_memgine = Some(engine);
        self
    }

    /// Override the shared approval-ledger journal path (default
    /// `~/.car/approvals.jsonl`).
    pub fn with_approval_journal(mut self, path: PathBuf) -> Self {
        self.approval_journal = Some(path);
        self
    }

    pub fn with_trajectory_dir(mut self, dir: PathBuf) -> Self {
        self.trajectory_dir = Some(dir);
        self
    }

    pub fn with_run_store_failures(
        mut self,
        failures: crate::run_store::RunStoreFailureInjector,
    ) -> Self {
        self.run_store_failures = Some(failures);
        self
    }

    #[doc(hidden)]
    pub fn with_run_store_summary_read_gate(
        mut self,
        gate: crate::run_store::RunStoreSummaryReadGate,
    ) -> Self {
        self.run_store_summary_read_gate = Some(gate);
        self
    }

    #[doc(hidden)]
    pub fn with_run_store_summary_write_gate(
        mut self,
        gate: crate::run_store::RunStoreSummaryWriteGate,
    ) -> Self {
        self.run_store_summary_write_gate = Some(gate);
        self
    }

    #[doc(hidden)]
    pub fn with_run_store_append_gate(
        mut self,
        gate: crate::run_store::RunStoreAppendGate,
    ) -> Self {
        self.run_store_append_gate = Some(gate);
        self
    }

    #[doc(hidden)]
    pub fn with_run_completion_fence_gate(mut self, gate: RunCompletionFenceGate) -> Self {
        self.run_completion_fence_gate = Some(gate);
        self
    }

    #[doc(hidden)]
    pub fn with_run_resume_lease(mut self, lease: std::time::Duration) -> Self {
        self.run_resume_lease = lease;
        self
    }

    #[doc(hidden)]
    pub fn with_run_store_lookup_gate(
        mut self,
        gate: crate::run_store::RunStoreLookupGate,
    ) -> Self {
        self.run_store_lookup_gate = Some(gate);
        self
    }

    pub fn with_run_store_private_path_failures(
        mut self,
        failures: car_secrets::PrivatePathDurabilityFailureInjector,
    ) -> Self {
        self.run_store_private_path_failures = Some(failures);
        self
    }

    pub fn with_journal_failures(mut self, failures: car_eventlog::JournalFailureInjector) -> Self {
        self.journal_failures = Some(failures);
        self
    }

    pub fn with_selfheal_ledger(mut self, path: PathBuf) -> Self {
        self.selfheal_ledger = Some(path);
        self
    }

    pub fn with_selfheal_interval_secs(mut self, interval_secs: u64) -> Self {
        self.selfheal_interval_secs = interval_secs.max(1);
        self
    }

    #[doc(hidden)]
    pub fn with_selfheal_evidence(mut self, evidence: crate::selfheal::SelfhealEvidence) -> Self {
        self.selfheal_evidence = Some(evidence);
        self
    }

    pub fn with_selfheal_source_probe(
        mut self,
        probe: crate::selfheal::SelfhealSourceProbe,
    ) -> Self {
        self.selfheal_source_probe = probe;
        self
    }

    #[doc(hidden)]
    pub fn with_startup_reconciliation_acknowledgement_timeout(
        mut self,
        timeout: std::time::Duration,
    ) -> Self {
        self.startup_reconciliation_acknowledgement_timeout = timeout;
        self
    }

    pub fn with_inference(mut self, engine: Arc<car_inference::InferenceEngine>) -> Self {
        self.inference = Some(engine);
        self
    }

    /// Plug in an embedder-supplied runtime for the A2A dispatcher.
    /// Use case: tokhn-daemon wants peers to see its OPA preflight
    /// tooling, not just CAR's `register_agent_basics` defaults.
    pub fn with_a2a_runtime(mut self, runtime: Arc<car_engine::Runtime>) -> Self {
        self.a2a_runtime = Some(runtime);
        self
    }

    /// Plug in an embedder-supplied task store for the A2A
    /// dispatcher. Use case: tokhn's polling-friendly persistent
    /// store keyed by their session id.
    pub fn with_a2a_store(mut self, store: Arc<dyn car_a2a::TaskStore>) -> Self {
        self.a2a_store = Some(store);
        self
    }

    /// Plug in an embedder-supplied agent card factory. The factory
    /// is invoked on every `agent/getAuthenticatedExtendedCard`
    /// dispatch, so embedders can reflect runtime tool changes.
    pub fn with_a2a_card_source(mut self, source: Arc<car_a2a::AgentCardSource>) -> Self {
        self.a2a_card_source = Some(source);
        self
    }

    /// Override the approval-gate policy. Pass
    /// [`ApprovalGate::disabled`] to skip the gate entirely (only
    /// appropriate when no untrusted caller can reach the WS port);
    /// pass a customised [`ApprovalGate`] to add or remove methods
    /// or to change the timeout.
    pub fn with_approval_gate(mut self, gate: ApprovalGate) -> Self {
        self.approval_gate = Some(gate);
        self
    }
}

/// The opted-in shared org-scope delivery subsystem + which org it serves. See
/// [`ServerState::org_sync`]. Held behind the same `std::sync::Mutex` as the
/// personal subsystem holder (init is serialized by the oplog's advisory lock);
/// the subsystem itself lives behind a `tokio::sync::Mutex`.
pub struct OrgSyncHolder {
    pub org: String,
    pub subsystem: Arc<tokio::sync::Mutex<crate::sync::SyncSubsystem>>,
}

/// Which subsystem a write routes to. Pure decision, extracted so it is unit-
/// testable without a full daemon.
#[derive(Debug, PartialEq, Eq)]
enum SyncRoute {
    User,
    Org,
}

/// Route by scope: a `Scope::Shared{org}` whose org matches the opted-in
/// `org_holder` goes to the org delivery subsystem; EVERYTHING else — `Personal`,
/// a different org, or no opted-in holder — goes to the personal subsystem. So
/// with `org_holder = None` (org-scope off) every scope routes `User`, byte-
/// identical to before.
fn route_for_scope(scope: &car_sync::Scope, org_holder: Option<&str>) -> SyncRoute {
    if let car_sync::Scope::Shared { org } = scope {
        if org_holder == Some(org.as_str()) {
            return SyncRoute::Org;
        }
    }
    SyncRoute::User
}

/// Global server state shared across all connections.
pub struct ServerState {
    pub journal_dir: PathBuf,
    pub sessions: Mutex<HashMap<String, Arc<ClientSession>>>,
    /// Serializes only session registry insertion/removal with the final
    /// liveness check + owner swap in `runs.resume`. This avoids holding the
    /// global session map while waiting on per-run durability and turns the
    /// replacement-present / old-owner-absent decision into one atomic fence.
    run_resume_liveness: Mutex<()>,
    run_resume_lease: std::time::Duration,
    run_completion_fence_gate: Option<RunCompletionFenceGate>,
    pub inference: std::sync::OnceLock<Arc<car_inference::InferenceEngine>>,
    pub host: Arc<crate::host::HostState>,
    /// When `Some`, `create_session` clones this handle into every new
    /// `ClientSession.memgine` — embedders that want a single shared
    /// memgine across all WS sessions set this. Standalone car-server
    /// leaves it `None`, which gives each session its own engine
    /// (preserving today's behavior).
    pub shared_memgine: Option<Arc<Mutex<car_memgine::MemgineEngine>>>,
    /// Daemon-wide trajectory store, cloned into **every** session's
    /// `Runtime` so execution traces persist to one place.
    ///
    /// Shared rather than per-session on purpose. A trajectory is evidence
    /// about how a tool behaves, and that evidence does not belong to the
    /// connection that happened to produce it — per-session stores would
    /// scatter the history across connections and reset it on every
    /// reconnect, leaving the derived success rates built from a handful of
    /// samples. This is the same reasoning that makes the approval ledger
    /// daemon-wide.
    ///
    /// Until now nothing called `Runtime::with_trajectory_store`, so
    /// `persist_trajectory` returned early on every execution and the store
    /// was dead code: `ToolFeedback::from_trajectories` existed with no data
    /// to read. Wiring it here is what makes the feedback loop real.
    pub trajectory_store: Arc<car_memgine::TrajectoryStore>,
    /// Daemon-wide watch-only self-healing detector + CAR_HOME ledger. Shared
    /// across sessions so `selfheal.run` and the cadence cannot overlap and
    /// every caller reads the same dismissals/detections.
    pub selfheal: Arc<crate::selfheal::SelfhealService>,
    /// Process-wide voice session registry. Each
    /// `voice.transcribe_stream.start` call registers its own per-client
    /// [`WsVoiceEventSink`] so events route back to the originating WS
    /// connection only.
    pub voice_sessions: Arc<car_voice::VoiceSessionRegistry>,
    /// Process-wide meeting registry. Meeting ids are global; each
    /// meeting binds to the originating client's WS for upstream
    /// events but persists transcripts to the resolved
    /// `.car/meetings/<id>/` regardless of which client started it.
    pub meetings: Arc<car_meeting::MeetingRegistry>,
    /// Process-wide A2UI surface store. Agent-produced surfaces are
    /// visible to every host UI subscriber, independent of the
    /// WebSocket session that applied the update.
    pub a2ui: car_a2ui::A2uiSurfaceStore,
    /// In-process UI-improvement agent. Invoked from
    /// `handle_a2ui_render_report` with each inbound report; returned
    /// `Decision::Patch` envelopes are applied via the standard
    /// `apply_a2ui_envelope` path so all subscribers see the patch.
    /// `Arc` so the agent's interior `DashMap` state survives across
    /// handler calls even when `ServerState` is cheap-cloned.
    pub ui_agent: Arc<car_ui_agent::UIImprovementAgent>,
    /// Per-surface oscillation detector for the UI-improvement
    /// loop. Sits between the agent's `Decision::Patch` and the
    /// apply path so A→B→A patch cycles get cooled down without
    /// the agent itself having to track history. neo's review:
    /// "controllers use workqueue backoff; reconcilers stay
    /// stateless."
    pub ui_agent_oscillation: Arc<crate::ui_agent_loop::OscillationDetector>,
    /// Per-surface iteration budget. Backstop against runaway
    /// loops the oscillation detector misses — caps total agent-
    /// driven patches per surface at `DEFAULT_MAX_ITERATIONS`.
    pub ui_agent_budget: Arc<crate::ui_agent_loop::IterationBudget>,
    /// Process-wide concurrency gate for inference RPC handlers. Sized
    /// from host RAM at startup, overridable via
    /// [`crate::admission::ENV_MAX_CONCURRENT`]. Without this, N
    /// concurrent users multiply KV-cache and activation memory and
    /// take the host out (#114-adjacent: filed alongside the daemon
    /// always-on rework). The semaphore lives on `ServerState` so it
    /// is shared across every WebSocket session in the same process.
    pub admission: Arc<crate::admission::InferenceAdmission>,
    /// Server-side A2A continuation auth keyed by A2UI surface id.
    /// Kept out of `A2uiSurface.owner` so host renderers never see
    /// bearer/API-key material.
    pub a2ui_route_auth: Mutex<HashMap<String, A2aRouteAuth>>,
    /// Lifecycle-managed agents — declarative manifest at
    /// `~/.car/agents.json` driving spawn/restart/stop. Closes
    /// Parslee-ai/car-releases#27. Lazy-initialized so embedders that
    /// don't want process supervision don't pay the disk-touch cost
    /// at server start.
    pub supervisor: std::sync::OnceLock<Arc<car_registry::supervisor::Supervisor>>,
    /// Declarative (in-daemon) agent registry — a parallel store to the
    /// supervisor for agents the coder→agent loop builds. Lazy-initialized on
    /// first use (`~/.car/declagents.json`).
    pub declagents: std::sync::OnceLock<Arc<car_registry::declarative::DeclRegistry>>,
    /// Routing learning state — per-agent success stats + agent→agent edge
    /// weights for capability-similarity routing (`~/.car/routing.json`).
    /// Lazy-initialized on first use, sibling to [`Self::declagents`].
    pub routing: std::sync::OnceLock<Arc<car_registry::routing::RoutingStore>>,
    /// Multi-device sync + execution-lease subsystem (B6). Lazily opened on
    /// first `sync.*`/`lease.*` contact, rooted at `<journal_dir>/sync/` — a
    /// daemon-held [`car_sync::SyncSession`] over an `FsRelay` (the single-user
    /// two-device loopback works out of the box) plus an in-process
    /// linearizable [`car_sync::InMemoryLeaseCoordinator`]. **One subsystem per
    /// daemon = one device in the sync fleet.** The `std::sync::Mutex<Option>`
    /// serializes the fallible first-open (the oplog journal holds an exclusive
    /// advisory lock, so a double-open would fail) without holding the lock
    /// across the subsystem's async work — the subsystem itself lives behind a
    /// `tokio::sync::Mutex` inside the `Arc`. A distributed relay + a
    /// cross-daemon linearizable `LeaseCoordinator` backend is the documented
    /// B6 follow-up; the traits are its contract.
    pub sync: std::sync::Mutex<Option<Arc<tokio::sync::Mutex<crate::sync::SyncSubsystem>>>>,
    /// The SHARED org-scope delivery subsystem — a SECOND device oplog whose relay
    /// scope is `org:{orgId}`, so `Scope::Shared{org}` ops converge across an org's
    /// members. `None` unless org-scope is opted in (the same gated
    /// `PARSLEE_SYNC_ORG_SCOPE` branch that resolves the org key builds it);
    /// populated by `open_sync_subsystem` alongside the personal subsystem. Only
    /// the `Scope::Shared` write sites + `pump` route to it — the personal path
    /// (`sync_subsystem()` + all readers) is untouched. Slice 8a.
    pub org_sync: std::sync::Mutex<Option<OrgSyncHolder>>,
    /// Manifest path this daemon is *observing* but does NOT own.
    /// Set by `car-server` when boot-time supervisor construction
    /// fails with [`car_registry::supervisor::SupervisorError::AlreadyRunning`]
    /// — another car-server process on the host holds the exclusive
    /// lock on this manifest. In that state, `supervisor()` returns a
    /// clear "observe-only" error so mutation handlers refuse
    /// (preventing the duplicate-spawn bug from
    /// Parslee-ai/car-releases#44), while read-only handlers
    /// (`agents.list`, `agents.health`) fall back to
    /// [`car_registry::supervisor::Supervisor::list_from_manifest`] /
    /// [`car_registry::supervisor::Supervisor::health_from_manifest`]
    /// so operators can still inspect what the primary daemon is
    /// supervising.
    pub observer_manifest_path: std::sync::OnceLock<PathBuf>,
    /// In-core A2A dispatcher — embedders that consume `car-server-core`
    /// get A2A reachability "for free" without standing up a separate
    /// HTTP listener. Closes Parslee-ai/car-releases#28. Lazy-init so
    /// the embedder can override the runtime / task store / agent card
    /// via [`ServerStateConfig::with_a2a_runtime`] etc. before the
    /// first dispatch.
    pub a2a_dispatcher: std::sync::OnceLock<Arc<car_a2a::A2aDispatcher>>,
    /// WS clients subscribed to A2UI envelope events. After every
    /// successful `a2ui.apply` / `a2ui.ingest`, the resulting
    /// `A2uiApplyResult` is broadcast to every subscriber as an
    /// `a2ui.event` JSON-RPC notification. Closes
    /// Parslee-ai/car-releases#29. Subscribers register via the
    /// `a2ui/subscribe` method and are auto-cleaned on WS disconnect.
    pub a2ui_subscribers: Mutex<HashMap<String, Arc<WsChannel>>>,
    /// Per-launch auth token. When `Some`, the WS dispatcher rejects
    /// non-auth methods on unauthenticated sessions until the client
    /// calls `session.auth` with the matching value. When `None`,
    /// auth is disabled and every connection works as before. Set
    /// at startup by `car-server` unless `--no-auth` is passed
    /// (default flipped 2026-05); embedders that want to enable
    /// auth call [`ServerState::install_auth_token`]. Closes
    /// Parslee-ai/car-releases#32.
    /// Shared MCP tool executor backing remote connectors. One per
    /// process: every WS session's runtime executor is a
    /// [`car_engine::McpToolExecutor::share_with_fallback`] view over
    /// this, so connector tools route here while non-connector tools
    /// fall back to the per-session [`WsToolExecutor`]. Connectors
    /// (their live sessions + routes) live on this shared instance, so
    /// a connector enabled in one WS session is reachable from all.
    pub mcp_executor: Arc<car_engine::McpToolExecutor>,
    /// Remote MCP connector manager (`~/.car/connectors.json`).
    /// Lazy-initialized over [`mcp_executor`](Self::mcp_executor) so
    /// embedders that never touch connectors pay no disk cost. See
    /// [`ServerState::connectors`].
    pub connectors: std::sync::OnceLock<Arc<car_connectors::ConnectorManager>>,
    /// One-shot guard so persisted connectors are loaded + dialed at
    /// most once per process (see [`ServerState::ensure_connectors_loaded`]).
    connectors_loaded: std::sync::atomic::AtomicBool,
    /// Runtime approval-transport channel supervisor (Units 1/2/3). Lazy-
    /// initialized at daemon boot by `spawn_channel_pollers`, which records the
    /// channels it spawned at startup. The host-gated `messaging.config.set`
    /// handler reaches it via `ServerState` to spawn a channel's watcher the
    /// instant the user enables it — no restart (U1). Also holds the per-channel
    /// liveness `messaging.status` reads (U2) and the send path writes (U3).
    /// `None` for embedders that never boot the channel pollers.
    pub channel_supervisor: std::sync::OnceLock<Arc<crate::channel_supervisor::ChannelSupervisor>>,
    /// Daemon-owned operations which must outlive the WebSocket request that
    /// started them. The tasks hold no [`ClientSession`] or [`WsChannel`]; only
    /// their oneshot response waiters are connection-scoped.
    ///
    /// Two kinds of work live here. Credential-backed auth operations (attempt
    /// completion and ordinary auth mutations), so a socket close can release
    /// its sink immediately without cancelling a coordinator/keychain operation
    /// halfway through its durable result. And `coder.start`, which registers a
    /// session and provisions a worktree *before* a multi-minute contract
    /// derivation: once that side effect exists on disk, the run's lifetime
    /// must not be owned by the board that asked for it.
    pub durable_tasks: Mutex<tokio::task::JoinSet<String>>,
    /// Per-daemon incarnation token persisted into redeeming auth leases.
    /// Awaiting-callback reservations intentionally have no daemon owner and
    /// survive a restart; a redeeming lease owned by a different incarnation
    /// is an orphan and is terminally failed by completion-status reconciliation.
    pub auth_completion_owner_id: String,
    pub auth_token: std::sync::OnceLock<String>,
    /// Per-launch **host** token — a credential distinct from
    /// `auth_token`, granting the host-management role (host-class
    /// reads like cross-agent run traces). Critically it is **never**
    /// served over `GET /auth-token`: a session becomes host-role only
    /// by presenting this via `session.auth { host_token }`, and the
    /// only way to obtain it is reading the `0600` `host-token` file,
    /// which a different local user cannot. This is what stops any
    /// authenticated local client from self-elevating to host and
    /// reading every agent's run traces (Parslee-ai/car#254). When
    /// `None`, host-role can't be granted (no host reads).
    pub host_token: std::sync::OnceLock<String>,
    /// Mobile Parslee Core runtime URL this daemon can hand to consumer host apps.
    /// Set by `car-server` at startup from the daemon bind address or an
    /// explicit public URL override. The token remains the per-launch auth
    /// token; this field only names the WebSocket endpoint.
    pub mobile_runtime_url: std::sync::OnceLock<String>,
    /// Explicit public mobile runtime URL to register after an authorized
    /// credential action succeeds. Absent for the default loopback URL, which
    /// is discoverable locally but must not be advertised to Parslee cloud.
    pub mobile_registration_url: std::sync::OnceLock<String>,
    /// Parslee cloud identity activated by an explicit credential-backed
    /// request after `car auth login` has been completed.
    pub parslee_session: std::sync::OnceLock<crate::parslee_auth::ParsleeSession>,
    /// `agent_id -> client_id` map of currently-attached lifecycle
    /// agents (#169). Populated by the `session.auth` handler when a
    /// supervised child presents its `agent_id` + per-agent token;
    /// drained on disconnect by `remove_session`. Single-claim:
    /// a second connection presenting the same `agent_id` is
    /// rejected so the daemon-side per-agent state stays unambiguous.
    pub attached_agents: Mutex<HashMap<String, String>>,
    /// `agent_id -> persistent memgine` map (#170). Lazy-loaded on
    /// first connection per id from `~/.car/memory/agents/<id>.jsonl`,
    /// retained across daemon restart, surviving any single
    /// disconnect/reconnect of the supervised child. Connections
    /// that auth without an `agent_id` (browser, host, ad-hoc CLI)
    /// keep the per-WS ephemeral memgine on `ClientSession.memgine`
    /// — no behaviour change.
    pub agent_memgines: Mutex<HashMap<String, Arc<Mutex<car_memgine::MemgineEngine>>>>,
    /// Daemon-owned memgines keyed by a host-declared **memory namespace**
    /// (`session.auth { memory_namespace }`), lazily loaded from
    /// `~/.car/memory/memory-namespaces/<encoded-ns>.json` — the filename is a
    /// percent-encoding of the namespace, injective so two namespaces can never
    /// share a snapshot file (#891).
    ///
    /// A separate axis from `agent_memgines`, deliberately: one agent may work
    /// across several namespaces, and two hosts may share a namespace without
    /// sharing an identity (Parslee-ai/car-releases#79). A host that binds none
    /// keeps today's behaviour — the daemon's shared graph — so the MCP-to-WS
    /// shared knowledge base is untouched.
    pub namespace_memgines: Mutex<HashMap<String, Arc<Mutex<car_memgine::MemgineEngine>>>>,
    /// Live coder sessions keyed by `session_id` (built-in coding agent).
    /// Process-wide so a session outlives the WS connection that started
    /// it; terminal sessions stay listed for `coder.get`/`coder.list`
    /// history until daemon restart (snapshots persist under
    /// `~/.car/coder/`).
    pub coder_sessions: Mutex<crate::coder::rpc::CoderSessionMap>,
    /// Live `coder.event` subscribers keyed by `(session_id, client_id)`
    /// — explicit fanout, same shape as `run_subscribers`. Lock order:
    /// the session's event buffer → this map; never the reverse.
    pub coder_subscribers: Mutex<HashMap<(String, String), Arc<WsChannel>>>,
    /// `coder.watch` board subscribers keyed by `client_id` — one
    /// subscription per connection covering EVERY session, so a board learns
    /// about runs started by any other client (`car code`, CarHost, milo)
    /// without polling. Fanned to as `coder.session_changed`. Lock order:
    /// a session's event buffer → `coder_subscribers` → this map; never the
    /// reverse.
    ///
    /// The value carries a **registration generation** alongside the channel.
    /// A board re-registers periodically (its `coder.watch` is idempotent and
    /// returns a fresh snapshot), and the fanout sheds a wedged watcher *after*
    /// releasing this lock — so a shed that removed by `client_id` alone would
    /// delete a registration made in that window and silently unwatch a healthy
    /// board. The shed compares the generation and removes only the entry it
    /// actually timed out on.
    pub coder_watchers: Mutex<HashMap<String, (u64, Arc<WsChannel>)>>,
    /// The one long-lived task that fans `coder.session_changed` out to
    /// [`coder_watchers`](Self::coder_watchers), lazily started on the first
    /// notification. Deliberately a single drain rather than a task per event:
    /// a running session emits on every tool call, and spawn-per-event against
    /// a half-open board socket accumulated blocked tasks without bound, each
    /// pinning the socket's write half past teardown (car#209's shape on a new
    /// path). Daemon-owned, so it is correctly absent from any connection's
    /// `conn_tasks`; it exits when this `ServerState` drops.
    pub(crate) coder_watch_notify: std::sync::OnceLock<mpsc::UnboundedSender<String>>,
    /// Live `coder.discuss.*` conversations keyed by `discussion_id`. In-memory
    /// only and deliberately NOT persisted across daemon restarts: a discussion
    /// is a thinking surface, not a run, and resuming one would mean rebuilding
    /// an assistant runtime and a model thread from a snapshot that never
    /// existed.
    pub coder_discussions: Mutex<crate::coder::discuss::DiscussionMap>,
    /// Open-discussion slots, one permit per allowed discussion. A permit is
    /// taken BEFORE `coder.discuss.start` does any of its expensive async work
    /// and lives inside the [`DiscussionEntry`](crate::coder::discuss::DiscussionEntry)
    /// it admitted, so it comes back when the discussion is dropped. Counting
    /// the map instead was a TOCTOU check — the count was read, the lock
    /// released, and two awaits (substrate bind + runtime build) ran before the
    /// insert, so N pipelined starts all passed a cap of N=8.
    pub(crate) coder_discussion_slots: Arc<tokio::sync::Semaphore>,
    /// In-flight `agents.chat` sessions keyed by `session_id`. See
    /// [`ChatSession`] for shape. Populated by `agents.chat`,
    /// cleared on terminal `agent.chat.event` or
    /// `agents.chat.cancel`. Disconnect cleanup happens in
    /// `remove_session` — any in-flight session bound to either the
    /// disconnecting host or agent client is dropped so subsequent
    /// stray notifications from a respawned agent fall on the floor
    /// rather than racing into a stale stream.
    pub chat_sessions: Mutex<HashMap<String, ChatSession>>,
    /// Per-recipient channel guards for peer messaging (`agents.message`).
    ///
    /// Keyed by recipient agent id. Bounds the *channel* rather than the
    /// sender's authority: size, identical repeats, per-sender rate, and queue
    /// depth. Two agents answering each other form a loop that no policy rule
    /// catches, because neither is misbehaving — this is what terminates it.
    ///
    /// Entries are dropped in `remove_session` when the agent detaches, so a
    /// restarted agent starts with a clean budget rather than inheriting the
    /// rate history of the process that used to own its name.
    pub peer_guards: Mutex<HashMap<String, car_peers::DeliveryGuard>>,
    /// Peer messages held for operator approval, oldest first.
    ///
    /// A message is held when the sender's `read_only` posture is
    /// `require_approval`. Bounded at [`car_peers::HOLD_CAP`]; past that the
    /// oldest is dropped, so a stream of held messages cannot grow without
    /// limit. Kept flat rather than per-recipient because approval is an
    /// operator action over the whole queue, not a per-agent one.
    ///
    /// Deliberately in memory only: a held message is a live decision awaiting a
    /// human, and one that outlived a daemon restart would be delivered into a
    /// world that had moved on.
    pub held_peer_messages: Mutex<std::collections::VecDeque<crate::peers::HeldPeerMessage>>,
    /// Live view of CAR daemons discovered on the local network.
    ///
    /// `None` when mDNS could not start (no multicast, a locked-down sandbox) —
    /// LAN discovery is then simply absent rather than the daemon failing to
    /// boot. A network with no peers and a network CAR cannot browse look the
    /// same to a caller, which is why the status surface reports which it is.
    pub lan_discovery: std::sync::Mutex<Option<car_a2a::lan::LanDirectory>>,
    /// This daemon's ed25519 peer identity, and the keys it accepts.
    ///
    /// The identity signs outbound cross-host requests; the trust set decides
    /// which inbound signers are CAR. Both are `None` until the A2A surface
    /// starts, because neither is meaningful without a network surface to
    /// authenticate.
    pub peer_identity: std::sync::Mutex<Option<Arc<car_a2a::peer_auth::PeerIdentity>>>,
    pub peer_trust: car_a2a::peer_auth::PeerTrust,
    /// In-process collectors for `agent.chat` streams that have **no** host UI
    /// to forward to — currently the A2A conversational bridge, which reverse-
    /// calls `agent.chat` on a host session and aggregates the streamed deltas
    /// into a single reply. Keyed by `session_id`. When a collector exists for a
    /// session, `try_forward_agent_chat_event` feeds chunks here instead of
    /// forwarding to a host channel. The collecting task owns the entry's
    /// lifetime (inserts before the reverse-call, removes when done/timed out).
    pub chat_collectors: Mutex<HashMap<String, ChatCollector>>,
    /// Standing deterministic chat goals keyed by `session_id`. This is a
    /// host-facing status/control registry, separate from ephemeral
    /// `chat_sessions` routing so a goal can be inspected after a turn finishes.
    pub chat_goals: Mutex<HashMap<String, ChatGoalState>>,
    /// Agent runs keyed by `run_id` (agent run tracing, U1). Process-
    /// wide (not per-session) so a run's record outlives the WS
    /// connection that produced it — the durable, connection-
    /// independent boundary `client_id` cannot be (R1). Populated by
    /// `runs.start`, made terminal by `runs.complete`, and swept to
    /// `Incomplete` on a mid-run disconnect past the grace window
    /// (R5). U2/U3 build the per-turn recorder and disk store on top
    /// of this registry.
    pub runs: Mutex<HashMap<String, RunMeta>>,
    /// Per-run serialization for trace append/fsync and terminal persistence.
    /// Filesystem work never holds the global [`runs`](Self::runs) lock.
    run_durability_locks: Mutex<HashMap<String, std::sync::Weak<Mutex<()>>>>,
    /// Live `runs.trace.event` subscribers keyed by `(run_id,
    /// host_client_id)` (agent run tracing, U4). Each value is a
    /// [`crate::host::RunTraceSubscriber`] — the producer side of a
    /// bounded channel whose dedicated drain task writes frames to that
    /// connection's WS. Two CarHost windows on one run register two
    /// distinct entries (explicit fanout — the built-in notification
    /// registry is single-subscriber-per-method).
    ///
    /// **Lock contract (invariant #1):** `runs.subscribe` snapshots the
    /// run's turns AND inserts the subscriber while holding the
    /// [`runs`](Self::runs) lock; the recorder (`record_run_turns`),
    /// `start_run`, `complete_run`, and `mark_run_incomplete` append to
    /// `runs` and push to `run_subscribers` while holding the SAME
    /// `runs` lock — so snapshot/register and append/notify are
    /// serialized. No turn appended in the snapshot/register window is
    /// dropped (gap) or double-delivered (dup). The lock order is always
    /// `runs` → `run_subscribers`; never the reverse.
    pub run_subscribers: Mutex<HashMap<(String, String), crate::host::RunTraceSubscriber>>,
    /// Every browser the drawer can reach (`browser.view.*`), keyed by
    /// conversation/agent-session — plus the one standing user session every
    /// conversation without an agent-attached browser shares. Owns the
    /// per-view snapshot/cursor/subscriber fanout; see
    /// [`crate::browser_view`].
    pub browser_views: Arc<crate::browser_view::BrowserViewRegistry>,
    /// Disk-backed run-trace store (agent run tracing, U3). Source of
    /// truth for REPLAY (U5) — persists each run's `RunStarted`, turns,
    /// and terminal record as JSONL under `~/.car/runs/{agent_id}/` so a
    /// run survives daemon restarts (R4), bounded by retention/GC (R6)
    /// and protected at rest with `0600`/`0700` perms + backup exclusion
    /// (R14). The in-memory [`runs`](Self::runs) buffer stays the source
    /// for the LIVE stream (U4); this store mirrors what was recorded.
    /// Derived from `journal_dir` at construction (sibling `runs/`).
    pub run_store: crate::run_store::RunStore,
    journal_failures: Option<car_eventlog::JournalFailureInjector>,
    /// Bound MCP HTTP-streamable URL (e.g.
    /// `"http://127.0.0.1:9102/mcp"`) — `car-server` installs this
    /// after binding the listener. Used by the
    /// `agents.invoke_external` handler to default
    /// `InvokeOptions.mcp_endpoint` so external agents
    /// (Claude Code today) load the daemon's CAR namespace via
    /// `--mcp-config` automatically. `None` when MCP isn't bound
    /// (e.g. `--mcp-bind disabled`).
    pub mcp_url: std::sync::OnceLock<String>,
    /// Registry of connected MCP SSE sessions. Populated alongside
    /// [`mcp_url`] when `car-server` boots the MCP listener. Public
    /// so handlers can call `crate::mcp::push_to_session` to send
    /// server-initiated requests to a specific MCP-connected
    /// client (MCP-3 foundation; MCP-3b will wire host-owned tool
    /// dispatch through this).
    pub mcp_sessions: std::sync::OnceLock<Arc<crate::mcp::SessionMap>>,
    /// Approval gate for high-risk WS methods (audit 2026-05). The
    /// gate intercepts `automation.run_applescript`,
    /// `automation.shortcuts.run`, `messages.send`, `mail.send`, and
    /// `vision.ocr` before they dispatch, raises a
    /// `host.create_approval` for the user to act on, and waits
    /// (with a timeout) for `host.resolve_approval`. Approve →
    /// dispatch continues; deny / timeout → JSON-RPC error code
    /// `-32003`. The set of gated methods and the wait timeout are
    /// embedder-overridable via
    /// [`ServerStateConfig::with_approval_gate`].
    pub approval_gate: ApprovalGate,
    /// Out-of-process supervision (proposal item 5). Holds the subscribed
    /// supervisors and the intents parked on their verdicts; the matching
    /// [`crate::supervision::SupervisionGate`] is registered as an
    /// `AdmissionGate` on every session runtime alongside
    /// `StaticVerificationGate`.
    ///
    /// Daemon-wide rather than per-session, deliberately: a supervisor connects
    /// on its OWN WebSocket and supervises proposals submitted on OTHER
    /// sessions. A per-session registry would only ever see the supervisor's
    /// own (empty) traffic.
    ///
    /// Costs nothing until someone subscribes — the gate returns `Allow`
    /// without building an intent while the subscriber set is empty.
    pub supervision: Arc<crate::supervision::SupervisionRegistry>,
    /// The daemon-wide shared HITL approval ledger (kernel review C1) —
    /// journal-backed at `~/.car/approvals.jsonl` (override via
    /// [`ServerStateConfig::with_approval_journal`]). ONE ledger for all
    /// sessions: `permission.approve`/`reject` on any connection records
    /// here, and every ledger consumer (`permission.evaluate`/`pending`,
    /// `evolution.run`, `cascade.run`, `skill.enforce_deployment`/
    /// `ingest_governed`) reads here — so an approval granted by a host
    /// connection is visible to the agent connection that surfaced it and
    /// survives daemon restart. Per-session gates keep only tier/classifier
    /// state (see [`ClientSession::permission_gate`]). Falls back to an
    /// in-memory ledger (loudly warned, approvals NOT restart-durable) only
    /// when the journal is unopenable.
    pub approval_ledger: Arc<tokio::sync::RwLock<car_policy::ApprovalLedger>>,
    /// The in-process harness evaluator, when the binary installed one — the
    /// thing that lets `evolution.run` grade a harness candidate ITSELF
    /// instead of waiting for an operator to run `car-bench-harness` twice and
    /// hand both files back in.
    ///
    /// `None` on a build that did not install one (every embedder, and any
    /// binary other than `car-server`): `evolution.run` with `harness_measure`
    /// then ERRS rather than quietly falling back to HITL, because an opt-in
    /// that silently does nothing would report an unattended cycle that never
    /// measured anything.
    ///
    /// A `std::sync::RwLock` on purpose — no caller holds it across an await;
    /// they clone the `Arc` out and drop the guard.
    pub harness_measurer: std::sync::RwLock<Option<Arc<dyn crate::evolution::HarnessMeasurer>>>,
    /// A2A-runtime / store / card factory carried over from the
    /// embedder's [`ServerStateConfig`]. Consumed lazily on first
    /// `a2a_dispatcher()` call so embedders can construct
    /// `ServerState` without paying the runtime spin-up cost when
    /// they don't actually use the A2A surface.
    pub(crate) a2a_runtime: std::sync::Mutex<Option<Arc<car_engine::Runtime>>>,
    pub(crate) a2a_store: std::sync::Mutex<Option<Arc<dyn car_a2a::TaskStore>>>,
    pub(crate) a2a_card_source: std::sync::Mutex<Option<Arc<car_a2a::AgentCardSource>>>,
}

/// Default location of the shared approval-ledger journal: `approvals.jsonl`
/// directly under the CAR state root, so it follows `CAR_HOME` rather than
/// pinning itself to `~/.car` (which is still where it lands when `CAR_HOME` is
/// unset — HOME, or USERPROFILE on Windows). Creates the root best-effort;
/// `None` when neither `CAR_HOME` nor a home directory is resolvable.
fn default_approval_journal_path() -> Option<PathBuf> {
    let dir = car_home::root()?;
    let _ = std::fs::create_dir_all(&dir);
    Some(dir.join("approvals.jsonl"))
}

/// The daemon's state root: `CAR_HOME` when set, otherwise `~/.car` (HOME, or
/// USERPROFILE on Windows).
///
/// One convention, used by [`default_approval_journal_path`], the `messaging.*`
/// handlers and now the project policy loader — a second one would mean an
/// operator's rules living somewhere their pairing config does not. Going
/// through `car_home` is what keeps that true once an operator relocates the
/// root: everything moves together or nothing does.
fn car_home_dir() -> Option<PathBuf> {
    car_home::root()
}

/// Register the declarative rules in `<car_dir>/policies/*.toml` on `runtime`.
///
/// **A malformed file is fatal, on purpose. Do not downgrade this to a
/// warning.** The loader's own words (`car_policy::load_policy_dir`): "a
/// security rule that fails to parse must surface, never be silently skipped."
/// A rule file exists to *stop* something; a typo that turns "never message
/// anyone outside the allowlist" into a no-op fails open, and it fails open
/// silently, at exactly the moment the operator believes they are protected. A
/// daemon that refuses to open the session is loud, immediate, and trivially
/// fixed; a daemon that logs a warning nobody reads is neither. The error text
/// names the offending file (`PolicyLoadError`'s `Display`) so the fix is
/// mechanical.
///
/// A **missing** directory is not a failure — most projects have no rules, and
/// `load_policy_dir` already returns an empty set for it.
///
/// Split out of [`ServerState::create_session`] so the fatal-vs-silent
/// behaviour is testable against a temp directory instead of the real `$HOME`.
pub(crate) async fn apply_project_policies(
    runtime: &Runtime,
    car_dir: &std::path::Path,
) -> Result<(), String> {
    match runtime.load_project_policies(car_dir).await {
        Ok(_) => Ok(()),
        Err(e) => Err(format!(
            "refusing to start with unreadable project policy rules: {e}. \
             Fix or remove the file — a policy rule that fails to parse is a \
             security control that would silently not exist."
        )),
    }
}

impl ServerState {
    async fn run_durability_lock(&self, run_id: &str) -> Arc<Mutex<()>> {
        let mut locks = self.run_durability_locks.lock().await;
        if let Some(lock) = locks.get(run_id).and_then(std::sync::Weak::upgrade) {
            return lock;
        }
        locks.retain(|_, lock| lock.strong_count() > 0);
        let lock = Arc::new(Mutex::new(()));
        locks.insert(run_id.to_string(), Arc::downgrade(&lock));
        lock
    }

    async fn quarantine_run_trace_locked(
        &self,
        runs: &mut HashMap<String, RunMeta>,
        run_id: &str,
        detail: String,
    ) -> String {
        let message = run_trace_corruption_message(run_id, detail);
        if let Some(meta) = runs.get_mut(run_id) {
            if meta.trace_corruption.as_deref() != Some(message.as_str()) {
                meta.durability_generation = meta.durability_generation.wrapping_add(1);
            }
            meta.trace_corruption = Some(message.clone());
        }
        let mut subs = self.run_subscribers.lock().await;
        subs.retain(|(subscribed_run, _), _| subscribed_run != run_id);
        message
    }

    /// Publish corruption found by a strict replay read into live lifecycle
    /// state. Disk scanning and marker persistence happen before this call;
    /// only the per-run durability gate and short registry/subscriber mutation
    /// execute here, so a corrupt trace cannot remain writable or subscribed.
    pub(crate) async fn quarantine_run_trace_from_read(
        &self,
        run_id: &str,
        detail: String,
    ) -> String {
        let durability_lock = self.run_durability_lock(run_id).await;
        let _durability_guard = durability_lock.lock().await;
        let mut runs = self.runs.lock().await;
        self.quarantine_run_trace_locked(&mut runs, run_id, detail)
            .await
    }

    /// Constructor for the standalone `car-server` binary. Each WS
    /// connection gets its own per-session memgine — matches the
    /// pre-extraction default and is correct for a single-process
    /// daemon serving one user at a time.
    ///
    /// **Embedders must not call this.** It silently leaves
    /// `shared_memgine = None`, which re-introduces the dual-memgine
    /// bug U7 was created to prevent (one engine in the embedder, a
    /// fresh one inside every WS session). Embedders use
    /// [`ServerState::embedded`] instead, which makes the shared
    /// engine handle a required argument so it cannot be forgotten.
    pub fn standalone(journal_dir: PathBuf) -> Self {
        Self::with_config(ServerStateConfig::new(journal_dir))
    }

    /// The daemon-held multi-device-sync + execution-lease subsystem (B6),
    /// lazily opened on first `sync.*`/`lease.*` contact and rooted at
    /// `<journal_dir>/sync/`. Serialized first-open (see the [`Self::sync`]
    /// field docs): the oplog journal's exclusive advisory lock means only one
    /// open may succeed, so the fallible init runs under the field's
    /// `std::sync::Mutex` while the subsystem itself is shared behind a
    /// `tokio::sync::Mutex`.
    pub fn sync_subsystem(
        &self,
    ) -> Result<Arc<tokio::sync::Mutex<crate::sync::SyncSubsystem>>, String> {
        let mut guard = self
            .sync
            .lock()
            .map_err(|_| "sync subsystem init lock poisoned".to_string())?;
        if let Some(existing) = guard.as_ref() {
            return Ok(existing.clone());
        }
        let root = self.journal_dir.join("sync");
        let subsystem = self.open_sync_subsystem(&root)?;
        let arc = Arc::new(tokio::sync::Mutex::new(subsystem));
        *guard = Some(arc.clone());
        Ok(arc)
    }

    /// Route a write to the right subsystem BY SCOPE: an opted-in
    /// `Scope::Shared{org}` goes to the shared org delivery subsystem (relay scope
    /// `org:{orgId}`, so it converges cross-member); everything else — `Personal`,
    /// or an org with no opted-in delivery — goes to the personal subsystem, which
    /// is byte-identical to before when org-scope is off. Used ONLY at the write
    /// sites; every reader keeps calling [`Self::sync_subsystem`]. (8a)
    pub fn subsystem_for_scope(
        &self,
        scope: &car_sync::Scope,
    ) -> Result<Arc<tokio::sync::Mutex<crate::sync::SyncSubsystem>>, String> {
        // Init the personal subsystem first — this also builds + stores the org
        // subsystem when opted in (via `open_sync_subsystem`).
        let user = self.sync_subsystem()?;
        let guard = self
            .org_sync
            .lock()
            .map_err(|_| "org sync lock poisoned".to_string())?;
        match route_for_scope(scope, guard.as_ref().map(|h| h.org.as_str())) {
            SyncRoute::Org => Ok(guard
                .as_ref()
                .expect("route_for_scope returns Org only when a holder exists")
                .subsystem
                .clone()),
            SyncRoute::User => Ok(user),
        }
    }

    /// The opted-in shared org delivery subsystem, if any — for pumping both scopes.
    pub fn org_subsystem(&self) -> Option<Arc<tokio::sync::Mutex<crate::sync::SyncSubsystem>>> {
        match self.org_sync.lock() {
            Ok(g) => g.as_ref().map(|h| h.subsystem.clone()),
            // Surface the poison loudly — do NOT silently degrade to personal-only
            // sync (linus B). Poison only occurs on a panic while locked.
            Err(_) => {
                tracing::error!(
                    "org_sync lock poisoned — skipping the org pump (org-scope delivery degraded)"
                );
                None
            }
        }
    }

    /// Build the sync subsystem for `root`, selecting the backend from config.
    ///
    /// Default is the local shared-directory `FsRelay`. When
    /// `PARSLEE_SYNC_BACKEND=parslee` (env or keychain), it builds the
    /// Parslee-backed device: a `ParsleeSyncTransport` at
    /// `PARSLEE_CAR_SYNC_BASE_URL` (default `{api_base}/sync`), authed with the
    /// live refresh-aware bearer, scoped `user:{account_id}`, E2E-encrypted under
    /// a passphrase-derived key (`PARSLEE_SYNC_PASSPHRASE`, the zero-knowledge
    /// cross-device key — same passphrase → same key on every device, never sent
    /// to the server). Missing a Parslee login falls back to local (logged);
    /// missing the passphrase is a hard error (refusing to sync unencrypted).
    fn open_sync_subsystem(
        &self,
        root: &std::path::Path,
    ) -> Result<crate::sync::SyncSubsystem, String> {
        let backend = car_secrets::resolve_env_or_keychain("PARSLEE_SYNC_BACKEND")
            .unwrap_or_default()
            .to_lowercase();
        if backend != "parslee" {
            return crate::sync::SyncSubsystem::open(root);
        }

        let Some(account_id) = self
            .parslee_session
            .get()
            .map(|s| s.identity.account_id.clone())
            .filter(|a| !a.is_empty())
        else {
            tracing::warn!(
                "sync backend=parslee but no Parslee login is present; using local sync until login"
            );
            return crate::sync::SyncSubsystem::open(root);
        };

        let Some(passphrase) = car_secrets::resolve_env_or_keychain("PARSLEE_SYNC_PASSPHRASE")
            .filter(|p| !p.is_empty())
        else {
            return Err(
                "sync backend=parslee requires PARSLEE_SYNC_PASSPHRASE (the zero-knowledge \
                 cross-device sync key) — set it in the keychain; refusing to sync unencrypted"
                    .to_string(),
            );
        };

        let base_url = car_secrets::resolve_env_or_keychain("PARSLEE_CAR_SYNC_BASE_URL")
            .filter(|u| !u.is_empty())
            .unwrap_or_else(|| {
                let api_base = car_auth::api_base(None);
                format!("{}/sync", api_base.trim_end_matches('/'))
            });

        let bearer: car_parslee::sync_transport::BearerFn = Arc::new(car_auth::access_token);
        let transport = Arc::new(
            car_parslee::sync_transport::ParsleeSyncTransport::new(base_url, bearer)
                .map_err(|e| format!("sync: build Parslee transport: {e}"))?,
        );
        // Argon2id-stretch the passphrase into the master ONCE at open (memory-hard
        // work factor for a password-equivalent secret), then feed BOTH the
        // per-audience provider and the org identity from it — so Argon2id runs
        // exactly once, not once per derivation.
        let master = car_sync::StretchedMaster::from_passphrase(passphrase.as_bytes(), &account_id);
        let base: Arc<dyn car_sync::SyncKeyProvider> =
            Arc::new(car_sync::DerivedKeyProvider::from_master(&master));

        // Org-scope shared-brain activation — REAL provisioning, but DEFAULT-OFF.
        // `PARSLEE_SYNC_ORG_SCOPE=<orgId>:<granter_hex>[,<granter_hex>...]` names the
        // org AND the admin-designated trusted granter key(s), held LOCALLY (never
        // backend-fetched — see `parse_org_scope_config`). Unset → `base`
        // (byte-identical to the personal-only path). Malformed → hard error above.
        // The configured org MUST match the signed-in `active_organization` or we
        // refuse loudly. Turning it ON for a real tenant is slice 12, still gated on
        // the cryptographer audit of the login_secret→identity entropy dependency.
        let mut pending_org: Option<OrgSyncHolder> = None;
        let provider: Arc<dyn car_sync::SyncKeyProvider> = match Self::parse_org_scope_config(
            car_secrets::resolve_env_or_keychain("PARSLEE_SYNC_ORG_SCOPE"),
        )? {
            None => base,
            Some((org_id, granters)) => {
                // Bind the configured org to the signed-in active organization — never
                // encrypt org ops for an org this account is not actually a member of.
                let active_org = self
                    .parslee_session
                    .get()
                    .and_then(|s| s.identity.active_organization.clone())
                    .filter(|o| !o.is_empty());
                if active_org.as_deref() != Some(org_id.as_str()) {
                    return Err(format!(
                        "PARSLEE_SYNC_ORG_SCOPE org {org_id:?} does not match the signed-in active \
                         organization {active_org:?} — refusing to activate org scope"
                    ));
                }

                // Bring the directory trait into scope for `publish_pubkey`.
                use car_sync::OrgKeyDirectory as _;
                let my_secret = car_sync::derive_x25519_identity(&master, &account_id);
                let mut directory = car_sync::NetworkOrgKeyDirectory::new(
                    transport.clone(),
                    format!("org:{org_id}"),
                );

                // Announce our X25519 identity pubkey so a granter can wrap K_org for
                // us (idempotent LWW). Non-fatal: a failure just means future grants
                // may not reach us; any already-resolved roots still work. This routes
                // to the bearer-authenticated backend, but whether the backend
                // ENFORCES "only this account may publish its own pubkey" (and "only a
                // K_org holder may publish a wrap") is the confidentiality-critical,
                // AUDIT-OWNED check — this is not a guarantee against a hostile backend.
                let my_pub = car_sync::x25519_public(&my_secret);
                let my_pub_hex: String = my_pub
                    .as_bytes()
                    .iter()
                    .map(|b| format!("{b:02x}"))
                    .collect();
                if let Err(e) = directory.publish_pubkey(&account_id, &my_pub_hex) {
                    tracing::warn!(org = %org_id, error = %e, "org-scope: failed to publish member pubkey — future grants may not reach this member");
                }

                // Resolve EVERY epoch of K_org we can unwrap (slice 10) → the
                // multi-epoch map the cipher selects by kid. `org → { epoch → K_org }`.
                // Fail-closed: Err (directory unreachable) and empty (ungranted) both
                // leave org scope on DenyCipher, NEVER demoted to the personal key.
                let mut roots = std::collections::HashMap::new();
                match car_sync::resolve_all_org_roots(
                    &directory,
                    &org_id,
                    &my_secret,
                    &account_id,
                    &granters,
                ) {
                    Ok(map) if !map.is_empty() => {
                        tracing::info!(org = %org_id, epochs = ?map.keys().collect::<Vec<_>>(), "org-scope: resolved K_org generations");
                        roots.insert(org_id.clone(), map);
                    }
                    Ok(_) => tracing::warn!(
                        org = %org_id,
                        "org-scope: no trusted grant for this member — org scope fails closed (DenyCipher)"
                    ),
                    Err(e) => tracing::warn!(
                        org = %org_id, error = %e,
                        "org-scope: org-key directory unreachable — org scope fails closed (DenyCipher), NOT demoted to the personal key"
                    ),
                }
                // ALWAYS OrgAwareKeyProvider once opted in — NEVER the bare
                // DerivedKeyProvider, which has no Scope::Shared case and would
                // silently encrypt org ops under the PERSONAL key (a fail-open).
                let provider: Arc<dyn car_sync::SyncKeyProvider> =
                    Arc::new(car_sync::OrgAwareKeyProvider::new(base, roots));

                // Open the SHARED org delivery subsystem: a SECOND device oplog
                // under `root/org-{orgId}/` whose relay scope is `org:{orgId}`, with
                // the SAME OrgAwareKeyProvider (it keys Scope::Shared{org} on K_org).
                // Scope::Shared{org} writes route here (subsystem_for_scope) and
                // converge across the org's members. `org_id` is a canonical slug
                // (validated on parse), so it is filesystem-safe. (8a)
                let org_root = root.join(format!("org-{org_id}"));
                let org_subsystem = crate::sync::SyncSubsystem::open_remote(
                    &org_root,
                    transport.clone(),
                    format!("org:{org_id}"),
                    provider.clone(),
                )?;
                // Do NOT commit to `org_sync` yet — hold the org subsystem in a
                // local and only install it AFTER the personal open below succeeds.
                // Else a failed personal open would orphan the org journal's
                // exclusive lock and permanently wedge sync retry (linus A).
                pending_org = Some(OrgSyncHolder {
                    org: org_id.clone(),
                    subsystem: Arc::new(tokio::sync::Mutex::new(org_subsystem)),
                });
                tracing::info!(org = %org_id, "org-scope: opened shared org delivery subsystem (relay org:{org_id})");
                provider
            }
        };
        let user = crate::sync::SyncSubsystem::open_remote(
            root,
            transport,
            format!("user:{account_id}"),
            provider,
        )?;
        // Personal open succeeded — NOW commit the org holder. On the failure path
        // above, `pending_org` drops here, releasing the org journal lock so a
        // retry can re-open cleanly.
        if let Some(holder) = pending_org {
            *self
                .org_sync
                .lock()
                .map_err(|_| "org sync init lock poisoned".to_string())? = Some(holder);
        }
        Ok(user)
    }

    /// Parse `PARSLEE_SYNC_ORG_SCOPE` (`<orgId>:<granter_hex>[,<granter_hex>...]`).
    /// `Ok(None)` when unset/empty → the personal-only path. `Err` on ANY malformed
    /// value — a typo must be a loud failure, never a silent fall-through to a
    /// wrong or unauthenticated state. Returns the org id + the (≥1) admin-designated
    /// trusted granter verifying keys.
    ///
    /// The granter set is held LOCALLY (operator-provisioned), NEVER fetched from
    /// the backend: whoever fills the `trusted` slice holds the authority, and if
    /// the same backend that serves the directory could name the granters it could
    /// name ITSELF a granter and substitute `K_org`. A backend-sourced / signed
    /// granter manifest is an audit-owned trust decision (slice 12), not this.
    /// Pure over the raw value so it is unit-testable without the environment.
    fn parse_org_scope_config(
        raw: Option<String>,
    ) -> Result<Option<(String, Vec<car_sync::OrgVerifyingKey>)>, String> {
        let Some(raw) = raw.filter(|s| !s.is_empty()) else {
            return Ok(None);
        };
        let (org_id, granters_csv) = raw.split_once(':').ok_or_else(|| {
            "PARSLEE_SYNC_ORG_SCOPE must be <orgId>:<granter_ed25519_pubkey_hex>[,...]".to_string()
        })?;
        // Honor the "loud failure" rule: a non-canonical org id (space, Unicode,
        // the platform's raw tenant form) must error here, not silently resolve
        // nothing — it would never match a real wrap's canonical `org` anyway.
        car_sync::require_canonical_org(org_id)
            .map_err(|e| format!("PARSLEE_SYNC_ORG_SCOPE: bad orgId: {e}"))?;
        let granters = granters_csv
            .split(',')
            .map(str::trim)
            .filter(|h| !h.is_empty())
            .map(|hex| {
                car_sync::parse_ed25519_verifying(hex)
                    .map_err(|e| format!("PARSLEE_SYNC_ORG_SCOPE: bad granter key {hex:?}: {e}"))
            })
            .collect::<Result<Vec<_>, _>>()?;
        if granters.is_empty() {
            return Err(
                "PARSLEE_SYNC_ORG_SCOPE: at least one granter ed25519 pubkey is required".into(),
            );
        }
        Ok(Some((org_id.to_string(), granters)))
    }

    /// Persist the daemon-wide standing chat-goal registry. The snapshot is
    /// taken under the async mutex, then serialized and atomically rewritten on
    /// the blocking pool so chat/event dispatch does not block a Tokio worker.
    pub async fn persist_chat_goals(&self) -> Result<(), String> {
        let path = chat_goals_path_from_journal_dir(&self.journal_dir);
        let goals = self.chat_goals.lock().await.clone();
        tokio::task::spawn_blocking(move || {
            write_chat_goals_atomic(&path, &goals)
                .map_err(|e| format!("write {}: {e}", path.display()))
        })
        .await
        .map_err(|e| format!("persist chat goals join: {e}"))?
    }

    /// Constructor for embedders (e.g. `tokhn-daemon`). The shared
    /// memgine handle is **required**: every WS session created by
    /// this state will reuse the same engine, preventing the
    /// dual-memgine bug.
    ///
    /// For embedders that also want to inject a pre-warmed inference
    /// engine or other advanced wiring, build a [`ServerStateConfig`]
    /// directly and call [`ServerState::with_config`].
    pub fn embedded(
        journal_dir: PathBuf,
        shared_memgine: Arc<Mutex<car_memgine::MemgineEngine>>,
    ) -> Self {
        Self::with_config(ServerStateConfig::new(journal_dir).with_shared_memgine(shared_memgine))
    }

    /// Build a `ServerState` from a [`ServerStateConfig`] — the path
    /// embedders use when they need to inject a shared memgine *and*
    /// a pre-warmed inference engine, or any other advanced wiring
    /// the convenience constructors don't cover.
    pub fn with_config(cfg: ServerStateConfig) -> Self {
        Self::try_with_config(cfg)
            .unwrap_or_else(|error| panic!("server state startup failed: {error}"))
    }

    /// Fallible startup surface used by the daemon before it begins listening.
    /// Durable lifecycle outbox replay is acknowledgement-bounded; any
    /// durability-unknown result stops construction and leaves its RunStore
    /// preimage/quarantine intact for an exact later retry.
    pub fn try_with_config(cfg: ServerStateConfig) -> Result<Self, String> {
        let startup_acknowledgement_timeout = cfg.startup_reconciliation_acknowledgement_timeout;
        if startup_acknowledgement_timeout.is_zero()
            || startup_acknowledgement_timeout > car_eventlog::MAX_CRITICAL_ACKNOWLEDGEMENT_TIMEOUT
        {
            return Err(format!(
                "startup reconciliation acknowledgement timeout must be between 1ns and {}ms, got {}ms",
                car_eventlog::MAX_CRITICAL_ACKNOWLEDGEMENT_TIMEOUT.as_millis(),
                startup_acknowledgement_timeout.as_millis()
            ));
        }
        let inference = std::sync::OnceLock::new();
        if let Some(eng) = cfg.inference {
            // OnceLock::set returns Err if already set — fresh OnceLock
            // means it's empty, so this is infallible here.
            let _ = inference.set(eng);
        }
        let voice_sessions = Arc::new(car_voice::VoiceSessionRegistry::new());
        // Reap sessions whose clients dropped without calling
        // voice.transcribe_stream.stop (WS disconnect, process exit,
        // etc.). Listener handles otherwise leak for the daemon's
        // lifetime. `with_config` is sync but always called from the
        // `#[tokio::main]` entry point, so `Handle::try_current()`
        // inside `start_sweeper` finds the runtime.
        voice_sessions.start_sweeper();
        // UI-improvement agent is pure decision logic — no I/O, no
        // persistence handle. Memgine ingest of strategy outcomes is
        // the caller's responsibility (handler.rs after a successful
        // Decision::Patch). Keeps the agent crate Mutex-flavor
        // agnostic so it can compose with std/tokio mutex callers.
        let ui_agent = Arc::new(car_ui_agent::UIImprovementAgent::with_default_strategies());
        let ui_agent_oscillation = Arc::new(crate::ui_agent_loop::OscillationDetector::new());
        let ui_agent_budget = Arc::new(crate::ui_agent_loop::IterationBudget::new());
        // Disk-backed run-trace store (U3). Rooted at the `runs/` sibling
        // of the journal dir (`~/.car/runs`), with retention read from
        // `~/.car/config.toml`. Boot-time GC enforces the retention cap
        // (R6) — best-effort; a failed GC must never block startup, so the
        // count is dropped. Never evicts an in-progress run.
        let mut run_store = crate::run_store::RunStore::from_journal_dir(&cfg.journal_dir);
        if let Some(failures) = cfg.run_store_failures.clone() {
            run_store = run_store.with_failure_injector(failures);
        }
        if let Some(gate) = cfg.run_store_summary_read_gate.clone() {
            run_store = run_store.with_summary_read_gate(gate);
        }
        if let Some(gate) = cfg.run_store_summary_write_gate.clone() {
            run_store = run_store.with_summary_write_gate(gate);
        }
        if let Some(gate) = cfg.run_store_append_gate.clone() {
            run_store = run_store.with_append_gate(gate);
        }
        if let Some(gate) = cfg.run_store_lookup_gate.clone() {
            run_store = run_store.with_lookup_gate(gate);
        }
        if let Some(failures) = cfg.run_store_private_path_failures.clone() {
            run_store = run_store.with_private_path_failure_injector(failures);
        }
        run_store
            .reconcile_proposal_retry_rollbacks()
            .map_err(|error| format!("proposal retry rollback reconciliation failed: {error}"))?;
        let chat_goals = load_chat_goals_from_disk(&cfg.journal_dir);
        let selfheal_ledger = cfg.selfheal_ledger.clone().unwrap_or_else(|| {
            cfg.journal_dir
                .parent()
                .unwrap_or(&cfg.journal_dir)
                .join("selfheal")
                .join("detections.jsonl")
        });
        let selfheal = Arc::new(crate::selfheal::SelfhealService::open(
            selfheal_ledger,
            cfg.selfheal_interval_secs,
            cfg.selfheal_evidence.clone(),
            cfg.selfheal_source_probe.clone(),
        )?);
        // Drop terminal history outside the configured retention window before
        // lifecycle replay. This keeps the reconciliation set bounded even
        // after years of newsroom runs; in-progress traces are never evicted.
        let _evicted_before_replay = run_store.gc();
        let mut startup_journals =
            StartupJournalCache::new(&cfg.journal_dir, cfg.journal_failures.as_ref());
        // Establish the durable RunStarted bracket before replaying a pending
        // proposal terminal. Then adopt the crash orphan and run lifecycle
        // reconciliation again to add RunCompleted. Both passes share the
        // per-client journal cache, are exact-row idempotent, and preserve:
        // RunStarted -> ProposalCompleted -> RunCompleted.
        reconcile_durable_run_journals(
            &mut startup_journals,
            &run_store,
            startup_acknowledgement_timeout,
        )?;
        reconcile_pending_proposal_journals(
            &mut startup_journals,
            &run_store,
            startup_acknowledgement_timeout,
        )?;
        reconcile_completed_proposal_guards(&run_store);
        // The in-memory map is empty at boot, so any remaining on-disk
        // InProgress run is a crashed prior process; mark it Incomplete so it
        // is terminal and age-GC-eligible (FIX 4).
        let _adopted = run_store.adopt_orphans();
        reconcile_durable_run_journals(
            &mut startup_journals,
            &run_store,
            startup_acknowledgement_timeout,
        )?;
        let _evicted = run_store.gc();
        // Adopt crash/restart-orphaned coder sessions the same way: the
        // in-memory `coder_sessions` map is empty at boot, so any non-terminal
        // snapshot under the coder state dir is a prior process's orphan. Mark
        // genuinely-stranded ones Failed ("daemon restarted mid-session") and
        // preserve a `needs_approval` orphan whose worktree still exists so the
        // user can inspect/approve it by hand. Best-effort — never blocks boot.
        if let Ok(coder_dir) = crate::coder::rpc::coder_state_dir() {
            let _coder_adopted = crate::coder::session::adopt_orphaned_sessions(&coder_dir);
        }
        // The daemon-wide shared approval ledger (kernel review C1): one
        // journal-backed store for every session's HITL records, so approvals
        // cross connections and survive restart. In-memory fallback is loud —
        // a silent downgrade here would fake durability.
        let approval_ledger = {
            let path = cfg.approval_journal.or_else(default_approval_journal_path);
            match path {
                Some(p) => {
                    // The journal's parent must exist before the first
                    // append-mode open (ApprovalLedger::record does not
                    // create directories).
                    if let Some(parent) = p.parent() {
                        let _ = std::fs::create_dir_all(parent);
                    }
                    match car_policy::ApprovalLedger::with_journal(&p) {
                        Ok(l) => {
                            if l.skipped_on_load() > 0 {
                                tracing::warn!(
                                    path = %p.display(),
                                    skipped = l.skipped_on_load(),
                                    "approval-ledger journal had unparseable lines; partial ledger loaded"
                                );
                            }
                            l
                        }
                        Err(e) => {
                            tracing::warn!(
                                path = %p.display(), error = %e,
                                "approval-ledger journal unopenable; falling back to IN-MEMORY \
                                 ledger — approvals will NOT survive restart"
                            );
                            car_policy::ApprovalLedger::new()
                        }
                    }
                }
                None => {
                    tracing::warn!(
                        "no CAR_HOME/HOME/USERPROFILE to place approvals.jsonl under the CAR \
                         state root; approval ledger is IN-MEMORY — approvals will NOT survive \
                         restart"
                    );
                    car_policy::ApprovalLedger::new()
                }
            }
        };
        // Built before the struct literal so the browser-view registry can be
        // pointed at the SAME `HostState` it will broadcast sign-in attention
        // on (`host.event`) — see `crate::browser_attention`.
        let host = Arc::new(crate::host::HostState::new());
        let browser_views = Arc::new(crate::browser_view::BrowserViewRegistry::default());
        browser_views.set_signin_attention(Arc::new(
            crate::browser_attention::HostSignInAttention::new(Arc::clone(&host)),
        ));
        Ok(Self {
            journal_dir: cfg.journal_dir,
            sessions: Mutex::new(HashMap::new()),
            inference,
            host,
            shared_memgine: cfg.shared_memgine,
            trajectory_store: Arc::new(car_memgine::TrajectoryStore::new(
                &cfg.trajectory_dir
                    .unwrap_or_else(car_memgine::TrajectoryStore::default_path),
            )),
            selfheal,
            voice_sessions,
            meetings: Arc::new(car_meeting::MeetingRegistry::new()),
            a2ui: car_a2ui::A2uiSurfaceStore::new(),
            ui_agent,
            ui_agent_oscillation,
            ui_agent_budget,
            admission: Arc::new(crate::admission::InferenceAdmission::new()),
            a2ui_route_auth: Mutex::new(HashMap::new()),
            supervisor: std::sync::OnceLock::new(),
            declagents: std::sync::OnceLock::new(),
            routing: std::sync::OnceLock::new(),
            sync: std::sync::Mutex::new(None),
            org_sync: std::sync::Mutex::new(None),
            observer_manifest_path: std::sync::OnceLock::new(),
            a2a_dispatcher: std::sync::OnceLock::new(),
            a2a_runtime: std::sync::Mutex::new(cfg.a2a_runtime),
            a2a_store: std::sync::Mutex::new(cfg.a2a_store),
            a2a_card_source: std::sync::Mutex::new(cfg.a2a_card_source),
            a2ui_subscribers: Mutex::new(HashMap::new()),
            mcp_executor: Arc::new(car_engine::McpToolExecutor::new()),
            connectors: std::sync::OnceLock::new(),
            connectors_loaded: std::sync::atomic::AtomicBool::new(false),
            channel_supervisor: std::sync::OnceLock::new(),
            durable_tasks: Mutex::new(tokio::task::JoinSet::new()),
            auth_completion_owner_id: uuid::Uuid::new_v4().simple().to_string(),
            auth_token: std::sync::OnceLock::new(),
            host_token: std::sync::OnceLock::new(),
            mobile_runtime_url: std::sync::OnceLock::new(),
            mobile_registration_url: std::sync::OnceLock::new(),
            parslee_session: std::sync::OnceLock::new(),
            attached_agents: Mutex::new(HashMap::new()),
            agent_memgines: Mutex::new(HashMap::new()),
            namespace_memgines: Mutex::new(HashMap::new()),
            coder_sessions: Mutex::new(HashMap::new()),
            coder_subscribers: Mutex::new(HashMap::new()),
            coder_watchers: Mutex::new(HashMap::new()),
            coder_watch_notify: std::sync::OnceLock::new(),
            coder_discussions: Mutex::new(HashMap::new()),
            coder_discussion_slots: Arc::new(tokio::sync::Semaphore::new(
                crate::coder::discuss::MAX_OPEN_DISCUSSIONS,
            )),
            chat_sessions: Mutex::new(HashMap::new()),
            peer_guards: Mutex::new(HashMap::new()),
            held_peer_messages: Mutex::new(std::collections::VecDeque::new()),
            lan_discovery: std::sync::Mutex::new(None),
            peer_identity: std::sync::Mutex::new(None),
            peer_trust: car_a2a::peer_auth::PeerTrust::new(),
            chat_collectors: Mutex::new(HashMap::new()),
            chat_goals: Mutex::new(chat_goals),
            runs: Mutex::new(HashMap::new()),
            run_resume_liveness: Mutex::new(()),
            run_resume_lease: cfg.run_resume_lease,
            run_completion_fence_gate: cfg.run_completion_fence_gate,
            run_durability_locks: Mutex::new(HashMap::new()),
            run_subscribers: Mutex::new(HashMap::new()),
            browser_views,
            run_store,
            journal_failures: cfg.journal_failures,
            mcp_url: std::sync::OnceLock::new(),
            mcp_sessions: std::sync::OnceLock::new(),
            approval_gate: cfg.approval_gate.unwrap_or_default(),
            supervision: Arc::new(crate::supervision::SupervisionRegistry::default()),
            approval_ledger: Arc::new(tokio::sync::RwLock::new(approval_ledger)),
            // Installed by the daemon binary at startup (see
            // `car-server/src/main.rs`), never here: the implementation lives
            // in `car-bench`, which depends on this crate.
            harness_measurer: std::sync::RwLock::new(None),
        })
    }

    /// Install the in-process harness evaluator `evolution.run`'s
    /// `harness_measure` path grades candidates with. Called once at daemon
    /// startup; a later call replaces it.
    pub fn set_harness_measurer(&self, m: Arc<dyn crate::evolution::HarnessMeasurer>) {
        match self.harness_measurer.write() {
            Ok(mut guard) => *guard = Some(m),
            // Poisoned only if a panic happened while the lock was held.
            // Refusing loudly beats leaving `evolution.run` to believe an
            // evaluator is installed when none is.
            Err(e) => tracing::error!(
                error = %e,
                "harness_measurer lock poisoned; the in-process harness evaluator was NOT installed"
            ),
        }
    }

    /// The installed harness evaluator, if any. Clones the `Arc` out and drops
    /// the guard so callers never hold a `std::sync` lock across an await.
    pub fn harness_measurer(&self) -> Option<Arc<dyn crate::evolution::HarnessMeasurer>> {
        match self.harness_measurer.read() {
            Ok(guard) => guard.clone(),
            Err(e) => {
                tracing::error!(error = %e, "harness_measurer lock poisoned; treating as absent");
                None
            }
        }
    }

    /// Start one daemon-owned operation and return a connection-scoped result
    /// receiver. Dropping the receiver never cancels the operation.
    ///
    /// The caller awaits the receiver exactly as it would have awaited the
    /// operation, so response shape and timing are unchanged for a connection
    /// that stays alive. What changes is the failure mode: when the connection
    /// goes away, the per-connection `JoinSet` aborts only this waiter, and the
    /// operation itself runs to completion on the daemon.
    pub async fn spawn_durable_operation<F, E>(
        &self,
        operation_name: impl Into<String>,
        operation: F,
    ) -> tokio::sync::oneshot::Receiver<Result<serde_json::Value, E>>
    where
        F: std::future::Future<Output = Result<serde_json::Value, E>> + Send + 'static,
        E: Send + 'static,
    {
        let (result_tx, result_rx) = tokio::sync::oneshot::channel();
        let operation_name = operation_name.into();
        let mut tasks = self.durable_tasks.lock().await;
        while let Some(result) = tasks.try_join_next() {
            if let Err(error) = result {
                tracing::warn!(error = %error, "daemon-owned task failed to join");
            }
        }
        tasks.spawn(async move {
            let result = operation.await;
            if result_tx.send(result).is_err() {
                tracing::debug!(
                    operation = %operation_name,
                    "daemon-owned operation finished after its response waiter closed"
                );
            }
            operation_name
        });
        result_rx
    }

    /// Start one fire-and-reconcile daemon-owned operation.
    pub async fn spawn_durable_task<F>(&self, operation_name: impl Into<String>, operation: F)
    where
        F: std::future::Future<Output = ()> + Send + 'static,
    {
        let operation_name = operation_name.into();
        let mut tasks = self.durable_tasks.lock().await;
        while let Some(result) = tasks.try_join_next() {
            if let Err(error) = result {
                tracing::warn!(error = %error, "daemon-owned task failed to join");
            }
        }
        tasks.spawn(async move {
            operation.await;
            operation_name
        });
    }

    /// Enable the per-launch auth handshake. After this call, every
    /// new WS connection must call `session.auth` with `token` as
    /// the first frame; otherwise the connection is closed. Called
    /// by `car-server` at startup unless `--no-auth` is set
    /// (default flipped 2026-05); embedders supply their own token
    /// if they want the same posture. Returns `Err(token)` when
    /// auth was already installed.
    pub fn install_auth_token(&self, token: String) -> Result<(), String> {
        self.auth_token.set(token)
    }

    /// Install the per-launch host token (Parslee-ai/car#254). A
    /// session that later presents this via `session.auth { host_token }`
    /// is granted the host-management role (`ClientSession::is_host`),
    /// which `authorize_run_access` requires for cross-agent run-trace
    /// reads. Set by `car-server` at startup (mints + writes the `0600`
    /// `host-token` file) unless `--no-auth` is set. Returns `Err(token)`
    /// when a host token was already installed.
    pub fn install_host_token(&self, token: String) -> Result<(), String> {
        self.host_token.set(token)
    }

    pub fn install_mobile_runtime_url(&self, url: String) -> Result<(), String> {
        self.mobile_runtime_url.set(url)
    }

    pub fn install_mobile_registration_url(&self, url: String) -> Result<(), String> {
        self.mobile_registration_url.set(url)
    }

    pub fn install_parslee_session(
        &self,
        session: crate::parslee_auth::ParsleeSession,
    ) -> Result<(), crate::parslee_auth::ParsleeSession> {
        self.parslee_session.set(session)
    }

    /// Install the runtime channel supervisor (Units 1/2/3). Called by
    /// `car-server` at boot right after `spawn_channel_pollers` builds it, so
    /// the host-gated `messaging.config.set` handler can reach it via
    /// `ServerState` to spawn a channel's watcher on enable. Idempotent on the
    /// first call; a second install is rejected (returns the supervisor back).
    pub fn install_channel_supervisor(
        &self,
        supervisor: Arc<crate::channel_supervisor::ChannelSupervisor>,
    ) -> Result<(), Arc<crate::channel_supervisor::ChannelSupervisor>> {
        self.channel_supervisor.set(supervisor)
    }

    /// Install the bound MCP URL after car-server's listener is up.
    /// Idempotent on the first call; subsequent calls are accepted
    /// silently (matches the supervisor / a2a_dispatcher install
    /// idiom). Returns `Err(())` when an MCP URL was already
    /// installed — embedders should treat this as "another
    /// component beat us to it" and use whichever value is now set.
    pub fn install_mcp_url(&self, url: String) -> Result<(), String> {
        self.mcp_url.set(url)
    }

    /// Install the MCP SSE session registry. Pairs with
    /// [`install_mcp_url`] — both come from the same `start_mcp`
    /// call and either both get installed or neither does (the
    /// daemon binds them together).
    pub fn install_mcp_sessions(
        &self,
        sessions: Arc<crate::mcp::SessionMap>,
    ) -> Result<(), Arc<crate::mcp::SessionMap>> {
        self.mcp_sessions.set(sessions)
    }

    /// Lazy-initialize and return the remote MCP connector manager,
    /// backed by `~/.car/connectors.json` over the shared
    /// [`mcp_executor`](Self::mcp_executor). Construction is sync and
    /// touches no network — call [`ensure_connectors_loaded`] to dial
    /// persisted connectors.
    pub fn connectors(&self) -> Arc<car_connectors::ConnectorManager> {
        self.connectors
            .get_or_init(|| {
                let mgr = car_connectors::ConnectorManager::new(self.mcp_executor.clone())
                    .unwrap_or_else(|_| {
                        // HOME unresolved (rare): fall back to a CWD-relative
                        // manifest so connectors still work this session.
                        car_connectors::ConnectorManager::with_path(
                            self.mcp_executor.clone(),
                            std::path::PathBuf::from("connectors.json"),
                        )
                    });
                Arc::new(mgr)
            })
            .clone()
    }

    /// Load persisted connectors and dial them, **once** per process.
    /// Idempotent: the first caller does the work; later callers return
    /// immediately. `car-server` calls this at boot so connector tools
    /// are live before clients connect; the `connectors.*` handlers call
    /// it too, so an embedder that skips the boot call still gets a
    /// lazy load on first use. Enabled tools discovered here are
    /// registered into any already-open sessions.
    pub async fn ensure_connectors_loaded(&self) {
        use std::sync::atomic::Ordering;
        if self.connectors_loaded.swap(true, Ordering::SeqCst) {
            return;
        }
        let mgr = self.connectors();
        match mgr.load_and_connect().await {
            Ok(entries) => self.register_connector_entries(&entries).await,
            Err(e) => tracing::warn!("connector load failed: {e}"),
        }
    }

    /// Register connector [`ToolEntry`]s into every currently-open
    /// session's runtime registry, so a tool enabled mid-session
    /// becomes visible without a reconnect. New sessions pick up the
    /// enabled set in [`create_session`].
    pub async fn register_connector_entries(&self, entries: &[car_engine::ToolEntry]) {
        if entries.is_empty() {
            return;
        }
        let sessions: Vec<Arc<ClientSession>> =
            self.sessions.lock().await.values().cloned().collect();
        for session in sessions {
            for entry in entries {
                session.runtime.register_tool_entry(entry.clone()).await;
            }
        }
    }

    /// Unregister connector tools (by canonical name) from every open
    /// session's runtime, so a disabled or removed connector's tools
    /// stop being visible to the model and accepted by the validator.
    pub async fn unregister_connector_tools(&self, canonical_names: &[String]) {
        if canonical_names.is_empty() {
            return;
        }
        let sessions: Vec<Arc<ClientSession>> =
            self.sessions.lock().await.values().cloned().collect();
        for session in sessions {
            for name in canonical_names {
                session.runtime.unregister_tool(name).await;
            }
        }
    }

    /// Whether any currently connected WS client holds the host-management
    /// role — authenticated via `session.auth { host_token }`
    /// (`ClientSession::is_host`), NOT the same thing as `host.subscribe`
    /// membership (see the doc comment on the auth check this mirrors,
    /// around `handle_session_auth`). In practice: is a CarHost / Command
    /// Deck app connected right now, so its browser drawer is a real
    /// visible surface?
    ///
    /// Scans the (small) live session set fresh on every call — no caching,
    /// so this is always current. The deciding signal for Task 7's
    /// headless/headed browser-launch default and its sign-in host-gone
    /// fallback (`assistant::browser_tools::HostConnectivity`).
    pub async fn any_host_connected(&self) -> bool {
        self.sessions
            .lock()
            .await
            .values()
            .any(|s| s.is_host.load(std::sync::atomic::Ordering::Acquire))
    }

    /// Lazy-initialize and return the agent supervisor. The first
    /// call constructs a [`car_registry::supervisor::Supervisor`] backed by
    /// `~/.car/agents.json` + `~/.car/logs/`. Embedders that need a
    /// non-default location should call
    /// [`ServerState::install_supervisor`] before any handler runs.
    ///
    /// In observer mode (set via [`install_observer_manifest`]),
    /// returns a clear error mentioning the manifest path the
    /// primary daemon owns. This prevents the second daemon from
    /// re-attempting `user_default()` (which would also fail with
    /// `AlreadyRunning`) on every WS call, and gives mutation
    /// handlers a stable refusal path. Read-only handlers
    /// (`agents.list`, `agents.health`) should call
    /// [`Self::observer_manifest_path`] first and fall back to
    /// [`car_registry::supervisor::Supervisor::list_from_manifest`] /
    /// `health_from_manifest` when set. Closes
    /// Parslee-ai/car-releases#44.
    /// The declarative-agent registry, lazy-initialized on first use.
    pub fn declagents(&self) -> Result<Arc<car_registry::declarative::DeclRegistry>, String> {
        if let Some(r) = self.declagents.get() {
            return Ok(r.clone());
        }
        let r = Arc::new(car_registry::declarative::DeclRegistry::user_default()?);
        let _ = self.declagents.set(r);
        Ok(self.declagents.get().expect("set or pre-existing").clone())
    }

    pub fn routing(&self) -> Result<Arc<car_registry::routing::RoutingStore>, String> {
        if let Some(r) = self.routing.get() {
            return Ok(r.clone());
        }
        let r = Arc::new(car_registry::routing::RoutingStore::user_default()?);
        let _ = self.routing.set(r);
        Ok(self.routing.get().expect("set or pre-existing").clone())
    }

    pub fn supervisor(&self) -> Result<Arc<car_registry::supervisor::Supervisor>, String> {
        if let Some(s) = self.supervisor.get() {
            return Ok(s.clone());
        }
        if let Some(p) = self.observer_manifest_path.get() {
            return Err(format!(
                "this car-server is observe-only — another car-server process \
                 holds the supervisor lock for {}. Mutations refuse here; route \
                 them to the primary daemon, or stop the other car-server first.",
                p.display()
            ));
        }
        let s = car_registry::supervisor::Supervisor::user_default()
            .map(Arc::new)
            .map_err(|e| e.to_string())?;
        // OnceLock::set returns the original arg back on collision —
        // a concurrent caller racing through user_default. Take
        // whichever wins.
        let _ = self.supervisor.set(s);
        Ok(self.supervisor.get().expect("set or pre-existing").clone())
    }

    /// Replace the lazy default with a caller-supplied supervisor.
    /// Returns `Err(())` when a supervisor was already installed.
    /// Used by the standalone `car-server` binary to call
    /// `start_all()` on a known-good handle without paying the
    /// lazy-init lookup cost.
    pub fn install_supervisor(
        &self,
        supervisor: Arc<car_registry::supervisor::Supervisor>,
    ) -> Result<(), Arc<car_registry::supervisor::Supervisor>> {
        self.supervisor.set(supervisor)
    }

    /// Non-acquiring read of the currently-installed supervisor.
    /// Unlike [`supervisor`](Self::supervisor), this does NOT lazy-
    /// init via `user_default()` — it returns `None` instead of
    /// constructing a fresh `Supervisor` and acquiring the
    /// `<manifest>.lock` as a side effect. Use this from read-only
    /// metadata paths (`host.subscribe` identity, status surfaces)
    /// where causing lock acquisition on observation would be a
    /// Heisenberg subscribe — the act of asking "do you own the
    /// lock?" must not be the act of taking it.
    pub fn supervisor_if_installed(&self) -> Option<Arc<car_registry::supervisor::Supervisor>> {
        self.supervisor.get().cloned()
    }

    /// Mark this daemon as *observing* a manifest owned by another
    /// car-server process. After this call, `supervisor()` returns
    /// an "observe-only" error and read-only handlers
    /// (`agents.list`, `agents.health`) fall back to the static
    /// `Supervisor::list_from_manifest` / `health_from_manifest`
    /// paths. Idempotent — subsequent calls with the same path are
    /// no-ops; a different path returns `Err(())`. Closes
    /// Parslee-ai/car-releases#44.
    pub fn install_observer_manifest(&self, path: PathBuf) -> Result<(), PathBuf> {
        self.observer_manifest_path.set(path)
    }

    /// Path of the manifest this daemon is observing but not
    /// supervising. `None` when this daemon owns the supervisor
    /// (the normal case) or when no manifest is configured at all
    /// (no `HOME`, embedder didn't install one).
    pub fn observer_manifest_path(&self) -> Option<&PathBuf> {
        self.observer_manifest_path.get()
    }

    /// Lazy-initialize and return the in-core A2A dispatcher. The
    /// first call constructs an [`car_a2a::A2aDispatcher`] from
    /// either the embedder's overrides (set via
    /// [`ServerStateConfig::with_a2a_runtime`] / `with_a2a_store` /
    /// `with_a2a_card_source`) or sensible defaults: a fresh
    /// `Runtime` with `register_agent_basics` registered, an
    /// `InMemoryTaskStore`, and a card built from the runtime's
    /// tool schemas advertising `ws://127.0.0.1:9100/` as the
    /// public URL. Closes Parslee-ai/car-releases#28.
    pub async fn a2a_dispatcher(&self) -> Arc<car_a2a::A2aDispatcher> {
        if let Some(d) = self.a2a_dispatcher.get() {
            return d.clone();
        }

        // Embedder overrides take precedence; fall back to defaults
        // for each slot independently (so an embedder that only
        // wants a custom card can leave the runtime + store at
        // defaults). `Mutex::take()` consumes the slot so the
        // defaults aren't reconstructed on a racing init that loses
        // the OnceLock::set call below.
        let runtime = self
            .a2a_runtime
            .lock()
            .expect("a2a_runtime mutex poisoned")
            .take();
        let runtime = match runtime {
            Some(r) => r,
            None => {
                let r = Arc::new(car_engine::Runtime::new());
                r.register_agent_basics().await;
                r
            }
        };

        let store = self
            .a2a_store
            .lock()
            .expect("a2a_store mutex poisoned")
            .take()
            .unwrap_or_else(|| Arc::new(car_a2a::InMemoryTaskStore::new()));

        let card_source = self
            .a2a_card_source
            .lock()
            .expect("a2a_card_source mutex poisoned")
            .take();
        let card_source = match card_source {
            Some(c) => c,
            None => {
                let card = car_a2a::build_default_agent_card(
                    &runtime,
                    car_a2a::AgentCardConfig::minimal(
                        "Common Agent Runtime",
                        "Embedded CAR daemon — A2A v1.0 reachable over WebSocket JSON-RPC.",
                        "ws://127.0.0.1:9100/",
                        car_a2a::AgentProvider {
                            organization: "Parslee".into(),
                            url: Some("https://github.com/Parslee-ai/car".into()),
                        },
                    ),
                )
                .await;
                Arc::new(move || card.clone()) as Arc<car_a2a::AgentCardSource>
            }
        };

        let dispatcher = Arc::new(car_a2a::A2aDispatcher::new(runtime, store, card_source));
        // OnceLock::set returns Err on race — accept whichever
        // dispatcher won and clone-return that one.
        let _ = self.a2a_dispatcher.set(dispatcher);
        self.a2a_dispatcher
            .get()
            .expect("a2a_dispatcher set or pre-existing")
            .clone()
    }

    /// Record the start of a run and return its [`RunMeta`]. The
    /// `run_id` is minted by the caller (the `runs.start` handler)
    /// so it can set `session.current_run_id` to the same value
    /// **before** responding (KTD3). Idempotent collision on an
    /// already-present `run_id` is treated as a fresh insert (uuids
    /// don't collide in practice; if one did, the latest start wins).
    ///
    /// U3: also writes the `RunStarted` line to the disk store + creates
    /// the run file, so the run is on disk from the first record (REPLAY
    /// survives a restart even before any turn lands). Disk failures are
    /// logged, never fatal — the in-memory registry is the live path.
    /// Fan one `runs.trace.event` out to every live subscriber of
    /// `run_id` (agent run tracing, U4). Called by the lifecycle methods
    /// **while they hold the `runs` lock** so append-and-notify is
    /// serialized with subscribe's snapshot-and-register (invariant #1).
    ///
    /// Each push is a non-blocking `try_send` onto the subscriber's
    /// bounded channel — the WS socket is written only by that
    /// subscriber's dedicated drain task, never here (invariant #2). A
    /// full channel drops the event (the slow-subscriber case); the
    /// client detects the cursor gap and re-subscribes (R8). Takes the
    /// already-acquired subscribers guard so the caller controls the lock
    /// scope and the `runs` → `run_subscribers` order.
    fn fanout_locked(
        subscribers: &HashMap<(String, String), crate::host::RunTraceSubscriber>,
        run_id: &str,
        agent_id: &str,
        record: car_proto::RunRecord,
        cursor: usize,
        status: car_proto::RunLiveStatus,
    ) {
        for ((sub_run, _client), sub) in subscribers.iter() {
            if sub_run != run_id {
                continue;
            }
            let event = car_proto::RunTraceEvent {
                run_id: run_id.to_string(),
                agent_id: agent_id.to_string(),
                record: record.clone(),
                cursor,
                status,
            };
            // Best-effort: a wedged subscriber's full channel drops the
            // event rather than stalling the producer. Logged at debug —
            // the client backfills via re-subscribe.
            if !sub.push(event) {
                tracing::debug!(
                    run_id,
                    "run-trace: dropped event for slow subscriber (channel full)"
                );
            }
        }
    }

    /// Atomically reserve a run id/idempotency key across every live session.
    /// A per-run durability lock serializes the disk absence check with the
    /// final process-wide map insert. The global `runs` lock is held only for
    /// the two bounded map checks, never for filesystem enumeration/replay.
    pub async fn reserve_run(&self, meta: RunMeta) -> Result<RunReservation, String> {
        fn existing_reservation(
            existing: &RunMeta,
            requested: &RunMeta,
        ) -> Result<RunReservation, String> {
            if existing.client_id != requested.client_id {
                return Err(format!(
                    "{} run `{}` belongs to client_id `{}`",
                    car_proto::RUN_OWNERSHIP_CONFLICT_MESSAGE_PREFIX,
                    requested.run_id,
                    existing.client_id
                ));
            }
            if existing.agent_id != requested.agent_id {
                return Err(format!(
                    "{} run `{}` was retried under a different agent",
                    car_proto::RUN_OWNERSHIP_CONFLICT_MESSAGE_PREFIX,
                    requested.run_id
                ));
            }
            let changed_fields = [
                (existing.intent != requested.intent).then_some("intent"),
                (existing.outcome_description != requested.outcome_description)
                    .then_some("outcome_description"),
            ]
            .into_iter()
            .flatten()
            .collect::<Vec<_>>();
            if !changed_fields.is_empty() {
                return Err(format!(
                    "{} idempotency conflict for run `{}`: retry changed occurrence-defining {}",
                    car_proto::RUN_OWNERSHIP_CONFLICT_MESSAGE_PREFIX,
                    requested.run_id,
                    changed_fields.join(" and ")
                ));
            }
            Ok(RunReservation::Existing(existing.clone()))
        }

        let durability_lock = self.run_durability_lock(&meta.run_id).await;
        let _durability_guard = durability_lock.lock().await;
        {
            let runs = self.runs.lock().await;
            if let Some(existing) = runs.get(&meta.run_id) {
                return existing_reservation(existing, &meta);
            }
        }
        let store = self.run_store.clone();
        let durable_run_id = meta.run_id.clone();
        let durable = tokio::task::spawn_blocking(move || store.run_started(&durable_run_id))
            .await
            .map_err(|error| format!("RunStarted reservation read task failed: {error}"))?
            .map_err(|error| format!("RunStarted reservation read failed: {error}"))?;
        if let Some(started) = durable {
            return Err(format!(
                "{} persisted run `{}` belongs to client_id `{}` and cannot be adopted",
                car_proto::RUN_OWNERSHIP_CONFLICT_MESSAGE_PREFIX,
                meta.run_id,
                started.client_id.as_deref().unwrap_or("historical/unknown")
            ));
        }
        let mut runs = self.runs.lock().await;
        if let Some(existing) = runs.get(&meta.run_id) {
            return existing_reservation(existing, &meta);
        }
        runs.insert(meta.run_id.clone(), meta);
        Ok(RunReservation::New)
    }

    /// Release a newly reserved candidate when `runs.start` rejects before it
    /// binds or persists that candidate. The exact pristine metadata and the
    /// absent RunStore boundary are both re-checked under the per-run durability
    /// lock. This deliberately does not touch `current_run_id` or the journal
    /// binding, which may still belong to an older live run on this session.
    pub(crate) async fn release_unpersisted_run_reservation(
        &self,
        session: &ClientSession,
        expected: &RunMeta,
    ) -> Result<(), String> {
        if expected.client_id != session.client_id
            || expected.start_committed
            || expected.termination.is_some()
            || expected.ended_at.is_some()
            || !expected.turns.is_empty()
            || expected.pending_terminal.is_some()
        {
            return Err(format!(
                "run `{}` is not a pristine reservation owned by this session",
                expected.run_id
            ));
        }
        let still_expected = |existing: &RunMeta| {
            existing.run_id == expected.run_id
                && existing.client_id == expected.client_id
                && existing.agent_id == expected.agent_id
                && existing.intent == expected.intent
                && existing.outcome_description == expected.outcome_description
                && existing.started_at == expected.started_at
                && existing.durability_generation == expected.durability_generation
                && !existing.start_committed
                && existing.termination.is_none()
                && existing.ended_at.is_none()
                && existing.turns.is_empty()
                && existing.pending_terminal.is_none()
        };
        let durability_lock = self.run_durability_lock(&expected.run_id).await;
        let _durability_guard = durability_lock.lock().await;
        {
            let runs = self.runs.lock().await;
            let existing = runs
                .get(&expected.run_id)
                .ok_or_else(|| format!("reserved run `{}` is absent", expected.run_id))?;
            if !still_expected(existing) {
                return Err(format!(
                    "reserved run `{}` changed before pre-start release",
                    expected.run_id
                ));
            }
        }
        let store = self.run_store.clone();
        let durable_run_id = expected.run_id.clone();
        if tokio::task::spawn_blocking(move || store.run_started(&durable_run_id))
            .await
            .map_err(|error| format!("RunStarted release read task failed: {error}"))?
            .map_err(|error| format!("RunStarted release read failed: {error}"))?
            .is_some()
        {
            return Err(format!(
                "reserved run `{}` reached durable storage and cannot be released",
                expected.run_id
            ));
        }
        let mut runs = self.runs.lock().await;
        let existing = runs
            .get(&expected.run_id)
            .ok_or_else(|| format!("reserved run `{}` is absent", expected.run_id))?;
        if !still_expected(existing) {
            return Err(format!(
                "reserved run `{}` changed during pre-start release",
                expected.run_id
            ));
        }
        runs.remove(&expected.run_id);
        Ok(())
    }

    pub async fn persist_run_start(&self, run_id: &str) -> Result<car_proto::RunStarted, String> {
        let durability_lock = self.run_durability_lock(run_id).await;
        let _durability_guard = durability_lock.lock().await;
        let started = {
            let runs = self.runs.lock().await;
            let meta = runs
                .get(run_id)
                .ok_or_else(|| format!("unknown reserved run_id `{run_id}`"))?;
            car_proto::RunStarted {
                run_id: meta.run_id.clone(),
                client_id: Some(meta.client_id.clone()),
                agent_id: meta.agent_id.clone(),
                intent: meta.intent.clone(),
                outcome_description: meta.outcome_description.clone(),
                started_at: meta.started_at,
            }
        };
        let store = self.run_store.clone();
        let durable_started = started.clone();
        tokio::task::spawn_blocking(move || store.write_started(&durable_started))
            .await
            .map_err(|error| format!("RunStarted durability task failed: {error}"))?
            .map_err(|error| format!("RunStarted durability failed: {error}"))?;
        Ok(started)
    }

    pub async fn commit_run_start(&self, run_id: &str) -> Result<(), String> {
        let mut runs = self.runs.lock().await;
        let meta = runs
            .get_mut(run_id)
            .ok_or_else(|| format!("unknown reserved run_id `{run_id}`"))?;
        if meta.start_committed {
            return Ok(());
        }
        meta.start_committed = true;
        meta.durability_generation = meta.durability_generation.wrapping_add(1);
        let started = car_proto::RunStarted {
            run_id: meta.run_id.clone(),
            client_id: Some(meta.client_id.clone()),
            agent_id: meta.agent_id.clone(),
            intent: meta.intent.clone(),
            outcome_description: meta.outcome_description.clone(),
            started_at: meta.started_at,
        };
        let subs = self.run_subscribers.lock().await;
        Self::fanout_locked(
            &subs,
            &started.run_id,
            &started.agent_id,
            car_proto::RunRecord::Started(started.clone()),
            0,
            car_proto::RunLiveStatus::InProgress,
        );
        Ok(())
    }

    /// Resolve a `runs.start` transaction whose acknowledgement failed before
    /// the owning socket disconnected. Returns `true` when no durable start
    /// boundary existed and the reservation was released. Returns `false`
    /// after reconciling an exact durable `RunStarted`, which the disconnect
    /// sweep must then close as one historical `Incomplete` occurrence.
    pub(crate) async fn reconcile_or_release_unacknowledged_start(
        &self,
        session: &ClientSession,
        run_id: &str,
    ) -> Result<bool, String> {
        let durability_lock = self.run_durability_lock(run_id).await;
        let _durability_guard = durability_lock.lock().await;
        let expected = {
            let runs = self.runs.lock().await;
            let meta = runs
                .get(run_id)
                .ok_or_else(|| format!("unknown reserved run_id `{run_id}`"))?;
            if meta.client_id != session.client_id {
                return Err(format!(
                    "{} run `{run_id}` belongs to client_id `{}`",
                    car_proto::RUN_OWNERSHIP_CONFLICT_MESSAGE_PREFIX,
                    meta.client_id
                ));
            }
            if meta.start_committed {
                return Ok(false);
            }
            if meta.is_terminal() || meta.pending_terminal.is_some() {
                return Err(format!(
                    "uncommitted run `{run_id}` unexpectedly has terminal state"
                ));
            }
            car_proto::RunStarted {
                run_id: meta.run_id.clone(),
                client_id: Some(meta.client_id.clone()),
                agent_id: meta.agent_id.clone(),
                intent: meta.intent.clone(),
                outcome_description: meta.outcome_description.clone(),
                started_at: meta.started_at,
            }
        };

        let store = self.run_store.clone();
        let durable_run_id = run_id.to_string();
        let durable = tokio::task::spawn_blocking(move || store.run_started(&durable_run_id))
            .await
            .map_err(|error| format!("RunStarted recovery read task failed: {error}"))?
            .map_err(|error| format!("RunStarted recovery read failed: {error}"))?;
        match durable {
            Some(durable) => {
                if durable != expected {
                    return Err(format!(
                        "durable RunStarted identity for `{run_id}` does not match its live reservation"
                    ));
                }
                // Re-run both exact idempotent durability barriers. The first
                // retry turns a prior write-without-fsync into an acknowledged
                // RunStore boundary; the second either appends RunStarted or
                // fsyncs the exact journal row already pending in memory.
                let store = self.run_store.clone();
                let durable_expected = expected.clone();
                tokio::task::spawn_blocking(move || store.write_started(&durable_expected))
                    .await
                    .map_err(|error| {
                        format!("RunStarted recovery durability task failed: {error}")
                    })?
                    .map_err(|error| format!("RunStarted recovery durability failed: {error}"))?;
                session.append_run_started_event(&expected).await?;
                self.commit_run_start(run_id).await?;
                Ok(false)
            }
            None => {
                // `runs.start` orders RunStore before journal append, so an
                // absent RunStore boundary proves no lifecycle row was ever
                // admitted to the journal. Remove only the zero-byte file an
                // open-before-write failure may have left behind.
                let store = self.run_store.clone();
                let durable_expected = expected.clone();
                tokio::task::spawn_blocking(move || {
                    store.rollback_empty_run_start(&durable_expected)
                })
                .await
                .map_err(|error| format!("empty RunStarted rollback task failed: {error}"))?
                .map_err(|error| format!("empty RunStarted rollback failed: {error}"))?;
                {
                    let mut runs = self.runs.lock().await;
                    let meta = runs.get(run_id).ok_or_else(|| {
                        format!("reserved run `{run_id}` disappeared during rollback")
                    })?;
                    let still_expected = meta.client_id == session.client_id
                        && !meta.start_committed
                        && !meta.is_terminal()
                        && meta.pending_terminal.is_none()
                        && meta.agent_id == expected.agent_id
                        && meta.intent == expected.intent
                        && meta.outcome_description == expected.outcome_description
                        && meta.started_at == expected.started_at;
                    if !still_expected {
                        return Err(format!(
                            "reserved run `{run_id}` changed during empty-start rollback"
                        ));
                    }
                    runs.remove(run_id);
                }
                session.clear_run_journal_binding(run_id).await?;
                let mut current = session.current_run_id.lock().await;
                if current.as_deref() == Some(run_id) {
                    *current = None;
                }
                Ok(true)
            }
        }
    }

    /// Compatibility/internal path for CAR-owned in-process callers that do
    /// not expose `runs.start`. The public RPC uses reserve/persist/journal/
    /// commit explicitly so its acknowledgement includes both durable stores.
    pub async fn start_run(&self, meta: RunMeta) -> Result<(), String> {
        let run_id = meta.run_id.clone();
        self.reserve_run(meta).await?;
        self.persist_run_start(&run_id).await?;
        self.commit_run_start(&run_id).await
    }

    /// Make a run terminal with a harness-reported outcome
    /// (`runs.complete`). Returns `Err` if the `run_id` is unknown or
    /// already terminal — the handler maps that to a JSON-RPC error so
    /// a double-complete or stale id is visible, not silently swallowed.
    ///
    /// U3: also appends the terminal `RunEnded` line to the run's JSONL
    /// file so REPLAY (U5) reports the right status (Completed) after a
    /// restart. The disk write happens after the lock is released; disk
    /// failures are logged, never fatal.
    pub async fn prepare_run_completion(
        &self,
        run_id: &str,
        termination: car_proto::RunTermination,
    ) -> Result<car_proto::RunEnded, String> {
        self.prepare_run_completion_for_active_owner(run_id, termination, None)
            .await
    }

    pub(crate) async fn prepare_run_completion_for_active_owner(
        &self,
        run_id: &str,
        termination: car_proto::RunTermination,
        expected_active_client_id: Option<&str>,
    ) -> Result<car_proto::RunEnded, String> {
        let completion_digest = run_completion_digest(&termination)?;
        if expected_active_client_id.is_some() {
            if let Some(gate) = &self.run_completion_fence_gate {
                gate.wait_if_armed().await;
            }
        }
        let durability_lock = self.run_durability_lock(run_id).await;
        let _durability_guard = durability_lock.lock().await;
        let ended = {
            let mut runs = self.runs.lock().await;
            let meta = runs
                .get_mut(run_id)
                .ok_or_else(|| format!("unknown run_id `{run_id}`"))?;
            if let Some(error) = &meta.trace_corruption {
                return Err(error.clone());
            }
            if !meta.start_committed {
                return Err(format!("run `{run_id}` start is not durably committed"));
            }
            if expected_active_client_id.is_some_and(|expected| meta.active_client_id != expected) {
                return Err(format!(
                    "run `{run_id}` active owner changed before terminalization"
                ));
            }
            if let Some(requested) = &meta.cancellation_pending {
                let matches_cancel = matches!(
                    &termination,
                    car_proto::RunTermination::Cancelled { cancellation }
                        if cancellation.run_id == requested.run_id
                            && cancellation.idempotency_key == requested.idempotency_key
                            && cancellation.reason_digest == requested.reason_digest
                            && cancellation.principal == requested.principal
                            && cancellation.action_id == requested.action_id
                            && cancellation.request_id == requested.request_id
                );
                if !matches_cancel {
                    return Err(format!(
                        "run `{run_id}` has a durable cancellation request and is quarantined"
                    ));
                }
            }
            if let Some(existing) = &meta.termination {
                if !same_run_termination(existing, &termination) {
                    return Err(format!("run `{run_id}` already has a different terminal"));
                }
                meta.resume_lease = None;
                car_proto::RunEnded {
                    run_id: run_id.to_string(),
                    client_id: Some(meta.client_id.clone()),
                    agent_id: meta.agent_id.clone(),
                    termination,
                    completion_digest: Some(completion_digest),
                    ended_at: meta
                        .ended_at
                        .ok_or_else(|| "terminal run is missing ended_at".to_string())?,
                }
            } else if let Some(pending) = &meta.pending_terminal {
                if !same_run_termination(&pending.termination, &termination)
                    || pending.completion_digest.as_deref() != Some(completion_digest.as_str())
                {
                    return Err(format!(
                        "run `{run_id}` has a different pending terminal transaction"
                    ));
                }
                meta.resume_lease = None;
                pending.clone()
            } else {
                let ended = car_proto::RunEnded {
                    run_id: run_id.to_string(),
                    client_id: Some(meta.client_id.clone()),
                    agent_id: meta.agent_id.clone(),
                    termination,
                    completion_digest: Some(completion_digest),
                    ended_at: chrono::Utc::now(),
                };
                meta.pending_terminal = Some(ended.clone());
                meta.resume_lease = None;
                meta.durability_generation = meta.durability_generation.wrapping_add(1);
                ended
            }
        };
        let store = self.run_store.clone();
        let durable_ended = ended.clone();
        let persisted = tokio::task::spawn_blocking(move || store.write_ended(&durable_ended))
            .await
            .map_err(|error| format!("RunEnded durability task failed: {error}"))?;
        if let Err(error) = persisted {
            if crate::run_store::is_trace_corruption_error(&error) {
                let mut runs = self.runs.lock().await;
                let message = self
                    .quarantine_run_trace_locked(&mut runs, run_id, error.to_string())
                    .await;
                return Err(message);
            }
            return Err(format!("RunEnded durability failed: {error}"));
        }
        Ok(ended)
    }

    pub async fn commit_run_completion(&self, ended: &car_proto::RunEnded) -> Result<(), String> {
        {
            let mut runs = self.runs.lock().await;
            let meta = runs
                .get_mut(&ended.run_id)
                .ok_or_else(|| format!("unknown run_id `{}`", ended.run_id))?;
            if let Some(existing) = &meta.termination {
                if same_run_termination(existing, &ended.termination)
                    && meta.ended_at == Some(ended.ended_at)
                {
                    return Ok(());
                }
                return Err(format!(
                    "run `{}` already has a different terminal",
                    ended.run_id
                ));
            }
            if !meta
                .pending_terminal
                .as_ref()
                .is_some_and(|pending| same_run_ended(pending, ended))
            {
                return Err(format!(
                    "run `{}` terminal commit does not match its prepared transaction",
                    ended.run_id
                ));
            }
            meta.termination = Some(ended.termination.clone());
            meta.ended_at = Some(ended.ended_at);
            meta.pending_terminal = None;
            meta.resume_lease = None;
            meta.cancellation_pending = None;
            meta.cancellation_receipt = None;
            meta.durability_generation = meta.durability_generation.wrapping_add(1);
            let cursor = meta.turn_cursor();
            let status = meta.live_status();
            let agent_id = meta.agent_id.clone();
            let subs = self.run_subscribers.lock().await;
            Self::fanout_locked(
                &subs,
                &ended.run_id,
                &agent_id,
                car_proto::RunRecord::Ended(ended.clone()),
                cursor,
                status,
            );
        }
        // Heap hygiene: the run is now terminal and fully flushed to disk,
        // and the terminal `runs.trace.event` has already fanned out to any
        // live subscriber. Drop the in-memory per-turn buffer (each entry
        // holds a full prompt + CLI output, potentially MBs) so completed
        // runs don't pin the heaviest payloads for the daemon's lifetime.
        // The lightweight `RunMeta` header (agent_id, status, started_at,
        // termination) stays resident so `run_meta`/status lookups still
        // work. A late `runs.subscribe` to this terminal run re-sources its
        // snapshot from disk (see `subscribe_run`). Only terminal runs are
        // cleared; an in-progress run keeps its turns (the live snapshot
        // source). `Vec::new()` frees the buffer's capacity, not just its
        // length.
        self.clear_terminal_run_turns(&ended.run_id).await;
        Ok(())
    }

    pub async fn complete_run(
        &self,
        run_id: &str,
        termination: car_proto::RunTermination,
    ) -> Result<car_proto::RunEnded, String> {
        let ended = self.prepare_run_completion(run_id, termination).await?;
        self.commit_run_completion(&ended).await?;
        Ok(ended)
    }

    /// Commit the body-free request to RunStore and the authenticated opening
    /// client's journal under the same per-run durability lock used by
    /// terminal completion. Memory changes only after both receipts exist.
    pub async fn persist_run_cancellation_requested(
        &self,
        owner_session: &ClientSession,
        requested: &car_proto::RunCancellationRequested,
    ) -> Result<(), String> {
        let run_id = requested.run_id.as_str();
        let durability_lock = self.run_durability_lock(run_id).await;
        let _durability_guard = durability_lock.lock().await;
        let agent_id = {
            let runs = self.runs.lock().await;
            let meta = runs
                .get(run_id)
                .ok_or_else(|| format!("unknown run_id `{run_id}`"))?;
            if meta.is_terminal() || meta.pending_terminal.is_some() {
                return Err(format!("run `{run_id}` is already terminal"));
            }
            if meta.client_id != owner_session.client_id || !meta.start_committed {
                return Err("run cancellation client/run binding mismatch".into());
            }
            if let Some(existing) = &meta.cancellation_pending {
                if existing != requested {
                    return Err(format!(
                        "run `{run_id}` already has a different cancellation request"
                    ));
                }
                return Ok(());
            }
            meta.agent_id.clone()
        };
        let store = self.run_store.clone();
        let durable_agent = agent_id.clone();
        let durable_requested = requested.clone();
        tokio::task::spawn_blocking(move || {
            store.write_cancellation_requested(&durable_agent, &durable_requested)
        })
        .await
        .map_err(|error| format!("cancellation request durability task failed: {error}"))?
        .map_err(|error| format!("cancellation request durability failed: {error}"))?;
        owner_session
            .append_run_cancellation_requested_event(requested)
            .await?;
        let mut runs = self.runs.lock().await;
        let meta = runs
            .get_mut(run_id)
            .ok_or_else(|| format!("unknown run_id `{run_id}`"))?;
        meta.cancellation_pending = Some(requested.clone());
        meta.durability_generation = meta.durability_generation.wrapping_add(1);
        let cursor = meta.turn_cursor();
        let agent_id = meta.agent_id.clone();
        let subs = self.run_subscribers.lock().await;
        Self::fanout_locked(
            &subs,
            run_id,
            &agent_id,
            car_proto::RunRecord::CancellationRequested(requested.clone()),
            cursor,
            car_proto::RunLiveStatus::CancellationPending,
        );
        Ok(())
    }

    pub async fn persist_run_cancellation_result(
        &self,
        owner_session: &ClientSession,
        result: &car_proto::RunCancelResponse,
    ) -> Result<(), String> {
        let run_id = result.run_id.as_str();
        let durability_lock = self.run_durability_lock(run_id).await;
        let _durability_guard = durability_lock.lock().await;
        let agent_id = {
            let runs = self.runs.lock().await;
            let meta = runs
                .get(run_id)
                .ok_or_else(|| format!("unknown run_id `{run_id}`"))?;
            if meta.is_terminal() {
                return Err(format!("run `{run_id}` is already terminal"));
            }
            let requested = meta
                .cancellation_pending
                .as_ref()
                .ok_or_else(|| format!("run `{run_id}` has no cancellation request"))?;
            if requested.idempotency_key != result.idempotency_key
                || requested.reason_digest != result.reason_digest
                || requested.principal != result.principal
            {
                return Err(format!("run `{run_id}` cancellation receipt mismatch"));
            }
            if let Some(existing) = &meta.cancellation_receipt {
                if existing != result {
                    return Err(format!(
                        "run `{run_id}` has a different cancellation receipt"
                    ));
                }
                return Ok(());
            }
            meta.agent_id.clone()
        };
        let store = self.run_store.clone();
        let durable_agent = agent_id.clone();
        let durable_result = result.clone();
        tokio::task::spawn_blocking(move || {
            store.write_cancellation_result(&durable_agent, &durable_result)
        })
        .await
        .map_err(|error| format!("cancellation result durability task failed: {error}"))?
        .map_err(|error| format!("cancellation result durability failed: {error}"))?;
        owner_session
            .append_run_cancellation_result_event(result)
            .await?;
        let mut runs = self.runs.lock().await;
        let meta = runs
            .get_mut(run_id)
            .ok_or_else(|| format!("unknown run_id `{run_id}`"))?;
        meta.cancellation_receipt = Some(result.clone());
        meta.durability_generation = meta.durability_generation.wrapping_add(1);
        let cursor = meta.turn_cursor();
        let agent_id = meta.agent_id.clone();
        let subs = self.run_subscribers.lock().await;
        Self::fanout_locked(
            &subs,
            run_id,
            &agent_id,
            car_proto::RunRecord::CancellationResult(result.clone()),
            cursor,
            car_proto::RunLiveStatus::CancellationPending,
        );
        Ok(())
    }

    /// Finish an unconfirmed cancellation after the opening socket has gone
    /// away. The durable `RunStarted` client id is the journal authority: a
    /// reconnect may request recovery, but it may not redirect the receipt to
    /// its own journal. RunStore, that authenticated journal, live state, and
    /// subscribers advance as one retryable transaction before success is
    /// returned.
    pub async fn persist_recovered_run_cancellation_result(
        &self,
        agent_id: &str,
        result: &car_proto::RunCancelResponse,
    ) -> Result<(), String> {
        if result.status != car_proto::RunCancellationStatus::TerminationUnconfirmed
            || result.terminal_digest.is_some()
        {
            return Err("recovered cancellation receipt must be termination_unconfirmed".into());
        }
        let run_id = result.run_id.as_str();
        let durability_lock = self.run_durability_lock(run_id).await;
        let _durability_guard = durability_lock.lock().await;
        let store = self.run_store.clone();
        let durable_agent = agent_id.to_string();
        let durable_run = run_id.to_string();
        let records = tokio::task::spawn_blocking(move || {
            store.get_run_trace_for_checked(&durable_agent, &durable_run)
        })
        .await
        .map_err(|error| format!("recovered cancellation trace task failed: {error}"))?
        .map_err(|error| format!("recovered cancellation trace read failed: {error}"))?
        .ok_or_else(|| "recovered cancellation durable owner is unavailable".to_string())?;
        let started = records
            .iter()
            .find_map(|record| match record {
                car_proto::RunRecord::Started(started) => Some(started),
                _ => None,
            })
            .ok_or_else(|| "recovered cancellation is missing durable RunStarted".to_string())?;
        if started.agent_id != agent_id {
            return Err("recovered cancellation durable agent mismatch".into());
        }
        let client_id = started
            .client_id
            .as_deref()
            .filter(|client_id| {
                !client_id.is_empty()
                    && !client_id.contains('/')
                    && !client_id.contains('\\')
                    && *client_id != "."
                    && *client_id != ".."
            })
            .ok_or_else(|| {
                "recovered cancellation has no authenticated durable journal owner".to_string()
            })?
            .to_string();
        if records
            .iter()
            .any(|record| matches!(record, car_proto::RunRecord::Ended(_)))
        {
            return Err(format!("run `{run_id}` is already terminal"));
        }
        let requested = records
            .iter()
            .find_map(|record| match record {
                car_proto::RunRecord::CancellationRequested(requested) => Some(requested),
                _ => None,
            })
            .ok_or_else(|| format!("run `{run_id}` has no durable cancellation request"))?;
        if requested.receipt_version != result.receipt_version
            || requested.run_id != result.run_id
            || requested.idempotency_key != result.idempotency_key
            || requested.reason_digest != result.reason_digest
            || requested.principal != result.principal
            || requested.action_id != result.action_id
            || requested.request_id != result.request_id
        {
            return Err(format!(
                "run `{run_id}` recovered cancellation receipt mismatch"
            ));
        }
        if let Some(existing) = records.iter().find_map(|record| match record {
            car_proto::RunRecord::CancellationResult(existing) => Some(existing),
            _ => None,
        }) {
            if existing != result {
                return Err(format!(
                    "run `{run_id}` has a different durable cancellation receipt"
                ));
            }
        }
        {
            let runs = self.runs.lock().await;
            if let Some(meta) = runs.get(run_id) {
                if meta.agent_id != agent_id
                    || meta.client_id != client_id
                    || !meta.start_committed
                    || meta.is_terminal()
                    || meta.cancellation_pending.as_ref() != Some(requested)
                {
                    return Err(format!(
                        "run `{run_id}` live state does not authenticate recovered cancellation"
                    ));
                }
                if meta
                    .cancellation_receipt
                    .as_ref()
                    .is_some_and(|existing| existing != result)
                {
                    return Err(format!(
                        "run `{run_id}` has a different live cancellation receipt"
                    ));
                }
            }
        }

        let store = self.run_store.clone();
        let durable_agent = agent_id.to_string();
        let durable_result = result.clone();
        tokio::task::spawn_blocking(move || {
            store.write_cancellation_result(&durable_agent, &durable_result)
        })
        .await
        .map_err(|error| format!("recovered cancellation durability task failed: {error}"))?
        .map_err(|error| format!("recovered cancellation durability failed: {error}"))?;

        let journal_dir = self.journal_dir.clone();
        let journal_failures = self.journal_failures.clone();
        let journal_client = client_id.clone();
        let journal_result = result.clone();
        let journal_was_new = tokio::task::spawn_blocking(move || {
            append_recovered_cancellation_result_journal(
                journal_dir,
                journal_failures,
                journal_client,
                journal_result,
            )
        })
        .await
        .map_err(|error| format!("recovered cancellation journal task failed: {error}"))??;

        let mut runs = self.runs.lock().await;
        let (cursor, fanout) = match runs.get_mut(run_id) {
            Some(meta) => {
                let fanout = meta.cancellation_receipt.is_none();
                meta.cancellation_receipt = Some(result.clone());
                if fanout {
                    meta.durability_generation = meta.durability_generation.wrapping_add(1);
                }
                (meta.turn_cursor(), fanout)
            }
            None => (
                records
                    .iter()
                    .filter(|record| matches!(record, car_proto::RunRecord::Turn(_)))
                    .count(),
                journal_was_new,
            ),
        };
        if fanout {
            let subscribers = self.run_subscribers.lock().await;
            Self::fanout_locked(
                &subscribers,
                run_id,
                agent_id,
                car_proto::RunRecord::CancellationResult(result.clone()),
                cursor,
                car_proto::RunLiveStatus::CancellationPending,
            );
        }
        Ok(())
    }

    /// Mark a run `Incomplete` (R5) — used by disconnect cleanup when a
    /// harness drops without `runs.complete`. No-op if the run is
    /// already terminal (the common healthy-close case where
    /// `runs.complete` won the race). Returns `true` if it actually
    /// wrote the `Incomplete` marker.
    ///
    /// U3: on the transition to `Incomplete`, appends the terminal
    /// `RunEnded { Incomplete }` line to disk so an orphaned run reports
    /// `Incomplete` (not `InProgress`) on REPLAY after a restart. Disk
    /// failures are logged, never fatal.
    pub async fn prepare_run_incomplete(&self, run_id: &str) -> Option<car_proto::RunEnded> {
        self.prepare_run_incomplete_for_active_owner(run_id, None)
            .await
    }

    async fn prepare_run_incomplete_for_active_owner(
        &self,
        run_id: &str,
        expected_active_client_id: Option<&str>,
    ) -> Option<car_proto::RunEnded> {
        // If a `runs.complete` transaction already prepared its exact outcome,
        // disconnect cleanup must finish that transaction, never overwrite it
        // with Incomplete.
        let termination = self
            .runs
            .lock()
            .await
            .get(run_id)
            .and_then(|meta| meta.pending_terminal.as_ref())
            .map(|ended| ended.termination.clone())
            .unwrap_or(car_proto::RunTermination::Incomplete);
        match self
            .prepare_run_completion_for_active_owner(run_id, termination, expected_active_client_id)
            .await
        {
            Ok(ended) => Some(ended),
            Err(error) => {
                tracing::error!(run_id, %error, "failed to prepare durable incomplete terminal");
                None
            }
        }
    }

    pub async fn mark_run_incomplete(&self, run_id: &str) -> Option<car_proto::RunEnded> {
        let ended = self.prepare_run_incomplete(run_id).await?;
        if let Err(error) = self.commit_run_completion(&ended).await {
            tracing::error!(run_id, %error, "failed to commit incomplete terminal");
            return None;
        }
        Some(ended)
    }

    /// Free the in-memory per-turn buffer of a TERMINAL run, keeping the
    /// lightweight `RunMeta` header resident (agent run tracing, heap
    /// hygiene follow-up). Called at the end of `complete_run` /
    /// `mark_run_incomplete`, AFTER the disk flush and terminal fanout, so
    /// the disk store remains the durable source of truth and any late
    /// `runs.subscribe` re-sources the snapshot from disk
    /// ([`Self::subscribe_run`]).
    ///
    /// Guards on `is_terminal()` so a same-id reuse that somehow re-opened
    /// the run (it shouldn't — ids are uuids) never has live turns dropped.
    /// `Vec::new()` releases the buffer's capacity, not just its length —
    /// the per-turn payloads (full prompt + CLI output) are the heavy part.
    async fn clear_terminal_run_turns(&self, run_id: &str) {
        let mut runs = self.runs.lock().await;
        if let Some(meta) = runs.get_mut(run_id) {
            if meta.is_terminal() {
                meta.turns = Vec::new();
            }
        }
    }

    /// Non-acquiring read of a run's current metadata (clone). Used by
    /// tests and by the U2 recorder to learn a run's owning `agent_id`.
    ///
    /// NOTE: this clones the ENTIRE `RunMeta`, including its `turns` buffer
    /// (full prompts + CLI output — up to the per-run caps). Hot callers
    /// that need only the run's header facts must use [`Self::run_header`]
    /// instead; cloning the whole buffer per batch RPC is a ~512 MB
    /// worst-case copy at this PR's caps (ADV-2).
    pub async fn run_meta(&self, run_id: &str) -> Option<RunMeta> {
        self.runs.lock().await.get(run_id).cloned()
    }

    /// Lightweight, non-cloning read of a run's header facts under the
    /// `runs` lock: `(agent_id, terminal, turn_count, trace_corruption)`.
    /// The owning agent id and optional error are the only heap allocations;
    /// This is what `handle_runs_record_turns` needs — owner for the write
    /// authz check, terminality for the distinguishable `run_terminal` drop,
    /// and the current turn count for the (fast-path) ceiling pre-check —
    /// without the `run_meta` deep clone of every recorded turn's payload.
    pub async fn run_header(&self, run_id: &str) -> Option<(String, bool, usize, Option<String>)> {
        self.runs.lock().await.get(run_id).map(|m| {
            (
                m.agent_id.clone(),
                m.is_terminal(),
                m.turns.len(),
                m.trace_corruption.clone(),
            )
        })
    }

    /// Lightweight authenticated producer binding for run lifecycle handlers:
    /// `(client_id, terminal)`. Never falls back to disk because an on-disk row
    /// has no live WebSocket producer to authorize new writes.
    pub async fn run_lifecycle_binding(&self, run_id: &str) -> Option<(String, bool)> {
        self.runs.lock().await.get(run_id).map(|meta| {
            (
                meta.active_client_id.clone(),
                meta.is_terminal() || meta.trace_corruption.is_some(),
            )
        })
    }

    /// `(client_id, terminal, start_committed, completion_pending)` for the
    /// authenticated producer path. Proposal writes require the exact owner,
    /// a committed start, and no prepared terminal transaction.
    pub async fn run_lifecycle_state(&self, run_id: &str) -> Option<(String, bool, bool, bool)> {
        self.run_lifecycle_state_with_corruption(run_id).await.map(
            |(client_id, terminal, committed, pending, _)| {
                (client_id, terminal, committed, pending)
            },
        )
    }

    pub async fn run_lifecycle_state_with_corruption(
        &self,
        run_id: &str,
    ) -> Option<(String, bool, bool, bool, Option<String>)> {
        self.runs.lock().await.get(run_id).map(|meta| {
            (
                meta.active_client_id.clone(),
                meta.is_terminal(),
                meta.start_committed,
                meta.pending_terminal.is_some() || meta.cancellation_pending.is_some(),
                meta.trace_corruption.clone(),
            )
        })
    }

    /// Atomically fence an orphaned live run to a replacement socket that has
    /// already authenticated as the run's agent. The immutable durable owner
    /// never changes: proposal and terminal artifacts continue to validate
    /// against the original `RunStarted` identity.
    pub(crate) async fn resume_run(
        &self,
        run_id: &str,
        agent_id: &str,
        replacement_client_id: &str,
    ) -> Result<RunResumeBinding, String> {
        // Serialize the ownership swap with every durable append/finalization
        // for this run. A task that already crossed this fence commits before
        // resume; one that has only captured the stale socket identity loses
        // after resume and cannot write any bytes.
        let durability_lock = self.run_durability_lock(run_id).await;
        let _durability_guard = durability_lock.lock().await;
        // Session creation/removal takes this same short fence around the map
        // mutation. Keep both guards through the run-owner compare-and-swap so
        // neither a late contender nor a disconnect can invalidate liveness
        // between validation and commit.
        let _liveness_guard = self.run_resume_liveness.lock().await;
        let live_clients = self
            .sessions
            .lock()
            .await
            .keys()
            .cloned()
            .collect::<std::collections::HashSet<_>>();
        if !live_clients.contains(replacement_client_id) {
            return Err("runs.resume replacement socket is no longer active".into());
        }
        let mut runs = self.runs.lock().await;
        let meta = runs
            .get_mut(run_id)
            .ok_or_else(|| "run not found or not authorized".to_string())?;
        if meta.agent_id != agent_id {
            return Err("run not found or not authorized".into());
        }
        if meta.is_terminal() {
            return Err(format!(
                "{} run `{run_id}` is already terminal",
                car_proto::RUN_OWNERSHIP_CONFLICT_MESSAGE_PREFIX
            ));
        }
        if !meta.start_committed {
            return Err(format!(
                "{} run `{run_id}` start is not durably committed",
                car_proto::RUN_OWNERSHIP_CONFLICT_MESSAGE_PREFIX
            ));
        }
        if meta.pending_terminal.is_some()
            || meta.cancellation_pending.is_some()
            || meta.trace_corruption.is_some()
        {
            return Err(format!(
                "{} run `{run_id}` is quarantined and cannot be resumed",
                car_proto::RUN_OWNERSHIP_CONFLICT_MESSAGE_PREFIX
            ));
        }

        if meta.active_client_id == replacement_client_id {
            let resumed_from_client_id =
                meta.resume_predecessor_client_id.clone().ok_or_else(|| {
                    format!(
                        "{} run `{run_id}` is already active on this socket",
                        car_proto::RUN_OWNERSHIP_CONFLICT_MESSAGE_PREFIX
                    )
                })?;
            return Ok(RunResumeBinding {
                run_id: meta.run_id.clone(),
                agent_id: meta.agent_id.clone(),
                active_client_id: meta.active_client_id.clone(),
                resumed_from_client_id,
            });
        }
        if live_clients.contains(&meta.active_client_id) {
            return Err(format!(
                "{} run `{run_id}` still has a live owner",
                car_proto::RUN_OWNERSHIP_CONFLICT_MESSAGE_PREFIX
            ));
        }
        let lease_is_current = meta.resume_lease.as_ref().is_some_and(|lease| {
            lease.disconnected_client_id == meta.active_client_id
                && tokio::time::Instant::now() < lease.expires_at
        });
        if !lease_is_current {
            return Err(format!(
                "{} run `{run_id}` resume lease is unavailable or expired",
                car_proto::RUN_OWNERSHIP_CONFLICT_MESSAGE_PREFIX
            ));
        }

        let resumed_from_client_id = std::mem::replace(
            &mut meta.active_client_id,
            replacement_client_id.to_string(),
        );
        meta.resume_predecessor_client_id = Some(resumed_from_client_id.clone());
        meta.resume_lease = None;
        meta.durability_generation = meta.durability_generation.wrapping_add(1);
        Ok(RunResumeBinding {
            run_id: meta.run_id.clone(),
            agent_id: meta.agent_id.clone(),
            active_client_id: meta.active_client_id.clone(),
            resumed_from_client_id,
        })
    }

    /// Returns the immutable durable owner and the currently fenced live owner.
    pub(crate) async fn run_owner_binding(&self, run_id: &str) -> Option<(String, String)> {
        self.runs
            .lock()
            .await
            .get(run_id)
            .map(|meta| (meta.client_id.clone(), meta.active_client_id.clone()))
    }

    /// Append per-turn trace records to a run's in-memory buffer (agent
    /// run tracing, U2). The recorder calls this after every
    /// `proposal.submit` on a session with a current run, passing the
    /// `RunRecord::Turn`s the recorder produced for that proposal.
    ///
    /// Returns a [`RecordRunTurnsOutcome`]:
    /// - `Appended { new_total }` — the batch landed; `new_total` is the
    ///   run's new total turn count (the caller computes its next
    ///   `start_index` from it).
    /// - `RefusedCeiling` — the batch would take the run past
    ///   [`RECORD_TURNS_RUN_CEILING`] and was refused WHOLE (the under-lock
    ///   runaway backstop, ADV-1). Nothing was appended.
    /// - `UnknownOrTerminal` — the `run_id` is unknown (the bracket was
    ///   never opened) or already terminal (a turn arriving after
    ///   `runs.complete` is dropped — the run is closed). Nothing was
    ///   appended.
    ///
    /// U3 flushes the in-memory buffer to disk; U4 broadcasts it. Both read
    /// via [`Self::run_turns`].
    ///
    /// U3: the same turn records are appended to the run's JSONL file so
    /// REPLAY (U5) sees them after a restart. Persistence is the commit point:
    /// only after the exact JSONL batch is durable do memory and subscribers
    /// advance. The operation runs in a daemon-owned task, so dropping a
    /// request waiter cannot strand a persisted batch between disk and memory.
    pub async fn record_run_turns(
        self: &Arc<Self>,
        run_id: &str,
        records: Vec<car_proto::RunRecord>,
    ) -> RecordRunTurnsOutcome {
        self.record_run_turns_for_owner(run_id, None, records).await
    }

    /// Append client-narrated turns only while `active_client_id` still owns
    /// the run. The socket id is daemon-minted and captured by the handler
    /// before it detaches the work into a daemon-owned task; a later resume
    /// therefore fences delayed work from the predecessor without trusting a
    /// caller-supplied credential.
    #[doc(hidden)]
    pub async fn record_run_turns_for_active_owner(
        self: &Arc<Self>,
        run_id: &str,
        active_client_id: &str,
        records: Vec<car_proto::RunRecord>,
    ) -> RecordRunTurnsOutcome {
        self.record_run_turns_for_owner(run_id, Some(active_client_id.to_string()), records)
            .await
    }

    async fn record_run_turns_for_owner(
        self: &Arc<Self>,
        run_id: &str,
        expected_active_client_id: Option<String>,
        records: Vec<car_proto::RunRecord>,
    ) -> RecordRunTurnsOutcome {
        let state = Arc::clone(self);
        let run_id = run_id.to_string();
        match tokio::spawn(async move {
            state
                .record_run_turns_owned(&run_id, expected_active_client_id.as_deref(), records)
                .await
        })
        .await
        {
            Ok(outcome) => outcome,
            Err(error) => RecordRunTurnsOutcome::PersistenceFailed(format!(
                "durable turn append task failed: {error}"
            )),
        }
    }

    /// Make a proposal's authenticated final trace durable exactly once and
    /// reflect a newly-appended batch into the live registry/subscription
    /// stream before proposal finalization can acknowledge success.
    pub async fn ensure_proposal_run_turns(
        self: &Arc<Self>,
        pending: &crate::run_store::PendingProposalFinalization,
    ) -> Result<(), String> {
        let state = Arc::clone(self);
        let pending = pending.clone();
        tokio::spawn(async move { state.ensure_proposal_run_turns_owned(&pending).await })
            .await
            .map_err(|error| format!("proposal trace durability task failed: {error}"))?
    }

    async fn ensure_proposal_run_turns_owned(
        &self,
        pending: &crate::run_store::PendingProposalFinalization,
    ) -> Result<(), String> {
        let durability_lock = self.run_durability_lock(&pending.run_id).await;
        let _durability_guard = durability_lock.lock().await;
        let (agent_id, status, generation) = {
            let runs = self.runs.lock().await;
            let meta = runs
                .get(&pending.run_id)
                .ok_or_else(|| format!("active run `{}` is absent", pending.run_id))?;
            if let Some(error) = &meta.trace_corruption {
                return Err(error.clone());
            }
            if !meta.accepts_proposals() {
                return Err(format!("active run `{}` is not writable", pending.run_id));
            }
            (
                meta.agent_id.clone(),
                meta.live_status(),
                meta.durability_generation,
            )
        };
        let store = self.run_store.clone();
        let durable = pending.clone();
        let durable_result =
            tokio::task::spawn_blocking(move || store.ensure_proposal_turns(&durable))
                .await
                .map_err(|error| format!("proposal trace durability task failed: {error}"))?;
        let mut runs = self.runs.lock().await;
        let ensured = match durable_result {
            Ok(ensured) => ensured,
            Err(error) if crate::run_store::is_trace_corruption_error(&error) => {
                let message = self
                    .quarantine_run_trace_locked(&mut runs, &pending.run_id, error.to_string())
                    .await;
                return Err(message);
            }
            Err(error) => return Err(format!("proposal trace durability failed: {error}")),
        };
        let meta = runs.get(&pending.run_id).ok_or_else(|| {
            format!(
                "active run `{}` disappeared after durability",
                pending.run_id
            )
        })?;
        if let Some(error) = &meta.trace_corruption {
            return Err(error.clone());
        }
        if meta.durability_generation != generation || !meta.accepts_proposals() {
            return Err(format!(
                "active run `{}` changed during proposal trace durability",
                pending.run_id
            ));
        }
        let already_live = !ensured.appended
            && meta.turns.iter().any(|record| match record {
                car_proto::RunRecord::Turn(turn) => {
                    turn.proposal_id.as_deref() == Some(pending.final_proposal_id.as_str())
                }
                _ => false,
            });
        if ensured.appended || !already_live {
            let meta = runs
                .get_mut(&pending.run_id)
                .expect("run registry remains locked during proposal trace commit");
            let base = ensured
                .records
                .first()
                .and_then(|record| match record {
                    car_proto::RunRecord::Turn(turn) => Some(turn.index),
                    _ => None,
                })
                .unwrap_or(meta.turns.len());
            meta.turns.extend(ensured.records.iter().cloned());
            meta.durability_generation = meta.durability_generation.wrapping_add(1);
            let subs = self.run_subscribers.lock().await;
            for (offset, record) in ensured.records.into_iter().enumerate() {
                Self::fanout_locked(
                    &subs,
                    &pending.run_id,
                    &agent_id,
                    record,
                    base + offset + 1,
                    status,
                );
            }
        }
        Ok(())
    }

    async fn record_run_turns_owned(
        &self,
        run_id: &str,
        expected_active_client_id: Option<&str>,
        mut records: Vec<car_proto::RunRecord>,
    ) -> RecordRunTurnsOutcome {
        // Serialize durability only with this run. The global registry lock is
        // held for the admission snapshot and memory/fanout commit, never for
        // trace scanning, append, flush, or fsync.
        let durability_lock = self.run_durability_lock(run_id).await;
        let _durability_guard = durability_lock.lock().await;
        let (agent_id, base, new_total, status, generation) = {
            let mut runs = self.runs.lock().await;
            match runs.get_mut(run_id) {
                Some(meta) => {
                    if let Some(error) = &meta.trace_corruption {
                        return RecordRunTurnsOutcome::PersistenceFailed(error.clone());
                    }
                    if expected_active_client_id
                        .is_some_and(|expected| meta.active_client_id != expected)
                    {
                        return RecordRunTurnsOutcome::UnknownOrTerminal;
                    }
                    if !meta.accepts_proposals() {
                        return RecordRunTurnsOutcome::UnknownOrTerminal;
                    }
                    // ADV-1: hard ceiling, under the lock. Count only the
                    // `Turn` records — Started/Ended are not buffered here,
                    // and the ceiling is a turn-count cap. Refuse the whole
                    // batch if appending it would cross the ceiling.
                    let incoming_turns = records
                        .iter()
                        .filter(|r| matches!(r, car_proto::RunRecord::Turn(_)))
                        .count();
                    if meta.turns.len() + incoming_turns > RECORD_TURNS_RUN_CEILING {
                        return RecordRunTurnsOutcome::RefusedCeiling;
                    } else {
                        // Position of the first record about to be appended —
                        // its post-append cursor is `base + 1`.
                        let base = meta.turns.len();
                        // FIX 6: re-stamp each turn's `index` from the LIVE
                        // append position under the `runs` lock. The caller
                        // computes a provisional `start_index` from a pre-read
                        // `run_turn_count` OUTSIDE this lock — a TOCTOU: two
                        // concurrent proposals on one connection can read the
                        // same `start_index` and bake overlapping indices.
                        // Authoritatively assigning the index here, from
                        // `base + offset`, makes it contiguous and correct
                        // regardless of the racing reads upstream.
                        for (offset, record) in records.iter_mut().enumerate() {
                            if let car_proto::RunRecord::Turn(turn) = record {
                                turn.index = base + offset;
                            }
                        }
                        let agent_id = meta.agent_id.clone();
                        let status = meta.live_status();
                        (
                            agent_id,
                            base,
                            base + records.len(),
                            status,
                            meta.durability_generation,
                        )
                    }
                }
                None => return RecordRunTurnsOutcome::UnknownOrTerminal,
            }
        };

        let store = self.run_store.clone();
        let durable_agent = agent_id.clone();
        let durable_run = run_id.to_string();
        let durable_records = records.clone();
        let append = tokio::task::spawn_blocking(move || {
            store.append_turns(&durable_agent, &durable_run, &durable_records)
        })
        .await;
        let mut runs = self.runs.lock().await;
        match append {
            Ok(Ok(())) => {}
            Ok(Err(error)) => {
                tracing::warn!(run_id, %error, "run-store: failed to persist turns");
                if crate::run_store::is_trace_corruption_error(&error) {
                    let message = self
                        .quarantine_run_trace_locked(&mut runs, run_id, error.to_string())
                        .await;
                    return RecordRunTurnsOutcome::PersistenceFailed(message);
                }
                return RecordRunTurnsOutcome::PersistenceFailed(error.to_string());
            }
            Err(error) => {
                tracing::warn!(run_id, %error, "run-store: durable turn task failed");
                return RecordRunTurnsOutcome::PersistenceFailed(error.to_string());
            }
        }

        let meta = runs
            .get_mut(run_id)
            .expect("run registry entry survives a committed turn append");
        if let Some(error) = &meta.trace_corruption {
            return RecordRunTurnsOutcome::PersistenceFailed(error.clone());
        }
        if meta.durability_generation != generation || !meta.accepts_proposals() {
            return RecordRunTurnsOutcome::PersistenceFailed(format!(
                "run `{run_id}` changed during durable turn append"
            ));
        }
        meta.turns.extend(records.iter().cloned());
        meta.durability_generation = meta.durability_generation.wrapping_add(1);
        let subs = self.run_subscribers.lock().await;
        for (offset, record) in records.into_iter().enumerate() {
            Self::fanout_locked(&subs, run_id, &agent_id, record, base + offset + 1, status);
        }
        RecordRunTurnsOutcome::Appended { new_total }
    }

    /// Number of turns already recorded for a run — the `start_index` the
    /// recorder passes so per-turn `index` stays monotonic across the
    /// run's proposals. `0` for an unknown run.
    pub async fn run_turn_count(&self, run_id: &str) -> usize {
        self.runs
            .lock()
            .await
            .get(run_id)
            .map(|m| m.turns.len())
            .unwrap_or(0)
    }

    /// Clone of a run's ordered per-turn trace (agent run tracing, U2).
    /// This is the accessor U3 reads to flush turns to disk and U4 reads
    /// to broadcast them. Empty Vec for an unknown run or a run with no
    /// turns yet.
    pub async fn run_turns(&self, run_id: &str) -> Vec<car_proto::RunRecord> {
        self.runs
            .lock()
            .await
            .get(run_id)
            .map(|m| m.turns.clone())
            .unwrap_or_default()
    }

    /// Atomically snapshot a run's turns AND register a live
    /// `runs.trace.event` subscriber for `(run_id, host_client_id)` —
    /// the load-bearing invariant #1 (agent run tracing, U4).
    ///
    /// Holds the `runs` lock across BOTH (a) reading the run's current
    /// turn buffer as the snapshot at cursor `C = turns.len()` and (b)
    /// inserting the subscriber into `run_subscribers`. Because the
    /// lifecycle path (`record_run_turns` / `complete_run` /
    /// `mark_run_incomplete`) appends-and-notifies under the SAME lock,
    /// the snapshot contains exactly the turns ≤ C and every turn appended
    /// after registration is delivered — no turn in the snapshot/register
    /// window is dropped (gap) or double-delivered (dup).
    ///
    /// Spawns the subscriber's drain task (bounded channel → WS) so the
    /// producer never writes the socket directly (invariant #2). Returns
    /// the snapshot response shape. `None` if the `run_id` is unknown (the
    /// handler maps that to a not-found error). A re-subscribe by the same
    /// `(run_id, host_client_id)` replaces the prior subscriber (its drain
    /// task ends when the old sender drops) and re-snapshots — the R8
    /// reconnect path: the fresh snapshot covers any turns emitted during
    /// the gap with no dup.
    ///
    /// Terminal-run disk fallback: a completed/incomplete run has its
    /// in-memory `turns` evicted for heap hygiene (see
    /// [`Self::clear_terminal_run_turns`]). When the run is terminal AND its
    /// in-memory buffer is empty, the snapshot is re-sourced from the disk
    /// store ([`crate::run_store::RunStore::get_run_trace`], the same source
    /// `runs.get_trace` reads), filtered to `RunRecord::Turn` so the shape
    /// matches the live in-memory snapshot. A terminal run takes no further
    /// appends, so the disk trail is final and there is no gap/dup concern —
    /// the snapshot/register atomicity the `runs` lock provides is only
    /// load-bearing for the in-progress path, which is unchanged.
    pub(crate) async fn subscribe_run_page(
        &self,
        run_id: &str,
        host_client_id: &str,
        channel: Arc<WsChannel>,
        cursor: usize,
        limit: usize,
    ) -> Result<Option<RunSubscribePageResult>, String> {
        // Build the subscriber (and its drain task) OUTSIDE the runs lock
        // so spawning never happens under the hot lock; registration
        // itself is the only thing serialized.
        let subscriber =
            crate::host::RunTraceSubscriber::spawn(host_client_id.to_string(), channel);

        let (agent_id, mut expected_state) = {
            let runs = self.runs.lock().await;
            let Some(meta) = runs.get(run_id) else {
                return Ok(None);
            };
            if let Some(error) = &meta.trace_corruption {
                return Err(error.clone());
            }
            if meta.is_terminal() {
                return Ok(Some(RunSubscribePageResult::Durable {
                    agent_id: meta.agent_id.clone(),
                    status: meta.live_status(),
                }));
            }
            (
                meta.agent_id.clone(),
                (meta.turns.len(), meta.pending_terminal.is_some()),
            )
        };

        for attempt in 0..=RUN_SUBSCRIBE_SUMMARY_STATE_RETRY_LIMIT {
            // Keep disk I/O off the global run-registry lock. A concurrent
            // mutation advances the captured state under that lock; retrying
            // the summary read then gives snapshot/registration one coherent
            // durable/live boundary.
            let store = self.run_store.clone();
            let durable_agent = agent_id.clone();
            let durable_run = run_id.to_string();
            let durable_corruption = tokio::task::spawn_blocking(move || {
                store.run_trace_corruption_for(&durable_agent, &durable_run)
            })
            .await
            .map_err(|error| format!("run trace corruption check task failed: {error}"))?
            .map_err(|error| format!("run trace corruption check failed: {error}"))?;
            let mut runs = self.runs.lock().await;
            let Some(meta) = runs.get(run_id) else {
                return Ok(None);
            };
            if let Some(error) = &meta.trace_corruption {
                return Err(error.clone());
            }
            if meta.agent_id != agent_id {
                return Err(format!("run `{run_id}` changed ownership during subscribe"));
            }
            if let Some(corruption) = durable_corruption {
                let message = self
                    .quarantine_run_trace_locked(
                        &mut runs,
                        run_id,
                        format!("malformed run trace record at line {}", corruption.line),
                    )
                    .await;
                return Err(message);
            }
            let status = meta.live_status();
            if meta.is_terminal() {
                return Ok(Some(RunSubscribePageResult::Durable { agent_id, status }));
            }
            let current_state = (meta.turns.len(), meta.pending_terminal.is_some());
            if current_state != expected_state {
                if attempt == RUN_SUBSCRIBE_SUMMARY_STATE_RETRY_LIMIT {
                    return Err(format!(
                        "run `{run_id}` changed repeatedly during durable subscribe validation; retry"
                    ));
                }
                expected_state = current_state;
                drop(runs);
                continue;
            }
            let live_cursor = meta.turns.len();
            if cursor > live_cursor {
                return Err(format!(
                    "runs.subscribe cursor {cursor} exceeds live_cursor {live_cursor}"
                ));
            }
            let end = cursor.saturating_add(limit).min(live_cursor);
            let turns = meta.turns[cursor..end].to_vec();
            let next_cursor = (end < live_cursor).then_some(end);
            let subscribed = next_cursor.is_none();
            if subscribed {
                let mut subs = self.run_subscribers.lock().await;
                subs.insert((run_id.to_string(), host_client_id.to_string()), subscriber);
            }
            drop(runs);

            return Ok(Some(RunSubscribePageResult::Ready(
                car_proto::RunSubscribeResponse {
                    run_id: run_id.to_string(),
                    agent_id,
                    turns,
                    cursor,
                    limit,
                    next_cursor,
                    live_cursor,
                    subscribed,
                    status,
                },
            )));
        }
        unreachable!("bounded subscribe validation loop always returns")
    }

    /// Remove a live run-trace subscriber for `(run_id, host_client_id)`
    /// (agent run tracing, U4). Returns `true` if a subscription existed.
    /// Dropping the [`crate::host::RunTraceSubscriber`] drops its channel
    /// sender, which ends the drain task.
    pub async fn unsubscribe_run(&self, run_id: &str, host_client_id: &str) -> bool {
        self.run_subscribers
            .lock()
            .await
            .remove(&(run_id.to_string(), host_client_id.to_string()))
            .is_some()
    }

    /// Drop every live run-trace subscription owned by `host_client_id`
    /// (agent run tracing, U4 — R8 cleanup). Called from
    /// [`remove_session`] on disconnect so a CarHost that drops doesn't
    /// leave dangling drain tasks. Reconnect-durability is client-side:
    /// the run stays subscribable while it lives, and the CarHost re-issues
    /// `runs.subscribe {run_id}` on its new connection — the server never
    /// synthesizes a failure on subscriber drop.
    pub async fn drop_run_subscribers_for_client(&self, host_client_id: &str) {
        self.run_subscribers
            .lock()
            .await
            .retain(|(_run, client), _| client != host_client_id);
    }

    /// Disconnect cleanup for agent runs (R5). Called from
    /// [`remove_session`] with the `client_id` of the dropping
    /// connection. For every run this connection owns that is **not**
    /// yet terminal, wait a short grace window then re-check: if a
    /// concurrently-dispatched `runs.complete` made it terminal in the
    /// meantime, leave it; otherwise mark it `Incomplete`. The grace
    /// window is the fix for the serve-mode false-positive where a
    /// healthy `runs.complete` is still in flight (in a spawned
    /// dispatch task) when the close frame arrives (Risk: run identity
    /// races).
    async fn sweep_runs_for_disconnect(&self, session: &ClientSession) {
        let client_id = session.client_id.as_str();
        // Snapshot the still-open runs owned by this connection.
        let pending: Vec<(String, Option<tokio::time::Instant>)> = {
            let runs = self.runs.lock().await;
            runs.values()
                .filter(|m| m.active_client_id == client_id && !m.is_terminal())
                .map(|m| {
                    let resume_deadline = m.resume_lease.as_ref().and_then(|lease| {
                        (lease.disconnected_client_id == client_id).then_some(lease.expires_at)
                    });
                    (m.run_id.clone(), resume_deadline)
                })
                .collect()
        };
        if pending.is_empty() {
            return;
        }
        for (run_id, resume_deadline) in pending {
            // The lease deadline was installed atomically with removal from
            // `sessions`; sleeping to that absolute instant keeps teardown
            // work before this point from silently extending the contract.
            match resume_deadline {
                Some(deadline) => tokio::time::sleep_until(deadline).await,
                None => tokio::time::sleep(RUN_COMPLETE_GRACE).await,
            }
            match self.run_lifecycle_state(&run_id).await {
                None | Some((_, true, _, _)) => continue,
                Some((owner, false, _, _)) if owner != client_id => {
                    // An authenticated replacement claimed this run during the
                    // disconnect grace window. The old owner is fenced and
                    // must not be allowed to close the replacement's run.
                    continue;
                }
                Some((_, false, true, _)) => {}
                Some((_, false, false, _)) => {
                    match self
                        .reconcile_or_release_unacknowledged_start(session, &run_id)
                        .await
                    {
                        Ok(true) => continue,
                        Ok(false) => {}
                        Err(error) => {
                            tracing::error!(run_id, client_id, %error, "failed to reconcile or release an unacknowledged RunStarted transaction");
                            continue;
                        }
                    }
                }
            }
            match self.run_store.pending_proposal(&run_id) {
                Ok(Some(_)) => {
                    tracing::warn!(
                        run_id,
                        client_id,
                        "leaving run open because proposal finalization is pending durable replay"
                    );
                    continue;
                }
                Err(error) => {
                    tracing::error!(run_id, client_id, %error, "leaving run open because proposal finalization state is unreadable and outcome is unknown");
                    continue;
                }
                Ok(None) => {}
            }
            match self.run_store.execution_marker(&run_id) {
                Ok(Some(marker)) => {
                    tracing::warn!(
                        run_id,
                        client_id,
                        original_proposal_id = marker.original_proposal_id,
                        "leaving run open because proposal execution outcome is unknown"
                    );
                    continue;
                }
                Err(error) => {
                    tracing::error!(run_id, client_id, %error, "leaving run open because proposal execution marker is unreadable and outcome is unknown");
                    continue;
                }
                Ok(None) => {}
            }
            if let Some(ended) = self
                .prepare_run_incomplete_for_active_owner(&run_id, Some(client_id))
                .await
            {
                let durable_client_id = ended.client_id.as_deref().unwrap_or(client_id);
                let append_result = match session
                    .append_run_terminal_event_once(&ended, durable_client_id)
                    .await
                {
                    Err(error) if error.is_retry_safe() => {
                        tracing::warn!(
                            run_id,
                            client_id,
                            %error,
                            "disconnect terminal durability is unknown; retrying the exact journal row once"
                        );
                        session
                            .append_run_terminal_event_once(&ended, durable_client_id)
                            .await
                    }
                    result => result,
                };
                if let Err(error) = append_result {
                    tracing::error!(run_id, client_id, %error, "failed to bind disconnect terminal journal event");
                    continue;
                }
                if let Err(error) = self.commit_run_completion(&ended).await {
                    tracing::error!(run_id, client_id, %error, "failed to commit disconnect terminal");
                    continue;
                }
                if let Err(error) = session.clear_run_journal_binding(&run_id).await {
                    tracing::error!(run_id, client_id, %error, "failed to clear durable disconnect journal binding");
                    continue;
                }
                let mut current = session.current_run_id.lock().await;
                if current.as_deref() == Some(run_id.as_str()) {
                    *current = None;
                }
            }
        }
    }

    /// Fails only when `<home>/.car/policies/` holds a malformed rule file —
    /// see [`apply_project_policies`] for why that is fatal rather than a
    /// warning.
    pub async fn create_session(
        &self,
        client_id: &str,
        channel: Arc<WsChannel>,
    ) -> Result<Arc<ClientSession>, String> {
        let journal_path = self.journal_dir.join(format!("{}.jsonl", client_id));
        let event_log = match self.journal_failures.clone() {
            Some(failures) => EventLog::with_journal_failure_injector(journal_path, failures),
            None => EventLog::with_journal(journal_path),
        };

        let negotiated_capabilities =
            Arc::new(std::sync::RwLock::new(std::collections::BTreeSet::new()));
        let ws_executor = Arc::new(WsToolExecutor::new(
            channel.clone(),
            negotiated_capabilities.clone(),
        ));

        // Compose: connector (remote MCP) tools route to the shared
        // process-wide executor; everything else falls back to this
        // session's WS tool executor. Both views share the same routing
        // table, so a connector enabled in any session is callable here.
        let executor: Arc<dyn ToolExecutor> =
            Arc::new(self.mcp_executor.share_with_fallback(ws_executor));

        // Outbound human messaging. Built-in iMessage adapter over the same
        // `RealMessageSender` + `~/.car` config store the daemon's
        // `messaging.*` handlers use, plus a host fallback so a channel CAR has
        // no built-in transport for (Teams, per Parslee-ai/car#885) is
        // delivered by the host rather than compiled into the runtime.
        //
        // NO APPROVAL LOOP. `RealMessageSender` deliberately calls the *un-gated*
        // `car_ffi_common::integrations::messages_send` so that the approval
        // transport's own prompt does not raise an approval for itself. That
        // choice is safe to reuse here, and the recursion this new path might
        // look like it opens does not exist, because the two directions never
        // meet: `messaging.send` → registry → adapter → `messages_send` is a
        // straight line into the JXA bridge with no re-entry into the runtime,
        // and an approval raised *for* `messaging.send` is delivered by the
        // orchestrator/fanout path, which calls `MessageSender` directly and
        // never dispatches a runtime tool. Neither leg can call the other, so
        // there is no cycle to bound. Keep it that way: routing approval
        // delivery through `messaging.send` WOULD create one.
        let outbound = Arc::new(car_messaging::outbound::OutboundRegistry::new());
        outbound.register(Arc::new(
            car_messaging::outbound::ImessageOutboundAdapter::new(
                Arc::new(crate::messaging_orchestrator::RealMessageSender),
                crate::messaging_config::MessagingConfigStore::from_home(),
            ),
        ));
        outbound.set_fallback(Arc::new(crate::host_channel::HostChannelAdapter::new(
            executor.clone(),
        )));

        // The trajectory store is the daemon-wide one, so traces from every
        // session accumulate in a single history. Without this the executor's
        // `persist_trajectory` short-circuits on its `self.trajectory_store
        // .as_ref()?` and nothing is ever recorded — which is why the
        // per-tool success rates had no data behind them.
        let runtime = Runtime::new()
            .with_event_log(event_log)
            .with_executor(executor)
            .with_trajectory_store(self.trajectory_store.clone())
            .with_message_sink(outbound);

        // Declarative rules from `~/.car/policies/*.toml`. Same `~/.car`
        // convention the messaging config and the approval journal use — the
        // daemon serves whatever project a client happens to be working in, so
        // its policy set is the operator's, not a per-project one. (The
        // assistant, which IS rooted in a project, loads `<root>/.car`.)
        if let Some(car_dir) = car_home_dir() {
            apply_project_policies(&runtime, &car_dir).await?;
        }

        // The session's permission-tier gate. Built here rather than inline in
        // the `ClientSession` literal below because the admission gate
        // registered a few lines down has to hold the SAME object the
        // `permission.*` RPCs mutate — see the field doc on
        // `ClientSession::permission_gate`.
        //
        // `SandboxEdit` is the default `docs/websocket-protocol.md` publishes,
        // and the code had drifted to `ReadOnly` while nothing enforced the
        // tier. `RiskClassifier::baseline` maps both `StateWrite` and
        // `ToolCall` to `SandboxEdit`, so keeping the drifted `ReadOnly`
        // default the moment enforcement lands would escalate essentially
        // every action every existing binding client submits. `full_access`
        // actions still escalate — mandatory HITL, exactly as documented.
        let permission_gate = Arc::new(tokio::sync::RwLock::new(car_policy::PermissionGate::new(
            car_policy::PermissionTier::SandboxEdit,
        )));
        // These handles are shared with the permission admission gate so it
        // observes the identity established later by `session.auth`; taking a
        // snapshot here would leave every newly-created session unbound.
        let authenticated = Arc::new(std::sync::atomic::AtomicBool::new(false));
        let agent_id = Arc::new(tokio::sync::Mutex::new(None));
        let callback_tool_schema_digests = Arc::new(tokio::sync::RwLock::new(HashMap::new()));

        // If the embedder supplied a shared memgine, every session uses it.
        // Otherwise each session gets its own — matches pre-extraction behavior.
        // Create it before the admission gate so proposal.submit and the
        // permission advisory RPCs consult the exact same live skill ceilings.
        let memgine = match &self.shared_memgine {
            Some(eng) => eng.clone(),
            None => Arc::new(Mutex::new(car_memgine::MemgineEngine::new(None))),
        };

        // Statically verify submitted proposals before any action dispatches.
        //
        // This is the surface that matters for it: `proposal.submit` accepts
        // caller-authored, **multi-action** proposals, and per-action validation
        // only rejects an action once execution reaches it — so a bad tool name
        // in action 5 is caught after actions 1–4 have already had their side
        // effects. The gate refuses the whole proposal up front, so nothing
        // partial happens. (On a runtime that submits one action at a time, like
        // the assistant loop, it adds nothing the validator wasn't already
        // doing; see `car_engine::verify_gate` for what blocks and why.)
        runtime
            .register_admission_gate(Arc::new(car_engine::StaticVerificationGate::new(
                runtime.tools.clone(),
            )))
            .await;

        // Out-of-process supervision (proposal item 5). Registered
        // unconditionally: with no supervisor subscribed the gate short-circuits
        // to Allow on an empty-map read, so leaving it on costs nothing and
        // means a supervisor can attach to a session that was already running
        // rather than only to sessions created after it connected.
        runtime
            .register_admission_gate(Arc::new(crate::supervision::SupervisionGate::new(
                self.supervision.clone(),
            )))
            .await;

        // Enforce the session's granted permission tier before any action
        // dispatches (Parslee-ai/car#890).
        //
        // Everything this needs already existed — the per-session tier gate,
        // the durable daemon-wide approval ledger, `permission.approve` — but
        // nothing consulted any of it on the way to the executor, so an action
        // the daemon's own `permission.evaluate` called `needs_approval`
        // executed anyway and its state write persisted. This is the missing
        // call site: it asks the SAME gate the same question `permission.
        // evaluate` asks, against the SAME shared ledger, so the advisory
        // answer and the enforced one cannot diverge. An escalation blocks the
        // whole proposal until an operator approves the named fingerprint;
        // approving it makes the next submit come back `Allow` with no
        // escalation raised at all.
        runtime
            .register_admission_gate(Arc::new(
                crate::permission_gate::PermissionAdmissionGate::new(
                    permission_gate.clone(),
                    self.approval_ledger.clone(),
                )
                .with_authenticated_agent(authenticated.clone(), agent_id.clone())
                .with_callback_tools(
                    callback_tool_schema_digests.clone(),
                    runtime.registry.clone(),
                )
                .with_skill_memgine(memgine.clone())
                .with_event_log(runtime.log.clone()),
            ))
            .await;

        let session = Arc::new(ClientSession {
            client_id: client_id.to_string(),
            runtime: Arc::new(runtime),
            channel,
            host: self.host.clone(),
            memgine,
            browser: car_ffi_common::browser::BrowserSessionSlot::new(),
            // When auth is disabled (no token installed), every
            // session is "authenticated" by default — preserves the
            // pre-#32 behaviour. When auth is enabled, the value is
            // ignored on creation; the dispatcher's gate checks
            // `ServerState::auth_token.is_some()` to decide whether
            // to enforce.
            authenticated,
            negotiated_protocol_version: std::sync::atomic::AtomicU32::new(0),
            negotiated_capabilities,
            inference_control: Arc::new(crate::inference_control::InferenceRegistry::default()),
            is_host: std::sync::atomic::AtomicBool::new(false),
            agent_id,
            callback_tool_schema_digests,
            memory_namespace: tokio::sync::Mutex::new(None),
            bound_memgine: tokio::sync::Mutex::new(None),
            current_run_id: tokio::sync::Mutex::new(None),
            run_lifecycle_guard: Arc::new(tokio::sync::Mutex::new(())),
            permission_gate,
            evolution_guard: crate::evolution::CycleGuard::default(),
            last_chat_turn: tokio::sync::Mutex::new(None),
            chat_inflight: std::sync::atomic::AtomicUsize::new(0),
            tenant: tokio::sync::Mutex::new(None),
            tool_stream_subscribed: std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)),
        });

        // Seed connector tools already enabled (in any session) into this
        // new session's runtime registry so the model sees them at once.
        // Only touches the manager if connectors have been initialized —
        // embedders that never use connectors pay nothing here.
        if self.connectors.get().is_some() {
            for entry in self.connectors().enabled_tool_entries().await {
                session.runtime.register_tool_entry(entry).await;
            }
        }

        {
            let _liveness_guard = self.run_resume_liveness.lock().await;
            self.sessions
                .lock()
                .await
                .insert(client_id.to_string(), session.clone());
        }

        Ok(session)
    }

    /// Bind a session's runtime to the [`McpSubstrate`] backed by an
    /// already-connected MCP connector, making that session **coherent**:
    /// its commodity built-ins (`read_file`/`write_file`/…/`run_command`)
    /// act on the *same* environment as its `mcp_{slug}_*` connector tools,
    /// instead of being split between the host/WS client and the remote MCP
    /// server (`docs/execution-substrate.md` §3, phase 3).
    ///
    /// This is **opt-in and explicit** — it is reached only via the
    /// `session.bindSubstrate` JSON-RPC method, never inferred from a
    /// connector merely being enabled. Sessions that never call it keep the
    /// exact historic composition (`share_with_fallback(ws_executor)` over a
    /// default `LocalSubstrate`), so GUI / A2A-over-WS / connectors-as-tools
    /// / coder behavior is unchanged.
    ///
    /// On bind we:
    /// 1. Wrap the live connector session (shared with the
    ///    `mcp_{slug}_*` routes — no second dial) as an [`McpSubstrate`] and
    ///    `set_substrate` it on the session runtime.
    /// 2. `register_agent_basics()` so the model sees portable names
    ///    (`read_file`/`run_command`/…) that the engine routes to the bound
    ///    substrate.
    /// 3. Swap the session executor to a [`SubstrateShadowExecutor`] over the
    ///    same `share_with_fallback(ws_executor)` composition, so the bare
    ///    substrate-owned names fall through to the substrate while connector
    ///    routes and host tool callbacks keep working unchanged (option (a)).
    ///
    /// Returns the substrate's environment name (the connector slug) on
    /// success, or an error string if the connector is not connected.
    pub async fn bind_substrate_to_connector(
        &self,
        session: &Arc<ClientSession>,
        slug: &str,
    ) -> Result<String, String> {
        self.ensure_connectors_loaded().await;

        // The connector's live MCP session is the same handle the
        // `mcp_{slug}_*` routes dispatch through — wrap it, don't re-dial.
        let mcp_session = self
            .mcp_executor
            .session(slug)
            .await
            .ok_or_else(|| format!("connector '{slug}' is not connected"))?;

        let substrate: Arc<dyn car_engine::Substrate> =
            Arc::new(car_engine::McpSubstrate::new(mcp_session, slug.to_string()));
        let name = substrate.name().to_string();

        // 1. Bind the substrate; 2. register substrate-backed built-ins.
        //
        // These three steps (set_substrate → register_agent_basics →
        // set_executor below) are not one critical section, so a tool `execute`
        // racing on this same session mid-swap could observe a half-applied
        // state. That's benign here: a WS connection processes its JSON-RPC
        // requests sequentially, and `bindSubstrate` is a client control call
        // issued before tool work — so no in-flight built-in overlaps the swap.
        session.runtime.set_substrate(substrate).await;
        session.runtime.register_agent_basics().await;

        // 3. Shadow the substrate-owned names so they fall through to the
        //    bound substrate, keeping connector routes + WS host callbacks.
        let ws_executor = Arc::new(WsToolExecutor::new(
            session.channel.clone(),
            session.negotiated_capabilities.clone(),
        ));
        let composed: Arc<dyn ToolExecutor> =
            Arc::new(self.mcp_executor.share_with_fallback(ws_executor));
        let shadowed: Arc<dyn ToolExecutor> = Arc::new(SubstrateShadowExecutor::new(composed));
        session.runtime.set_executor(shadowed).await;

        Ok(name)
    }

    /// Remove a per-client session from the registry on disconnect.
    /// Returns the removed session if present so callers can drop any
    /// remaining strong refs (e.g. drain pending tool callbacks). Fix
    /// for MULTI-4 / WS-3 — without this, `state.sessions` retains
    /// `Arc<ClientSession>` for every connection that ever existed.
    pub async fn remove_session(&self, client_id: &str) -> Option<Arc<ClientSession>> {
        // Cancel any live detached tool invocations on this session's
        // runtime (linus review D3): after disconnect the per-session
        // registry is unreachable, so a still-running tool would be
        // uncancellable until daemon restart. Same teardown discipline
        // as run subscribers / chat sessions below.
        if let Some(session) = self.sessions.lock().await.get(client_id).cloned() {
            let n = session.runtime.tool_handles.cancel_all().await;
            if n > 0 {
                tracing::info!(
                    client_id,
                    cancelled = n,
                    "cancelled detached tools at session teardown"
                );
            }
        }

        let removed = {
            let _liveness_guard = self.run_resume_liveness.lock().await;
            let removed = self.sessions.lock().await.remove(client_id);
            let resume_negotiated = removed.as_ref().is_some_and(|session| {
                session
                    .negotiated_capabilities
                    .read()
                    .map(|caps| caps.contains(car_proto::RUNS_RESUME_CAPABILITY))
                    .unwrap_or(false)
            });
            if resume_negotiated {
                let expires_at = tokio::time::Instant::now() + self.run_resume_lease;
                let mut runs = self.runs.lock().await;
                for meta in runs
                    .values_mut()
                    .filter(|meta| meta.active_client_id == client_id && !meta.is_terminal())
                {
                    meta.resume_lease = Some(RunResumeLease {
                        disconnected_client_id: client_id.to_string(),
                        expires_at,
                    });
                }
            }
            removed
        };
        if let Some(session) = &removed {
            // #169: drop the agent_id → client_id binding so a
            // disconnected lifecycle agent can reconnect (or its
            // supervisor-respawned replacement can take the slot)
            // without colliding with the stale claim.
            let bound = session.agent_id.lock().await.clone();
            if let Some(id) = bound {
                let mut attached = self.attached_agents.lock().await;
                if attached.get(&id).map(String::as_str) == Some(client_id) {
                    attached.remove(&id);
                }
            }
            // Drop the peer-messaging guard with the binding. The guard holds a
            // rate budget and a dedupe window keyed by agent *name*, so a
            // supervisor-respawned replacement taking the same name would
            // otherwise inherit the dead process's history — arriving
            // pre-throttled, or silently deduped against messages it never saw.
            let bound_for_guards = session.agent_id.lock().await.clone();
            if let Some(id) = bound_for_guards {
                self.peer_guards.lock().await.remove(&id);
            }
            // Drop any in-flight `agents.chat` sessions bound to this
            // client — either side disconnecting orphans the stream,
            // and a respawned agent's stray `agent.chat.event`
            // notifications must not race into a stale routing entry.
            // See `docs/proposals/agent-chat-surface.md`.
            let bound_agent = session.agent_id.lock().await.clone();
            let mut chats = self.chat_sessions.lock().await;
            chats.retain(|_, s| {
                if s.host_client_id == client_id {
                    return false;
                }
                if let Some(agent_id) = &bound_agent {
                    if &s.agent_id == agent_id {
                        return false;
                    }
                }
                true
            });
            // Drop the chat lock before the run sweep, which awaits the
            // grace window and re-locks `runs` — keeping locks disjoint
            // avoids holding `chat_sessions` across the sleep.
            drop(chats);
            // Same for in-process A2A chat collectors owned by this client.
            // Dropping the entry drops the sole sender, so the collecting task's
            // `recv()` returns `None` and it falls back immediately instead of
            // waiting out its 180s cap. Done after `drop(chats)` so we never
            // hold `chat_sessions` while locking `chat_collectors` (the
            // interceptor locks them collectors→sessions; staying disjoint here
            // avoids inverting that order).
            self.chat_collectors
                .lock()
                .await
                .retain(|_, c| c.host_client_id != client_id);
            // Agent run tracing (U4): drop this connection's live
            // run-trace subscriptions so its drain tasks end and the
            // registry doesn't accumulate stale `(run_id, client)` entries
            // across reconnects. Reconnect-durability is client-side: the
            // run stays subscribable; CarHost re-subscribes by `run_id` on
            // its new connection (R8). This is exempt from the chat
            // drain-and-synthesize-error path — a dropped trace subscriber
            // never marks the underlying run failed.
            self.drop_run_subscribers_for_client(client_id).await;
            // Browser-drawer subscriptions are per-connection too. Dropping
            // them also starts the control grace period on any view this
            // connection was DRIVING, so an agent is never parked forever
            // behind a controller whose app just quit (see
            // `browser_view::CONTROL_GRACE`).
            self.browser_views
                .drop_subscriptions_for_client(client_id)
                .await;
            // Only a host-client removal can change host connectivity. This
            // used to fan a redundant reverse call to every producer for every
            // agent/CLI disconnect too. The session is already out of
            // `sessions`, so a real host removal reads the post-removal truth.
            if session.is_host.load(std::sync::atomic::Ordering::Acquire) {
                let host_connected = self.any_host_connected().await;
                self.browser_views
                    .broadcast_host_connected(host_connected)
                    .await;
            }
            // And if this connection was a supervised agent PUBLISHING a
            // browser, that browser went with the process. Its views stay
            // registered reporting an empty browser, so the drawer sees the
            // change and a restarted process can replace them in place (see
            // `browser_relay::RelayProducer::note_disconnected`).
            self.browser_views
                .note_producer_disconnected(client_id)
                .await;
            // Coder event subscriptions are per-connection too; the session
            // itself keeps running (reconnect + `coder.subscribe { from_seq }`
            // replays what was missed).
            // Board watchers (`coder.watch`) and discussion streams are
            // per-connection too, and are dropped on the same boundary.
            crate::coder::rpc::drop_subscriptions_for_client(self, client_id).await;
            crate::coder::discuss::drop_subscriptions_for_client(self, client_id).await;
            // Agent run tracing (R5): any run this connection owns that
            // has no terminal record yet is swept to `Incomplete` after
            // a short grace window — long enough for an in-flight
            // `runs.complete` to land first so a healthy close is never
            // mislabeled.
            let _run_guard = session.run_lifecycle_guard.lock().await;
            self.sweep_runs_for_disconnect(session).await;
        }
        removed
    }
}

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

    #[tokio::test]
    async fn dropping_response_waiter_does_not_cancel_operation_or_release_its_guard() {
        let temp = tempfile::tempdir().unwrap();
        let state = Arc::new(ServerState::with_config(ServerStateConfig::new(
            temp.path().to_path_buf(),
        )));
        let overlap_guard = Arc::new(tokio::sync::Mutex::new(()));
        let operation_guard = overlap_guard.clone();
        let (started_tx, started_rx) = tokio::sync::oneshot::channel();
        let (release_tx, release_rx) = tokio::sync::oneshot::channel();

        let response = state
            .spawn_durable_operation("auth.test", async move {
                let _held = operation_guard.lock().await;
                let _ = started_tx.send(());
                let _ = release_rx.await;
                Ok::<_, String>(serde_json::json!({ "ok": true }))
            })
            .await;
        started_rx.await.unwrap();
        drop(response);

        assert!(
            overlap_guard.try_lock().is_err(),
            "dropping only the connection waiter must not release the operation guard"
        );
        release_tx.send(()).unwrap();
        let joined = tokio::time::timeout(
            std::time::Duration::from_secs(1),
            state.durable_tasks.lock().await.join_next(),
        )
        .await
        .expect("daemon-owned operation should finish")
        .expect("task should exist")
        .expect("task should join");
        assert_eq!(joined, "auth.test");
        assert!(
            overlap_guard.try_lock().is_ok(),
            "the guard releases only after the daemon-owned operation finishes"
        );
    }
}

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

    #[test]
    fn honors_action_timeout_over_default() {
        // The action's budget bounds the wait (+grace so the executor's own
        // deadline reaps first) — no 60s ceiling, no env read.
        assert_eq!(
            tool_callback_timeout(Some(180_000)),
            std::time::Duration::from_millis(180_000 + TOOL_TIMEOUT_GRACE_MS)
        );
        // A budget far above the old 60s ceiling is honored (the #259 bug),
        // and the wait always exceeds the budget (so the executor wins).
        assert!(tool_callback_timeout(Some(600_000)) > std::time::Duration::from_secs(600));
        assert!(tool_callback_timeout(Some(180_000)) >= std::time::Duration::from_millis(180_000));
    }

    #[test]
    fn default_is_not_the_old_60s() {
        // With no action budget and no env override, the fallback is the
        // raised default — not the hardcoded 60s that reaped real tools.
        // (Asserted only when CAR_TOOL_TIMEOUT is unset, which it is in CI.)
        if std::env::var_os("CAR_TOOL_TIMEOUT").is_none() {
            assert_eq!(
                tool_callback_timeout(None),
                std::time::Duration::from_millis(DEFAULT_TOOL_TIMEOUT_MS)
            );
            assert!(
                DEFAULT_TOOL_TIMEOUT_MS > 60_000,
                "default must exceed the old 60s"
            );
        }
    }
}

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

    fn journal_dir() -> PathBuf {
        let target = std::env::var_os("CARGO_TARGET_DIR")
            .map(std::path::PathBuf::from)
            .unwrap_or_else(|| {
                std::path::PathBuf::from(env!("CARGO_MANIFEST_DIR"))
                    .join("..")
                    .join("..")
                    .join("target")
            });
        std::fs::create_dir_all(&target).ok();
        let target = std::fs::canonicalize(&target).unwrap_or(target);
        let tmp = tempfile::TempDir::new_in(&target).unwrap();
        let p = tmp.path().to_path_buf();
        std::mem::forget(tmp); // keep the dir alive for the test
        p
    }

    #[test]
    fn supervisor_returns_observer_error_when_marker_set() {
        // Closes Parslee-ai/car-releases#44: the second car-server on
        // a host installs the observer marker after `with_paths`
        // returns AlreadyRunning. Subsequent `state.supervisor()`
        // calls must return a clear "observe-only" error mentioning
        // the manifest path — they must NOT retry user_default()
        // (which would re-acquire the lock and likely also fail).
        let state = ServerState::standalone(journal_dir());
        let fake_manifest = PathBuf::from("/tmp/fake-manifest-for-test.json");
        state
            .install_observer_manifest(fake_manifest.clone())
            .expect("install_observer_manifest succeeds on fresh state");
        assert_eq!(state.observer_manifest_path(), Some(&fake_manifest));

        let err = state.supervisor().map(|_| ()).unwrap_err();
        assert!(
            err.contains("observe-only"),
            "error must mention observe-only mode: {err}"
        );
        assert!(
            err.contains("fake-manifest-for-test.json"),
            "error must surface the manifest path so operators know which daemon owns it: {err}"
        );
    }

    #[test]
    fn install_observer_manifest_is_idempotent_per_path_collision() {
        let state = ServerState::standalone(journal_dir());
        let p = PathBuf::from("/tmp/manifest-a.json");
        let q = PathBuf::from("/tmp/manifest-b.json");
        state.install_observer_manifest(p.clone()).unwrap();
        // OnceLock::set returns the value back on collision.
        let err = state.install_observer_manifest(q.clone()).unwrap_err();
        assert_eq!(err, q);
        assert_eq!(state.observer_manifest_path(), Some(&p));
    }

    #[test]
    fn supervisor_if_installed_does_not_lazy_init() {
        // The Heisenberg-subscribe guard: `host.subscribe`'s
        // identity path must use the non-acquiring read so a
        // purely observational client can't cause the daemon to
        // claim `<manifest>.lock` as a side effect of asking
        // about it. Fresh state has no supervisor installed.
        let state = ServerState::standalone(journal_dir());
        assert!(state.supervisor_if_installed().is_none());
        // observer_manifest_path should remain unset too — no
        // implicit init.
        assert!(state.observer_manifest_path().is_none());
    }
}

#[cfg(test)]
mod substrate_binding_tests {
    use super::*;
    use car_engine::{McpSession, McpToolInfo};
    use car_ir::ActionProposal;
    use serde_json::json;
    use std::sync::Mutex as StdMutex;

    fn journal_dir() -> PathBuf {
        let target = std::env::var_os("CARGO_TARGET_DIR")
            .map(std::path::PathBuf::from)
            .unwrap_or_else(|| {
                std::path::PathBuf::from(env!("CARGO_MANIFEST_DIR"))
                    .join("..")
                    .join("..")
                    .join("target")
            });
        std::fs::create_dir_all(&target).ok();
        let target = std::fs::canonicalize(&target).unwrap_or(target);
        let tmp = tempfile::TempDir::new_in(&target).unwrap();
        let p = tmp.path().to_path_buf();
        std::mem::forget(tmp);
        p
    }

    /// A minimal in-memory MCP session standing in for a connected `vm`
    /// connector. `read_text`/`write_text`/`run_command` behave like the
    /// `vm` bridge so we can assert a substrate-bound session's built-ins
    /// land here rather than on the WS client (which has no responder in a
    /// test and would hang/fail).
    struct FakeVmSession {
        name: String,
        files: Arc<StdMutex<std::collections::HashMap<String, String>>>,
    }

    #[async_trait::async_trait]
    impl McpSession for FakeVmSession {
        async fn list_tools(&mut self) -> Result<Vec<McpToolInfo>, String> {
            Ok(vec![])
        }
        async fn call_tool(
            &mut self,
            name: &str,
            args: serde_json::Value,
        ) -> Result<serde_json::Value, String> {
            match name {
                "write_text" => {
                    let p = args["path"].as_str().unwrap().to_string();
                    let c = args["content"].as_str().unwrap().to_string();
                    self.files.lock().unwrap().insert(p, c);
                    Ok(json!("ok"))
                }
                "read_text" => {
                    let p = args["path"].as_str().unwrap();
                    let c = self
                        .files
                        .lock()
                        .unwrap()
                        .get(p)
                        .cloned()
                        .ok_or_else(|| "not found".to_string())?;
                    Ok(json!(c))
                }
                "run_command" => {
                    let command = args["command"].as_str().unwrap_or_default();
                    if let Some(path) = command
                        .split_once("[ -e '")
                        .map(|(_, path)| path)
                        .and_then(|path| path.split_once("' ]").map(|(path, _)| path))
                    {
                        let exists = self.files.lock().unwrap().contains_key(path);
                        return Ok(json!({
                            "stdout": "",
                            "stderr": "",
                            "exit_code": if exists { 0 } else { 1 }
                        }));
                    }
                    Ok(json!({
                        "stdout": "from-vm",
                        "stderr": "",
                        "exit_code": 0
                    }))
                }
                other => Err(format!("unknown tool {other}")),
            }
        }
        fn name(&self) -> &str {
            &self.name
        }
    }

    fn read_file_proposal(path: &str) -> ActionProposal {
        serde_json::from_value(json!({
            "source": "test",
            "actions": [{
                "id": "r0",
                "type": "tool_call",
                "tool": "read_file",
                "parameters": { "path": path },
                "dependencies": [],
            }],
        }))
        .expect("proposal deserializes")
    }

    /// The outbound message sink is attached to every session runtime, so
    /// `messaging.send` is a real, executable tool rather than a schema the
    /// runtime would refuse. `with_message_sink` registers the sink and the
    /// schema together, so seeing the schema proves the sink is there.
    #[tokio::test]
    async fn session_runtime_has_the_messaging_send_tool() {
        let state = Arc::new(ServerState::standalone(journal_dir()));
        let session = state
            .create_session("c-messaging", Arc::new(WsChannel::test_stub()))
            .await
            .unwrap();

        assert!(
            session
                .runtime
                .tools
                .read()
                .await
                .contains_key("messaging.send"),
            "every session must be able to execute messaging.send"
        );
    }

    /// Default (no bind): the session runtime keeps a `LocalSubstrate`.
    /// This is the backward-compat invariant — every existing consumer
    /// (GUI, A2A-over-WS, connectors-as-tools, coder) sees no change.
    #[tokio::test]
    async fn default_session_substrate_is_local() {
        let state = Arc::new(ServerState::standalone(journal_dir()));
        let channel = Arc::new(WsChannel::test_stub());
        let session = state.create_session("c-default", channel).await.unwrap();

        let sub = session.runtime.substrate().await;
        assert_eq!(
            sub.name(),
            "local",
            "an un-bound session must keep the default LocalSubstrate"
        );
        assert!(sub.is_local(), "default substrate must be the host");
    }

    /// Binding fails cleanly when the named connector is not connected —
    /// no panic, a descriptive error, and the substrate stays Local.
    #[tokio::test]
    async fn bind_unknown_connector_errors_and_keeps_local() {
        let state = Arc::new(ServerState::standalone(journal_dir()));
        let channel = Arc::new(WsChannel::test_stub());
        let session = state.create_session("c-missing", channel).await.unwrap();

        let err = state
            .bind_substrate_to_connector(&session, "nope")
            .await
            .unwrap_err();
        assert!(err.contains("not connected"), "got: {err}");
        assert_eq!(session.runtime.substrate().await.name(), "local");
    }

    /// A connector-driven session that binds its substrate routes the bare
    /// commodity built-ins (`read_file`/`write_file`) to that connector's
    /// environment via the engine fall-through, NOT to the WS client. We
    /// prove it end-to-end: write+read through the runtime land on the fake
    /// VM session's file map.
    #[tokio::test]
    async fn bound_session_routes_builtins_to_substrate() {
        let state = Arc::new(ServerState::standalone(journal_dir()));
        let channel = Arc::new(WsChannel::test_stub());
        let session = state.create_session("c-vm", channel).await.unwrap();

        // Register a connected "vm" connector session + its routes the way
        // the connector manager does on dial.
        let files = Arc::new(StdMutex::new(std::collections::HashMap::new()));
        let fake: Arc<Mutex<dyn McpSession>> = Arc::new(Mutex::new(FakeVmSession {
            name: "vm".into(),
            files: files.clone(),
        }));
        state.mcp_executor.add_session("vm", fake).await;

        // Bind the session to the vm connector's substrate.
        let bound = state
            .bind_substrate_to_connector(&session, "vm")
            .await
            .expect("bind succeeds for a connected connector");
        assert_eq!(bound, "vm");
        assert_eq!(session.runtime.substrate().await.name(), "vm");

        // write_file through the runtime → must hit the fake VM session,
        // not the WS client (which would hang/fail with no responder).
        let write = serde_json::from_value::<ActionProposal>(json!({
            "source": "test",
            "actions": [{
                "id": "w0",
                "type": "tool_call",
                "tool": "write_file",
                "parameters": { "path": "/vm/a.txt", "content": "vm-bytes" },
                "dependencies": [],
            }],
        }))
        .unwrap();
        let wres = session.runtime.execute(&write).await;
        assert!(
            wres.results[0].error.is_none(),
            "write_file via substrate failed: {:?}",
            wres.results[0].error
        );
        assert_eq!(
            files.lock().unwrap().get("/vm/a.txt").map(String::as_str),
            Some("vm-bytes"),
            "write_file must land on the bound VM substrate"
        );

        // read_file back must return the VM's content.
        let rres = session
            .runtime
            .execute(&read_file_proposal("/vm/a.txt"))
            .await;
        assert!(
            rres.results[0].error.is_none(),
            "read_file via substrate failed: {:?}",
            rres.results[0].error
        );
        let out = rres.results[0].output.clone().unwrap_or(Value::Null);
        let content = out.get("content").and_then(|v| v.as_str()).unwrap_or("");
        // read_file output is line-numbered `cat -n` style (H1/F4-remainder,
        // audit 2026-07-06); the VM's single-line content becomes "     1\t<line>".
        assert_eq!(
            content, "     1\tvm-bytes",
            "read_file must come from the VM substrate"
        );
    }

    /// The shadow executor: substrate-owned built-in names return the
    /// `"unknown tool"` prefix (so the engine falls through to the bound
    /// substrate), while any other tool name is delegated to the inner
    /// executor unchanged (connector `mcp_*` routes + host WS callbacks).
    #[tokio::test]
    async fn shadow_executor_only_shadows_builtin_names() {
        struct Inner;
        #[async_trait::async_trait]
        impl ToolExecutor for Inner {
            async fn execute(&self, tool: &str, _p: &Value) -> Result<Value, String> {
                Ok(json!({ "delegated": tool }))
            }
        }
        let shadow = SubstrateShadowExecutor::new(Arc::new(Inner));

        for name in SUBSTRATE_OWNED_TOOLS {
            let err = shadow.execute(name, &json!({})).await.unwrap_err();
            assert!(
                err.starts_with("unknown tool"),
                "{name} must be shadowed so the engine falls through: {err}"
            );
        }

        // A connector tool name and a pure tool pass through untouched.
        for passthrough in ["mcp_vm_run_command", "calculate", "browser.run"] {
            let out = shadow.execute(passthrough, &json!({})).await.unwrap();
            assert_eq!(out["delegated"], json!(passthrough));
        }
    }
}

/// `.car/policies/*.toml` reaching the runtime at all — the whole point of
/// wiring `load_project_policies` to a real call site. Exercised against a
/// temp directory rather than through `create_session`, which would read the
/// developer's real `$HOME`.
#[cfg(test)]
mod project_policy_loading_tests {
    use super::*;

    fn runtime() -> Runtime {
        Runtime::new()
    }

    #[tokio::test]
    async fn a_missing_policies_directory_is_silent() {
        let dir = tempfile::tempdir().unwrap();
        // No `.car` at all, let alone `.car/policies`.
        apply_project_policies(&runtime(), &dir.path().join(".car"))
            .await
            .expect("a project with no rules must start normally");
    }

    #[tokio::test]
    async fn an_empty_policies_directory_is_silent() {
        let dir = tempfile::tempdir().unwrap();
        let car = dir.path().join(".car");
        std::fs::create_dir_all(car.join("policies")).unwrap();
        apply_project_policies(&runtime(), &car).await.unwrap();
    }

    #[tokio::test]
    async fn a_well_formed_rule_file_loads_and_takes_effect() {
        let dir = tempfile::tempdir().unwrap();
        let car = dir.path().join(".car");
        std::fs::create_dir_all(car.join("policies")).unwrap();
        // Exactly the shape `car_policy::rules`' module docs advertise: a
        // top-level list for `deny_tool`, a table array for the per-parameter
        // rules.
        std::fs::write(
            car.join("policies").join("messaging.toml"),
            "deny_tool = [\"messaging.send\"]\n\n\
             [[deny_tool_param]]\n\
             tool = \"messaging.send\"\n\
             param = \"channel\"\n\
             equals = \"slack\"\n",
        )
        .unwrap();

        let rt = runtime();
        apply_project_policies(&rt, &car).await.unwrap();

        // Registered on the runtime's engine, not merely parsed.
        let names: Vec<String> = rt
            .list_policies(None)
            .await
            .unwrap()
            .into_iter()
            .map(|(name, _)| name)
            .collect();
        assert!(
            names.len() >= 2,
            "both declarative rules must reach the policy engine, got {names:?}"
        );
    }

    #[tokio::test]
    async fn a_malformed_rule_file_fails_loudly_and_names_the_file() {
        let dir = tempfile::tempdir().unwrap();
        let car = dir.path().join(".car");
        std::fs::create_dir_all(car.join("policies")).unwrap();
        std::fs::write(
            car.join("policies").join("broken.toml"),
            "[[deny_tool]\ntool = \"shell\"",
        )
        .unwrap();

        let err = apply_project_policies(&runtime(), &car)
            .await
            .expect_err("a malformed policy file must NOT be swallowed");
        assert!(
            err.contains("broken.toml"),
            "the operator has to know which file to fix: {err}"
        );
        assert!(err.contains("refusing to start"), "{err}");
    }
}

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

    fn granter_hex_for(secret: &[u8], user: &str) -> String {
        let master = car_sync::StretchedMaster::from_issued_high_entropy(secret, user);
        let vk = car_sync::ed25519_verifying(&car_sync::derive_ed25519_identity(&master, user));
        vk.to_bytes().iter().map(|b| format!("{b:02x}")).collect()
    }
    fn granter_hex() -> String {
        granter_hex_for(b"granter-login", "acc_granter")
    }

    #[test]
    fn unset_env_is_the_personal_only_path() {
        // The real-tenants-OFF invariant: no value → Ok(None) → base provider.
        assert!(ServerState::parse_org_scope_config(None).unwrap().is_none());
        assert!(ServerState::parse_org_scope_config(Some(String::new()))
            .unwrap()
            .is_none());
    }

    #[test]
    fn wellformed_env_parses_org_and_granter() {
        let hex = granter_hex();
        let (org, granters) = ServerState::parse_org_scope_config(Some(format!("orgtest:{hex}")))
            .unwrap()
            .expect("well-formed opt-in parses");
        assert_eq!(org, "orgtest");
        assert_eq!(granters.len(), 1);
        let expected_master =
            car_sync::StretchedMaster::from_issued_high_entropy(b"granter-login", "acc_granter");
        let expected = car_sync::ed25519_verifying(&car_sync::derive_ed25519_identity(
            &expected_master,
            "acc_granter",
        ));
        assert_eq!(granters[0].to_bytes(), expected.to_bytes());
    }

    #[test]
    fn multiple_granters_parse_in_order() {
        // The trusted set is plural: an org can designate several granters.
        let g1 = granter_hex_for(b"g1", "acc_g1");
        let g2 = granter_hex_for(b"g2", "acc_g2");
        let (org, granters) =
            ServerState::parse_org_scope_config(Some(format!("orgtest:{g1},{g2}")))
                .unwrap()
                .expect("well-formed multi-granter opt-in parses");
        assert_eq!(org, "orgtest");
        assert_eq!(granters.len(), 2);
        assert_ne!(granters[0].to_bytes(), granters[1].to_bytes());
        // Surrounding whitespace on a granter entry is tolerated.
        let spaced = ServerState::parse_org_scope_config(Some(format!("orgtest:{g1} , {g2}")))
            .unwrap()
            .unwrap();
        assert_eq!(spaced.1.len(), 2);
    }

    #[test]
    fn one_bad_granter_in_the_list_fails_the_whole_value() {
        // A typo in ANY granter is a loud failure — never silently drop it and
        // proceed with a partial trusted set.
        let good = granter_hex();
        assert!(
            ServerState::parse_org_scope_config(Some(format!("orgtest:{good},not-hex"))).is_err(),
            "a malformed granter among valid ones must be a hard error"
        );
        // An org id with no granters at all is rejected.
        assert!(ServerState::parse_org_scope_config(Some("orgtest:".into())).is_err());
        assert!(ServerState::parse_org_scope_config(Some("orgtest: , ".into())).is_err());
    }

    #[test]
    fn malformed_env_is_a_hard_error_never_silent() {
        // A typo must NOT silently fall through to a wrong/unauthenticated state.
        let hex = granter_hex();
        let bads = [
            "no-colon-here".to_string(),   // missing ':'
            ":deadbeef".to_string(),       // empty org
            "orgtest:not-hex".to_string(), // bad hex
            "orgtest:aabb".to_string(),    // wrong length
            format!("bad org:{hex}"),      // non-canonical org (space)
            format!("org/evil:{hex}"),     // non-canonical org (delimiter)
        ];
        for bad in &bads {
            assert!(
                ServerState::parse_org_scope_config(Some(bad.clone())).is_err(),
                "malformed value {bad:?} must be a hard error"
            );
        }
        // NOTE: an all-zero (identity/small-order) key PARSES — from_bytes accepts
        // the encoding; verify_strict rejects it at verification, so such a granter
        // simply grants nothing (fail-closed), which is correct, not a parse error.
    }

    #[test]
    fn scope_routing_sends_only_matching_org_to_the_org_subsystem() {
        use car_sync::Scope;
        let acme = Scope::Shared { org: "acme".into() };
        let globex = Scope::Shared {
            org: "globex".into(),
        };

        // org-scope OFF (no holder): EVERYTHING routes to the personal subsystem —
        // the byte-identical invariant.
        assert_eq!(route_for_scope(&Scope::Personal, None), SyncRoute::User);
        assert_eq!(route_for_scope(&acme, None), SyncRoute::User);

        // opted into "acme": only Scope::Shared{acme} routes to the org subsystem.
        assert_eq!(route_for_scope(&acme, Some("acme")), SyncRoute::Org);
        assert_eq!(
            route_for_scope(&Scope::Personal, Some("acme")),
            SyncRoute::User
        );
        assert_eq!(
            route_for_scope(&globex, Some("acme")),
            SyncRoute::User,
            "a different org must NOT go to acme's delivery subsystem"
        );
    }
}