car_server_core/

session.rs

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

use car_engine::{Runtime, ToolExecutor};
use car_eventlog::EventLog;
use car_proto::{ToolExecuteRequest, ToolExecuteResponse};
use futures::Sink;
use serde::{Deserialize, Serialize};
use serde_json::Value;
use std::collections::HashMap;
use std::path::PathBuf;
use std::pin::Pin;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::Arc;
use tokio::sync::{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);

/// 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 },
}

/// 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>>>,
    pub next_id: AtomicU64,
}

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()),
            next_id: AtomicU64::new(0),
        }
    }
}

/// 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,
}

/// 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,
    /// WS `client_id` that called `runs.start`. Lets disconnect cleanup
    /// find this run's owning connection.
    pub client_id: String,
    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>,
}

/// 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, Copy, 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,
}

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

    /// 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 => car_proto::RunLiveStatus::InProgress,
            Some(car_proto::RunTermination::Outcome { .. }) => {
                car_proto::RunLiveStatus::Completed
            }
            Some(car_proto::RunTermination::Incomplete) => car_proto::RunLiveStatus::Incomplete,
        }
    }

    /// 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)
}

/// Tool executor that sends callbacks to the client over WebSocket.
pub struct WsToolExecutor {
    pub channel: Arc<WsChannel>,
}

#[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> {
        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,
            attempt: 1,
        };

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

        // 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 = tokio::time::timeout(wait, rx)
            .await
            .map_err(|_| {
                format!(
                    "tool '{}' callback timed out ({}s)",
                    tool,
                    wait.as_secs()
                )
            })?
            .map_err(|_| format!("tool '{}' callback channel closed", tool))?;

        if let Some(err) = response.error {
            Err(err)
        } else {
            Ok(response.output.unwrap_or(Value::Null))
        }
    }
}

/// 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);
    }
}

/// 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: std::sync::atomic::AtomicBool,
    /// 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: 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>>,
}

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()
    }
}

/// 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 macOS-automation surface
/// (`automation.run_applescript`, `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.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>,
}

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,
        }
    }

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

/// Global server state shared across all connections.
pub struct ServerState {
    pub journal_dir: PathBuf,
    pub sessions: Mutex<HashMap<String, Arc<ClientSession>>>,
    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>>>,
    /// 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>>,
    /// 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.
    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>,
    /// Parslee cloud identity loaded from the user's OS keychain at
    /// daemon startup when `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>>>>,
    /// 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>>,
    /// 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>>,
    /// 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>>,
    /// Per-run disk-write serialization (FIX 6). The `runs` lock is
    /// released before [`record_run_turns`](Self::record_run_turns) does
    /// its blocking disk append (so I/O never stalls the global registry),
    /// so without this two concurrent batches for the same `run_id` could
    /// interleave their appended records on disk. Each run gets its own
    /// `Mutex<()>` held only across the append; entries are reclaimed when
    /// the run leaves the registry. Defense-in-depth: the canonical harness
    /// is sequential, but a misbehaving caller must not corrupt the trace.
    run_write_locks: Mutex<HashMap<String, Arc<Mutex<()>>>>,
    /// 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,
    /// 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,
    /// 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>>>,
}

impl ServerState {
    /// 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))
    }

    /// 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 {
        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 run_store = crate::run_store::RunStore::from_journal_dir(&cfg.journal_dir);
        // Adopt crash-orphaned runs BEFORE gc() (FIX 4). The in-memory map is
        // empty at boot, so any on-disk InProgress run is a crashed prior
        // process; mark it Incomplete so it's terminal and age-GC-eligible —
        // otherwise it leaks forever (GC never evicts in-progress runs).
        let _adopted = run_store.adopt_orphans();
        let _evicted = run_store.gc();
        Self {
            journal_dir: cfg.journal_dir,
            sessions: Mutex::new(HashMap::new()),
            inference,
            host: Arc::new(crate::host::HostState::new()),
            shared_memgine: cfg.shared_memgine,
            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(),
            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()),
            auth_token: std::sync::OnceLock::new(),
            host_token: std::sync::OnceLock::new(),
            parslee_session: std::sync::OnceLock::new(),
            attached_agents: Mutex::new(HashMap::new()),
            agent_memgines: Mutex::new(HashMap::new()),
            chat_sessions: Mutex::new(HashMap::new()),
            runs: Mutex::new(HashMap::new()),
            run_subscribers: Mutex::new(HashMap::new()),
            run_write_locks: Mutex::new(HashMap::new()),
            run_store,
            mcp_url: std::sync::OnceLock::new(),
            mcp_sessions: std::sync::OnceLock::new(),
            approval_gate: cfg.approval_gate.unwrap_or_default(),
        }
    }

    /// 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_parslee_session(
        &self,
        session: crate::parslee_auth::ParsleeSession,
    ) -> Result<(), crate::parslee_auth::ParsleeSession> {
        self.parslee_session.set(session)
    }

    /// 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 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.
    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)"
                );
            }
        }
    }

    pub async fn start_run(&self, meta: RunMeta) {
        let started = car_proto::RunStarted {
            run_id: meta.run_id.clone(),
            agent_id: meta.agent_id.clone(),
            intent: meta.intent.clone(),
            outcome_description: meta.outcome_description.clone(),
            started_at: meta.started_at,
        };
        // Insert under the runs lock; while still holding it, fan the
        // `Started` lifecycle event out to any subscribers. In practice
        // there are none yet (a client can only subscribe by `run_id`
        // after the run exists), but emitting keeps the broadcast path
        // uniform and correct if a subscribe ever races the insert.
        {
            let mut runs = self.runs.lock().await;
            runs.insert(meta.run_id.clone(), meta);
            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,
            );
        }
        if let Err(e) = self.run_store.write_started(&started) {
            tracing::warn!(run_id = %started.run_id, error = %e, "run-store: failed to persist RunStarted");
        }
    }

    /// 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 complete_run(
        &self,
        run_id: &str,
        termination: car_proto::RunTermination,
    ) -> Result<(), String> {
        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 meta.is_terminal() {
                return Err(format!(
                    "run `{run_id}` is already terminal — cannot record another outcome"
                ));
            }
            let ended_at = chrono::Utc::now();
            meta.termination = Some(termination.clone());
            meta.ended_at = Some(ended_at);
            let ended = car_proto::RunEnded {
                run_id: run_id.to_string(),
                agent_id: meta.agent_id.clone(),
                termination,
                ended_at,
            };
            // Fan the terminal event out under the runs lock so a
            // subscriber that registered just before this transition sees
            // the terminal record exactly once, after its last turn.
            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,
                run_id,
                &agent_id,
                car_proto::RunRecord::Ended(ended.clone()),
                cursor,
                status,
            );
            ended
        };
        if let Err(e) = self.run_store.write_ended(&ended) {
            tracing::warn!(run_id, error = %e, "run-store: failed to persist RunEnded");
        }
        // 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(run_id).await;
        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 mark_run_incomplete(&self, run_id: &str) -> bool {
        let ended = {
            let mut runs = self.runs.lock().await;
            match runs.get_mut(run_id) {
                Some(meta) if !meta.is_terminal() => {
                    let ended_at = chrono::Utc::now();
                    meta.termination = Some(car_proto::RunTermination::Incomplete);
                    meta.ended_at = Some(ended_at);
                    let ended = car_proto::RunEnded {
                        run_id: run_id.to_string(),
                        agent_id: meta.agent_id.clone(),
                        termination: car_proto::RunTermination::Incomplete,
                        ended_at,
                    };
                    // Notify subscribers of the terminal Incomplete under
                    // the runs lock (same contract as complete_run).
                    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::Ended(ended.clone()),
                        cursor,
                        car_proto::RunLiveStatus::Incomplete,
                    );
                    Some(ended)
                }
                _ => None,
            }
        };
        match ended {
            Some(ended) => {
                if let Err(e) = self.run_store.write_ended(&ended) {
                    tracing::warn!(run_id, error = %e, "run-store: failed to persist Incomplete");
                }
                // Heap hygiene (same rationale as `complete_run`): this run
                // just became terminal (`Incomplete`) and is flushed to disk
                // with its terminal event fanned out. Drop the in-memory turn
                // buffer; a late subscriber re-sources from disk. Only runs
                // that actually transitioned reach here (`Some(ended)`); a
                // no-op (already-terminal or unknown) leaves nothing to clear.
                self.clear_terminal_run_turns(run_id).await;
                true
            }
            None => false,
        }
    }

    /// 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();
            }
        }
        drop(runs);
        // FIX 6 hygiene: a terminal run accepts no more turns, so its
        // per-run disk-write lock is dead weight. Drop the map entry to keep
        // `run_write_locks` from growing unbounded. Any in-flight append
        // already holds its own `Arc<Mutex<()>>` clone, so removing the map
        // entry can't pull the rug out from under it.
        self.run_write_locks.lock().await.remove(run_id);
    }

    /// 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)`. The owning agent id
    /// is the only heap allocation; the turn BUFFER is never cloned (ADV-2).
    ///
    /// 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)> {
        self.runs
            .lock()
            .await
            .get(run_id)
            .map(|m| (m.agent_id.clone(), m.is_terminal(), m.turns.len()))
    }

    /// 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. The disk append happens for
    /// exactly the records that were accepted into the in-memory buffer
    /// (so disk and memory stay in lock-step), keyed by the run's owning
    /// `agent_id`. Disk failures are logged, never fatal.
    pub async fn record_run_turns(
        &self,
        run_id: &str,
        mut records: Vec<car_proto::RunRecord>,
    ) -> RecordRunTurnsOutcome {
        // Mutate the in-memory buffer under the lock, capturing the owning
        // agent_id and the accepted records so the disk append (which does
        // its own blocking I/O) happens after the lock is released.
        //
        // Invariant #1: the append AND the per-turn fanout both happen
        // while holding the `runs` lock, so they are serialized with
        // `runs.subscribe`'s snapshot+register (which holds the same
        // lock). A subscriber registered at cursor C therefore receives
        // exactly the turns appended after C — the snapshot covered ≤ C —
        // with no gap and no dup at the boundary.
        //
        // ADV-1: the runaway ceiling is enforced HERE, under the same lock,
        // BEFORE the append. The dispatcher runs a task per frame, so
        // pipelined batches each pass the handler's lock-free pre-check
        // against a stale snapshot; only this under-lock check is atomic
        // with the append, so it is the one that actually caps the run. A
        // batch that would take the run PAST the ceiling is refused WHOLE
        // (no partial admit) and reported back distinguishably so the
        // handler emits `run_turn_limit`, not `run_terminal`.
        enum Locked {
            Persist(String, Vec<car_proto::RunRecord>, usize),
            RefusedCeiling,
            UnknownOrTerminal,
        }
        let locked = {
            let mut runs = self.runs.lock().await;
            match runs.get_mut(run_id) {
                Some(meta) if !meta.is_terminal() => {
                    // 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 {
                        Locked::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;
                            }
                        }
                        meta.turns.extend(records.iter().cloned());
                        let len = meta.turns.len();
                        let agent_id = meta.agent_id.clone();
                        let status = meta.live_status();
                        // Fan each newly-appended record out AFTER it is in the
                        // buffer, in order, carrying its 1-based post-append
                        // cursor (`base + offset + 1`).
                        let subs = self.run_subscribers.lock().await;
                        for (offset, record) in records.iter().enumerate() {
                            Self::fanout_locked(
                                &subs,
                                run_id,
                                &agent_id,
                                record.clone(),
                                base + offset + 1,
                                status,
                            );
                        }
                        drop(subs);
                        Locked::Persist(agent_id, records, len)
                    }
                }
                _ => Locked::UnknownOrTerminal,
            }
        };
        match locked {
            Locked::Persist(agent_id, records, len) => {
                // FIX 6: serialize the disk append per run. The `runs` lock
                // is released above (so blocking I/O doesn't stall the
                // global registry), so two concurrent batches for the same
                // run could otherwise interleave their writes on disk. The
                // per-run write lock makes each batch's append atomic w.r.t.
                // other batches; `append_turns` itself writes the whole
                // batch in a single `write_all` so a batch is never torn
                // mid-record by a concurrent appender.
                let write_guard = self.run_write_lock(run_id).await;
                let _held = write_guard.lock().await;
                if let Err(e) = self.run_store.append_turns(&agent_id, run_id, &records) {
                    tracing::warn!(run_id, error = %e, "run-store: failed to persist turns");
                }
                RecordRunTurnsOutcome::Appended { new_total: len }
            }
            Locked::RefusedCeiling => RecordRunTurnsOutcome::RefusedCeiling,
            Locked::UnknownOrTerminal => RecordRunTurnsOutcome::UnknownOrTerminal,
        }
    }

    /// Per-run disk-write mutex (FIX 6). Returns (creating on first use) the
    /// `Mutex` that serializes appends for `run_id`, so concurrent
    /// `record_run_turns` batches for the same run can't interleave on disk.
    /// Keyed by `run_id`; the map entry is reclaimed when the run goes
    /// terminal (see [`Self::clear_terminal_run_turns`]).
    async fn run_write_lock(&self, run_id: &str) -> Arc<Mutex<()>> {
        let mut locks = self.run_write_locks.lock().await;
        locks
            .entry(run_id.to_string())
            .or_insert_with(|| Arc::new(Mutex::new(())))
            .clone()
    }

    /// 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 async fn subscribe_run(
        &self,
        run_id: &str,
        host_client_id: &str,
        channel: Arc<WsChannel>,
    ) -> Option<car_proto::RunSubscribeResponse> {
        // 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 runs = self.runs.lock().await;
        let meta = runs.get(run_id)?;
        let mut snapshot = meta.turns.clone();
        let agent_id = meta.agent_id.clone();
        let status = meta.live_status();
        // A TERMINAL run whose in-memory turns were evicted (heap hygiene at
        // `complete_run`/`mark_run_incomplete`) must re-source its snapshot
        // from disk, or a late subscriber to a just-completed run would get
        // an empty snapshot. This is sound WITHOUT widening the lock: a
        // terminal run takes no further appends, so its disk trail is final
        // and immutable — there is no snapshot/register race to guard, which
        // is the only thing the `runs` lock protects here. We therefore keep
        // the no-blocking-I/O-under-the-async-mutex discipline (matching the
        // write paths) and defer the disk read until after the lock drops.
        let needs_disk_snapshot = meta.is_terminal() && snapshot.is_empty();
        // Register under the same lock — snapshot+register stays atomic for
        // the live (in-progress) path; only the terminal-and-evicted SOURCE
        // moves to disk, and that case has no concurrent appends.
        {
            let mut subs = self.run_subscribers.lock().await;
            // A re-subscribe replaces the prior entry; dropping the old
            // RunTraceSubscriber drops its sender, ending its drain task.
            subs.insert((run_id.to_string(), host_client_id.to_string()), subscriber);
        }
        drop(runs);

        if needs_disk_snapshot {
            // Load the durable trail (the same source `runs.get_trace` uses)
            // and keep only `RunRecord::Turn`s — `turns_so_far` is turns-only
            // by contract (`Started`/`Ended` are conveyed by `agent_id` /
            // `status`), so this matches the live in-memory snapshot exactly.
            if let Some(records) = self.run_store.get_run_trace(run_id) {
                snapshot = records
                    .into_iter()
                    .filter(|r| matches!(r, car_proto::RunRecord::Turn(_)))
                    .collect();
            }
        }
        let cursor = snapshot.len();

        Some(car_proto::RunSubscribeResponse {
            run_id: run_id.to_string(),
            agent_id,
            turns_so_far: snapshot,
            cursor,
            status,
        })
    }

    /// 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, client_id: &str) {
        // Snapshot the still-open runs owned by this connection.
        let pending: Vec<String> = {
            let runs = self.runs.lock().await;
            runs.values()
                .filter(|m| m.client_id == client_id && !m.is_terminal())
                .map(|m| m.run_id.clone())
                .collect()
        };
        if pending.is_empty() {
            return;
        }
        // Short grace window so an in-flight `runs.complete` (dispatched
        // on a spawned task that may outrace the close frame) can land
        // its terminal record before we conclude the run was abandoned.
        tokio::time::sleep(RUN_COMPLETE_GRACE).await;
        for run_id in pending {
            self.mark_run_incomplete(&run_id).await;
        }
    }

    pub async fn create_session(
        &self,
        client_id: &str,
        channel: Arc<WsChannel>,
    ) -> Arc<ClientSession> {
        let journal_path = self.journal_dir.join(format!("{}.jsonl", client_id));
        let event_log = EventLog::with_journal(journal_path);

        let executor = Arc::new(WsToolExecutor {
            channel: channel.clone(),
        });

        let runtime = Runtime::new()
            .with_event_log(event_log)
            .with_executor(executor);

        // If the embedder supplied a shared memgine, every session uses it.
        // Otherwise each session gets its own — matches pre-extraction behavior.
        let memgine = match &self.shared_memgine {
            Some(eng) => eng.clone(),
            None => Arc::new(Mutex::new(car_memgine::MemgineEngine::new(None))),
        };

        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: std::sync::atomic::AtomicBool::new(false),
            is_host: std::sync::atomic::AtomicBool::new(false),
            agent_id: tokio::sync::Mutex::new(None),
            bound_memgine: tokio::sync::Mutex::new(None),
            current_run_id: tokio::sync::Mutex::new(None),
        });

        self.sessions
            .lock()
            .await
            .insert(client_id.to_string(), session.clone());

        session
    }

    /// 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>> {
        let removed = self.sessions.lock().await.remove(client_id);
        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 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);
            // 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;
            // 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.
            self.sweep_runs_for_disconnect(client_id).await;
        }
        removed
    }
}

#[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());
    }
}