car_server_core/

handler.rs

//! WebSocket connection handler — bidirectional JSON-RPC.
//!
//! Tool callback flow:
//! 1. Client submits proposal via proposal.submit
//! 2. Runtime encounters a ToolCall action
//! 3. WsToolExecutor sends tools.execute request to client via shared write half
//! 4. WsToolExecutor awaits response on a oneshot channel
//! 5. Client executes tool locally, sends JSON-RPC response back
//! 6. Handler receives the response, resolves the oneshot
//! 7. Runtime continues execution with the tool result

use crate::session::{A2aRouteAuth, ServerState, WsChannel};
use car_proto::*;
use car_verify;
use futures::StreamExt;
use serde::{Deserialize, Serialize};
use serde_json::Value;
use std::collections::HashMap;
use std::net::SocketAddr;
use std::sync::atomic::AtomicU64;
use std::sync::Arc;
use tokio::net::TcpStream;
use tokio::sync::Mutex;
use tokio_tungstenite::{accept_async, tungstenite::Message};
use tracing::{info, instrument};

#[derive(Debug, Deserialize)]
#[allow(dead_code)]
pub struct JsonRpcMessage {
    #[serde(default)]
    pub jsonrpc: String,
    #[serde(default)]
    pub method: Option<String>,
    #[serde(default)]
    pub params: Value,
    #[serde(default)]
    pub id: Value,
    // Response fields
    #[serde(default)]
    pub result: Option<Value>,
    #[serde(default)]
    pub error: Option<Value>,
}

#[derive(Debug, Serialize)]
pub struct JsonRpcResponse {
    pub jsonrpc: &'static str,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub result: Option<Value>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub error: Option<JsonRpcError>,
    pub id: Value,
}

#[derive(Debug, Serialize)]
pub struct JsonRpcError {
    pub code: i32,
    pub message: String,
}

impl JsonRpcResponse {
    pub fn success(id: Value, result: Value) -> Self {
        Self {
            jsonrpc: "2.0",
            result: Some(result),
            error: None,
            id,
        }
    }
    pub fn error(id: Value, code: i32, message: &str) -> Self {
        Self {
            jsonrpc: "2.0",
            result: None,
            error: Some(JsonRpcError {
                code,
                message: message.to_string(),
            }),
            id,
        }
    }
}

/// Convenience wrapper for the standalone `car-server` binary: accepts
/// the WebSocket handshake on a raw [`TcpStream`] then delegates to
/// [`run_dispatch`]. Embedders that already have a handshake-completed
/// `WebSocketStream` skip this and call `run_dispatch` directly.
#[instrument(
    name = "ws.connection",
    skip_all,
    fields(peer = %peer),
)]
pub async fn handle_connection(
    stream: TcpStream,
    peer: SocketAddr,
    state: Arc<ServerState>,
) -> Result<(), Box<dyn std::error::Error>> {
    let ws_stream = accept_async(stream).await?;
    let (write, read) = ws_stream.split();
    run_dispatch(read, Box::pin(write), peer.to_string(), state).await
}

/// Convenience wrapper for the daemon-as-default Unix-socket
/// listener. Same shape as [`handle_connection`] but accepts a
/// `UnixStream` — used by the per-user UDS listener in
/// `car-server::main` (default transport for FFI thin clients,
/// since UDS is faster + permission-scoped vs localhost TCP).
///
/// Unix-only — `tokio::net::UnixStream` is gated on
/// `cfg(all(unix, feature = "net"))`. On Windows the daemon binds
/// only the TCP listener (loopback) and this entry point is
/// compiled out; consumers must use `handle_connection` instead.
#[cfg(unix)]
#[instrument(
    name = "ws.connection",
    skip_all,
    fields(peer = %peer),
)]
pub async fn handle_connection_unix(
    stream: tokio::net::UnixStream,
    peer: String,
    state: Arc<ServerState>,
) -> Result<(), Box<dyn std::error::Error>> {
    let ws_stream = tokio_tungstenite::accept_async(stream).await?;
    let (write, read) = ws_stream.split();
    run_dispatch(read, Box::pin(write), peer, state).await
}

/// Transport-neutral entry point: drives the JSON-RPC dispatch loop
/// against an already-handshake-completed split WebSocket. Generic
/// over the read half (any `Stream<Item = Result<Message, WsError>>`)
/// and the write half (a [`WsSink`] — type-erased so this function
/// doesn't templatize every downstream consumer of `WsChannel`).
///
/// `peer` is a free-form string ("127.0.0.1:1234" for TCP,
/// "uds:/path/sock" for UDS, "axum:..." for embedders) — used only
/// for tracing fields, never for dispatch logic.
#[instrument(
    name = "ws.dispatch",
    skip_all,
    fields(client_id = tracing::field::Empty, peer = %peer),
)]
pub async fn run_dispatch<R>(
    mut read: R,
    write: crate::session::WsSink,
    peer: String,
    state: Arc<ServerState>,
) -> Result<(), Box<dyn std::error::Error>>
where
    R: futures::Stream<Item = Result<Message, tokio_tungstenite::tungstenite::Error>>
        + Unpin
        + Send,
{
    let client_id = uuid::Uuid::new_v4().simple().to_string()[..12].to_string();
    tracing::Span::current().record("client_id", &client_id.as_str());

    info!("New connection from {}", peer);

    let channel = Arc::new(WsChannel {
        write: Mutex::new(write),
        pending: Mutex::new(HashMap::new()),
        next_id: AtomicU64::new(1),
    });

    let session = state.create_session(&client_id, channel.clone()).await;

    // car#209: per-request handlers are spawned detached (below) and
    // each clones `Arc<ClientSession>` → `Arc<WsChannel>`. A bare
    // `tokio::spawn` outlives the connection, so a slow/hung handler
    // pins the split sink and the inbound socket lingers in CLOSED
    // until the daemon hits EMFILE. Own them in a per-connection
    // `JoinSet` so every in-flight handler is aborted the instant the
    // WS drops (`abort_all` in the cleanup block; `JoinSet`'s Drop
    // also aborts on any early return), releasing the FD immediately.
    let mut conn_tasks: tokio::task::JoinSet<()> = tokio::task::JoinSet::new();

    while let Some(msg) = read.next().await {
        // Reap finished handlers so a long-lived connection doesn't
        // retain their `JoinHandle`s unbounded (memory, not FDs).
        while conn_tasks.try_join_next().is_some() {}

        // car#209: on an abrupt transport error (peer reset / network
        // drop — the trader crash-loop case) `read.next()` yields
        // `Some(Err(_))`. The old `let msg = msg?;` propagated that
        // out of the function, *skipping the cleanup block below*, so
        // `state.sessions` (+ the host/a2ui/chat registries) kept the
        // session + channel — and thus the socket FD — forever. Break
        // instead: every disconnect, clean or not, runs the single
        // cleanup path.
        let msg = match msg {
            Ok(m) => m,
            Err(e) => {
                info!("read error from {}: {}; closing", client_id, e);
                break;
            }
        };
        if msg.is_text() {
            let text = match msg.to_text() {
                Ok(t) => t,
                Err(e) => {
                    info!("non-text frame from {}: {}; closing", client_id, e);
                    break;
                }
            };
            let parsed: JsonRpcMessage = match serde_json::from_str(text) {
                Ok(m) => m,
                Err(e) => {
                    send_response(
                        &session.channel,
                        JsonRpcResponse::error(Value::Null, -32700, &format!("Parse error: {}", e)),
                    )
                    .await
                    .ok();
                    continue;
                }
            };

            // Is this a response to a pending tool callback?
            if parsed.method.is_none() && (parsed.result.is_some() || parsed.error.is_some()) {
                if let Some(id_str) = parsed.id.as_str() {
                    let mut pending = session.channel.pending.lock().await;
                    if let Some(tx) = pending.remove(id_str) {
                        let tool_resp = if let Some(result) = parsed.result {
                            ToolExecuteResponse {
                                action_id: id_str.to_string(),
                                output: Some(result),
                                error: None,
                            }
                        } else {
                            let err_msg = parsed
                                .error
                                .as_ref()
                                .and_then(|e| e.get("message"))
                                .and_then(|m| m.as_str())
                                .unwrap_or("unknown error")
                                .to_string();
                            ToolExecuteResponse {
                                action_id: id_str.to_string(),
                                output: None,
                                error: Some(err_msg),
                            }
                        };
                        let _ = tx.send(tool_resp);
                        continue;
                    }
                }
            }

            // Agent → host chat-event interceptor. `agent.chat.event`
            // notifications coming up the WS from a connected agent
            // are forwarded to the originating host's channel as
            // `agents.chat.event`. Lives ahead of the regular method
            // dispatch so the dispatcher doesn't reply with
            // "method-not-found" on what is a fire-and-forget
            // notification (no id). See
            // `docs/proposals/agent-chat-surface.md`.
            if try_forward_agent_chat_event(&parsed, &state).await {
                continue;
            }

            // Otherwise it's a client request
            if let Some(method) = &parsed.method {
                info!(method = %method, "dispatching JSON-RPC method");

                // Auth gate (Parslee-ai/car-releases#32). When the
                // server has an auth token installed, every method
                // other than `session.auth` is rejected on
                // unauthenticated sessions and the connection is
                // closed after the error response goes out. When no
                // token is installed (default), this branch never
                // fires — preserves pre-#32 behaviour.
                if state.auth_token.get().is_some()
                    && !session
                        .authenticated
                        .load(std::sync::atomic::Ordering::Acquire)
                    && method != "session.auth"
                {
                    let resp = JsonRpcResponse::error(
                        parsed.id.clone(),
                        -32001,
                        "auth required: send `session.auth` with the per-launch token \
                         from ~/Library/Application Support/ai.parslee.car/auth-token \
                         (macOS), $XDG_RUNTIME_DIR/ai.parslee.car/auth-token (Linux), \
                         or %LOCALAPPDATA%\\ai.parslee.car\\auth-token (Windows) \
                         as the first frame on this connection",
                    );
                    let _ = send_response(&session.channel, resp).await;
                    info!(client = %client_id, method = %method,
                        "rejecting non-auth method on unauthenticated session; closing");
                    break;
                }

                // Approval gate (audit 2026-05). High-risk methods —
                // anything that drives macOS automation or sends
                // messages on the user's behalf — must be acked by
                // the user via `host.resolve_approval` before they
                // dispatch. The gate raises an `approval.requested`
                // host event the local UI can render approve/deny on,
                // then parks until resolved or the configured
                // timeout fires. Returns a JSON-RPC error and
                // continues the dispatch loop on deny / timeout —
                // no connection close, since the caller may want to
                // retry with revised parameters.
                if state.approval_gate.requires_approval(method.as_str()) {
                    match gate_high_risk_method(method.as_str(), &parsed.params, &state).await {
                        Ok(()) => {}
                        Err(reason) => {
                            let resp = JsonRpcResponse::error(parsed.id.clone(), -32003, &reason);
                            let _ = send_response(&session.channel, resp).await;
                            info!(
                                client = %client_id,
                                method = %method,
                                reason = %reason,
                                "approval gate blocked dispatch"
                            );
                            continue;
                        }
                    }
                }

                // Spawn the per-method dispatch in a task so the read
                // loop keeps reading frames. Without this, methods
                // that trigger server-initiated `tools.execute`
                // callbacks (`proposal.submit`, `workflow.run`,
                // `multi.*` paths that fire a registered tool)
                // deadlock the connection: the handler awaits the
                // callback response on a oneshot, but the response is
                // another frame on this same read half — which the
                // synchronous `.await` here would prevent the loop
                // from ever picking up. Surfaced by the
                // `executeProposal: echo tool via JS callback` smoke
                // (#173). Response ordering becomes id-keyed (the
                // JSON-RPC demuxing contract) rather than
                // arrival-ordered.
                let session_task = session.clone();
                let state_task = state.clone();
                let method_owned = method.clone();
                let parsed_task = parsed;
                // car#209: owned by the per-connection JoinSet so it's
                // aborted on disconnect instead of leaking the channel.
                conn_tasks.spawn(async move {
                    let session = session_task;
                    let state = state_task;
                    let parsed = parsed_task;
                    let result = match method_owned.as_str() {
                        "session.auth" => handle_session_auth(&parsed, &session, &state).await,
                        "parslee.auth" => handle_parslee_auth().await,
                        "auth.start" => handle_auth_start(&parsed).await,
                        "auth.complete" => handle_auth_complete(&parsed).await,
                        "auth.status" => handle_auth_status().await,
                        "auth.logout" => handle_auth_logout().await,
                        "session.init" => handle_session_init(&parsed, &session).await,
                        "host.subscribe" => handle_host_subscribe(&session, &state).await,
                        "host.agents" => handle_host_agents(&session).await,
                        "host.events" => handle_host_events(&parsed, &session).await,
                        "host.approvals" => handle_host_approvals(&session).await,
                        "host.register_agent" => {
                            handle_host_register_agent(&parsed, &session).await
                        }
                        "host.unregister_agent" => {
                            handle_host_unregister_agent(&parsed, &session).await
                        }
                        "host.set_status" => handle_host_set_status(&parsed, &session).await,
                        "host.notify" => handle_host_notify(&parsed, &session).await,
                        "host.request_approval" => {
                            handle_host_request_approval(&parsed, &session).await
                        }
                        "host.resolve_approval" => {
                            handle_host_resolve_approval(&parsed, &session).await
                        }
                        "tools.register" => handle_tools_register(&parsed, &session).await,
                        "proposal.submit" => {
                            handle_proposal_submit(&parsed, &session, &state).await
                        }
                        "policy.register" => handle_policy_register(&parsed, &session).await,
                        "session.policy.open" => handle_session_policy_open(&session).await,
                        "session.policy.close" => {
                            handle_session_policy_close(&parsed, &session).await
                        }
                        "verify" => handle_verify(&parsed, &session).await,
                        "state.get" => handle_state_get(&parsed, &session).await,
                        "state.set" => handle_state_set(&parsed, &session).await,
                        "state.exists" => handle_state_exists(&parsed, &session).await,
                        "state.keys" => handle_state_keys(&parsed, &session).await,
                        "state.snapshot" => handle_state_snapshot(&parsed, &session).await,
                        "memory.add_fact" => handle_memory_add_fact(&parsed, &session).await,
                        "memory.query" => handle_memory_query(&parsed, &session).await,
                        "memory.build_context" => {
                            handle_memory_build_context(&parsed, &session).await
                        }
                        "memory.build_context_fast" => {
                            handle_memory_build_context_fast(&parsed, &session).await
                        }
                        "memory.consolidate" => handle_memory_consolidate(&session).await,
                        "memory.fact_count" => handle_memory_fact_count(&session).await,
                        "memory.persist" => handle_memory_persist(&parsed, &session).await,
                        "memory.load" => handle_memory_load(&parsed, &session).await,
                        "skill.ingest" => handle_skill_ingest(&parsed, &session).await,
                        "skill.find" => handle_skill_find(&parsed, &session).await,
                        "skill.report" => handle_skill_report(&parsed, &session).await,
                        "skill.repair" => handle_skill_repair(&parsed, &session).await,
                        "skills.ingest_distilled" => {
                            handle_skills_ingest_distilled(&parsed, &session).await
                        }
                        "skills.evolve" => handle_skills_evolve(&parsed, &session).await,
                        "skills.domains_needing_evolution" => {
                            handle_skills_domains_needing_evolution(&parsed, &session).await
                        }
                        "skills.ingest_provisional" => {
                            handle_skills_ingest_provisional(&parsed, &session).await
                        }
                        "skills.gate" => handle_skills_gate(&parsed, &session).await,
                        "skill.meta" => handle_skill_meta(&parsed, &session).await,
                        "skill.export" => handle_skill_export(&parsed, &session).await,
                        "skill.import" => handle_skill_import(&parsed, &session).await,
                        "multi.swarm" => handle_multi_swarm(&parsed, &session).await,
                        "multi.pipeline" => handle_multi_pipeline(&parsed, &session).await,
                        "multi.supervisor" => handle_multi_supervisor(&parsed, &session).await,
                        "multi.map_reduce" => handle_multi_map_reduce(&parsed, &session).await,
                        "multi.vote" => handle_multi_vote(&parsed, &session).await,
                        "multi.tournament" => handle_multi_tournament(&parsed, &session).await,
                        "multi.subtask" => handle_multi_subtask(&parsed, &session).await,
                        "scheduler.create" => handle_scheduler_create(&parsed),
                        "scheduler.run" => handle_scheduler_run(&parsed, &session).await,
                        "scheduler.run_loop" => handle_scheduler_run_loop(&parsed, &session).await,
                        "infer" => handle_infer(&parsed, &state, &session).await,
                        "image.generate" => handle_image_generate(&parsed, &state).await,
                        "video.generate" => handle_video_generate(&parsed, &state).await,
                        "embed" => handle_embed(&parsed, &state).await,
                        "classify" => handle_classify(&parsed, &state).await,
                        "tokenize" => handle_tokenize(&parsed, &state).await,
                        "detokenize" => handle_detokenize(&parsed, &state).await,
                        "rerank" => handle_rerank(&parsed, &state).await,
                        "transcribe" => handle_transcribe(&parsed, &state).await,
                        "synthesize" => handle_synthesize(&parsed, &state).await,
                        "infer_stream" => handle_infer_stream(&parsed, &session, &state).await,
                        "speech.prepare" => handle_speech_prepare(&state).await,
                        "models.route" => handle_models_route(&parsed, &state).await,
                        "models.stats" => handle_models_stats(&state).await,
                        "outcomes.resolve_pending" => {
                            handle_outcomes_resolve_pending(&parsed, &state).await
                        }
                        "events.count" => handle_events_count(&session).await,
                        "events.stats" => handle_events_stats(&session).await,
                        "events.truncate" => handle_events_truncate(&parsed, &session).await,
                        "events.clear" => handle_events_clear(&session).await,
                        // Agent run tracing, U1 — run lifecycle bracket.
                        // `runs.start` mints a run_id and tags the
                        // session's current run BEFORE responding (KTD3);
                        // `runs.complete` records the terminal outcome.
                        // The per-turn recorder (U2) and the read/
                        // subscribe RPCs (U4/U5) build on these.
                        "runs.start" => handle_runs_start(&parsed, &session, &state).await,
                        "runs.complete" => {
                            handle_runs_complete(&parsed, &session, &state).await
                        }
                        // Client-narrated turn append (feedback-agent
                        // A2UI/runs plan, U1). WS-only (no FFI method, like
                        // runs.subscribe) — an out-of-pipeline agent whose
                        // work happens inside its own subprocess pushes full
                        // RunTurns it built itself. Authorized to the run's
                        // OWNING agent only (host-token/unbound writers are
                        // rejected as forgery); appends through the shared
                        // `record_run_turns` (index re-stamp, persist, fanout).
                        "runs.record_turns" => {
                            handle_runs_record_turns(&parsed, &session, &state).await
                        }
                        // Live run-trace subscribe/unsubscribe (U4). WS-only
                        // (no FFI method) — CarHost consumes the
                        // `runs.trace.event` notification. Authorization
                        // gated per-run (R16/KTD10).
                        "runs.subscribe" => {
                            handle_runs_subscribe(&parsed, &session, &state).await
                        }
                        "runs.unsubscribe" => {
                            handle_runs_unsubscribe(&parsed, &session, &state).await
                        }
                        // Replay reads (U5). WS-only (no FFI method, like
                        // `runs.subscribe`) — CarHost lists an agent's runs
                        // and fetches a completed run's full trace from the
                        // disk store (works across restart / client_id
                        // churn). Authorization gated per-agent/run (R16).
                        "runs.list" => handle_runs_list(&parsed, &session, &state).await,
                        "runs.get_trace" => {
                            handle_runs_get_trace(&parsed, &session, &state).await
                        }
                        "replan.set_config" => handle_replan_set_config(&parsed, &session).await,
                        "models.list" => handle_models_list(&state),
                        "models.register" => handle_models_register(&parsed, &state).await,
                        "models.unregister" => handle_models_unregister(&parsed, &state).await,
                        "models.list_unified" => handle_models_list_unified(&state),
                        "models.search" => handle_models_search(&parsed, &state),
                        "models.recommend" => handle_models_recommend(&parsed, &state),
                        "models.setup_plan" => handle_models_setup_plan(&parsed, &state),
                        "models.upgrades" => handle_models_upgrades(&state),
                        "models.detect_upgrades" => handle_models_detect_upgrades(&state).await,
                        "models.check_upgrade_nudge" => {
                            handle_models_check_upgrade_nudge(&parsed, &state).await
                        }
                        "models.dismiss_upgrade" => {
                            handle_models_dismiss_upgrade(&parsed, &state)
                        }
                        "models.check_concierge" => {
                            handle_models_check_concierge(&parsed, &state).await
                        }
                        "models.dismiss_suggestion" => {
                            handle_models_dismiss_suggestion(&parsed, &state)
                        }
                        "models.update_prefs_get" => handle_models_update_prefs_get(&state),
                        "models.update_prefs_set" => {
                            handle_models_update_prefs_set(&parsed, &state)
                        }
                        "models.pull" => handle_models_pull(&parsed, &state).await,
                        "models.install" => handle_models_pull(&parsed, &state).await,
                        "skills.distill" => handle_skills_distill(&parsed, &state).await,
                        "skills.list" => handle_skills_list(&parsed, &session).await,
                        "browser.run" => handle_browser_run(&parsed, &session).await,
                        "browser.close" => handle_browser_close(&session).await,
                        "secret.put" => handle_secret_put(&parsed),
                        "secret.get" => handle_secret_get(&parsed),
                        "secret.delete" => handle_secret_delete(&parsed),
                        "secret.status" => handle_secret_status(&parsed),
                        "secret.available" => Ok(car_ffi_common::secrets::is_available()),
                        "permissions.status" => handle_perm_status(&parsed),
                        "permissions.request" => handle_perm_request(&parsed),
                        "permissions.explain" => handle_perm_explain(&parsed),
                        "permissions.domains" => Ok(car_ffi_common::permissions::domains()),
                        "accounts.list" => car_ffi_common::accounts::list(),
                        "accounts.open" => {
                            #[derive(serde::Deserialize, Default)]
                            struct OpenParams {
                                #[serde(default)]
                                account_id: Option<String>,
                            }
                            let p: OpenParams =
                                serde_json::from_value(parsed.params.clone()).unwrap_or_default();
                            car_ffi_common::accounts::open_settings(p.account_id.as_deref())
                        }
                        "calendar.list" => car_ffi_common::integrations::calendar_list(),
                        "calendar.events" => handle_calendar_events(&parsed),
                        "calendar.create_event" => handle_calendar_create_event(&parsed),
                        "calendar.update_event" => handle_calendar_update_event(&parsed),
                        "calendar.delete_event" => handle_calendar_delete_event(&parsed),
                        "contacts.containers" => {
                            car_ffi_common::integrations::contacts_containers()
                        }
                        "contacts.find" => handle_contacts_find(&parsed),
                        "mail.accounts" => car_ffi_common::integrations::mail_accounts(),
                        "mail.inbox" => handle_mail_inbox(&parsed),
                        "mail.send" => handle_mail_send(&parsed),
                        "messages.services" => car_ffi_common::integrations::messages_services(),
                        "messages.chats" => handle_messages_chats(&parsed),
                        "messages.send" => handle_messages_send(&parsed),
                        "notes.accounts" => car_ffi_common::integrations::notes_accounts(),
                        "notes.find" => handle_notes_find(&parsed),
                        "reminders.lists" => car_ffi_common::integrations::reminders_lists(),
                        "reminders.items" => handle_reminders_items(&parsed),
                        "photos.albums" => car_ffi_common::integrations::photos_albums(),
                        "bookmarks.list" => handle_bookmarks_list(&parsed),
                        "files.locations" => car_ffi_common::integrations::files_locations(),
                        "keychain.status" => car_ffi_common::integrations::keychain_status(),
                        "health.status" => car_ffi_common::health::status(),
                        "health.sleep" => handle_health_sleep(&parsed),
                        "health.workouts" => handle_health_workouts(&parsed),
                        "health.activity" => handle_health_activity(&parsed),
                        "voice.transcribe_stream.start" => {
                            handle_voice_transcribe_stream_start(&parsed, &state, &session).await
                        }
                        "voice.transcribe_stream.stop" => {
                            handle_voice_transcribe_stream_stop(&parsed, &state).await
                        }
                        "voice.transcribe_stream.push" => {
                            handle_voice_transcribe_stream_push(&parsed, &state).await
                        }
                        "voice.tts_stream.start" => {
                            handle_voice_tts_stream_start(&parsed, &session).await
                        }
                        "voice.tts_stream.cancel" => handle_voice_tts_stream_cancel(&parsed).await,
                        "voice.tts_stream.list" => Ok(handle_voice_tts_stream_list()),
                        "voice.sessions.list" => Ok(handle_voice_sessions_list(&state)),
                        "voice.dispatch_turn" => {
                            handle_voice_dispatch_turn(&parsed, &state, &session).await
                        }
                        "voice.cancel_turn" => handle_voice_cancel_turn().await,
                        "voice.prewarm_turn" => handle_voice_prewarm_turn(&state).await,
                        "inference.register_runner" => {
                            handle_inference_register_runner(&session).await
                        }
                        "inference.runner.event" => handle_inference_runner_event(&parsed).await,
                        "inference.runner.complete" => {
                            handle_inference_runner_complete(&parsed).await
                        }
                        "inference.runner.fail" => handle_inference_runner_fail(&parsed).await,
                        "voice.providers.list" => {
                            // Stateless: enumerates STT/TTS providers compiled into
                            // this build. Runtime readiness (API key, permission,
                            // model download) is reported via per-provider errors.
                            serde_json::from_str::<serde_json::Value>(
                                &car_voice::list_voice_providers_json(),
                            )
                            .map_err(|e| e.to_string())
                        }
                        "voice.prepare_parakeet" => car_ffi_common::voice::prepare_parakeet()
                            .await
                            .and_then(|j| serde_json::from_str(&j).map_err(|e| e.to_string())),
                        "voice.prepare_diarizer" => car_ffi_common::voice::prepare_diarizer()
                            .await
                            .and_then(|j| serde_json::from_str(&j).map_err(|e| e.to_string())),
                        "voice.enroll_speaker" => handle_enroll_speaker(&parsed).await,
                        "voice.list_enrollments" => car_ffi_common::voice::list_enrollments()
                            .and_then(|j| serde_json::from_str(&j).map_err(|e| e.to_string())),
                        "voice.remove_enrollment" => handle_remove_enrollment(&parsed),
                        "workflow.run" => handle_workflow_run(&parsed, &session).await,
                        "workflow.resume" => handle_workflow_resume(&parsed, &session).await,
                        "builder.build" => handle_builder_build(&parsed, &state, &session).await,
                        "workflow.verify" => handle_workflow_verify(&parsed),
                        "meeting.start" => handle_meeting_start(&parsed, &state, &session).await,
                        "meeting.stop" => handle_meeting_stop(&parsed, &state, &session).await,
                        "meeting.list" => handle_meeting_list(&parsed),
                        "meeting.get" => handle_meeting_get(&parsed),
                        "registry.register" => handle_registry_register(&parsed),
                        "registry.heartbeat" => handle_registry_heartbeat(&parsed),
                        "registry.unregister" => handle_registry_unregister(&parsed),
                        "registry.list" => handle_registry_list(&parsed),
                        "registry.reap" => handle_registry_reap(&parsed),
                        "admission.status" => handle_admission_status(&state),
                        "a2a.start" => handle_a2a_start(&parsed, &session).await,
                        "a2a.stop" => handle_a2a_stop(),
                        "a2a.status" => handle_a2a_status(),
                        "a2a.send" => handle_a2a_send(&parsed, &state).await,
                        "a2ui.apply" => handle_a2ui_apply(&parsed, &state).await,
                        "a2ui.ingest" => handle_a2ui_ingest(&parsed, &state).await,
                        "a2ui.capabilities" => handle_a2ui_capabilities(&state),
                        "a2ui.reap" => handle_a2ui_reap(&state).await,
                        "a2ui.surfaces" => handle_a2ui_surfaces(&state).await,
                        "a2ui.get" => handle_a2ui_get(&parsed, &state).await,
                        "a2ui.action" => handle_a2ui_action(&parsed, &state).await,
                        "a2ui.render_report" => handle_a2ui_render_report(&parsed, &state).await,
                        "a2ui/subscribe" => handle_a2ui_subscribe(&session, &state).await,
                        "a2ui/unsubscribe" => handle_a2ui_unsubscribe(&session, &state).await,
                        "a2ui/replay" => handle_a2ui_replay(&parsed, &state).await,
                        "automation.run_applescript" => handle_run_applescript(&parsed).await,
                        "automation.shortcuts.list" => handle_list_shortcuts(&parsed).await,
                        "automation.shortcuts.run" => handle_run_shortcut(&parsed).await,
                        "notifications.local" => handle_local_notification(&parsed).await,
                        "vision.ocr" => handle_vision_ocr(&parsed).await,
                        "agents.list" => handle_agents_list(&state).await,
                        "agents.health" => handle_agents_health(&state).await,
                        "agents.upsert" => handle_agents_upsert(&parsed, &state).await,
                        "agents.install" => handle_agents_install(&parsed, &state).await,
                        "agents.remove" => handle_agents_remove(&parsed, &state).await,
                        "agents.start" => handle_agents_start(&parsed, &state).await,
                        "agents.stop" => handle_agents_stop(&parsed, &state).await,
                        "agents.restart" => handle_agents_restart(&parsed, &state).await,
                        "agents.tail_log" => handle_agents_tail_log(&parsed, &state).await,
                        "agents.list_external" => handle_agents_list_external(&parsed).await,
                        "agents.detect_external" => handle_agents_detect_external(&parsed).await,
                        "agents.health_external" => handle_agents_health_external(&parsed).await,
                        "agents.invoke_external" => {
                            handle_agents_invoke_external(&parsed, &state, &session).await
                        }
                        "agents.chat" => handle_agents_chat(&parsed, &state, &session).await,
                        "agents.chat.cancel" => handle_agents_chat_cancel(&parsed, &state).await,
                        // A2A v1.0 (PascalCase) + v0.3 (slash form) — both
                        // alias to the same in-core dispatcher per
                        // Parslee-ai/car-releases#28. Embedders that need a
                        // custom AgentCardSource / TaskStore plug them in
                        // via ServerStateConfig::with_a2a_card_source /
                        // with_a2a_store before any handler runs.
                        "message/send"
                        | "SendMessage"
                        | "message/stream"
                        | "SendStreamingMessage"
                        | "tasks/get"
                        | "GetTask"
                        | "tasks/list"
                        | "ListTasks"
                        | "tasks/cancel"
                        | "CancelTask"
                        | "tasks/resubscribe"
                        | "SubscribeToTask"
                        | "tasks/pushNotificationConfig/set"
                        | "CreateTaskPushNotificationConfig"
                        | "tasks/pushNotificationConfig/get"
                        | "GetTaskPushNotificationConfig"
                        | "tasks/pushNotificationConfig/list"
                        | "ListTaskPushNotificationConfigs"
                        | "tasks/pushNotificationConfig/delete"
                        | "DeleteTaskPushNotificationConfig"
                        | "agent/getAuthenticatedExtendedCard"
                        | "GetExtendedAgentCard" => {
                            handle_a2a_dispatch(method_owned.as_str(), &parsed, &state).await
                        }
                        _ => Err(format!("unknown method: {}", method_owned)),
                    };

                    let resp = match result {
                        Ok(value) => JsonRpcResponse::success(parsed.id, value),
                        Err(e) => JsonRpcResponse::error(parsed.id, -32603, &e),
                    };
                    let _ = send_response(&session.channel, resp).await;
                });
            }
        } else if msg.is_binary() {
            // CAR binary frame transport — see `car_ffi_common::voice::binary`
            // for the canonical header definition. 26-byte fixed header
            // followed by an opaque payload. Inbound type 0x01 carries
            // 16-bit signed LE PCM into a `pcm_push` session; other
            // types (0x02 TTS chunk, 0x03 final marker, 0x04 error)
            // are server-emitted and rejected here.
            let bytes = msg.into_data();
            let parsed = match car_ffi_common::voice::binary::parse_frame(&bytes) {
                Ok(p) => p,
                Err(e) => {
                    tracing::warn!("binary frame from {} rejected: {}", client_id, e);
                    continue;
                }
            };
            match parsed.frame_type {
                car_ffi_common::voice::binary::FRAME_TYPE_INBOUND_PCM => {
                    let registry = state.voice_sessions.clone();
                    let payload_owned = parsed.payload.to_vec();
                    let session_id_owned = parsed.session_id_hex.clone();
                    conn_tasks.spawn(async move {
                        if let Err(e) = car_ffi_common::voice::transcribe_stream_push(
                            &session_id_owned,
                            &payload_owned,
                            registry,
                        )
                        .await
                        {
                            tracing::warn!(
                                "binary PCM push to session {} failed: {}",
                                session_id_owned,
                                e
                            );
                        }
                    });
                }
                other => {
                    tracing::debug!(
                        "binary frame type {:#04x} from {} not accepted server-side",
                        other,
                        client_id
                    );
                }
            }
        } else if msg.is_close() {
            info!("Client {} disconnected", client_id);
            break;
        }
    }

    // car#209: abort every in-flight handler for this connection
    // *first* — they each hold an `Arc<ClientSession>` → `Arc<WsChannel>`
    // clone; until they're gone the split sink (and the inbound socket
    // FD) can't drop, even after the registries below are cleared.
    conn_tasks.abort_all();

    session.host.unsubscribe(&client_id).await;
    // Auto-cancel this session's pending approvals so the queue stays
    // in sync with what's actually decidable — covers graceful
    // unregister+close, hard crash (TCP reset), and ping timeout in
    // one place. System-level gate approvals (client_id None) are not
    // touched. car-releases#48.
    session.host.reap_session_approvals(&client_id).await;
    state.a2ui_subscribers.lock().await.remove(&client_id);

    // Fix for MULTI-4 / WS-3: drop the session from the registry and
    // drain any pending tool callbacks. Without this, every connection
    // we ever accepted keeps an `Arc<ClientSession>` alive in
    // `state.sessions`, and outstanding `oneshot::Sender`s in
    // `session.channel.pending` outlive the closed connection until
    // their per-call timeout (the action budget, or `DEFAULT_TOOL_TIMEOUT_MS`
    // — no longer a hardcoded 60s, see car#259). Dropping the senders here causes any
    // awaiting `recv()` in `WsToolExecutor::execute` to return
    // `RecvError` immediately, which the existing error-handler path
    // already maps to "callback channel closed" — same shape as the
    // timeout path, just faster.
    let _removed = state.remove_session(&client_id).await;
    {
        let mut pending = session.channel.pending.lock().await;
        pending.clear();
    }

    Ok(())
}

async fn send_response(
    channel: &WsChannel,
    resp: JsonRpcResponse,
) -> Result<(), Box<dyn std::error::Error>> {
    use futures::SinkExt;
    let json = serde_json::to_string(&resp)?;
    channel
        .write
        .lock()
        .await
        .send(Message::Text(json.into()))
        .await?;
    Ok(())
}

// --- Request handlers ---

async fn handle_host_subscribe(
    session: &crate::session::ClientSession,
    state: &Arc<ServerState>,
) -> Result<Value, String> {
    session
        .host
        .subscribe(&session.client_id, session.channel.clone())
        .await;
    serde_json::to_value(HostSnapshot {
        subscribed: true,
        agents: session.host.agents().await,
        approvals: session.host.approvals().await,
        events: session.host.events(50).await,
        identity: Some(daemon_identity(state)),
    })
    .map_err(|e| e.to_string())
}

/// Snapshot the daemon-identity facts for a fresh subscriber.
/// Cheap: non-acquiring reads on `OnceLock`s + a single
/// `to_string_lossy` on the manifest path. Critically uses
/// [`ServerState::supervisor_if_installed`] — not the lazy-init
/// `supervisor()` — so a Heisenberg subscribe can't *cause* the
/// daemon to acquire the manifest lock just by asking whether it
/// owns one.
fn daemon_identity(state: &Arc<ServerState>) -> car_proto::HostIdentity {
    // Observer takes precedence: when both supervisor and observer
    // markers are set (currently unreachable through the standalone
    // binary, but an embedder could install both racily), the
    // observer marker is the authoritative role since the
    // supervisor handle is only installed when this daemon owns
    // the lock.
    let (manifest_path, manifest_role) = if let Some(p) = state.observer_manifest_path() {
        (
            Some(p.to_string_lossy().into_owned()),
            car_proto::HostManifestRole::Observer,
        )
    } else if let Some(s) = state.supervisor_if_installed() {
        (
            Some(s.manifest_path().to_string_lossy().into_owned()),
            car_proto::HostManifestRole::Owner,
        )
    } else {
        (None, car_proto::HostManifestRole::None)
    };
    car_proto::HostIdentity {
        version: env!("CARGO_PKG_VERSION").to_string(),
        pid: std::process::id(),
        manifest_path,
        manifest_role,
        parslee: state
            .parslee_session
            .get()
            .map(|session| session.identity.clone()),
    }
}

/// Return the Parslee cloud credential for an authenticated CAR
/// connection. Managed agents already receive `CAR_AUTH_TOKEN` and
/// authenticate to the local daemon with `session.auth`; this method is
/// their supported bridge from local CAR auth to the user's Parslee
/// backend auth.
///
/// The bearer token is intentionally not injected into every managed
/// child process environment. Agents ask for it only when they need it,
/// through the same local auth gate that protects the rest of the
/// daemon.
async fn handle_parslee_auth() -> Result<Value, String> {
    let session = crate::parslee_auth::load_or_refresh()
        .await?
        .ok_or_else(|| "Parslee account not authenticated; run `car auth login`".to_string())?;
    Ok(serde_json::json!({
        "authenticated": true,
        "token_type": "Bearer",
        "access_token": session.access_token,
        "authorization_header": format!("Bearer {}", session.access_token),
        "identity": session.identity,
    }))
}

// --- auth.* : GUI-driven Parslee sign-in (CAR Host.app).
// Shares one implementation with `car auth login` via the car-auth
// crate. Stateless: the trusted in-process GUI holds the PKCE
// verifier + state between auth.start and auth.complete (same trust
// boundary as the keychain it ultimately writes).
async fn handle_auth_start(req: &JsonRpcMessage) -> Result<Value, String> {
    let api_base = car_auth::api_base(req.params.get("api_base").and_then(|v| v.as_str()));
    let client_id = req
        .params
        .get("client_id")
        .and_then(|v| v.as_str())
        .unwrap_or("parslee-car");
    let redirect_uri = req
        .params
        .get("redirect_uri")
        .and_then(|v| v.as_str())
        .ok_or_else(|| "redirect_uri is required".to_string())?;
    let provider = req.params.get("provider").and_then(|v| v.as_str());
    let state = car_auth::new_state();
    let verifier = car_auth::pkce_verifier();
    let challenge = car_auth::pkce_challenge(&verifier);
    let url =
        car_auth::authorize_url(&api_base, client_id, redirect_uri, &state, &challenge, provider)?;
    Ok(serde_json::json!({
        "authorize_url": url,
        "state": state,
        "verifier": verifier,
    }))
}

async fn handle_auth_complete(req: &JsonRpcMessage) -> Result<Value, String> {
    let api_base = car_auth::api_base(req.params.get("api_base").and_then(|v| v.as_str()));
    let client_id = req
        .params
        .get("client_id")
        .and_then(|v| v.as_str())
        .unwrap_or("parslee-car");
    let redirect_uri = req
        .params
        .get("redirect_uri")
        .and_then(|v| v.as_str())
        .ok_or_else(|| "redirect_uri is required".to_string())?;
    let code = req
        .params
        .get("code")
        .and_then(|v| v.as_str())
        .ok_or_else(|| "code is required".to_string())?;
    let verifier = req
        .params
        .get("verifier")
        .and_then(|v| v.as_str())
        .ok_or_else(|| "verifier is required".to_string())?;
    let token =
        car_auth::exchange_code(&api_base, client_id, redirect_uri, code, verifier).await?;
    car_auth::store_tokens(&api_base, &token)?;
    Ok(serde_json::json!({ "ok": true }))
}

async fn handle_auth_status() -> Result<Value, String> {
    match car_auth::fetch_status(None).await? {
        Some(session_json) => {
            let session: Value = serde_json::from_str(&session_json).unwrap_or(Value::Null);
            Ok(serde_json::json!({ "authenticated": true, "session": session }))
        }
        None => Ok(serde_json::json!({ "authenticated": false })),
    }
}

async fn handle_auth_logout() -> Result<Value, String> {
    car_auth::clear_tokens()?;
    Ok(serde_json::json!({ "ok": true }))
}

async fn handle_host_agents(session: &crate::session::ClientSession) -> Result<Value, String> {
    serde_json::to_value(session.host.agents().await).map_err(|e| e.to_string())
}

async fn handle_host_events(
    req: &JsonRpcMessage,
    session: &crate::session::ClientSession,
) -> Result<Value, String> {
    let limit = req
        .params
        .get("limit")
        .and_then(|v| v.as_u64())
        .unwrap_or(100) as usize;
    serde_json::to_value(session.host.events(limit).await).map_err(|e| e.to_string())
}

async fn handle_host_approvals(session: &crate::session::ClientSession) -> Result<Value, String> {
    serde_json::to_value(session.host.approvals().await).map_err(|e| e.to_string())
}

async fn handle_a2ui_apply(
    req: &JsonRpcMessage,
    state: &Arc<ServerState>,
) -> Result<Value, String> {
    #[derive(Deserialize)]
    struct Params {
        #[serde(default)]
        envelope: Option<car_a2ui::A2uiEnvelope>,
        #[serde(default)]
        message: Option<car_a2ui::A2uiEnvelope>,
    }

    let envelope = if req.params.get("createSurface").is_some()
        || req.params.get("updateComponents").is_some()
        || req.params.get("updateDataModel").is_some()
        || req.params.get("deleteSurface").is_some()
    {
        serde_json::from_value::<car_a2ui::A2uiEnvelope>(req.params.clone())
            .map_err(|e| e.to_string())?
    } else {
        match serde_json::from_value::<Params>(req.params.clone()) {
            Ok(params) => params
                .envelope
                .or(params.message)
                .ok_or_else(|| "`a2ui.apply` requires an A2UI envelope".to_string())?,
            Err(_) => serde_json::from_value::<car_a2ui::A2uiEnvelope>(req.params.clone())
                .map_err(|e| e.to_string())?,
        }
    };

    apply_a2ui_envelope(state, envelope, None, None).await
}

async fn handle_a2ui_ingest(
    req: &JsonRpcMessage,
    state: &Arc<ServerState>,
) -> Result<Value, String> {
    #[derive(Deserialize)]
    #[serde(rename_all = "camelCase")]
    struct Params {
        #[serde(default)]
        endpoint: Option<String>,
        #[serde(default)]
        a2a_endpoint: Option<String>,
        #[serde(default)]
        owner: Option<car_a2ui::A2uiSurfaceOwner>,
        #[serde(default)]
        route_auth: Option<A2aRouteAuth>,
        #[serde(default)]
        allow_untrusted_endpoint: bool,
    }

    let params = serde_json::from_value::<Params>(req.params.clone()).unwrap_or(Params {
        endpoint: None,
        a2a_endpoint: None,
        owner: None,
        route_auth: None,
        allow_untrusted_endpoint: false,
    });
    let payload = req.params.get("payload").unwrap_or(&req.params);
    state
        .a2ui
        .validate_payload(payload)
        .map_err(|e| e.to_string())?;
    let envelopes = car_a2ui::envelopes_from_value(payload).map_err(|e| e.to_string())?;
    if envelopes.is_empty() {
        return Err("no A2UI envelopes found in payload".into());
    }
    let endpoint = params.endpoint.or(params.a2a_endpoint);
    let endpoint = trusted_route_endpoint(endpoint, params.allow_untrusted_endpoint);
    let owner = params
        .owner
        .or_else(|| car_a2ui::owner_from_value(payload))
        .map(|owner| match endpoint.clone() {
            Some(endpoint) => owner.with_endpoint(Some(endpoint)),
            None => owner,
        });

    let mut results = Vec::new();
    for envelope in envelopes {
        let value =
            apply_a2ui_envelope(state, envelope, owner.clone(), params.route_auth.clone()).await?;
        results.push(value);
    }
    Ok(serde_json::json!({ "applied": results }))
}

async fn apply_a2ui_envelope(
    state: &Arc<ServerState>,
    envelope: car_a2ui::A2uiEnvelope,
    owner: Option<car_a2ui::A2uiSurfaceOwner>,
    route_auth: Option<A2aRouteAuth>,
) -> Result<Value, String> {
    let result = state
        .a2ui
        .apply_with_owner(envelope, owner)
        .await
        .map_err(|e| e.to_string())?;
    update_a2ui_route_auth(state, &result, route_auth).await;
    let kind = if result.deleted {
        "a2ui.surface_deleted"
    } else {
        "a2ui.surface_updated"
    };
    let message = if result.deleted {
        format!("A2UI surface {} deleted", result.surface_id)
    } else {
        format!("A2UI surface {} updated", result.surface_id)
    };
    let payload = serde_json::to_value(&result).map_err(|e| e.to_string())?;
    state
        .host
        .record_event(kind, None, message, payload.clone())
        .await;
    // Push the envelope result to every WS subscriber as an
    // `a2ui.event` notification — Parslee-ai/car-releases#29. Late
    // joiners catch up via `a2ui/replay` (or `a2ui.surfaces`).
    broadcast_a2ui_event(state, kind, &payload).await;
    serde_json::to_value(result).map_err(|e| e.to_string())
}

async fn broadcast_a2ui_event(state: &Arc<ServerState>, kind: &str, result: &Value) {
    use futures::SinkExt;
    use tokio_tungstenite::tungstenite::Message;
    let subscribers: Vec<Arc<crate::session::WsChannel>> = state
        .a2ui_subscribers
        .lock()
        .await
        .values()
        .cloned()
        .collect();
    if subscribers.is_empty() {
        return;
    }
    let Ok(json) = serde_json::to_string(&serde_json::json!({
        "jsonrpc": "2.0",
        "method": "a2ui.event",
        "params": {
            "kind": kind,
            "result": result,
        },
    })) else {
        return;
    };
    for channel in subscribers {
        let _ = channel
            .write
            .lock()
            .await
            .send(Message::Text(json.clone().into()))
            .await;
    }
}

async fn update_a2ui_route_auth(
    state: &Arc<ServerState>,
    result: &car_a2ui::A2uiApplyResult,
    route_auth: Option<A2aRouteAuth>,
) {
    let mut auth = state.a2ui_route_auth.lock().await;
    if result.deleted {
        auth.remove(&result.surface_id);
        return;
    }

    let has_route_endpoint = result
        .surface
        .as_ref()
        .and_then(|surface| surface.owner.as_ref())
        .and_then(|owner| owner.endpoint.as_ref())
        .is_some();
    match (has_route_endpoint, route_auth) {
        (true, Some(route_auth)) => {
            auth.insert(result.surface_id.clone(), route_auth);
        }
        _ => {
            auth.remove(&result.surface_id);
        }
    }
}

fn handle_a2ui_capabilities(state: &Arc<ServerState>) -> Result<Value, String> {
    serde_json::to_value(state.a2ui.capabilities()).map_err(|e| e.to_string())
}

async fn handle_a2ui_reap(state: &Arc<ServerState>) -> Result<Value, String> {
    let removed = state.a2ui.reap_expired(chrono::Utc::now()).await;
    if !removed.is_empty() {
        let mut auth = state.a2ui_route_auth.lock().await;
        for surface_id in &removed {
            auth.remove(surface_id);
        }
    }
    Ok(serde_json::json!({ "removed": removed }))
}

async fn handle_a2ui_surfaces(state: &Arc<ServerState>) -> Result<Value, String> {
    serde_json::to_value(state.a2ui.list().await).map_err(|e| e.to_string())
}

async fn handle_a2ui_get(req: &JsonRpcMessage, state: &Arc<ServerState>) -> Result<Value, String> {
    let surface_id = req
        .params
        .get("surface_id")
        .or_else(|| req.params.get("surfaceId"))
        .and_then(Value::as_str)
        .ok_or_else(|| "`a2ui.get` requires surface_id".to_string())?;
    serde_json::to_value(state.a2ui.get(surface_id).await).map_err(|e| e.to_string())
}

/// `a2ui/subscribe` — opt this WS connection into `a2ui.event`
/// notifications. Subscribers receive every `apply_a2ui_envelope`
/// result for as long as they're connected; the cleanup hook in
/// `run_dispatch` removes them on disconnect. Closes
/// Parslee-ai/car-releases#29.
async fn handle_a2ui_subscribe(
    session: &crate::session::ClientSession,
    state: &Arc<ServerState>,
) -> Result<Value, String> {
    state
        .a2ui_subscribers
        .lock()
        .await
        .insert(session.client_id.clone(), session.channel.clone());
    Ok(serde_json::json!({ "subscribed": true }))
}

/// `a2ui/unsubscribe` — opt out of `a2ui.event` notifications.
/// Idempotent: returns `{ subscribed: false }` regardless of prior
/// state.
async fn handle_a2ui_unsubscribe(
    session: &crate::session::ClientSession,
    state: &Arc<ServerState>,
) -> Result<Value, String> {
    state
        .a2ui_subscribers
        .lock()
        .await
        .remove(&session.client_id);
    Ok(serde_json::json!({ "subscribed": false }))
}

/// `a2ui/replay` — fetch the current state of one surface. Intended
/// for late joiners and reconnect: a client calls `subscribe`, then
/// `replay` once per surface it's tracking, and from then on
/// notifications keep it in sync. Equivalent to `a2ui.get` on the
/// surface store; lives in the subscribe namespace for
/// discoverability.
async fn handle_a2ui_replay(
    req: &JsonRpcMessage,
    state: &Arc<ServerState>,
) -> Result<Value, String> {
    let surface_id = req
        .params
        .get("surface_id")
        .or_else(|| req.params.get("surfaceId"))
        .and_then(Value::as_str)
        .ok_or_else(|| "`a2ui/replay` requires surface_id".to_string())?;
    serde_json::to_value(state.a2ui.get(surface_id).await).map_err(|e| e.to_string())
}

async fn handle_a2ui_action(
    req: &JsonRpcMessage,
    state: &Arc<ServerState>,
) -> Result<Value, String> {
    let action: car_a2ui::ClientAction =
        serde_json::from_value(req.params.clone()).map_err(|e| e.to_string())?;
    let owner = state.a2ui.owner(&action.surface_id).await;

    // Deliver the interaction to A2UI subscribers on the same channel that
    // carries surface updates — Parslee-ai/car-releases#58. The owning agent
    // listens for `a2ui.event` (that's how it gets `surface_updated`); before
    // this it saw nothing on click, so its surfaces were display-only.
    // Broadcast BEFORE the A2A route below: the local notification must not be
    // gated behind a (possibly slow) outbound A2A round-trip, and `route` is
    // deliberately excluded so an A2A endpoint URL/response isn't fanned out to
    // unrelated subscribers — it stays in the `host.event` record and the RPC
    // return for the privileged consumers that already see it.
    let action_result = serde_json::json!({
        "surfaceId": action.surface_id,
        "action": action,
        "owner": owner,
    });
    broadcast_a2ui_event(state, "a2ui.action", &action_result).await;

    let route = route_a2ui_action(state, &action, owner.clone()).await;
    let payload = serde_json::json!({
        "action": action,
        "owner": owner,
        "route": route,
    });
    let event = state
        .host
        .record_event(
            "a2ui.action",
            None,
            format!(
                "A2UI action {} from {}",
                action.name, action.source_component_id
            ),
            payload,
        )
        .await;
    Ok(serde_json::json!({
        "event": event,
        "route": route,
    }))
}

/// `a2ui.render_report` — renderer-emitted telemetry envelope
/// (Parslee-ai/car#180). Fire-and-forget; we record an event in
/// the host log (so dev tools and the conversation log see it) and
/// broadcast as `a2ui.event { kind: "a2ui.render_report", result }`
/// to every WS subscriber. The improvement agent reads the report
/// and decides whether to issue a follow-up `patchComponents`.
async fn handle_a2ui_render_report(
    req: &JsonRpcMessage,
    state: &Arc<ServerState>,
) -> Result<Value, String> {
    // Parse into the typed struct to enforce the schema; we
    // re-serialize for the event/broadcast payload so downstream
    // consumers don't have to defensively re-validate.
    let report: car_a2ui::RenderReport =
        serde_json::from_value(req.params.clone()).map_err(|e| e.to_string())?;
    let payload = serde_json::to_value(&report).map_err(|e| e.to_string())?;
    let kind = "a2ui.render_report";
    let message = format!("A2UI render report for surface {}", report.surface_id);
    let event = state
        .host
        .record_event(kind, None, message, payload.clone())
        .await;
    broadcast_a2ui_event(state, kind, &payload).await;

    // Hand the report to the in-process UI-improvement agent. The
    // agent is sync but cheap; we await the surface lookup before
    // calling it so the strategies see the same surface state the
    // renderer saw at report-emit time. Best-effort: surface lookup
    // misses (the surface might have been deleted between
    // report-emit and report-receive) are logged and skipped, never
    // surfaced as JSON-RPC errors — render_report is fire-and-forget.
    if let Some(surface) = state.a2ui.get(&report.surface_id).await {
        // Iteration budget — runaway-loop backstop. try_consume
        // claims the slot atomically; if the surface is already at
        // the cap, we short-circuit before consulting the agent.
        // The slot stays consumed only if the patch actually
        // applies — failure paths refund.
        if !state.ui_agent_budget.try_consume(&report.surface_id) {
            tracing::warn!(
                surface_id = %report.surface_id,
                count = state.ui_agent_budget.count(&report.surface_id),
                max = state.ui_agent_budget.max(),
                "ui-agent iteration budget exhausted; skipping agent invocation"
            );
            return Ok(serde_json::json!({ "event": event }));
        }
        // From here on, every non-applied branch must `refund` the
        // slot we just claimed. Only the successful-apply branch
        // keeps it consumed.
        match state.ui_agent.on_render_report(&report, &surface) {
            car_ui_agent::Decision::Patch {
                envelope,
                strategy_id,
                patch_hash,
                elapsed_ns,
            } => {
                // Runtime-side convergence monitor — neo's deferred
                // ask. The agent's no-double-patch guard catches
                // same-sequence repeats; this catches A→B→A across
                // sequences by tracking recent patch hashes per
                // surface. When a proposal would repeat one in the
                // window, drop it; the loop waits for the next
                // signature change.
                if !state
                    .ui_agent_oscillation
                    .check_and_record(&report.surface_id, patch_hash)
                {
                    tracing::warn!(
                        surface_id = %report.surface_id,
                        strategy = %strategy_id,
                        patch_hash,
                        "ui-agent oscillation detected; suppressing patch"
                    );
                    // Suppressed patch never applied → release the
                    // budget slot we claimed above.
                    state.ui_agent_budget.refund(&report.surface_id);
                    return Ok(serde_json::json!({ "event": event }));
                }
                let a2ui_envelope = car_a2ui::A2uiEnvelope {
                    patch_components: Some(envelope),
                    ..Default::default()
                };
                if let Err(e) = apply_a2ui_envelope(state, a2ui_envelope, None, None).await {
                    tracing::warn!(
                        surface_id = %report.surface_id,
                        strategy = %strategy_id,
                        patch_hash,
                        elapsed_ns,
                        error = %e,
                        "ui-agent patch apply failed",
                    );
                    // Apply failed → release the budget slot.
                    state.ui_agent_budget.refund(&report.surface_id);
                } else {
                    tracing::debug!(
                        surface_id = %report.surface_id,
                        strategy = %strategy_id,
                        patch_hash,
                        elapsed_ns,
                        iteration = state.ui_agent_budget.count(&report.surface_id),
                        "ui-agent patch applied",
                    );
                    // Memgine trace: one Conversation node per
                    // successful patch, tagged "ui-agent/<surface>".
                    // Spreading activation can surface these on
                    // future renders of related surfaces. Spawned
                    // off the render-report hot path — ingest walks
                    // the graph for spreading activation and can
                    // take real time; we don't want two surfaces
                    // churning patches to serialize through the
                    // memgine mutex behind the WS handler.
                    if let Some(memgine) = state.shared_memgine.clone() {
                        let speaker = format!("ui-agent/{}", report.surface_id);
                        let text = format!("strategy applied: {}", strategy_id);
                        tokio::spawn(async move {
                            let mut guard = memgine.lock().await;
                            guard.ingest_conversation(&speaker, &text, chrono::Utc::now());
                        });
                    }
                }
            }
            car_ui_agent::Decision::StableNoChange => {
                // No patch this round → release the slot.
                state.ui_agent_budget.refund(&report.surface_id);
            }
            car_ui_agent::Decision::HardStop { reason } => {
                state.ui_agent_budget.refund(&report.surface_id);
                // Renderer painted unknown components — contract
                // violation between server and renderer. Per
                // types.rs docs: "loud failure" + "MUST pause."
                // `error!` not `warn!` so it surfaces in production
                // logs at the right severity.
                tracing::error!(
                    surface_id = %report.surface_id,
                    reason = %reason,
                    "ui-agent hard-stopped improvement loop",
                );
            }
        }
    } else {
        tracing::debug!(
            surface_id = %report.surface_id,
            "ui-agent skipped — surface not found in store",
        );
    }

    Ok(serde_json::json!({ "event": event }))
}

async fn route_a2ui_action(
    state: &Arc<ServerState>,
    action: &car_a2ui::ClientAction,
    owner: Option<car_a2ui::A2uiSurfaceOwner>,
) -> Value {
    let Some(owner) = owner else {
        return serde_json::json!({ "delivered": false, "reason": "surface has no owner" });
    };
    if owner.kind != "a2a" {
        return serde_json::json!({ "delivered": false, "reason": "unsupported owner kind", "owner": owner });
    }
    let Some(endpoint) = owner.endpoint.clone() else {
        return serde_json::json!({
            "delivered": false,
            "reason": "surface owner has no endpoint",
            "owner": owner
        });
    };

    let message = car_a2a::Message {
        message_id: format!("a2ui-action-{}", uuid::Uuid::new_v4().simple()),
        role: car_a2a::MessageRole::User,
        parts: vec![car_a2a::Part::Data(car_a2a::types::DataPart {
            data: serde_json::json!({
                "a2uiAction": action,
            }),
            metadata: Default::default(),
        })],
        task_id: owner.task_id.clone(),
        context_id: owner.context_id.clone(),
        metadata: Default::default(),
    };

    let auth = state
        .a2ui_route_auth
        .lock()
        .await
        .get(&action.surface_id)
        .cloned()
        .map(client_auth_from_route_auth)
        .unwrap_or(car_a2a::ClientAuth::None);

    match car_a2a::A2aClient::new(endpoint.clone())
        .with_auth(auth)
        .send_message(message, false)
        .await
    {
        Ok(result) => serde_json::json!({
            "delivered": true,
            "owner": owner,
            "endpoint": endpoint,
            "result": result,
        }),
        Err(error) => serde_json::json!({
            "delivered": false,
            "owner": owner,
            "endpoint": endpoint,
            "error": error.to_string(),
        }),
    }
}

fn client_auth_from_route_auth(auth: A2aRouteAuth) -> car_a2a::ClientAuth {
    match auth {
        A2aRouteAuth::None => car_a2a::ClientAuth::None,
        A2aRouteAuth::Bearer { token } => car_a2a::ClientAuth::Bearer(token),
        A2aRouteAuth::Header { name, value } => car_a2a::ClientAuth::Header { name, value },
    }
}

fn trusted_route_endpoint(endpoint: Option<String>, allow_untrusted: bool) -> Option<String> {
    let endpoint = endpoint?;
    if allow_untrusted || is_loopback_http_endpoint(&endpoint) {
        Some(endpoint)
    } else {
        None
    }
}

fn is_loopback_http_endpoint(endpoint: &str) -> bool {
    endpoint == "http://localhost"
        || endpoint.starts_with("http://localhost:")
        || endpoint.starts_with("http://localhost/")
        || endpoint == "http://127.0.0.1"
        || endpoint.starts_with("http://127.0.0.1:")
        || endpoint.starts_with("http://127.0.0.1/")
        || endpoint == "http://[::1]"
        || endpoint.starts_with("http://[::1]:")
        || endpoint.starts_with("http://[::1]/")
}

async fn handle_host_register_agent(
    req: &JsonRpcMessage,
    session: &crate::session::ClientSession,
) -> Result<Value, String> {
    let request: RegisterHostAgentRequest =
        serde_json::from_value(req.params.clone()).map_err(|e| e.to_string())?;
    serde_json::to_value(
        session
            .host
            .register_agent(&session.client_id, request)
            .await?,
    )
    .map_err(|e| e.to_string())
}

async fn handle_host_unregister_agent(
    req: &JsonRpcMessage,
    session: &crate::session::ClientSession,
) -> Result<Value, String> {
    let agent_id = req
        .params
        .get("agent_id")
        .and_then(|v| v.as_str())
        .ok_or("missing agent_id")?;
    session
        .host
        .unregister_agent(&session.client_id, agent_id)
        .await?;
    Ok(serde_json::json!({"ok": true}))
}

async fn handle_host_set_status(
    req: &JsonRpcMessage,
    session: &crate::session::ClientSession,
) -> Result<Value, String> {
    let request: SetHostAgentStatusRequest =
        serde_json::from_value(req.params.clone()).map_err(|e| e.to_string())?;
    serde_json::to_value(session.host.set_status(&session.client_id, request).await?)
        .map_err(|e| e.to_string())
}

async fn handle_host_notify(
    req: &JsonRpcMessage,
    session: &crate::session::ClientSession,
) -> Result<Value, String> {
    let kind = req
        .params
        .get("kind")
        .and_then(|v| v.as_str())
        .unwrap_or("host.notification");
    let agent_id = req
        .params
        .get("agent_id")
        .and_then(|v| v.as_str())
        .map(str::to_string);
    let message = req
        .params
        .get("message")
        .and_then(|v| v.as_str())
        .unwrap_or("");
    let payload = req.params.get("payload").cloned().unwrap_or(Value::Null);
    serde_json::to_value(
        session
            .host
            .record_event(kind, agent_id, message, payload)
            .await,
    )
    .map_err(|e| e.to_string())
}

async fn handle_host_request_approval(
    req: &JsonRpcMessage,
    session: &crate::session::ClientSession,
) -> Result<Value, String> {
    let request: CreateHostApprovalRequest =
        serde_json::from_value(req.params.clone()).map_err(|e| e.to_string())?;
    if let Some(agent_id) = &request.agent_id {
        // Best-effort. If the caller doesn't own this agent the
        // ACL added 2026-05 will refuse the status update — that
        // is the correct semantics; we still want the approval row
        // itself to land so the UI can render the request.
        let _ = session
            .host
            .set_status(
                &session.client_id,
                SetHostAgentStatusRequest {
                    agent_id: agent_id.clone(),
                    status: HostAgentStatus::WaitingForApproval,
                    current_task: None,
                    message: Some("Waiting for approval".to_string()),
                    payload: Value::Null,
                },
            )
            .await;
    }
    // `system_level: true` opts the approval out of per-session
    // ownership. The host-side ACL then allows any authenticated
    // session (typically CarHost or `car-host approve`) to resolve.
    // Agents requesting user approval should always set this — the
    // session-owned mode is only correct when the requesting session
    // is also the resolving session, which approval-via-UI never is.
    let owner_client_id = if request.system_level {
        None
    } else {
        Some(session.client_id.as_str())
    };
    serde_json::to_value(session.host.create_approval(owner_client_id, request).await?)
        .map_err(|e| e.to_string())
}

async fn handle_host_resolve_approval(
    req: &JsonRpcMessage,
    session: &crate::session::ClientSession,
) -> Result<Value, String> {
    let request: ResolveHostApprovalRequest =
        serde_json::from_value(req.params.clone()).map_err(|e| e.to_string())?;
    serde_json::to_value(
        session
            .host
            .resolve_approval(&session.client_id, request)
            .await?,
    )
    .map_err(|e| e.to_string())
}

/// `session.auth` — present the per-launch token to unlock the
/// connection. When `state.auth_token` is unset, this method is a
/// no-op success (auth is disabled). When set, the supplied token
/// must equal it (constant-time comparison) — a successful auth
/// flips `session.authenticated` to `true` so subsequent methods
/// pass the gate. Wrong token returns an error AND leaves the
/// session unauthenticated; the dispatcher loop's gate then closes
/// the connection on the next non-auth method.
///
/// Closes Parslee-ai/car-releases#32.
async fn handle_session_auth(
    req: &JsonRpcMessage,
    session: &crate::session::ClientSession,
    state: &Arc<ServerState>,
) -> Result<Value, String> {
    // #254: optional `host_token` elevates the connection to the
    // host-management role. It is a *distinct* credential from the
    // daemon `token` — only readable from the `0600` host-token file,
    // never served over `GET /auth-token` — so a generic authenticated
    // client (or another local user who scraped the auth token via the
    // HTTP endpoint) cannot self-elevate to host and read other agents'
    // run traces. Validated first, before the `token` requirement,
    // because presenting a valid host token both authenticates and
    // elevates the session (a host client sends only `host_token`).
    if let Some(host_supplied) = req.params.get("host_token").and_then(Value::as_str) {
        let expected = state.host_token.get().ok_or_else(|| {
            "host auth unavailable: this daemon has no host token (started with --no-auth?)"
                .to_string()
        })?;
        if !constant_time_eq(host_supplied.as_bytes(), expected.as_bytes()) {
            return Err("auth failed: host token mismatch".to_string());
        }
        session
            .authenticated
            .store(true, std::sync::atomic::Ordering::Release);
        session
            .is_host
            .store(true, std::sync::atomic::Ordering::Release);
        return Ok(serde_json::json!({
            "ok": true,
            "auth_enabled": true,
            "role": "host",
        }));
    }

    let supplied = req
        .params
        .get("token")
        .and_then(Value::as_str)
        .ok_or_else(|| "session.auth requires { token: string }".to_string())?;
    // #169: optional `agent_id` binds the WS connection to a
    // supervised lifecycle agent. When present, the supplied token
    // must equal the per-agent token the supervisor minted at upsert
    // (NOT the daemon-wide auth token). When absent, fall back to
    // the daemon-wide token — preserves the legacy unbound-token
    // path for browser/host/CLI clients.
    let agent_id = req
        .params
        .get("agent_id")
        .and_then(Value::as_str)
        .map(str::to_string);

    if let Some(id) = agent_id {
        let supervisor = state.supervisor()?;
        if !supervisor.validate_agent_token(&id, supplied).await {
            return Err(format!(
                "auth failed: agent_id `{id}` is not supervised, or token mismatch"
            ));
        }
        // Single-claim: only one connection at a time per
        // agent_id. A second claim is rejected so the daemon-side
        // per-agent state stays unambiguous.
        {
            let mut attached = state.attached_agents.lock().await;
            if let Some(prior) = attached.get(&id) {
                if prior != &session.client_id {
                    return Err(format!(
                        "auth failed: agent_id `{id}` is already attached on \
                         another connection (client_id={prior})"
                    ));
                }
            }
            attached.insert(id.clone(), session.client_id.clone());
        }
        // #170: attach the daemon-owned persistent memgine for
        // this agent. Lazy-loaded on first connection per id from
        // `~/.car/memory/agents/<id>.jsonl`; retained across
        // disconnect so the next session sees the same state.
        let agent_eng = get_or_load_agent_memgine(state, &id).await?;
        *session.bound_memgine.lock().await = Some(agent_eng);
        *session.agent_id.lock().await = Some(id.clone());
        session
            .authenticated
            .store(true, std::sync::atomic::Ordering::Release);
        return Ok(serde_json::json!({
            "ok": true,
            "auth_enabled": true,
            "agent_id": id,
        }));
    }

    let expected = match state.auth_token.get() {
        Some(t) => t,
        None => {
            // Auth disabled — accept any token politely so callers
            // that always include a session.auth handshake (e.g. the
            // FFI proxy) don't fail when the daemon happens to be
            // unauthed. Mark the session authenticated anyway so the
            // gate is a no-op below.
            session
                .authenticated
                .store(true, std::sync::atomic::Ordering::Release);
            return Ok(serde_json::json!({ "ok": true, "auth_enabled": false }));
        }
    };
    if !constant_time_eq(supplied.as_bytes(), expected.as_bytes()) {
        return Err("auth failed: token mismatch".to_string());
    }
    session
        .authenticated
        .store(true, std::sync::atomic::Ordering::Release);
    Ok(serde_json::json!({
        "ok": true,
        "auth_enabled": true,
        "parslee": state.parslee_session.get().map(|session| session.identity.clone()),
    }))
}

/// Length-checked constant-time byte comparison. Returns false when
/// lengths differ (so length itself is the only timing leak — fine
/// for our 43-char fixed-length tokens).
fn constant_time_eq(a: &[u8], b: &[u8]) -> bool {
    if a.len() != b.len() {
        return false;
    }
    let mut diff: u8 = 0;
    for (x, y) in a.iter().zip(b.iter()) {
        diff |= x ^ y;
    }
    diff == 0
}

/// Block dispatch of `method` until the user resolves the approval
/// raised on [`HostState`].
///
/// Called from the dispatcher loop only when
/// [`crate::session::ApprovalGate::requires_approval`] returns true.
/// `Ok(())` means the user picked "approve"; `Err(reason)` is sent
/// to the caller as JSON-RPC error code `-32003` with the supplied
/// reason. On timeout, the approval row stays in `Pending` so the
/// UI keeps a record of the unanswered request.
async fn gate_high_risk_method(
    method: &str,
    params: &Value,
    state: &Arc<ServerState>,
) -> Result<(), String> {
    let timeout = state.approval_gate.timeout;
    let req = CreateHostApprovalRequest {
        agent_id: None,
        action: format!("ws.method:{method}"),
        details: serde_json::json!({
            "method": method,
            // Truncate params for the UI — full payload is recoverable
            // via the request-time host event log if needed. The cap
            // keeps a malicious caller from drowning the UI in JSON.
            "params_preview": preview_params(params, 2_000),
        }),
        options: vec!["approve".to_string(), "deny".to_string()],
        // The high-risk-method gate is already system-level (it
        // passes None as the owner via request_and_wait_approval's
        // internal call). This field is informational here.
        system_level: true,
    };
    match state
        .host
        .request_and_wait_approval(req, "approve", timeout)
        .await
    {
        Ok(crate::host::ApprovalOutcome::Approved) => Ok(()),
        Ok(crate::host::ApprovalOutcome::Denied) => Err(format!(
            "{method} denied by user (approval gate, audit 2026-05). \
             To call this method without an interactive prompt, start \
             car-server with --no-approvals on a trusted machine."
        )),
        Ok(crate::host::ApprovalOutcome::TimedOut) => Err(format!(
            "{method} approval timed out after {}s with no resolution. \
             The approval is still visible in `host.approvals` for \
             forensics; resubmit the request to retry.",
            timeout.as_secs()
        )),
        Err(e) => Err(format!("approval gate error: {e}")),
    }
}

fn preview_params(value: &Value, max_chars: usize) -> Value {
    let s = value.to_string();
    if s.len() <= max_chars {
        value.clone()
    } else {
        Value::String(format!("{}… (truncated)", &s[..max_chars]))
    }
}

async fn handle_session_init(
    req: &JsonRpcMessage,
    session: &crate::session::ClientSession,
) -> Result<Value, String> {
    let init: SessionInitRequest =
        serde_json::from_value(req.params.clone()).map_err(|e| e.to_string())?;

    for tool in &init.tools {
        register_from_definition(&session.runtime, tool).await;
    }

    let mut policy_count = 0;
    {
        let mut policies = session.runtime.policies.write().await;
        for policy_def in &init.policies {
            if let Some(check) = build_policy_check(policy_def) {
                policies.register(&policy_def.name, check, "");
                policy_count += 1;
            }
        }
    }

    serde_json::to_value(SessionInitResponse {
        session_id: session.client_id.clone(),
        tools_registered: init.tools.len(),
        policies_registered: policy_count,
    })
    .map_err(|e| e.to_string())
}

fn build_policy_check(def: &PolicyDefinition) -> Option<car_policy::PolicyCheck> {
    match def.rule.as_str() {
        "deny_tool" => {
            let target = def.target.clone();
            Some(Box::new(
                move |action: &car_ir::Action, _: &car_state::StateStore| {
                    if action.tool.as_deref() == Some(&target) {
                        Some(format!("tool '{}' denied", target))
                    } else {
                        None
                    }
                },
            ))
        }
        "require_state" => {
            let key = def.key.clone();
            let value = def.value.clone();
            Some(Box::new(
                move |_: &car_ir::Action, state: &car_state::StateStore| {
                    if state.get(&key).as_ref() != Some(&value) {
                        Some(format!("state['{}'] must be {:?}", key, value))
                    } else {
                        None
                    }
                },
            ))
        }
        "deny_tool_param" => {
            let target = def.target.clone();
            let param = def.key.clone();
            let pattern = def.pattern.clone();
            Some(Box::new(
                move |action: &car_ir::Action, _: &car_state::StateStore| {
                    if action.tool.as_deref() != Some(&target) {
                        return None;
                    }
                    if let Some(val) = action.parameters.get(&param) {
                        let s = val.as_str().unwrap_or(&val.to_string()).to_string();
                        if s.contains(&pattern) {
                            return Some(format!("param '{}' matches '{}'", param, pattern));
                        }
                    }
                    None
                },
            ))
        }
        _ => None,
    }
}

async fn handle_tools_register(
    req: &JsonRpcMessage,
    session: &crate::session::ClientSession,
) -> Result<Value, String> {
    let tools: Vec<ToolDefinition> =
        serde_json::from_value(req.params.clone()).map_err(|e| e.to_string())?;
    for tool in &tools {
        register_from_definition(&session.runtime, tool).await;
    }
    Ok(Value::from(tools.len()))
}

/// Bridge a wire-protocol `ToolDefinition` to the engine's
/// schema-aware registration. Carries the full ToolSchema shape
/// (description, parameters, returns, idempotency, caching, rate
/// limit) through to the validator. An empty `parameters` object is
/// the legacy schemaless registration — the validator no-ops for
/// those, so pre-v0.5.x callers see no change.
async fn register_from_definition(runtime: &car_engine::Runtime, def: &ToolDefinition) {
    runtime
        .register_tool_schema(car_ir::ToolSchema {
            name: def.name.clone(),
            description: def.description.clone(),
            parameters: def.parameters.clone(),
            returns: def.returns.clone(),
            idempotent: def.idempotent,
            cache_ttl_secs: def.cache_ttl_secs,
            rate_limit: def.rate_limit.as_ref().map(|rl| car_ir::ToolRateLimit {
                max_calls: rl.max_calls,
                interval_secs: rl.interval_secs,
            }),
        })
        .await;
}

async fn handle_proposal_submit(
    req: &JsonRpcMessage,
    session: &crate::session::ClientSession,
    state: &Arc<ServerState>,
) -> Result<Value, String> {
    let submit: ProposalSubmitRequest =
        serde_json::from_value(req.params.clone()).map_err(|e| e.to_string())?;
    // `session_id` is sibling to `proposal` in the params object —
    // not part of `ProposalSubmitRequest` (kept proto-compatible). When
    // present, executes the proposal under the named session so any
    // session-scoped policies layer on top of global ones.
    // See docs/proposals/per-session-policy-scoping.md.
    let session_id = req
        .params
        .get("session_id")
        .and_then(|v| v.as_str())
        .map(str::to_string);

    // `scope` is the optional per-execution caller / tenant surface
    // added in Parslee-ai/car#187 phase 3. Shape mirrors
    // `car_engine::RuntimeScope` ({ callerId, tenantId, claims });
    // serde's camelCase rename keeps the wire form consistent with
    // the rest of the JSON-RPC surface. When present, the proposal
    // routes through `execute_scoped*` so the runtime records the
    // identity on the event log and state R/W ops apply the tenant
    // prefix (#187 phase 3-B).
    let scope: Option<car_engine::RuntimeScope> = match req.params.get("scope") {
        Some(v) if !v.is_null() => {
            Some(serde_json::from_value(v.clone()).map_err(|e| format!("invalid scope: {e}"))?)
        }
        _ => None,
    };

    let result = match (session_id, scope) {
        // session_id + scope: no single combined entry point on
        // Runtime today. Scope takes precedence because it's the
        // tenant-isolation surface; per-session policies layer in
        // a follow-up when there's a real caller asking for both.
        (Some(_sid), Some(s)) => session.runtime.execute_scoped(&submit.proposal, &s).await,
        (Some(sid), None) => {
            session
                .runtime
                .execute_with_session(&submit.proposal, &sid)
                .await
        }
        (None, Some(s)) => session.runtime.execute_scoped(&submit.proposal, &s).await,
        (None, None) => session.runtime.execute(&submit.proposal).await,
    };

    // Agent run tracing (U2): when this connection has a current run open
    // (set by `runs.start` before its ack — KTD3), record one per-turn
    // trace record per action. The recorder joins the submitted proposal's
    // actions to `result.results` by `action_id` (ActionResult carries no
    // tool/params) and classifies Bulldozer's `drive_cli`/`check_outcome`
    // shapes generically. The turn stream lives on the run's in-memory
    // buffer for U3 (disk) / U4 (broadcast) to consume; recording never
    // affects the proposal response.
    // Snapshot the current run id and release the per-session lock before
    // touching the process-wide `runs` registry — keeps the two locks
    // disjoint (no `current_run_id` held across a `runs` acquisition).
    let current_run = session.current_run_id.lock().await.clone();
    if let Some(run_id) = current_run {
        let start_index = state.run_turn_count(&run_id).await;
        let records =
            crate::run_trace::record_turns(&submit.proposal, &result.results, start_index);
        if !records.is_empty() {
            // The proposal recorder flows through the SAME ceiling-enforcing
            // path as the WS handler (ADV-1): a runaway main-agent loop is
            // backstopped here too. The outcome is intentionally not
            // surfaced on the `proposal.submit` response — tracing never
            // affects the proposal result — but a ceiling refusal is logged
            // so a runaway is observable. The cap (2000 turns) is far above
            // any proposal-driven run a test seeds, so existing run-trace
            // tests are unaffected.
            match state.record_run_turns(&run_id, records).await {
                crate::session::RecordRunTurnsOutcome::RefusedCeiling => {
                    tracing::warn!(
                        run_id = %run_id,
                        "proposal-recorder: run hit the turn ceiling; trace append refused (runaway backstop)"
                    );
                }
                crate::session::RecordRunTurnsOutcome::Appended { .. }
                | crate::session::RecordRunTurnsOutcome::UnknownOrTerminal => {}
            }
        }
    }

    serde_json::to_value(result).map_err(|e| e.to_string())
}

async fn handle_session_policy_open(
    session: &crate::session::ClientSession,
) -> Result<Value, String> {
    let id = session.runtime.open_session().await;
    Ok(serde_json::json!({ "session_id": id }))
}

async fn handle_session_policy_close(
    req: &JsonRpcMessage,
    session: &crate::session::ClientSession,
) -> Result<Value, String> {
    let sid = req
        .params
        .get("session_id")
        .and_then(|v| v.as_str())
        .ok_or("missing 'session_id'")?;
    let closed = session.runtime.close_session(sid).await;
    Ok(serde_json::json!({ "closed": closed }))
}

/// `policy.register` — register one policy against this WebSocket
/// session's runtime. Mirrors the `PolicyDefinition` shape used by
/// `session.init`. When `session_id` is present, the policy is scoped
/// to the named in-runtime session opened via `session.policy.open`;
/// otherwise it is global.
async fn handle_policy_register(
    req: &JsonRpcMessage,
    session: &crate::session::ClientSession,
) -> Result<Value, String> {
    let def: PolicyDefinition = serde_json::from_value(req.params.clone())
        .map_err(|e| format!("invalid policy params: {e}"))?;
    let session_id = req
        .params
        .get("session_id")
        .and_then(|v| v.as_str())
        .map(str::to_string);
    let check = build_policy_check(&def)
        .ok_or_else(|| format!("unsupported policy rule '{}'", def.rule))?;
    match session_id {
        Some(sid) => session
            .runtime
            .register_policy_in_session(&sid, &def.name, check, "")
            .await
            .map(|_| serde_json::json!({ "registered": def.name, "scope": { "session_id": sid } })),
        None => {
            let mut policies = session.runtime.policies.write().await;
            policies.register(&def.name, check, "");
            Ok(serde_json::json!({ "registered": def.name, "scope": "global" }))
        }
    }
}

async fn handle_verify(
    req: &JsonRpcMessage,
    session: &crate::session::ClientSession,
) -> Result<Value, String> {
    let vr: VerifyRequest =
        serde_json::from_value(req.params.clone()).map_err(|e| e.to_string())?;
    // Verify against the full registered schemas so tool_call
    // parameters are checked for type mismatches + missing required
    // fields, not just tool existence (register_tool_schema's
    // documented contract; car-releases#56). The read guard is held
    // across the synchronous verify call.
    let tools_guard = session.runtime.tools.read().await;
    let result =
        car_verify::verify_with_schemas(&vr.proposal, Some(&vr.initial_state), Some(&tools_guard), 30);
    drop(tools_guard);
    serde_json::to_value(VerifyResponse {
        valid: result.valid,
        issues: result
            .issues
            .iter()
            .map(|i| VerifyIssueProto {
                action_id: i.action_id.clone(),
                severity: i.severity.clone(),
                message: i.message.clone(),
            })
            .collect(),
        simulated_state: result.simulated_state,
    })
    .map_err(|e| e.to_string())
}

/// Parse the optional `tenant_id` sibling field from JSON-RPC
/// params (Parslee-ai/car#187 phase 3-E). When set and non-empty,
/// state R/W routes through `StateStore::scoped(tenant_id)` so
/// distinct tenants can't see each other's keys over the WS
/// surface — symmetric to the proposal.submit scope plumbing.
/// When absent / empty, the legacy unscoped namespace applies.
fn tenant_from_params(req: &JsonRpcMessage) -> Option<String> {
    req.params
        .get("tenant_id")
        .and_then(|v| v.as_str())
        .filter(|s| !s.is_empty())
        .map(str::to_string)
}

async fn handle_state_get(
    req: &JsonRpcMessage,
    session: &crate::session::ClientSession,
) -> Result<Value, String> {
    let key = req
        .params
        .get("key")
        .and_then(|v| v.as_str())
        .ok_or("missing 'key'")?;
    let tenant = tenant_from_params(req);
    Ok(session
        .runtime
        .state
        .scoped(tenant.as_deref())
        .get(key)
        .unwrap_or(Value::Null))
}

async fn handle_state_set(
    req: &JsonRpcMessage,
    session: &crate::session::ClientSession,
) -> Result<Value, String> {
    let key = req
        .params
        .get("key")
        .and_then(|v| v.as_str())
        .ok_or("missing 'key'")?;
    let value = req.params.get("value").cloned().unwrap_or(Value::Null);
    let tenant = tenant_from_params(req);
    session
        .runtime
        .state
        .scoped(tenant.as_deref())
        .set(key, value, "client");
    Ok(Value::from("ok"))
}

/// `state.exists` — true if the key is set in this session's state
/// store, false otherwise. Cheaper than `state.get` + null-check on
/// the client side because it doesn't serialize the value.
async fn handle_state_exists(
    req: &JsonRpcMessage,
    session: &crate::session::ClientSession,
) -> Result<Value, String> {
    let key = req
        .params
        .get("key")
        .and_then(|v| v.as_str())
        .ok_or("missing 'key'")?;
    let tenant = tenant_from_params(req);
    Ok(Value::Bool(
        session.runtime.state.scoped(tenant.as_deref()).exists(key),
    ))
}

/// `state.keys` — list every key currently set in this session's
/// state store. Returns a JSON array of strings.
async fn handle_state_keys(
    req: &JsonRpcMessage,
    session: &crate::session::ClientSession,
) -> Result<Value, String> {
    let tenant = tenant_from_params(req);
    Ok(Value::Array(
        session
            .runtime
            .state
            .scoped(tenant.as_deref())
            .keys()
            .into_iter()
            .map(Value::String)
            .collect(),
    ))
}

/// `state.snapshot` — return the entire session state store as a
/// JSON object (`{ key: value, ... }`). Equivalent to iterating
/// `state.keys` + `state.get` but in a single round-trip; for
/// inspectors/dashboards.
///
/// Tenant-scoped variant: when `tenant_id` is set, only that
/// tenant's keys are returned (prefix stripped on the way out).
/// `state.snapshot` with no `tenant_id` returns only unscoped
/// keys; consistent with `state.keys`'s filter behaviour and the
/// strict-isolation contract from phase 3-B.
async fn handle_state_snapshot(
    req: &JsonRpcMessage,
    session: &crate::session::ClientSession,
) -> Result<Value, String> {
    let tenant = tenant_from_params(req);
    let view = session.runtime.state.scoped(tenant.as_deref());
    let mut map = serde_json::Map::new();
    for key in view.keys() {
        if let Some(value) = view.get(&key) {
            map.insert(key, value);
        }
    }
    Ok(Value::Object(map))
}

// --- Per-agent persistent memgine (#170) ---

/// `~/.car/memory/agents/<id>.json` — the per-agent snapshot file.
/// Mirrors the existing `memory.persist` shape (flat JSON array of
/// fact objects) so the same loader path works.
fn agent_memgine_snapshot_path(agent_id: &str) -> Result<std::path::PathBuf, String> {
    let base = car_ffi_common::memory_path::ensure_base()
        .map_err(|e| format!("memory base unavailable: {e}"))?;
    let dir = base.join("agents");
    std::fs::create_dir_all(&dir).map_err(|e| format!("create agents dir: {e}"))?;
    Ok(dir.join(format!("{agent_id}.json")))
}

/// Acquire (or lazy-create + load from disk) the daemon-owned
/// persistent memgine for `agent_id`. First call per id reads
/// `~/.car/memory/agents/<id>.json` if it exists; subsequent calls
/// share the in-memory engine across sessions. Caller stores the
/// returned `Arc` on `ClientSession.bound_memgine` so memory.*
/// handlers route through it via [`ClientSession::effective_memgine`].
async fn get_or_load_agent_memgine(
    state: &Arc<ServerState>,
    agent_id: &str,
) -> Result<Arc<tokio::sync::Mutex<car_memgine::MemgineEngine>>, String> {
    {
        let map = state.agent_memgines.lock().await;
        if let Some(eng) = map.get(agent_id) {
            return Ok(eng.clone());
        }
    }
    // Build a fresh engine and try to load from disk.
    let engine = Arc::new(tokio::sync::Mutex::new(car_memgine::MemgineEngine::new(
        None,
    )));
    let path = agent_memgine_snapshot_path(agent_id)?;
    if path.exists() {
        let content = std::fs::read_to_string(&path)
            .map_err(|e| format!("read {}: {}", path.display(), e))?;
        let facts: Vec<Value> = serde_json::from_str(&content).unwrap_or_default();
        let mut g = engine.lock().await;
        let mut loaded: u32 = 0;
        for fact in &facts {
            let subject = fact.get("subject").and_then(|v| v.as_str()).unwrap_or("");
            let body = fact.get("body").and_then(|v| v.as_str()).unwrap_or("");
            let kind = fact
                .get("kind")
                .and_then(|v| v.as_str())
                .unwrap_or("pattern");
            let fid = format!("loaded-{loaded}");
            g.ingest_fact(
                &fid,
                subject,
                body,
                "user",
                "peer",
                chrono::Utc::now(),
                "global",
                None,
                vec![],
                kind == "constraint",
            );
            loaded += 1;
        }
    }
    let mut map = state.agent_memgines.lock().await;
    let stored = map.entry(agent_id.to_string()).or_insert(engine).clone();
    Ok(stored)
}

/// Snapshot the agent's memgine to its disk file. Same on-wire shape
/// as `memory.persist` so manual snapshots and the daemon-owned
/// persistence stay interoperable.
async fn persist_agent_memgine(
    agent_id: &str,
    engine: &Arc<tokio::sync::Mutex<car_memgine::MemgineEngine>>,
) -> Result<(), String> {
    let path = agent_memgine_snapshot_path(agent_id)?;
    let g = engine.lock().await;
    let facts: Vec<Value> = g
        .graph
        .inner
        .node_indices()
        .filter_map(|nix| {
            let node = g.graph.inner.node_weight(nix)?;
            if !node.is_valid() {
                return None;
            }
            if node.kind == car_memgine::MemKind::Identity
                || node.kind == car_memgine::MemKind::Environment
            {
                return None;
            }
            Some(serde_json::json!({
                "subject": node.key,
                "body": node.value,
                "kind": match node.kind {
                    car_memgine::MemKind::Fact => if node.is_constraint { "constraint" } else { "pattern" },
                    car_memgine::MemKind::Conversation => "outcome",
                    _ => "pattern",
                },
                "confidence": 0.5,
                "content_type": node.content_type.as_label(),
            }))
        })
        .collect();
    let json = serde_json::to_string(&facts).map_err(|e| e.to_string())?;
    std::fs::write(&path, json).map_err(|e| format!("write {}: {}", path.display(), e))?;
    Ok(())
}

// --- Memory handlers ---

/// `memory.fact_count` — return `valid_fact_count()` of the
/// session's memgine. Used by FFI bindings to mirror their
/// embedded `fact_count()` accessor without round-tripping a full
/// query. No params.
async fn handle_memory_fact_count(
    session: &crate::session::ClientSession,
) -> Result<Value, String> {
    let engine_arc = session.effective_memgine().await;
    let engine = engine_arc.lock().await;
    Ok(Value::from(engine.valid_fact_count()))
}

async fn handle_memory_add_fact(
    req: &JsonRpcMessage,
    session: &crate::session::ClientSession,
) -> Result<Value, String> {
    let subject = req
        .params
        .get("subject")
        .and_then(|v| v.as_str())
        .ok_or("missing subject")?;
    let body = req
        .params
        .get("body")
        .and_then(|v| v.as_str())
        .ok_or("missing body")?;
    let kind = req
        .params
        .get("kind")
        .and_then(|v| v.as_str())
        .unwrap_or("pattern");
    // Route through `effective_memgine` so connections bound to a
    // lifecycle agent (#169) write into the daemon-owned per-agent
    // memgine instead of the per-WS ephemeral one (#170).
    let engine_arc = session.effective_memgine().await;
    let count = {
        let mut engine = engine_arc.lock().await;
        let fid = format!("ws-{}", engine.valid_fact_count());
        engine.ingest_fact(
            &fid,
            subject,
            body,
            "user",
            "peer",
            chrono::Utc::now(),
            "global",
            None,
            vec![],
            kind == "constraint",
        );
        engine.valid_fact_count()
    };
    // Persist after every add when the session is bound to a
    // supervised agent. Synchronous write — small JSON snapshot.
    if let Some(id) = session.agent_id.lock().await.clone() {
        if let Err(e) = persist_agent_memgine(&id, &engine_arc).await {
            tracing::warn!(agent_id = %id, error = %e,
                "agent memgine persist failed; in-memory state is canonical");
        }
    }
    Ok(Value::from(count))
}

async fn handle_memory_query(
    req: &JsonRpcMessage,
    session: &crate::session::ClientSession,
) -> Result<Value, String> {
    let query = req
        .params
        .get("query")
        .and_then(|v| v.as_str())
        .ok_or("missing query")?;
    let k = req.params.get("k").and_then(|v| v.as_u64()).unwrap_or(5) as usize;
    let engine_arc = session.effective_memgine().await;
    let engine = engine_arc.lock().await;
    let seeds = engine.graph.find_seeds(query, 5);
    // FFI parity with NAPI `query_facts` (car-ffi-napi/src/lib.rs:577) —
    // both use Personalized PageRank so transport choice doesn't shift
    // ranking semantics. Result shape (subject/body/kind/confidence)
    // also mirrors NAPI for the same reason.
    let hits = if !seeds.is_empty() {
        engine.graph.retrieve_ppr(&seeds, None, 0.5, k)
    } else {
        vec![]
    };
    let results: Vec<Value> = hits
        .iter()
        .filter_map(|hit| {
            let node = engine.graph.inner.node_weight(hit.node_ix)?;
            Some(serde_json::json!({
                "subject": node.key,
                "body": node.value,
                "kind": format!("{:?}", node.kind).to_lowercase(),
                "confidence": hit.activation,
            }))
        })
        .collect();
    serde_json::to_value(results).map_err(|e| e.to_string())
}

async fn handle_memory_build_context(
    req: &JsonRpcMessage,
    session: &crate::session::ClientSession,
) -> Result<Value, String> {
    let query = req
        .params
        .get("query")
        .and_then(|v| v.as_str())
        .unwrap_or("");
    // FFI parity with NAPI `build_context(query, model_context_window)`.
    // When supplied, sizes the assembly budget against the model's window
    // instead of the fixed 8K default.
    let model_context_window = req
        .params
        .get("model_context_window")
        .and_then(|v| v.as_u64())
        .map(|w| w as usize);
    let mut engine = session.memgine.lock().await;
    Ok(Value::from(
        engine.build_context_for_model(query, model_context_window),
    ))
}

/// `memory.build_context_fast` — Fast-mode context assembly for
/// latency-sensitive paths (voice, real-time). Skips embedding flush,
/// skill lookup, PPR-based scoring, inline repairs, known-unknowns
/// extraction. Keeps identity, constraints, facts (creation order),
/// conversation, environment.
async fn handle_memory_build_context_fast(
    req: &JsonRpcMessage,
    session: &crate::session::ClientSession,
) -> Result<Value, String> {
    let query = req
        .params
        .get("query")
        .and_then(|v| v.as_str())
        .unwrap_or("");
    let model_context_window = req
        .params
        .get("model_context_window")
        .and_then(|v| v.as_u64())
        .map(|w| w as usize);
    let mut engine = session.memgine.lock().await;
    Ok(Value::from(engine.build_context_with_options(
        query,
        model_context_window,
        car_memgine::ContextMode::Fast,
        None,
    )))
}

/// `memory.persist` — write the session's memgine to a JSON file
/// at `path`. Mirrors NAPI `persist_memory` (car-ffi-napi/src/lib.rs:797)
/// so daemon-mode clients can drive checkpoint/restore symmetrically
/// with embedded mode. Returns the number of facts written.
///
/// Filesystem caveat: `path` is interpreted on the daemon's filesystem,
/// not the caller's. Since the 2026-05 audit, `path` is also
/// sandboxed under `~/.car/memory/` via
/// [`car_ffi_common::memory_path::resolve`] — relative paths land
/// under the base, absolute paths must already be under the base,
/// `..` segments are rejected, symlinks pointing out are rejected.
/// Pre-2026-05 the path was passed straight to `std::fs::write` and
/// became an arbitrary file-write primitive. The base64-blob escape
/// hatch tracked in `Parslee-ai/car-releases#31` will plug into the
/// same resolver when it lands.
async fn handle_memory_persist(
    req: &JsonRpcMessage,
    session: &crate::session::ClientSession,
) -> Result<Value, String> {
    let path = req
        .params
        .get("path")
        .and_then(|v| v.as_str())
        .ok_or("missing path")?;
    let resolved = car_ffi_common::memory_path::resolve(path)
        .map_err(|e| format!("memory.persist rejected path {path:?}: {e}"))?;
    let engine = session.memgine.lock().await;
    let facts: Vec<Value> = engine
        .graph
        .inner
        .node_indices()
        .filter_map(|nix| {
            let node = engine.graph.inner.node_weight(nix)?;
            if !node.is_valid() {
                return None;
            }
            if node.kind == car_memgine::MemKind::Identity
                || node.kind == car_memgine::MemKind::Environment
            {
                return None;
            }
            Some(serde_json::json!({
                "subject": node.key,
                "body": node.value,
                "kind": match node.kind {
                    car_memgine::MemKind::Fact => if node.is_constraint { "constraint" } else { "pattern" },
                    car_memgine::MemKind::Conversation => "outcome",
                    _ => "pattern",
                },
                "confidence": 0.5,
                "content_type": node.content_type.as_label(),
            }))
        })
        .collect();
    let count = facts.len();
    let json = serde_json::to_string(&facts).map_err(|e| e.to_string())?;
    std::fs::write(&resolved, json)
        .map_err(|e| format!("failed to write {}: {}", resolved.display(), e))?;
    Ok(Value::from(count as u64))
}

/// `memory.load` — replace the session's memgine with facts from the
/// JSON file at `path`. Mirrors NAPI `load_memory`
/// (car-ffi-napi/src/lib.rs:121). Same `~/.car/memory/` sandboxing
/// as `memory.persist` since the 2026-05 audit — relative paths
/// land under the base, anything that escapes is rejected.
async fn handle_memory_load(
    req: &JsonRpcMessage,
    session: &crate::session::ClientSession,
) -> Result<Value, String> {
    let path = req
        .params
        .get("path")
        .and_then(|v| v.as_str())
        .ok_or("missing path")?;
    let resolved = car_ffi_common::memory_path::resolve(path)
        .map_err(|e| format!("memory.load rejected path {path:?}: {e}"))?;
    let content = std::fs::read_to_string(&resolved)
        .map_err(|e| format!("failed to read {}: {}", resolved.display(), e))?;
    let facts: Vec<Value> =
        serde_json::from_str(&content).map_err(|e| format!("invalid JSON: {}", e))?;
    let mut engine = session.memgine.lock().await;
    engine.reset();
    let mut count: u32 = 0;
    for fact in &facts {
        let subject = fact.get("subject").and_then(|v| v.as_str()).unwrap_or("");
        let body = fact.get("body").and_then(|v| v.as_str()).unwrap_or("");
        let kind = fact
            .get("kind")
            .and_then(|v| v.as_str())
            .unwrap_or("pattern");
        let fid = format!("loaded-{}", count);
        engine.ingest_fact(
            &fid,
            subject,
            body,
            "user",
            "peer",
            chrono::Utc::now(),
            "global",
            None,
            vec![],
            kind == "constraint",
        );
        count += 1;
    }
    Ok(Value::from(count))
}

// --- Skill handlers ---

async fn handle_skill_ingest(
    req: &JsonRpcMessage,
    session: &crate::session::ClientSession,
) -> Result<Value, String> {
    let name = req
        .params
        .get("name")
        .and_then(|v| v.as_str())
        .ok_or("missing name")?;
    let code = req
        .params
        .get("code")
        .and_then(|v| v.as_str())
        .ok_or("missing code")?;
    let platform = req
        .params
        .get("platform")
        .and_then(|v| v.as_str())
        .unwrap_or("unknown");
    let persona = req
        .params
        .get("persona")
        .and_then(|v| v.as_str())
        .unwrap_or("");
    let url_pattern = req
        .params
        .get("url_pattern")
        .and_then(|v| v.as_str())
        .unwrap_or("");
    let description = req
        .params
        .get("description")
        .and_then(|v| v.as_str())
        .unwrap_or("");
    let supersedes = req.params.get("supersedes").and_then(|v| v.as_str());
    let keywords: Vec<String> = req
        .params
        .get("task_keywords")
        .and_then(|v| v.as_array())
        .map(|arr| {
            arr.iter()
                .filter_map(|v| v.as_str().map(String::from))
                .collect()
        })
        .unwrap_or_default();

    let trigger = car_memgine::SkillTrigger {
        persona: persona.into(),
        url_pattern: url_pattern.into(),
        task_keywords: keywords,
        structured: None,
    };
    let mut engine = session.memgine.lock().await;
    let node = engine.ingest_skill(
        name,
        code,
        platform,
        trigger,
        description,
        supersedes,
        vec![],
        vec![],
    );
    Ok(Value::from(node.index() as u64))
}

async fn handle_skill_find(
    req: &JsonRpcMessage,
    session: &crate::session::ClientSession,
) -> Result<Value, String> {
    let persona = req
        .params
        .get("persona")
        .and_then(|v| v.as_str())
        .unwrap_or("");
    let url = req.params.get("url").and_then(|v| v.as_str()).unwrap_or("");
    let task = req
        .params
        .get("task")
        .and_then(|v| v.as_str())
        .unwrap_or("");
    let max = req
        .params
        .get("max_results")
        .and_then(|v| v.as_u64())
        .unwrap_or(1) as usize;
    let engine = session.memgine.lock().await;
    let results = engine.find_skill(persona, url, task, max);
    let json: Vec<Value> = results
        .iter()
        .map(|(m, s)| {
            serde_json::json!({
                "name": m.name, "code": m.code, "platform": m.platform,
                "description": m.description, "stats": m.stats, "match_score": s,
            })
        })
        .collect();
    serde_json::to_value(json).map_err(|e| e.to_string())
}

async fn handle_skill_report(
    req: &JsonRpcMessage,
    session: &crate::session::ClientSession,
) -> Result<Value, String> {
    let name = req
        .params
        .get("skill_name")
        .and_then(|v| v.as_str())
        .ok_or("missing skill_name")?;
    let outcome_str = req
        .params
        .get("outcome")
        .and_then(|v| v.as_str())
        .ok_or("missing outcome")?;
    let outcome = match outcome_str {
        "success" => car_memgine::SkillOutcome::Success,
        _ => car_memgine::SkillOutcome::Fail,
    };
    let mut engine = session.memgine.lock().await;
    let stats = engine
        .report_outcome(name, outcome)
        .ok_or(format!("skill '{}' not found", name))?;
    serde_json::to_value(stats).map_err(|e| e.to_string())
}

// ---------------------------------------------------------------------------
// Multi-agent coordination handlers
//
// The WsAgentRunner sends a `multi.run_agent` JSON-RPC request to the client.
// The client runs the model loop and responds with AgentOutput JSON.
// ---------------------------------------------------------------------------

/// AgentRunner backed by WebSocket callback to the client.
struct WsAgentRunner {
    channel: Arc<WsChannel>,
    host: Arc<crate::host::HostState>,
    client_id: String,
}

#[async_trait::async_trait]
impl car_multi::AgentRunner for WsAgentRunner {
    async fn run(
        &self,
        spec: &car_multi::AgentSpec,
        task: &str,
        _runtime: &car_engine::Runtime,
        _mailbox: &car_multi::Mailbox,
    ) -> std::result::Result<car_multi::AgentOutput, car_multi::MultiError> {
        use futures::SinkExt;

        let request_id = self.channel.next_request_id();
        let agent_id = agent_id_for_run(&self.client_id, &spec.name, &request_id);
        let agent = self
            .host
            .register_agent(
                &self.client_id,
                RegisterHostAgentRequest {
                    id: Some(agent_id.clone()),
                    name: spec.name.clone(),
                    kind: "callback".to_string(),
                    capabilities: spec.tools.clone(),
                    project: spec
                        .metadata
                        .get("project")
                        .and_then(|v| v.as_str())
                        .map(str::to_string),
                    pid: None,
                    display: serde_json::from_value(
                        spec.metadata
                            .get("display")
                            .cloned()
                            .unwrap_or(serde_json::Value::Null),
                    )
                    .unwrap_or_default(),
                    metadata: serde_json::to_value(&spec.metadata).unwrap_or(Value::Null),
                },
            )
            .await
            .map_err(|e| car_multi::MultiError::AgentFailed(spec.name.clone(), e))?;
        let _ = self
            .host
            .set_status(
                &self.client_id,
                SetHostAgentStatusRequest {
                    agent_id: agent.id.clone(),
                    status: HostAgentStatus::Running,
                    current_task: Some(task.to_string()),
                    message: Some(format!("{} started", spec.name)),
                    payload: serde_json::json!({ "task": task }),
                },
            )
            .await;

        let rpc_request = serde_json::json!({
            "jsonrpc": "2.0",
            "method": "multi.run_agent",
            "params": {
                "spec": spec,
                "task": task,
            },
            "id": request_id,
        });

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

        let msg = Message::Text(
            serde_json::to_string(&rpc_request)
                .map_err(|e| car_multi::MultiError::AgentFailed(spec.name.clone(), e.to_string()))?
                .into(),
        );
        if let Err(e) = self.channel.write.lock().await.send(msg).await {
            let _ = self
                .host
                .set_status(
                    &self.client_id,
                    SetHostAgentStatusRequest {
                        agent_id: agent_id.clone(),
                        status: HostAgentStatus::Errored,
                        current_task: None,
                        message: Some(format!("{} failed to start", spec.name)),
                        payload: serde_json::json!({ "error": e.to_string() }),
                    },
                )
                .await;
            return Err(car_multi::MultiError::AgentFailed(
                spec.name.clone(),
                format!("ws send error: {}", e),
            ));
        }

        // Wait for client response (5 min timeout for model loops)
        let response = match tokio::time::timeout(std::time::Duration::from_secs(300), rx).await {
            Ok(Ok(response)) => response,
            Ok(Err(_)) => {
                let _ = self
                    .host
                    .set_status(
                        &self.client_id,
                        SetHostAgentStatusRequest {
                            agent_id: agent_id.clone(),
                            status: HostAgentStatus::Errored,
                            current_task: None,
                            message: Some(format!("{} callback channel closed", spec.name)),
                            payload: Value::Null,
                        },
                    )
                    .await;
                return Err(car_multi::MultiError::AgentFailed(
                    spec.name.clone(),
                    "agent callback channel closed".into(),
                ));
            }
            Err(_) => {
                let _ = self
                    .host
                    .set_status(
                        &self.client_id,
                        SetHostAgentStatusRequest {
                            agent_id: agent_id.clone(),
                            status: HostAgentStatus::Errored,
                            current_task: None,
                            message: Some(format!("{} timed out", spec.name)),
                            payload: Value::Null,
                        },
                    )
                    .await;
                return Err(car_multi::MultiError::AgentFailed(
                    spec.name.clone(),
                    "agent callback timed out (300s)".into(),
                ));
            }
        };

        if let Some(err) = response.error {
            let _ = self
                .host
                .set_status(
                    &self.client_id,
                    SetHostAgentStatusRequest {
                        agent_id: agent_id.clone(),
                        status: HostAgentStatus::Errored,
                        current_task: None,
                        message: Some(format!("{} errored", spec.name)),
                        payload: serde_json::json!({ "error": err }),
                    },
                )
                .await;
            return Err(car_multi::MultiError::AgentFailed(spec.name.clone(), err));
        }

        let output_value = response.output.unwrap_or(Value::Null);
        let output: car_multi::AgentOutput = serde_json::from_value(output_value).map_err(|e| {
            car_multi::MultiError::AgentFailed(
                spec.name.clone(),
                format!("invalid AgentOutput: {}", e),
            )
        })?;
        let status = if output.error.is_some() {
            HostAgentStatus::Errored
        } else {
            HostAgentStatus::Completed
        };
        let message = if output.error.is_some() {
            format!("{} errored", spec.name)
        } else {
            format!("{} completed", spec.name)
        };
        let _ = self
            .host
            .set_status(
                &self.client_id,
                SetHostAgentStatusRequest {
                    agent_id,
                    status,
                    current_task: None,
                    message: Some(message),
                    payload: serde_json::to_value(&output).unwrap_or(Value::Null),
                },
            )
            .await;

        Ok(output)
    }
}

fn agent_id_for_run(client_id: &str, name: &str, request_id: &str) -> String {
    let safe_name: String = name
        .chars()
        .map(|c| {
            if c.is_ascii_alphanumeric() || c == '-' || c == '_' {
                c
            } else {
                '-'
            }
        })
        .collect();
    format!("{}:{}:{}", client_id, safe_name, request_id)
}

/// Build a [`car_multi::SharedInfra`], attaching a coordination budget when the
/// request carries a `budget` object (a `car_multi::BudgetLimits`, e.g.
/// `{"max_total_tokens": 200000, "max_agents": 12}`). Omitted fields are
/// unbounded; absent `budget` means no limits.
fn multi_infra_with_budget(req: &JsonRpcMessage) -> Result<car_multi::SharedInfra, String> {
    let infra = car_multi::SharedInfra::new();
    match req.params.get("budget") {
        None | Some(Value::Null) => Ok(infra),
        Some(v) => {
            let limits: car_multi::BudgetLimits =
                serde_json::from_value(v.clone()).map_err(|e| format!("invalid budget: {}", e))?;
            Ok(infra.with_budget(limits))
        }
    }
}

async fn handle_multi_swarm(
    req: &JsonRpcMessage,
    session: &crate::session::ClientSession,
) -> Result<Value, String> {
    let mode_str = req
        .params
        .get("mode")
        .and_then(|v| v.as_str())
        .ok_or("missing 'mode'")?;
    let agents_val = req.params.get("agents").ok_or("missing 'agents'")?;
    let task = req
        .params
        .get("task")
        .and_then(|v| v.as_str())
        .ok_or("missing 'task'")?;

    let swarm_mode: car_multi::SwarmMode = serde_json::from_str(&format!("\"{}\"", mode_str))
        .map_err(|e| format!("invalid mode '{}': {}", mode_str, e))?;
    let agent_specs: Vec<car_multi::AgentSpec> =
        serde_json::from_value(agents_val.clone()).map_err(|e| format!("invalid agents: {}", e))?;
    let synth: Option<car_multi::AgentSpec> = req
        .params
        .get("synthesizer")
        .map(|v| serde_json::from_value(v.clone()))
        .transpose()
        .map_err(|e| format!("invalid synthesizer: {}", e))?;

    let runner: Arc<dyn car_multi::AgentRunner> = Arc::new(WsAgentRunner {
        channel: session.channel.clone(),
        host: session.host.clone(),
        client_id: session.client_id.clone(),
    });
    let infra = multi_infra_with_budget(req)?;

    let mut swarm = car_multi::Swarm::new(agent_specs, swarm_mode);
    if let Some(s) = synth {
        swarm = swarm.with_synthesizer(s);
    }

    let result = swarm
        .run(task, &runner, &infra)
        .await
        .map_err(|e| format!("swarm error: {}", e))?;
    serde_json::to_value(result).map_err(|e| e.to_string())
}

async fn handle_multi_pipeline(
    req: &JsonRpcMessage,
    session: &crate::session::ClientSession,
) -> Result<Value, String> {
    let stages_val = req.params.get("stages").ok_or("missing 'stages'")?;
    let task = req
        .params
        .get("task")
        .and_then(|v| v.as_str())
        .ok_or("missing 'task'")?;

    let stage_specs: Vec<car_multi::AgentSpec> =
        serde_json::from_value(stages_val.clone()).map_err(|e| format!("invalid stages: {}", e))?;

    let runner: Arc<dyn car_multi::AgentRunner> = Arc::new(WsAgentRunner {
        channel: session.channel.clone(),
        host: session.host.clone(),
        client_id: session.client_id.clone(),
    });
    let infra = multi_infra_with_budget(req)?;

    let result = car_multi::Pipeline::new(stage_specs)
        .run(task, &runner, &infra)
        .await
        .map_err(|e| format!("pipeline error: {}", e))?;
    serde_json::to_value(result).map_err(|e| e.to_string())
}

async fn handle_multi_supervisor(
    req: &JsonRpcMessage,
    session: &crate::session::ClientSession,
) -> Result<Value, String> {
    let workers_val = req.params.get("workers").ok_or("missing 'workers'")?;
    let supervisor_val = req.params.get("supervisor").ok_or("missing 'supervisor'")?;
    let task = req
        .params
        .get("task")
        .and_then(|v| v.as_str())
        .ok_or("missing 'task'")?;
    let max_rounds = req
        .params
        .get("max_rounds")
        .and_then(|v| v.as_u64())
        .unwrap_or(3) as u32;

    let worker_specs: Vec<car_multi::AgentSpec> = serde_json::from_value(workers_val.clone())
        .map_err(|e| format!("invalid workers: {}", e))?;
    let supervisor_spec: car_multi::AgentSpec = serde_json::from_value(supervisor_val.clone())
        .map_err(|e| format!("invalid supervisor: {}", e))?;

    let runner: Arc<dyn car_multi::AgentRunner> = Arc::new(WsAgentRunner {
        channel: session.channel.clone(),
        host: session.host.clone(),
        client_id: session.client_id.clone(),
    });
    let infra = multi_infra_with_budget(req)?;

    let result = car_multi::Supervisor::new(worker_specs, supervisor_spec)
        .with_max_rounds(max_rounds)
        .run(task, &runner, &infra)
        .await
        .map_err(|e| format!("supervisor error: {}", e))?;
    serde_json::to_value(result).map_err(|e| e.to_string())
}

async fn handle_multi_map_reduce(
    req: &JsonRpcMessage,
    session: &crate::session::ClientSession,
) -> Result<Value, String> {
    let mapper_val = req.params.get("mapper").ok_or("missing 'mapper'")?;
    let reducer_val = req.params.get("reducer").ok_or("missing 'reducer'")?;
    let task = req
        .params
        .get("task")
        .and_then(|v| v.as_str())
        .ok_or("missing 'task'")?;
    let items_val = req.params.get("items").ok_or("missing 'items'")?;

    let mapper_spec: car_multi::AgentSpec =
        serde_json::from_value(mapper_val.clone()).map_err(|e| format!("invalid mapper: {}", e))?;
    let reducer_spec: car_multi::AgentSpec = serde_json::from_value(reducer_val.clone())
        .map_err(|e| format!("invalid reducer: {}", e))?;
    let items: Vec<String> =
        serde_json::from_value(items_val.clone()).map_err(|e| format!("invalid items: {}", e))?;

    let runner: Arc<dyn car_multi::AgentRunner> = Arc::new(WsAgentRunner {
        channel: session.channel.clone(),
        host: session.host.clone(),
        client_id: session.client_id.clone(),
    });
    let infra = multi_infra_with_budget(req)?;

    let result = car_multi::MapReduce::new(mapper_spec, reducer_spec)
        .run(task, &items, &runner, &infra)
        .await
        .map_err(|e| format!("map_reduce error: {}", e))?;
    serde_json::to_value(result).map_err(|e| e.to_string())
}

async fn handle_multi_vote(
    req: &JsonRpcMessage,
    session: &crate::session::ClientSession,
) -> Result<Value, String> {
    let agents_val = req.params.get("agents").ok_or("missing 'agents'")?;
    let task = req
        .params
        .get("task")
        .and_then(|v| v.as_str())
        .ok_or("missing 'task'")?;

    let agent_specs: Vec<car_multi::AgentSpec> =
        serde_json::from_value(agents_val.clone()).map_err(|e| format!("invalid agents: {}", e))?;
    let synth: Option<car_multi::AgentSpec> = req
        .params
        .get("synthesizer")
        .map(|v| serde_json::from_value(v.clone()))
        .transpose()
        .map_err(|e| format!("invalid synthesizer: {}", e))?;

    let runner: Arc<dyn car_multi::AgentRunner> = Arc::new(WsAgentRunner {
        channel: session.channel.clone(),
        host: session.host.clone(),
        client_id: session.client_id.clone(),
    });
    let infra = multi_infra_with_budget(req)?;

    let mut vote = car_multi::Vote::new(agent_specs);
    if let Some(s) = synth {
        vote = vote.with_synthesizer(s);
    }

    let result = vote
        .run(task, &runner, &infra)
        .await
        .map_err(|e| format!("vote error: {}", e))?;
    serde_json::to_value(result).map_err(|e| e.to_string())
}

async fn handle_multi_tournament(
    req: &JsonRpcMessage,
    session: &crate::session::ClientSession,
) -> Result<Value, String> {
    let competitors_val = req.params.get("competitors").ok_or("missing 'competitors'")?;
    let judge_val = req.params.get("judge").ok_or("missing 'judge'")?;
    let task = req
        .params
        .get("task")
        .and_then(|v| v.as_str())
        .ok_or("missing 'task'")?;

    let competitors: Vec<car_multi::AgentSpec> = serde_json::from_value(competitors_val.clone())
        .map_err(|e| format!("invalid competitors: {}", e))?;
    let judge: car_multi::AgentSpec =
        serde_json::from_value(judge_val.clone()).map_err(|e| format!("invalid judge: {}", e))?;

    let runner: Arc<dyn car_multi::AgentRunner> = Arc::new(WsAgentRunner {
        channel: session.channel.clone(),
        host: session.host.clone(),
        client_id: session.client_id.clone(),
    });
    let infra = multi_infra_with_budget(req)?;

    let result = car_multi::Tournament::new(competitors, judge)
        .run(task, &runner, &infra)
        .await
        .map_err(|e| format!("tournament error: {}", e))?;
    serde_json::to_value(result).map_err(|e| e.to_string())
}

async fn handle_multi_subtask(
    req: &JsonRpcMessage,
    session: &crate::session::ClientSession,
) -> Result<Value, String> {
    let main_val = req.params.get("main").ok_or("missing 'main'")?;
    let task = req
        .params
        .get("task")
        .and_then(|v| v.as_str())
        .ok_or("missing 'task'")?;

    let main_spec: car_multi::AgentSpec =
        serde_json::from_value(main_val.clone()).map_err(|e| format!("invalid main: {}", e))?;

    let runner: Arc<dyn car_multi::AgentRunner> = Arc::new(WsAgentRunner {
        channel: session.channel.clone(),
        host: session.host.clone(),
        client_id: session.client_id.clone(),
    });
    let infra = multi_infra_with_budget(req)?;

    let result = car_multi::SpawnSubtask::new(main_spec)
        .run(task, &runner, &infra)
        .await
        .map_err(|e| format!("spawn_subtask error: {}", e))?;
    serde_json::to_value(result).map_err(|e| e.to_string())
}

// ---------------------------------------------------------------------------
// Scheduler handlers
// ---------------------------------------------------------------------------

fn handle_scheduler_create(req: &JsonRpcMessage) -> Result<Value, String> {
    let name = req
        .params
        .get("name")
        .and_then(|v| v.as_str())
        .ok_or("scheduler.create requires 'name'")?;
    let prompt = req
        .params
        .get("prompt")
        .and_then(|v| v.as_str())
        .ok_or("scheduler.create requires 'prompt'")?;

    let mut task = car_scheduler::Task::new(name, prompt);

    if let Some(t) = req.params.get("trigger").and_then(|v| v.as_str()) {
        let trigger = match t {
            "once" => car_scheduler::TaskTrigger::Once,
            "cron" => car_scheduler::TaskTrigger::Cron,
            "interval" => car_scheduler::TaskTrigger::Interval,
            "file_watch" => car_scheduler::TaskTrigger::FileWatch,
            _ => car_scheduler::TaskTrigger::Manual,
        };
        let schedule = req
            .params
            .get("schedule")
            .and_then(|v| v.as_str())
            .unwrap_or("");
        task = task.with_trigger(trigger, schedule);
    }

    if let Some(sp) = req.params.get("system_prompt").and_then(|v| v.as_str()) {
        task = task.with_system_prompt(sp);
    }

    serde_json::to_value(&task).map_err(|e| e.to_string())
}

async fn handle_scheduler_run(
    req: &JsonRpcMessage,
    session: &crate::session::ClientSession,
) -> Result<Value, String> {
    let task_val = req
        .params
        .get("task")
        .ok_or("scheduler.run requires 'task'")?;
    let mut task: car_scheduler::Task =
        serde_json::from_value(task_val.clone()).map_err(|e| format!("invalid task: {}", e))?;

    let runner: Arc<dyn car_multi::AgentRunner> = Arc::new(WsAgentRunner {
        channel: session.channel.clone(),
        host: session.host.clone(),
        client_id: session.client_id.clone(),
    });
    let executor = car_scheduler::Executor::new(runner);
    let execution = executor.run_once(&mut task).await;

    serde_json::to_value(&execution).map_err(|e| e.to_string())
}

async fn handle_scheduler_run_loop(
    req: &JsonRpcMessage,
    session: &crate::session::ClientSession,
) -> Result<Value, String> {
    let task_val = req
        .params
        .get("task")
        .ok_or("scheduler.run_loop requires 'task'")?;
    let mut task: car_scheduler::Task =
        serde_json::from_value(task_val.clone()).map_err(|e| format!("invalid task: {}", e))?;
    let max_iterations = req
        .params
        .get("max_iterations")
        .and_then(|v| v.as_u64())
        .map(|v| v as u32);

    let runner: Arc<dyn car_multi::AgentRunner> = Arc::new(WsAgentRunner {
        channel: session.channel.clone(),
        host: session.host.clone(),
        client_id: session.client_id.clone(),
    });
    let executor = car_scheduler::Executor::new(runner);
    let (_cancel_tx, cancel_rx) = tokio::sync::watch::channel(false);
    let executions = executor
        .run_loop(&mut task, max_iterations, cancel_rx)
        .await;

    serde_json::to_value(&executions).map_err(|e| e.to_string())
}

// ---------------------------------------------------------------------------
// Inference handlers
// ---------------------------------------------------------------------------

fn get_inference_engine(state: &ServerState) -> &Arc<car_inference::InferenceEngine> {
    state.inference.get_or_init(|| {
        Arc::new(car_inference::InferenceEngine::new(
            car_inference::InferenceConfig::default(),
        ))
    })
}

async fn handle_infer(
    msg: &JsonRpcMessage,
    state: &ServerState,
    session: &crate::session::ClientSession,
) -> Result<Value, String> {
    let engine = get_inference_engine(state);
    let mut req: car_inference::GenerateRequest =
        serde_json::from_value(msg.params.clone()).map_err(|e| format!("invalid params: {}", e))?;

    // If context_query is provided, build context from memgine and inject it
    if let Some(cq) = msg.params.get("context_query").and_then(|v| v.as_str()) {
        let mut memgine = session.memgine.lock().await;
        let ctx = memgine.build_context(cq);
        if !ctx.is_empty() {
            req.context = Some(ctx);
        }
    }

    // Process-wide admission gate. Held for the duration of the
    // generation so a burst of concurrent infer RPCs can't multiply
    // KV-cache + activation memory and take the host out. The
    // `_permit` binding is intentional — its `Drop` releases the slot
    // when this future returns.
    let _permit = state.admission.acquire().await;

    // Use generate_tracked() so tool_calls, usage, model_used, trace_id, and
    // latency_ms are preserved in the response. Plain `generate()` discards
    // everything except `.text`, which silently breaks tool-use over the
    // WebSocket protocol (issue #43).
    //
    // NOTE: This directly serializes `InferenceResult`. Any field added to
    // that struct in `car-inference` becomes part of the public WebSocket
    // protocol. The shape is locked by `inference_result_serializes_*` tests
    // in car-inference; updating those tests is part of intentionally
    // changing the wire contract.
    let result = engine
        .generate_tracked(req)
        .await
        .map_err(|e| e.to_string())?;
    serde_json::to_value(&result).map_err(|e| format!("serialize result: {}", e))
}

/// Streaming inference — mirrors NAPI `inferStream`. Closes
/// Parslee-ai/car-releases#30. Same `GenerateRequest` shape as
/// `infer`; emits `inference.stream.event` JSON-RPC notifications
/// during the run, and returns the final `InferenceResult` as the
/// JSON-RPC response when the stream completes.
///
/// Notification shape (server → client):
/// ```jsonc
/// {
///   "jsonrpc": "2.0",
///   "method": "inference.stream.event",
///   "params": {
///     "request_id": "<original RPC id>",
///     "event": { "type": "text" | "tool_start" | "tool_delta" | "usage", ... }
///   }
/// }
/// ```
///
/// The final `done` event is not pushed as a notification — it's
/// the JSON-RPC response with the accumulated `InferenceResult`.
/// `video.generate` — daemon-side wrapper for
/// `InferenceEngine::generate_video`. Mirrors `handle_infer`'s
/// admission gate + JSON request shape (Parslee-ai/car#185).
///
/// Previously the CLI's `cmd_video` constructed an in-process
/// engine and called `generate_video` directly — a v0.7 holdover
/// that bypassed the daemon. With this handler the CLI proxies
/// here, so the engine-level audio_passthrough gate fires
/// inside the daemon process where all FFI surfaces converge.
async fn handle_image_generate(msg: &JsonRpcMessage, state: &ServerState) -> Result<Value, String> {
    let engine = get_inference_engine(state);
    let req: car_inference::GenerateImageRequest =
        serde_json::from_value(msg.params.clone()).map_err(|e| format!("invalid params: {}", e))?;
    // Share the same admission gate as text/video generation — a burst
    // of image requests shouldn't smuggle around the concurrency cap.
    let _permit = state.admission.acquire().await;
    let result = engine
        .generate_image(req)
        .await
        .map_err(|e| e.to_string())?;
    serde_json::to_value(&result).map_err(|e| format!("serialize result: {}", e))
}

async fn handle_video_generate(msg: &JsonRpcMessage, state: &ServerState) -> Result<Value, String> {
    let engine = get_inference_engine(state);
    let req: car_inference::GenerateVideoRequest =
        serde_json::from_value(msg.params.clone()).map_err(|e| format!("invalid params: {}", e))?;
    let _permit = state.admission.acquire().await;
    let result = engine
        .generate_video(req)
        .await
        .map_err(|e| e.to_string())?;
    serde_json::to_value(&result).map_err(|e| format!("serialize result: {}", e))
}

async fn handle_infer_stream(
    msg: &JsonRpcMessage,
    session: &crate::session::ClientSession,
    state: &ServerState,
) -> Result<Value, String> {
    use futures::SinkExt;
    use tokio_tungstenite::tungstenite::Message;

    let engine = get_inference_engine(state);
    let mut req: car_inference::GenerateRequest =
        serde_json::from_value(msg.params.clone()).map_err(|e| format!("invalid params: {}", e))?;

    // Same context-injection convenience as non-streaming `infer` so
    // the two methods have parity on the call shape.
    if let Some(cq) = msg.params.get("context_query").and_then(|v| v.as_str()) {
        let mut memgine = session.memgine.lock().await;
        let ctx = memgine.build_context(cq);
        if !ctx.is_empty() {
            req.context = Some(ctx);
        }
    }

    let _permit = state.admission.acquire().await;
    let mut rx = engine
        .generate_tracked_stream(req)
        .await
        .map_err(|e| e.to_string())?;

    let mut accumulator = car_inference::StreamAccumulator::default();
    let request_id = msg.id.clone();

    while let Some(event) = rx.recv().await {
        let event_payload = match &event {
            car_inference::StreamEvent::TextDelta(text) => {
                serde_json::json!({"type": "text", "data": text})
            }
            car_inference::StreamEvent::ToolCallStart { name, index, .. } => {
                serde_json::json!({"type": "tool_start", "name": name, "index": index})
            }
            car_inference::StreamEvent::ToolCallDelta {
                index,
                arguments_delta,
            } => serde_json::json!({
                "type": "tool_delta",
                "index": index,
                "data": arguments_delta,
            }),
            car_inference::StreamEvent::Usage {
                input_tokens,
                output_tokens,
            } => serde_json::json!({
                "type": "usage",
                "input_tokens": input_tokens,
                "output_tokens": output_tokens,
            }),
            // Done is delivered as the JSON-RPC response, not a
            // notification — matches the NAPI contract where the
            // standalone function's return value is the accumulated
            // result and the callback only sees in-progress events.
            car_inference::StreamEvent::Done { .. } => {
                accumulator.push(&event);
                continue;
            }
        };

        let notif = serde_json::json!({
            "jsonrpc": "2.0",
            "method": "inference.stream.event",
            "params": {
                "request_id": request_id,
                "event": event_payload,
            },
        });
        if let Ok(text) = serde_json::to_string(&notif) {
            let _ = session
                .channel
                .write
                .lock()
                .await
                .send(Message::Text(text.into()))
                .await;
        }
        accumulator.push(&event);
    }

    let (text, tool_calls, usage) = accumulator.finish_with_usage();
    Ok(serde_json::json!({
        "text": text,
        "tool_calls": tool_calls,
        "usage": usage,
    }))
}

async fn handle_embed(msg: &JsonRpcMessage, state: &ServerState) -> Result<Value, String> {
    let engine = get_inference_engine(state);
    let req: car_inference::EmbedRequest =
        serde_json::from_value(msg.params.clone()).map_err(|e| format!("invalid params: {}", e))?;
    // Embeds load their own model weights; share the same admission
    // gate as generations so a burst of embed requests can't smuggle
    // around the concurrency cap.
    let _permit = state.admission.acquire().await;
    let result = engine.embed(req).await.map_err(|e| e.to_string())?;
    Ok(serde_json::json!({"embeddings": result}))
}

async fn handle_classify(msg: &JsonRpcMessage, state: &ServerState) -> Result<Value, String> {
    let engine = get_inference_engine(state);
    let req: car_inference::ClassifyRequest =
        serde_json::from_value(msg.params.clone()).map_err(|e| format!("invalid params: {}", e))?;
    let _permit = state.admission.acquire().await;
    let result = engine.classify(req).await.map_err(|e| e.to_string())?;
    Ok(serde_json::json!({"classifications": result}))
}

/// Surface the current admission state so the menubar tray and
/// `car daemon status` can show "queued: N" / "permits: P/T". Read-only
/// snapshot — racy by definition but correct enough for status panels.
fn handle_admission_status(state: &ServerState) -> Result<Value, String> {
    let total = state.admission.permits();
    let available = state.admission.permits_available();
    let in_use = total.saturating_sub(available);
    Ok(serde_json::json!({
        "permits_total": total,
        "permits_available": available,
        "permits_in_use": in_use,
        "env_override": crate::admission::ENV_MAX_CONCURRENT,
    }))
}

async fn handle_tokenize(msg: &JsonRpcMessage, state: &ServerState) -> Result<Value, String> {
    let model = msg
        .params
        .get("model")
        .and_then(|v| v.as_str())
        .ok_or("missing 'model' parameter")?;
    let text = msg
        .params
        .get("text")
        .and_then(|v| v.as_str())
        .ok_or("missing 'text' parameter")?;
    let engine = get_inference_engine(state);
    let ids = engine
        .tokenize(model, text)
        .await
        .map_err(|e| e.to_string())?;
    Ok(serde_json::json!({"tokens": ids}))
}

async fn handle_detokenize(msg: &JsonRpcMessage, state: &ServerState) -> Result<Value, String> {
    let model = msg
        .params
        .get("model")
        .and_then(|v| v.as_str())
        .ok_or("missing 'model' parameter")?;
    let tokens: Vec<u32> = msg
        .params
        .get("tokens")
        .and_then(|v| v.as_array())
        .ok_or("missing 'tokens' parameter")?
        .iter()
        .map(|t| {
            t.as_u64()
                .and_then(|n| u32::try_from(n).ok())
                .ok_or_else(|| "tokens[] must be u32 values".to_string())
        })
        .collect::<Result<Vec<_>, _>>()?;
    let engine = get_inference_engine(state);
    let text = engine
        .detokenize(model, &tokens)
        .await
        .map_err(|e| e.to_string())?;
    Ok(serde_json::json!({"text": text}))
}

/// `models.register` — persist a user-supplied `ModelSchema` to
/// `~/.car/models.json` (Parslee-ai/car-releases#39). Replaces any
/// existing entry with the same `id`. Returns `{id, registered}`.
///
/// **Phase 1 limitation**: the daemon's live `UnifiedRegistry` is
/// not updated in-process — the new model becomes visible to
/// `models.list`, `infer`, `infer_stream` on the **next daemon
/// boot** when `load_user_config` re-reads the file. This is
/// enough to unblock opencode's setup flow (register ahead of
/// time, then start the daemon). Hot-update requires either an
/// `RwLock<InferenceEngine>` on `ServerState` or an
/// interior-mutable `UnifiedRegistry`; both touch 20+ call sites
/// and are tracked as a follow-up.
///
/// Until hot-update lands, callers SHOULD register their models
/// before issuing `infer` calls against them, and operators
/// SHOULD restart the daemon after batches of model
/// registrations.
async fn handle_models_register(
    req: &JsonRpcMessage,
    _state: &Arc<ServerState>,
) -> Result<Value, String> {
    // The params shape mirrors v0.7's FFI `rt.registerModel(schemaJson)`:
    // either the bare `ModelSchema` value, OR `{ schema: ModelSchema }`.
    // Honor both so existing in-process callers don't have to reshape.
    let schema_value = match req.params.get("schema") {
        Some(v) => v.clone(),
        None => req.params.clone(),
    };
    let schema: car_inference::ModelSchema =
        serde_json::from_value(schema_value).map_err(|e| format!("invalid ModelSchema: {e}"))?;
    let id = schema.id.clone();

    // Resolve the models.json path the same way UnifiedRegistry does:
    // `<models_dir>/../models.json` where models_dir defaults to
    // `~/.car/models/`. We read whatever's there, swap in the new
    // entry, and write back atomically.
    let home = std::env::var_os("HOME")
        .or_else(|| std::env::var_os("USERPROFILE"))
        .ok_or_else(|| "no HOME / USERPROFILE in env".to_string())?;
    let car_dir = std::path::PathBuf::from(home).join(".car");
    std::fs::create_dir_all(&car_dir).map_err(|e| format!("create {}: {e}", car_dir.display()))?;
    let path = car_dir.join("models.json");

    let mut models: Vec<car_inference::ModelSchema> = if path.exists() {
        let text =
            std::fs::read_to_string(&path).map_err(|e| format!("read {}: {e}", path.display()))?;
        if text.trim().is_empty() {
            Vec::new()
        } else {
            serde_json::from_str(&text).map_err(|e| format!("parse {}: {e}", path.display()))?
        }
    } else {
        Vec::new()
    };
    // Replace existing entry with the same id, else append.
    if let Some(slot) = models.iter_mut().find(|m| m.id == id) {
        *slot = schema;
    } else {
        models.push(schema);
    }
    let json =
        serde_json::to_string_pretty(&models).map_err(|e| format!("serialize models.json: {e}"))?;
    let tmp = path.with_extension("json.tmp");
    std::fs::write(&tmp, json).map_err(|e| format!("write {}: {e}", tmp.display()))?;
    std::fs::rename(&tmp, &path)
        .map_err(|e| format!("rename {} -> {}: {e}", tmp.display(), path.display()))?;
    Ok(serde_json::json!({
        "id": id,
        "registered": true,
        "path": path.to_string_lossy(),
        "note": "Daemon restart required for live UnifiedRegistry visibility \
                 (Parslee-ai/car-releases#39 phase 1). The model is persisted; \
                 next car-server boot loads it via UnifiedRegistry::load_user_config.",
    }))
}

/// `models.unregister` — remove an entry from `~/.car/models.json`
/// by id (Parslee-ai/car#186 — symmetric to `models.register`).
/// Returns `{ id, unregistered, path }` on success. Returns an error
/// when the model isn't present.
///
/// **Phase 1 limitation** (same as `models.register`): the daemon's
/// live `UnifiedRegistry` is not rebuilt — the removal takes effect
/// on the next daemon boot. Callers SHOULD restart the daemon after
/// a batch of unregistrations if they expect `models.list_unified`
/// to reflect the change immediately.
async fn handle_models_unregister(
    req: &JsonRpcMessage,
    _state: &Arc<ServerState>,
) -> Result<Value, String> {
    // Params shape mirrors the CLI flag: `{ id: string }`. Bare-string
    // params are honored for symmetry with the register handler's
    // tolerant shape (`{schema: ...}` OR bare schema).
    let id = match req.params.get("id") {
        Some(v) => v
            .as_str()
            .ok_or_else(|| "`id` must be a string".to_string())?
            .to_string(),
        None => match req.params.as_str() {
            Some(s) => s.to_string(),
            None => return Err("missing `id` parameter".to_string()),
        },
    };

    let home = std::env::var_os("HOME")
        .or_else(|| std::env::var_os("USERPROFILE"))
        .ok_or_else(|| "no HOME / USERPROFILE in env".to_string())?;
    let car_dir = std::path::PathBuf::from(home).join(".car");
    let path = car_dir.join("models.json");

    if !path.exists() {
        return Err(format!(
            "no models.json at {} — nothing to unregister",
            path.display()
        ));
    }
    let text =
        std::fs::read_to_string(&path).map_err(|e| format!("read {}: {e}", path.display()))?;
    let mut models: Vec<car_inference::ModelSchema> = if text.trim().is_empty() {
        Vec::new()
    } else {
        serde_json::from_str(&text).map_err(|e| format!("parse {}: {e}", path.display()))?
    };
    let before = models.len();
    models.retain(|m| m.id != id);
    if models.len() == before {
        return Err(format!("model {} not found in {}", id, path.display()));
    }
    let json =
        serde_json::to_string_pretty(&models).map_err(|e| format!("serialize models.json: {e}"))?;
    let tmp = path.with_extension("json.tmp");
    std::fs::write(&tmp, json).map_err(|e| format!("write {}: {e}", tmp.display()))?;
    std::fs::rename(&tmp, &path)
        .map_err(|e| format!("rename {} -> {}: {e}", tmp.display(), path.display()))?;
    Ok(serde_json::json!({
        "id": id,
        "unregistered": true,
        "path": path.to_string_lossy(),
        "note": "Daemon restart required for live UnifiedRegistry visibility \
                 (phase 1, matching models.register).",
    }))
}

fn handle_models_list(state: &ServerState) -> Result<Value, String> {
    let engine = get_inference_engine(state);
    let models = engine.list_models();
    serde_json::to_value(&models).map_err(|e| e.to_string())
}

fn handle_models_list_unified(state: &ServerState) -> Result<Value, String> {
    let engine = get_inference_engine(state);
    let models = engine.list_models_unified();
    serde_json::to_value(&models).map_err(|e| e.to_string())
}

#[derive(Debug, Deserialize)]
#[serde(rename_all = "camelCase")]
struct ModelSearchParams {
    #[serde(default)]
    query: Option<String>,
    #[serde(default)]
    capability: Option<car_inference::ModelCapability>,
    #[serde(default)]
    provider: Option<String>,
    #[serde(default)]
    local_only: bool,
    #[serde(default)]
    available_only: bool,
    #[serde(default)]
    limit: Option<usize>,
}

#[derive(Debug, Serialize)]
#[serde(rename_all = "camelCase")]
struct ModelSearchEntry {
    #[serde(flatten)]
    info: car_inference::ModelInfo,
    family: String,
    version: String,
    tags: Vec<String>,
    pullable: bool,
    upgrade: Option<car_inference::ModelUpgrade>,
}

#[derive(Debug, Serialize)]
#[serde(rename_all = "camelCase")]
struct ModelSearchResponse {
    models: Vec<ModelSearchEntry>,
    upgrades: Vec<car_inference::ModelUpgrade>,
    total: usize,
    available: usize,
    local: usize,
    remote: usize,
}

fn handle_models_search(req: &JsonRpcMessage, state: &ServerState) -> Result<Value, String> {
    let params: ModelSearchParams =
        serde_json::from_value(req.params.clone()).unwrap_or(ModelSearchParams {
            query: None,
            capability: None,
            provider: None,
            local_only: false,
            available_only: false,
            limit: None,
        });
    let engine = get_inference_engine(state);
    let upgrades = engine.available_model_upgrades();
    let upgrades_by_from: HashMap<String, car_inference::ModelUpgrade> = upgrades
        .iter()
        .cloned()
        .map(|upgrade| (upgrade.from_id.clone(), upgrade))
        .collect();
    let query = params
        .query
        .as_deref()
        .map(str::trim)
        .filter(|q| !q.is_empty())
        .map(|q| q.to_ascii_lowercase());
    let provider = params
        .provider
        .as_deref()
        .map(str::trim)
        .filter(|p| !p.is_empty())
        .map(|p| p.to_ascii_lowercase());

    let mut entries: Vec<ModelSearchEntry> = engine
        .list_schemas()
        .into_iter()
        .filter(|schema| {
            if let Some(capability) = params.capability {
                if !schema.has_capability(capability) {
                    return false;
                }
            }
            if let Some(provider) = provider.as_deref() {
                if schema.provider.to_ascii_lowercase() != provider {
                    return false;
                }
            }
            if params.local_only && !schema.is_local() {
                return false;
            }
            if params.available_only && !schema.available {
                return false;
            }
            if let Some(query) = query.as_deref() {
                let capability_text = schema
                    .capabilities
                    .iter()
                    .map(|cap| format!("{cap:?}").to_ascii_lowercase())
                    .collect::<Vec<_>>()
                    .join(" ");
                let haystack = format!(
                    "{} {} {} {} {} {}",
                    schema.id,
                    schema.name,
                    schema.provider,
                    schema.family,
                    schema.tags.join(" "),
                    capability_text
                )
                .to_ascii_lowercase();
                if !haystack.contains(query) {
                    return false;
                }
            }
            true
        })
        .map(|schema| {
            let pullable = !schema.available
                && matches!(
                    schema.source,
                    car_inference::ModelSource::Local { .. }
                        | car_inference::ModelSource::Mlx { .. }
                );
            let info = car_inference::ModelInfo::from(&schema);
            let upgrade = upgrades_by_from.get(&schema.id).cloned();
            ModelSearchEntry {
                info,
                family: schema.family,
                version: schema.version,
                tags: schema.tags,
                pullable,
                upgrade,
            }
        })
        .collect();
    entries.sort_by(|a, b| {
        b.info
            .available
            .cmp(&a.info.available)
            .then(b.info.is_local.cmp(&a.info.is_local))
            .then(a.info.name.cmp(&b.info.name))
    });
    if let Some(limit) = params.limit {
        entries.truncate(limit);
    }

    let total = entries.len();
    let available = entries.iter().filter(|entry| entry.info.available).count();
    let local = entries.iter().filter(|entry| entry.info.is_local).count();
    let response = ModelSearchResponse {
        models: entries,
        upgrades,
        total,
        available,
        local,
        remote: total.saturating_sub(local),
    };
    serde_json::to_value(response).map_err(|e| e.to_string())
}

/// Parse an optional serde enum param: absent/null → `None` (caller defaults);
/// present-but-unparseable → a clear invalid-params error rather than a silent
/// fallback, so a host app learns it sent a bad value.
fn optional_enum_param<T: serde::de::DeserializeOwned>(
    req: &JsonRpcMessage,
    key: &str,
) -> Result<Option<T>, String> {
    match req.params.get(key) {
        None | Some(Value::Null) => Ok(None),
        Some(v) => serde_json::from_value(v.clone())
            .map(Some)
            .map_err(|_| format!("invalid '{key}': {v}")),
    }
}

/// Parse `use_case`/`tier`/`cloud_ok` from JSON-RPC params and run the
/// recommender. `HardwareInfo::detect()` runs per call — acceptable because
/// recommend/setup_plan are onboarding-frequency (not the inference hot path),
/// and detection is a couple of cheap, non-blocking system probes.
fn recommend_from_params(
    req: &JsonRpcMessage,
    engine: &car_inference::InferenceEngine,
) -> Result<car_inference::RecommendationSet, String> {
    let use_case = optional_enum_param::<car_inference::UseCase>(req, "use_case")?.unwrap_or_default();
    let tier = optional_enum_param::<car_inference::QualityTier>(req, "tier")?.unwrap_or_default();
    let privacy = if req
        .params
        .get("cloud_ok")
        .and_then(|v| v.as_bool())
        .unwrap_or(false)
    {
        car_inference::Privacy::CloudOk
    } else {
        car_inference::Privacy::OnDevice
    };
    let hw = car_inference::HardwareInfo::detect();
    let schemas = engine.list_schemas();
    let refs: Vec<&car_inference::ModelSchema> = schemas.iter().collect();
    Ok(car_inference::recommend(&refs, &hw, use_case, tier, privacy))
}

/// `models.recommend` — ranked, explained model picks for this machine + intent.
fn handle_models_recommend(req: &JsonRpcMessage, state: &ServerState) -> Result<Value, String> {
    let engine = get_inference_engine(state);
    let set = recommend_from_params(req, engine)?;
    serde_json::to_value(set).map_err(|e| e.to_string())
}

/// `models.setup_plan` — a concrete onboarding plan the host/SDK can present:
/// machine description, the top pick, alternatives, and what needs more memory.
fn handle_models_setup_plan(req: &JsonRpcMessage, state: &ServerState) -> Result<Value, String> {
    let engine = get_inference_engine(state);
    let set = recommend_from_params(req, engine)?;
    let hw = car_inference::HardwareInfo::detect();
    let mut picks = set.picks.into_iter();
    let recommended = picks.next();
    let alternatives: Vec<_> = picks.collect();
    serde_json::to_value(serde_json::json!({
        "machine": describe_machine_for_plan(&hw),
        "recommended": recommended,
        "alternatives": alternatives,
        "needs_more_memory": set.not_enough_memory,
        "note": set.note,
    }))
    .map_err(|e| e.to_string())
}

/// Plain-language one-liner about the host machine for setup plans.
fn describe_machine_for_plan(hw: &car_inference::HardwareInfo) -> String {
    use car_inference::hardware::SupportedAcceleration::*;
    match hw.supported_acceleration() {
        Apple { unified_memory_mb } => format!(
            "Apple Silicon, {} GB unified memory (Metal)",
            unified_memory_mb / 1024
        ),
        Cuda { device_memory_mb } => match device_memory_mb {
            Some(mb) => format!("NVIDIA GPU, {} GB VRAM (CUDA)", mb / 1024),
            None => "NVIDIA GPU (CUDA)".to_string(),
        },
        UnsupportedDiscreteGpu { name, .. } => format!(
            "{} GB RAM, CPU inference ({name} not yet supported)",
            hw.total_ram_mb / 1024
        ),
        Cpu => format!("{} GB RAM, CPU inference", hw.total_ram_mb / 1024),
    }
}

fn handle_models_upgrades(state: &ServerState) -> Result<Value, String> {
    let engine = get_inference_engine(state);
    serde_json::to_value(serde_json::json!({
        "upgrades": engine.available_model_upgrades()
    }))
    .map_err(|e| e.to_string())
}

/// `models.detect_upgrades` — curated + upstream-aware findings (channel-gated,
/// cached, offline-safe).
async fn handle_models_detect_upgrades(state: &ServerState) -> Result<Value, String> {
    let engine = get_inference_engine(state);
    let findings = engine.detect_upgrades().await;
    serde_json::to_value(serde_json::json!({ "upgrades": findings })).map_err(|e| e.to_string())
}

/// `models.check_upgrade_nudge` — the current nudge decision (poll form). The
/// daemon also pushes `models.upgrade_available` proactively; this lets a
/// client ask on demand. `inference_active` defaults to false.
async fn handle_models_check_upgrade_nudge(
    req: &JsonRpcMessage,
    state: &ServerState,
) -> Result<Value, String> {
    let engine = get_inference_engine(state);
    let inference_active = req
        .params
        .get("inference_active")
        .and_then(|v| v.as_bool())
        .unwrap_or(false);
    let (decision, _state) = engine.check_upgrade_nudge(inference_active).await;
    serde_json::to_value(decision).map_err(|e| e.to_string())
}

/// `models.dismiss_upgrade` — remember that the user waved away a nudge.
fn handle_models_dismiss_upgrade(
    req: &JsonRpcMessage,
    state: &ServerState,
) -> Result<Value, String> {
    let key = req
        .params
        .get("dismiss_key")
        .and_then(|v| v.as_str())
        .map(str::trim)
        .filter(|s| !s.is_empty())
        .ok_or("missing or empty 'dismiss_key' parameter")?;
    let engine = get_inference_engine(state);
    engine.dismiss_upgrade_nudge(key).map_err(|e| e.to_string())?;
    Ok(serde_json::json!({ "dismissed": key }))
}

/// `models.check_concierge` — the current concierge suggestions (poll form).
/// The daemon also pushes `models.suggestion_available` proactively; this lets
/// a client ask on demand. `inference_active` defaults to false.
async fn handle_models_check_concierge(
    req: &JsonRpcMessage,
    state: &ServerState,
) -> Result<Value, String> {
    let engine = get_inference_engine(state);
    let inference_active = req
        .params
        .get("inference_active")
        .and_then(|v| v.as_bool())
        .unwrap_or(false);
    let (suggestions, _state) = engine.check_concierge(inference_active).await;
    serde_json::to_value(serde_json::json!({ "suggestions": suggestions }))
        .map_err(|e| e.to_string())
}

/// `models.dismiss_suggestion` — remember that the user waved away a concierge
/// suggestion, so it is never surfaced again.
fn handle_models_dismiss_suggestion(
    req: &JsonRpcMessage,
    state: &ServerState,
) -> Result<Value, String> {
    let key = req
        .params
        .get("dismiss_key")
        .and_then(|v| v.as_str())
        .map(str::trim)
        .filter(|s| !s.is_empty())
        .ok_or("missing or empty 'dismiss_key' parameter")?;
    let engine = get_inference_engine(state);
    engine
        .dismiss_concierge_suggestion(key)
        .map_err(|e| e.to_string())?;
    Ok(serde_json::json!({ "dismissed": key }))
}

/// `models.update_prefs_get` — current update preferences.
fn handle_models_update_prefs_get(state: &ServerState) -> Result<Value, String> {
    let engine = get_inference_engine(state);
    serde_json::to_value(engine.update_prefs()).map_err(|e| e.to_string())
}

/// `models.update_prefs_set` — replace update preferences. Params are the
/// `UpdatePreferences` shape (all fields optional; missing → defaults).
fn handle_models_update_prefs_set(
    req: &JsonRpcMessage,
    state: &ServerState,
) -> Result<Value, String> {
    let prefs: car_inference::UpdatePreferences =
        serde_json::from_value(req.params.clone()).map_err(|e| format!("invalid preferences: {e}"))?;
    let engine = get_inference_engine(state);
    engine.set_update_prefs(&prefs).map_err(|e| e.to_string())?;
    serde_json::to_value(prefs).map_err(|e| e.to_string())
}

/// Run one proactive upgrade check and push a `models.upgrade_available`
/// nudge to subscribers if warranted. Stamps `last_nudge_secs` after sending
/// so the per-day throttle holds across ticks. The daemon calls this on a
/// periodic timer; the nudge logic itself decides whether to actually surface
/// anything (policy/throttle/dismissals).
pub async fn run_upgrade_nudge_check(state: &Arc<ServerState>) {
    let engine = get_inference_engine(state);
    // Defer the nudge if any inference is in flight (held admission permits) —
    // the real "machine is busy" signal, not a guess.
    let inference_active = state.admission.in_flight() > 0;
    let (decision, mut nstate) = engine.check_upgrade_nudge(inference_active).await;
    if let Some(nudge) = decision.nudge {
        let delivered = broadcast_upgrade_nudge(state, &nudge).await;
        // Only burn the once-per-day throttle when the nudge actually reached
        // someone — otherwise a nudge fired with no UI connected would silence
        // the user for a day without them ever seeing it.
        if delivered > 0 {
            let now = std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .map(|d| d.as_secs())
                .unwrap_or(0);
            nstate.last_nudge_secs = now;
            let _ = nstate.save_to(&car_inference::NudgeState::default_path());
        }
    }
}

/// Push a `models.upgrade_available` notification to all subscribed UI clients,
/// returning how many it reached. Targets the same subscriber set the macOS
/// host already uses for pushed events (`a2ui_subscribers`) — the UI-push
/// channel — rather than standing up a parallel subscription. Mirrors
/// [`broadcast_a2ui_event`].
pub async fn broadcast_upgrade_nudge(
    state: &Arc<ServerState>,
    nudge: &car_inference::UpgradeNudge,
) -> usize {
    use futures::SinkExt;
    use tokio_tungstenite::tungstenite::Message;
    let subscribers: Vec<Arc<crate::session::WsChannel>> = state
        .a2ui_subscribers
        .lock()
        .await
        .values()
        .cloned()
        .collect();
    if subscribers.is_empty() {
        return 0;
    }
    let Ok(json) = serde_json::to_string(&serde_json::json!({
        "jsonrpc": "2.0",
        "method": "models.upgrade_available",
        "params": nudge,
    })) else {
        return 0;
    };
    let mut delivered = 0;
    for channel in subscribers {
        if channel
            .write
            .lock()
            .await
            .send(Message::Text(json.clone().into()))
            .await
            .is_ok()
        {
            delivered += 1;
        }
    }
    delivered
}

/// Run one proactive concierge check and push a `models.suggestion_available`
/// notification per unserved lane to subscribers. Stamps `last_concierge_secs`
/// after sending so the throttle holds across ticks. Mirrors
/// [`run_upgrade_nudge_check`] but on the concierge's independent cadence and
/// throttle field, so the two never starve each other.
pub async fn run_concierge_check(state: &Arc<ServerState>) {
    let engine = get_inference_engine(state);
    let inference_active = state.admission.in_flight() > 0;
    let (suggestions, mut nstate) = engine.check_concierge(inference_active).await;
    if suggestions.is_empty() {
        return;
    }
    let mut any_delivered = false;
    for suggestion in &suggestions {
        if broadcast_concierge_suggestion(state, suggestion).await > 0 {
            any_delivered = true;
        }
    }
    // Only burn the throttle when a suggestion actually reached someone —
    // otherwise a check that fired with no UI connected would silence the user
    // for a week without them ever seeing it.
    if any_delivered {
        let now = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .map(|d| d.as_secs())
            .unwrap_or(0);
        nstate.last_concierge_secs = now;
        let _ = nstate.save_to(&car_inference::NudgeState::default_path());
    }
}

/// Push a `models.suggestion_available` notification to subscribed UI clients,
/// returning how many it reached. Targets the same `a2ui_subscribers` UI-push
/// channel as [`broadcast_upgrade_nudge`].
pub async fn broadcast_concierge_suggestion(
    state: &Arc<ServerState>,
    suggestion: &car_inference::ConciergeSuggestion,
) -> usize {
    use futures::SinkExt;
    use tokio_tungstenite::tungstenite::Message;
    let subscribers: Vec<Arc<crate::session::WsChannel>> = state
        .a2ui_subscribers
        .lock()
        .await
        .values()
        .cloned()
        .collect();
    if subscribers.is_empty() {
        return 0;
    }
    let Ok(json) = serde_json::to_string(&serde_json::json!({
        "jsonrpc": "2.0",
        "method": "models.suggestion_available",
        "params": suggestion,
    })) else {
        return 0;
    };
    let mut delivered = 0;
    for channel in subscribers {
        if channel
            .write
            .lock()
            .await
            .send(Message::Text(json.clone().into()))
            .await
            .is_ok()
        {
            delivered += 1;
        }
    }
    delivered
}

/// Bridges the (sync) download progress sink to async WS broadcasts: each
/// `DownloadEvent` is pushed onto an unbounded channel that a concurrent task
/// drains and broadcasts as `models.pull_progress`.
struct PullProgressSink {
    tx: tokio::sync::mpsc::UnboundedSender<car_inference::DownloadEvent>,
}

impl car_inference::DownloadProgress for PullProgressSink {
    fn on_event(&self, event: &car_inference::DownloadEvent) {
        // Unbounded send from a sync context; ignore if the receiver is gone.
        let _ = self.tx.send(event.clone());
    }
}

async fn handle_models_pull(msg: &JsonRpcMessage, state: &Arc<ServerState>) -> Result<Value, String> {
    let name = msg
        .params
        .get("name")
        .or_else(|| msg.params.get("id"))
        .or_else(|| msg.params.get("model"))
        .and_then(|v| v.as_str())
        .ok_or("missing 'name' parameter")?
        .to_string();
    let engine = get_inference_engine(state);

    // Stream progress live: events flow sink → channel → broadcaster task,
    // which runs concurrently with the download. Unbounded is safe here because
    // progress is file-level (O(files): Started + per-file Started/Completed +
    // Completed — a few dozen events per pull), not byte-level, so the channel
    // can't grow without bound.
    let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel::<car_inference::DownloadEvent>();
    let sink = car_inference::ProgressSink::new(Arc::new(PullProgressSink { tx }));
    let broadcaster_state = state.clone();
    let model_label = name.clone();
    let broadcaster = tokio::spawn(async move {
        while let Some(event) = rx.recv().await {
            broadcast_pull_progress(&broadcaster_state, &model_label, &event).await;
        }
    });

    let result = engine.pull_model_with_progress(&name, &sink).await;
    // Drop the sink so its sender closes, ending the broadcaster cleanly.
    drop(sink);
    // A broadcaster panic must not poison the (successful) pull — log and move on.
    if let Err(e) = broadcaster.await {
        tracing::warn!(error = %e, "pull-progress broadcaster task failed");
    }

    let path = result.map_err(|e| e.to_string())?;
    Ok(serde_json::json!({"path": path.display().to_string()}))
}

/// Push a `models.pull_progress` notification to subscribed UI clients.
/// Mirrors [`broadcast_upgrade_nudge`]; best-effort, no-op with no subscribers.
async fn broadcast_pull_progress(
    state: &Arc<ServerState>,
    model: &str,
    event: &car_inference::DownloadEvent,
) {
    use futures::SinkExt;
    use tokio_tungstenite::tungstenite::Message;
    let subscribers: Vec<Arc<crate::session::WsChannel>> = state
        .a2ui_subscribers
        .lock()
        .await
        .values()
        .cloned()
        .collect();
    if subscribers.is_empty() {
        return;
    }
    let Ok(json) = serde_json::to_string(&serde_json::json!({
        "jsonrpc": "2.0",
        "method": "models.pull_progress",
        "params": { "model": model, "event": event },
    })) else {
        return;
    };
    for channel in subscribers {
        let _ = channel
            .write
            .lock()
            .await
            .send(Message::Text(json.clone().into()))
            .await;
    }
}

async fn handle_skills_distill(msg: &JsonRpcMessage, state: &ServerState) -> Result<Value, String> {
    let events: Vec<car_memgine::TraceEvent> = serde_json::from_value(
        msg.params
            .get("events")
            .cloned()
            .unwrap_or(msg.params.clone()),
    )
    .map_err(|e| format!("invalid events: {}", e))?;

    let inference = get_inference_engine(state).clone();
    let engine = car_memgine::MemgineEngine::new(None).with_inference(inference);

    let skills = engine.distill_skills(&events).await;
    serde_json::to_value(&skills).map_err(|e| e.to_string())
}

/// Run memory consolidation against this client's session memgine
/// (or the daemon-owned per-agent memgine when bound — #170).
/// Returns the JSON `ConsolidationReport`.
async fn handle_memory_consolidate(
    session: &crate::session::ClientSession,
) -> Result<Value, String> {
    let engine_arc = session.effective_memgine().await;
    let report = {
        let mut engine = engine_arc.lock().await;
        engine.consolidate().await
    };
    if let Some(id) = session.agent_id.lock().await.clone() {
        if let Err(e) = persist_agent_memgine(&id, &engine_arc).await {
            tracing::warn!(agent_id = %id, error = %e,
                "agent memgine persist after consolidate failed");
        }
    }
    serde_json::to_value(&report).map_err(|e| e.to_string())
}

/// Repair a degraded skill on this client's session memgine.
/// Returns `{ code: "..." }` on success, `null` if the skill
/// isn't broken or repair failed.
async fn handle_skill_repair(
    msg: &JsonRpcMessage,
    session: &crate::session::ClientSession,
) -> Result<Value, String> {
    let name = msg
        .params
        .get("skill_name")
        .and_then(|v| v.as_str())
        .ok_or("missing 'skill_name' parameter")?;
    let mut engine = session.memgine.lock().await;
    let code = engine.repair_skill(name).await;
    Ok(match code {
        Some(c) => serde_json::json!({ "code": c }),
        None => Value::Null,
    })
}

/// Ingest distilled skills into this client's session memgine.
/// Returns the number of nodes inserted.
async fn handle_skills_ingest_distilled(
    msg: &JsonRpcMessage,
    session: &crate::session::ClientSession,
) -> Result<Value, String> {
    let skills: Vec<car_memgine::DistilledSkill> = serde_json::from_value(
        msg.params
            .get("skills")
            .cloned()
            .unwrap_or(msg.params.clone()),
    )
    .map_err(|e| format!("invalid skills: {}", e))?;
    let mut engine = session.memgine.lock().await;
    let nodes = engine.ingest_distilled_skills(&skills);
    Ok(serde_json::json!({ "ingested": nodes.len() }))
}

/// Run skill evolution against this session's memgine for a
/// specified domain.  Returns the resulting `DistilledSkill` array.
async fn handle_skills_evolve(
    msg: &JsonRpcMessage,
    session: &crate::session::ClientSession,
) -> Result<Value, String> {
    let domain = msg
        .params
        .get("domain")
        .and_then(|v| v.as_str())
        .ok_or("missing 'domain' parameter")?
        .to_string();
    let events: Vec<car_memgine::TraceEvent> = serde_json::from_value(
        msg.params
            .get("events")
            .cloned()
            .unwrap_or(Value::Array(vec![])),
    )
    .map_err(|e| format!("invalid events: {}", e))?;
    let mut engine = session.memgine.lock().await;
    let skills = engine.evolve_skills(&events, &domain).await;
    serde_json::to_value(&skills).map_err(|e| e.to_string())
}

/// List domains whose skills are underperforming on this session.
async fn handle_skills_domains_needing_evolution(
    msg: &JsonRpcMessage,
    session: &crate::session::ClientSession,
) -> Result<Value, String> {
    let threshold = msg
        .params
        .get("threshold")
        .and_then(|v| v.as_f64())
        .unwrap_or(0.6);
    let engine = session.memgine.lock().await;
    let domains = engine.domains_needing_evolution(threshold);
    serde_json::to_value(&domains).map_err(|e| e.to_string())
}

/// Ingest distilled/evolved skills as PROVISIONAL candidates on trial
/// (validation-gated optimization — see docs/solutions/gated-skill-optimization.md).
/// Unlike `skills.ingest_distilled`, these must prove themselves before the
/// promotion gate makes them Active. Returns `{ ingested: n }` (drops
/// already-rejected or already-trialing candidates).
async fn handle_skills_ingest_provisional(
    msg: &JsonRpcMessage,
    session: &crate::session::ClientSession,
) -> Result<Value, String> {
    let skills: Vec<car_memgine::DistilledSkill> = serde_json::from_value(
        msg.params
            .get("skills")
            .cloned()
            .unwrap_or(msg.params.clone()),
    )
    .map_err(|e| format!("invalid skills: {}", e))?;
    let tenant = msg.params.get("tenant").and_then(|v| v.as_str());
    let mut engine = session.memgine.lock().await;
    let ingested = engine.ingest_provisional_candidates(&skills, tenant);
    Ok(serde_json::json!({ "ingested": ingested }))
}

/// Run the skill promotion gate against this session's memgine: provisional
/// candidates with enough trial outcomes are promoted (strictly-better Wilson
/// lower bound) or rejected. Normally fires automatically in `consolidate()`;
/// this exposes a manual trigger. Returns `{ promoted: [...], rejected: [...] }`.
async fn handle_skills_gate(
    _msg: &JsonRpcMessage,
    session: &crate::session::ClientSession,
) -> Result<Value, String> {
    let mut engine = session.memgine.lock().await;
    let (promoted, rejected) = engine.gate_skill_candidates();
    Ok(serde_json::json!({ "promoted": promoted, "rejected": rejected }))
}

/// Fetch a skill's full `SkillMeta` by key — including lifecycle `status`
/// (active/provisional), `incumbent`, `version`, and `stats`. Returns the JSON
/// SkillMeta, or `null` if no active skill node holds the key.
async fn handle_skill_meta(
    msg: &JsonRpcMessage,
    session: &crate::session::ClientSession,
) -> Result<Value, String> {
    let key = msg
        .params
        .get("key")
        .and_then(|v| v.as_str())
        .ok_or("missing 'key' parameter")?;
    let engine = session.memgine.lock().await;
    match engine.skill_meta(key) {
        Some(meta) => serde_json::to_value(&meta).map_err(|e| e.to_string()),
        None => Ok(Value::Null),
    }
}

/// Export a VALIDATED skill as a portable markdown document (the SkillOpt
/// best_skill.md analog). Only Active, healthy skills export; returns the
/// markdown string, or `null` if the key is absent / not exportable.
async fn handle_skill_export(
    msg: &JsonRpcMessage,
    session: &crate::session::ClientSession,
) -> Result<Value, String> {
    let key = msg
        .params
        .get("key")
        .and_then(|v| v.as_str())
        .ok_or("missing 'key' parameter")?;
    let engine = session.memgine.lock().await;
    Ok(engine.export_skill(key).map(Value::String).unwrap_or(Value::Null))
}

/// Import a skill from a portable markdown document (digest-verified). Ingests
/// as a fresh Active skill. Returns `{ imported: true }`, or a JSON-RPC error if
/// the document is malformed or its content digest doesn't verify.
async fn handle_skill_import(
    msg: &JsonRpcMessage,
    session: &crate::session::ClientSession,
) -> Result<Value, String> {
    let md = msg
        .params
        .get("markdown")
        .and_then(|v| v.as_str())
        .ok_or("missing 'markdown' parameter")?;
    let mut engine = session.memgine.lock().await;
    engine.import_skill_markdown(md)?;
    Ok(serde_json::json!({ "imported": true }))
}

/// Rerank documents against a query using a cross-encoder model.
async fn handle_rerank(msg: &JsonRpcMessage, state: &ServerState) -> Result<Value, String> {
    let engine = get_inference_engine(state);
    let req: car_inference::RerankRequest =
        serde_json::from_value(msg.params.clone()).map_err(|e| format!("invalid params: {}", e))?;
    let _permit = state.admission.acquire().await;
    let result = engine.rerank(req).await.map_err(|e| e.to_string())?;
    serde_json::to_value(&result).map_err(|e| e.to_string())
}

/// Transcribe audio at the given path. The path is interpreted on
/// the daemon's filesystem, not the FFI caller's — Daemon-mode
/// callers must pass a path the daemon can read (typically a
/// shared `~/.car/...` location or stdin push via the streaming
/// API).
async fn handle_transcribe(msg: &JsonRpcMessage, state: &ServerState) -> Result<Value, String> {
    use base64::Engine as _;
    let engine = get_inference_engine(state);

    // Sandbox-crossing escape hatch (Parslee-ai/car-releases#31): when
    // the caller can't share a filesystem view with the daemon (e.g.
    // unsandboxed Milo talking to a sandboxed car-host), they pass
    // `audio_b64` instead of `audio_path`. We decode to a tempfile,
    // run transcribe against the path the engine expects, and clean up
    // on drop. Accepts either form; `audio_b64` wins if both are set.
    let mut params = msg.params.clone();
    let audio_b64 = params
        .as_object_mut()
        .and_then(|m| m.remove("audio_b64"))
        .and_then(|v| v.as_str().map(str::to_string));
    let _tmp_audio = if let Some(b64) = audio_b64 {
        let bytes = base64::engine::general_purpose::STANDARD
            .decode(b64.as_bytes())
            .map_err(|e| format!("audio_b64 decode failed: {e}"))?;
        let tmp = tempfile::NamedTempFile::new().map_err(|e| e.to_string())?;
        std::fs::write(tmp.path(), &bytes).map_err(|e| e.to_string())?;
        let path = tmp.path().to_string_lossy().into_owned();
        if let Some(obj) = params.as_object_mut() {
            obj.insert("audio_path".to_string(), Value::String(path));
        }
        Some(tmp)
    } else {
        None
    };

    let req: car_inference::TranscribeRequest =
        serde_json::from_value(params).map_err(|e| format!("invalid params: {}", e))?;
    let _permit = state.admission.acquire().await;
    let result = engine.transcribe(req).await.map_err(|e| e.to_string())?;
    serde_json::to_value(&result).map_err(|e| e.to_string())
}

/// Synthesize speech. By default writes to `output_path` on the
/// daemon's filesystem; when `return_b64: true` (or no `output_path`
/// was supplied) the result also includes an `audio_b64` field with
/// the rendered bytes inline so cross-sandbox callers can avoid
/// filesystem coordination. Closes Parslee-ai/car-releases#31.
async fn handle_synthesize(msg: &JsonRpcMessage, state: &ServerState) -> Result<Value, String> {
    use base64::Engine as _;
    let engine = get_inference_engine(state);

    let mut params = msg.params.clone();
    let return_b64 = params
        .as_object_mut()
        .and_then(|m| m.remove("return_b64"))
        .and_then(|v| v.as_bool())
        .unwrap_or(false);
    let no_output_path = params
        .as_object()
        .map(|m| !m.contains_key("output_path"))
        .unwrap_or(true);

    let req: car_inference::SynthesizeRequest =
        serde_json::from_value(params).map_err(|e| format!("invalid params: {}", e))?;
    let _permit = state.admission.acquire().await;
    let result = engine.synthesize(req).await.map_err(|e| e.to_string())?;
    let mut value = serde_json::to_value(&result).map_err(|e| e.to_string())?;

    // Inline the bytes when the caller asked for them OR when no
    // output_path was specified (typical sandbox-crossing case —
    // they didn't pick a path because they have no shared one).
    if return_b64 || no_output_path {
        let bytes = std::fs::read(&result.audio_path).map_err(|e| {
            format!(
                "synthesize: failed to read rendered audio at {}: {e}",
                result.audio_path
            )
        })?;
        let encoded = base64::engine::general_purpose::STANDARD.encode(&bytes);
        if let Some(obj) = value.as_object_mut() {
            obj.insert("audio_b64".to_string(), Value::String(encoded));
        }
    }
    Ok(value)
}

/// Prepare the speech runtime (downloads / warmup). Returns a
/// JSON status string, mirroring the embedded
/// `prepare_speech_runtime` shape.
async fn handle_speech_prepare(state: &ServerState) -> Result<Value, String> {
    let engine = get_inference_engine(state);
    let status = engine
        .prepare_speech_runtime()
        .await
        .map_err(|e| e.to_string())?;
    serde_json::to_value(&status).map_err(|e| e.to_string())
}

/// Adaptive route decision for a prompt — returns the routing
/// JSON the FFI's `route_model` returns.
async fn handle_models_route(msg: &JsonRpcMessage, state: &ServerState) -> Result<Value, String> {
    let prompt = msg
        .params
        .get("prompt")
        .and_then(|v| v.as_str())
        .ok_or("missing 'prompt' parameter")?;
    let engine = get_inference_engine(state);
    let decision = engine.route_adaptive(prompt).await;
    serde_json::to_value(&decision).map_err(|e| e.to_string())
}

/// Model performance profiles snapshot.
async fn handle_models_stats(state: &ServerState) -> Result<Value, String> {
    let engine = get_inference_engine(state);
    let profiles = engine.export_profiles().await;
    serde_json::to_value(&profiles).map_err(|e| e.to_string())
}

#[derive(Deserialize)]
#[serde(rename_all = "camelCase")]
struct OutcomesResolvePendingParams {
    /// Flat `(trace_id, success, confidence, output)` tuples from the
    /// caller. Same shape `car-reason`'s session produces from its
    /// `ActionOutcome` vector. Daemon side runs the inference rules
    /// and writes resolved outcomes back to the shared tracker.
    action_results: Vec<(String, bool, f64, String)>,
}

/// `outcomes.resolve_pending` — write inferred outcomes back to the
/// shared engine's `OutcomeTracker` (Parslee-ai/car#189 follow-up).
///
/// Symmetric to the in-process path
/// `ReasoningInferenceHandle::record_inferred_outcomes` on
/// `InferenceEngine`: takes the per-action result tuples the
/// reasoning session produces, runs
/// `OutcomeTracker::infer_outcomes_from_action_sequence` to convert
/// them into `InferredOutcome` records, and calls
/// `resolve_pending_from_signals` under the tracker write lock. The
/// learning loop that adjusts routing decisions therefore survives
/// daemon-routed reasoning runs (previously a best-effort no-op on
/// the daemon side).
///
/// Returns `{ recorded: N }` where N is the number of action results
/// the caller passed. The tracker doesn't surface how many of those
/// actually had pending entries to resolve; that count would require
/// expanding the tracker API and isn't load-bearing for any caller
/// yet.
async fn handle_outcomes_resolve_pending(
    req: &JsonRpcMessage,
    state: &ServerState,
) -> Result<Value, String> {
    let params: OutcomesResolvePendingParams =
        serde_json::from_value(req.params.clone()).map_err(|e| format!("invalid params: {e}"))?;
    let engine = get_inference_engine(state);
    let mut tracker = engine.outcome_tracker.write().await;
    let inferred = tracker.infer_outcomes_from_action_sequence(&params.action_results);
    tracker.resolve_pending_from_signals(inferred);
    Ok(serde_json::json!({ "recorded": params.action_results.len() }))
}

/// Per-session event log size.
async fn handle_events_count(session: &crate::session::ClientSession) -> Result<Value, String> {
    let n = session.runtime.log.lock().await.len();
    Ok(Value::from(n as u64))
}

async fn handle_events_stats(session: &crate::session::ClientSession) -> Result<Value, String> {
    let stats = session.runtime.log.lock().await.stats();
    serde_json::to_value(stats).map_err(|e| e.to_string())
}

#[derive(Deserialize)]
#[serde(rename_all = "camelCase")]
struct EventsTruncateParams {
    #[serde(default)]
    max_events: Option<usize>,
    #[serde(default)]
    max_spans: Option<usize>,
}

async fn handle_events_truncate(
    msg: &JsonRpcMessage,
    session: &crate::session::ClientSession,
) -> Result<Value, String> {
    let params: EventsTruncateParams =
        serde_json::from_value(msg.params.clone()).unwrap_or(EventsTruncateParams {
            max_events: None,
            max_spans: None,
        });
    let mut log = session.runtime.log.lock().await;
    let removed_events = params
        .max_events
        .map(|max| log.truncate_events_keep_last(max))
        .unwrap_or(0);
    let removed_spans = params
        .max_spans
        .map(|max| log.truncate_spans_keep_last(max))
        .unwrap_or(0);
    let stats = log.stats();
    Ok(serde_json::json!({
        "removedEvents": removed_events,
        "removedSpans": removed_spans,
        "stats": stats,
    }))
}

async fn handle_events_clear(session: &crate::session::ClientSession) -> Result<Value, String> {
    let mut log = session.runtime.log.lock().await;
    let removed = log.clear();
    Ok(serde_json::json!({ "removed": removed, "stats": log.stats() }))
}

// ---------------------------------------------------------------------------
// Agent run tracing — run lifecycle (U1).
//
// `runs.start` brackets the beginning of an agent run: it mints a durable
// `run_id`, resolves the owning `agent_id`, tags it as the session's
// current run BEFORE responding (so the U2 per-turn recorder always reads
// the run_id the bracket set — KTD3), records `RunStarted`, and returns
// `{ run_id, agent_id }`. `runs.complete` records the terminal
// `AgentOutcome` and acks. On a mid-run disconnect with no
// `runs.complete`, `remove_session` sweeps the run to `Incomplete` after
// a short grace window (R5) so a healthy in-flight complete is not raced.
// ---------------------------------------------------------------------------

/// Resolve the owning `agent_id` for a `runs.start` call, in priority
/// order: the connection's bound agent (`session.auth {agent_id}`), the
/// `CAR_AGENT_ID` env (supervised one-shot), then a deterministic id
/// synthesized from the supplied `agent_name` (unsupervised one-shot, so
/// `run_scenarios.py` runs still record). Returns `Err` only when none of
/// these resolve — a run with no identity has no durable key and is
/// rejected rather than recorded under an ambiguous id.
async fn resolve_run_agent_id(
    session: &crate::session::ClientSession,
    req: &car_proto::RunStartRequest,
) -> Result<String, String> {
    // 1. The connection's bound agent from `session.auth {agent_id}` is
    //    AUTHORITATIVE and wins over any caller-supplied `agent_id` (FIX 5).
    //    A bound session must not be able to record a run under a DIFFERENT
    //    agent — that's trace forgery (writing into another agent's run
    //    history). A matching explicit param is fine (redundant); a
    //    mismatching one is rejected so the forgery attempt is loud rather
    //    than silently misattributed.
    if let Some(bound) = session.agent_id.lock().await.clone() {
        if let Some(id) = req.agent_id.as_deref() {
            let id = id.trim();
            if !id.is_empty() && id != bound {
                return Err(format!(
                    "runs.start `agent_id` (`{id}`) does not match this session's bound \
                     agent: a bound connection can only record runs under its own agent"
                ));
            }
        }
        return Ok(bound);
    }
    // 2. UNBOUND session only: an explicit `agent_id` param wins. This is
    //    the one-shot path that legitimately names its agent (no binding to
    //    derive from).
    if let Some(id) = req.agent_id.as_deref() {
        let id = id.trim();
        if !id.is_empty() {
            return Ok(id.to_string());
        }
    }
    // 3. The supervisor-injected env for supervised one-shot runs.
    if let Ok(env_id) = std::env::var("CAR_AGENT_ID") {
        let env_id = env_id.trim().to_string();
        if !env_id.is_empty() {
            return Ok(env_id);
        }
    }
    // 4. Unsupervised one-shot fallback: synthesize a deterministic id
    //    from the agent's name so the run still has a stable key.
    if let Some(name) = req.agent_name.as_deref() {
        if let Some(synth) = synthesize_agent_id(name) {
            return Ok(synth);
        }
    }
    Err(
        "runs.start could not resolve an agent_id: no `agent_id` param, no bound \
         session.auth {agent_id}, no CAR_AGENT_ID env, and no usable `agent_name` \
         to synthesize one from"
            .to_string(),
    )
}

/// Deterministically derive a stable agent id from a display name for
/// the unsupervised one-shot path. Lowercases, collapses any run of
/// non-alphanumeric characters to a single `-`, trims leading/trailing
/// `-`, and prefixes `name:` so a synthesized id is recognizable as a
/// name-derived fallback (and never collides with a real supervised
/// agent id, which has no `name:` prefix). Returns `None` when the name
/// has no alphanumeric content to key on.
fn synthesize_agent_id(name: &str) -> Option<String> {
    let mut slug = String::new();
    let mut prev_dash = false;
    for ch in name.chars() {
        if ch.is_ascii_alphanumeric() {
            slug.push(ch.to_ascii_lowercase());
            prev_dash = false;
        } else if !prev_dash {
            slug.push('-');
            prev_dash = true;
        }
    }
    let slug = slug.trim_matches('-');
    if slug.is_empty() {
        return None;
    }
    Some(format!("name:{slug}"))
}

async fn handle_runs_start(
    req: &JsonRpcMessage,
    session: &crate::session::ClientSession,
    state: &Arc<ServerState>,
) -> Result<Value, String> {
    let params: car_proto::RunStartRequest = serde_json::from_value(req.params.clone())
        .map_err(|e| format!("runs.start requires {{ intent, agent_id?, agent_name?, outcome_description? }}: {e}"))?;
    if params.intent.trim().is_empty() {
        return Err("runs.start requires a non-empty `intent`".to_string());
    }

    let agent_id = resolve_run_agent_id(session, &params).await?;
    let run_id = uuid::Uuid::new_v4().to_string();
    let started_at = chrono::Utc::now();

    // KTD3: tag the session's current run BEFORE we respond and BEFORE
    // we register the run, so the moment the harness gets its ack and
    // submits a proposal, the U2 recorder reads exactly this run_id.
    *session.current_run_id.lock().await = Some(run_id.clone());

    state
        .start_run(crate::session::RunMeta {
            run_id: run_id.clone(),
            agent_id: agent_id.clone(),
            client_id: session.client_id.clone(),
            intent: params.intent.clone(),
            outcome_description: params.outcome_description.clone(),
            started_at,
            termination: None,
            ended_at: None,
            turns: Vec::new(),
        })
        .await;

    serde_json::to_value(car_proto::RunStartResponse { run_id, agent_id })
        .map_err(|e| e.to_string())
}

async fn handle_runs_complete(
    req: &JsonRpcMessage,
    session: &crate::session::ClientSession,
    state: &Arc<ServerState>,
) -> Result<Value, String> {
    let params: car_proto::RunCompleteRequest = serde_json::from_value(req.params.clone())
        .map_err(|e| format!("runs.complete requires {{ run_id, outcome }}: {e}"))?;

    let termination = car_proto::RunTermination::Outcome {
        status: params.outcome.status,
        outcome: params.outcome.clone(),
    };
    state.complete_run(&params.run_id, termination).await?;

    // Clear the session's current run when it matches — the bracket is
    // closed. A non-matching current run (the harness completing an
    // older run id) is left alone.
    {
        let mut cur = session.current_run_id.lock().await;
        if cur.as_deref() == Some(params.run_id.as_str()) {
            *cur = None;
        }
    }

    serde_json::to_value(car_proto::RunCompleteResponse {
        run_id: params.run_id,
        ok: true,
    })
    .map_err(|e| e.to_string())
}

/// Daemon-owned size invariants for `runs.record_turns` (R8a). The daemon
/// owns turn size the way it owns the turn index (`record_run_turns`,
/// session.rs FIX 6) and the way `A2uiLimits` owns surface size — it never
/// trusts the client's truncation for data that lands on disk.

/// Per-string-field byte cap. Each turn's `prompt` and `output` text and
/// any oversized `parameters` string is truncated to this with a marker
/// before the turn is appended. Sized to match U3's client-side cap with
/// headroom (the client caps at ~8 KB; the daemon's 16 KB ceiling catches
/// a misbehaving or older client without trimming healthy turns).
const RECORD_TURN_FIELD_CAP_BYTES: usize = 16 * 1024;

/// Marker appended to a daemon-truncated field so a reader can tell the
/// value was cut, not merely short.
const RECORD_TURN_TRUNC_MARKER: &str = "…[truncated]";

/// Aggregate per-turn byte cap. The per-field 16 KiB cap bounds each leaf
/// string, but a turn can carry MANY leaves (e.g. a `parameters` array of
/// thousands of sub-cap strings), so the per-field cap alone does not
/// bound the persisted JSONL line. After the typed decode each candidate
/// turn is serialized; one whose encoded form exceeds this cap gets its
/// heavy free-form fields (`parameters`, `output`, `prompt`) replaced
/// whole with [`RECORD_TURN_OVERSIZE_MARKER`] — see
/// [`enforce_turn_byte_cap`].
const RECORD_TURN_MAX_BYTES: usize = 256 * 1024;

/// Replacement value for a free-form field dropped whole by the aggregate
/// per-turn cap ([`RECORD_TURN_MAX_BYTES`]) — distinct from the per-field
/// marker so a reader can tell WHICH invariant fired.
const RECORD_TURN_OVERSIZE_MARKER: &str = "…[truncated: turn exceeded 256 KiB]";

/// Max turns in a single `runs.record_turns` batch. A larger batch is a
/// client bug (the agent flushes a bounded queue per cycle); reject it
/// loudly rather than admit an unbounded append under one lock.
const RECORD_TURNS_MAX_BATCH: usize = 256;

/// 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: a batch that would take the run PAST this many
/// recorded turns is refused whole with `dropped: "run_turn_limit"`, which
/// the agent treats as stop-sending. (Losing the straddling batch is
/// correct — at this depth the run is a runaway, not a healthy cycle.)
///
/// The AUTHORITATIVE enforcement lives in
/// [`crate::session::ServerState::record_run_turns`], under the `runs` lock,
/// because the dispatcher spawns a task per frame and the handler's
/// pre-check below reads a lock-free snapshot that pipelined batches all
/// pass before any append lands (ADV-1). The handler pre-check is kept only
/// as a fast path that avoids a doomed decode+append for the obvious
/// already-over case. We alias the single source of truth here.
const RECORD_TURNS_RUN_CEILING: usize = crate::session::RECORD_TURNS_RUN_CEILING;

/// Truncate one string in place to the daemon field-byte cap, appending
/// [`RECORD_TURN_TRUNC_MARKER`] when it was cut. Cuts on a UTF-8 char
/// boundary at or below the cap so the result is always valid.
fn truncate_turn_string(s: &mut String) {
    if s.len() > RECORD_TURN_FIELD_CAP_BYTES {
        // Find the largest char boundary ≤ the cap.
        let mut end = RECORD_TURN_FIELD_CAP_BYTES;
        while end > 0 && !s.is_char_boundary(end) {
            end -= 1;
        }
        s.truncate(end);
        s.push_str(RECORD_TURN_TRUNC_MARKER);
    }
}

/// Truncate a JSON string value in place to the daemon field-byte cap,
/// appending [`RECORD_TURN_TRUNC_MARKER`] when it was cut. Non-string
/// values are recursively walked (arrays/objects) so a large blob nested
/// inside `parameters`/`output` is still bounded; numbers/bools are left
/// as-is.
fn truncate_turn_value(value: &mut Value) {
    match value {
        Value::String(s) => truncate_turn_string(s),
        Value::Array(items) => {
            for item in items.iter_mut() {
                truncate_turn_value(item);
            }
        }
        Value::Object(map) => {
            for (_k, v) in map.iter_mut() {
                truncate_turn_value(v);
            }
        }
        _ => {}
    }
}

/// Headroom (bytes) subtracted from [`RECORD_TURN_MAX_BYTES`] when MEASURING
/// a turn, to cover the re-stamp slack (ADV-4). The cap is enforced here with
/// the turn's placeholder `index: 0`, but `record_run_turns` re-stamps the
/// index to the live append position under the `runs` lock — up to a
/// 4-digit-plus number near the run ceiling. So `"index":0` (9 bytes) can
/// grow to e.g. `"index":1999` (12 bytes): a turn measured at exactly the cap
/// would persist a few bytes OVER it. 16 bytes is comfortably above the
/// worst-case index-digit growth (the ceiling is 2000 → 4 digits → +3 bytes),
/// with margin to spare.
const RECORD_TURN_REINDEX_HEADROOM: usize = 16;

/// Effective measurement cap: a turn is treated as "within the persisted
/// cap" only when its encoded length (with the placeholder index) is at or
/// below this, leaving [`RECORD_TURN_REINDEX_HEADROOM`] for the live re-stamp.
const RECORD_TURN_MEASURE_CAP: usize = RECORD_TURN_MAX_BYTES - RECORD_TURN_REINDEX_HEADROOM;

/// Encoded size of a turn as it would persist (one JSONL line, minus the
/// trailing newline). A turn that fails to serialize reports `usize::MAX`
/// so the cap path treats it as oversized rather than waving it through.
fn turn_encoded_len(turn: &car_proto::RunTurn) -> usize {
    serde_json::to_vec(turn).map_or(usize::MAX, |v| v.len())
}

/// Enforce the aggregate per-turn byte cap ([`RECORD_TURN_MAX_BYTES`]) on
/// a decoded turn — the invariant the per-field pass cannot give: the
/// per-field cap bounds each leaf string, but not how many leaves a turn
/// carries, so a `parameters` array of thousands of sub-cap strings (or a
/// multi-MB `tool` scalar the per-field pass never visits) would still
/// produce an unbounded JSONL line.
///
/// When the encoded turn exceeds the cap, the heavy free-form fields
/// (`parameters`, `output`, `prompt`) are replaced WHOLE with
/// [`RECORD_TURN_OVERSIZE_MARKER`] (no partial salvage — at this size the
/// content is a misbehaving client, not signal). If the turn is somehow
/// STILL over cap, the remaining string scalars the per-field pass never
/// truncates (`tool`, `policy_rejected.rule`/`param`) are field-capped too.
///
/// Returns `true` when the turn is within the persisted cap (keep it) and
/// `false` when it is STILL over cap after both passes (ADV-5: reject/drop
/// it — a serialize failure or a future free-form field this pass doesn't
/// bound). The caller drops a rejected turn with a logged reason and a
/// structured drop rather than appending an over-cap line. The measurement
/// uses [`RECORD_TURN_MEASURE_CAP`] (ADV-4 re-stamp headroom).
fn enforce_turn_byte_cap(turn: &mut car_proto::RunTurn) -> bool {
    // ADV-5: exhaustively destructure so a future new `RunTurn` field is a
    // COMPILE error here, not a silently-unbounded persisted line. Every
    // field below is either a free-form string/JSON we bound, or a
    // number/C-like enum that is bounded by construction.
    let car_proto::RunTurn {
        index: _,           // usize — bounded; daemon re-stamps it anyway
        prompt: _,          // bounded below (replaced whole when oversize)
        tool: _,            // bounded below (field-capped when oversize)
        parameters: _,      // bounded below (replaced whole when oversize)
        output: _,          // bounded below (replaced whole when oversize)
        cli_outcome: _,     // C-like enum + i64 — bounded
        verifier_verdict: _, // C-like enum — bounded
        policy_rejected: _, // bounded below (field-capped when oversize)
    } = turn;

    if turn_encoded_len(turn) <= RECORD_TURN_MEASURE_CAP {
        return true;
    }
    // Replace the big three free-form fields whole.
    if !turn.parameters.is_null() {
        turn.parameters = Value::String(RECORD_TURN_OVERSIZE_MARKER.to_string());
    }
    if turn.output.is_some() {
        turn.output = Some(Value::String(RECORD_TURN_OVERSIZE_MARKER.to_string()));
    }
    if turn.prompt.is_some() {
        turn.prompt = Some(RECORD_TURN_OVERSIZE_MARKER.to_string());
    }
    if turn_encoded_len(turn) <= RECORD_TURN_MEASURE_CAP {
        return true;
    }
    // Still over cap: the bytes live in the scalars the per-field pass
    // never visits. Field-cap them.
    if let Some(tool) = turn.tool.as_mut() {
        truncate_turn_string(tool);
    }
    if let Some(pr) = turn.policy_rejected.as_mut() {
        truncate_turn_string(&mut pr.rule);
        if let Some(param) = pr.param.as_mut() {
            truncate_turn_string(param);
        }
    }
    // ADV-5: if the turn is STILL over cap, every free-form field has already
    // been replaced/capped, so the only ways to land here are a serialize
    // failure (`turn_encoded_len` → usize::MAX) or a future free-form field
    // the exhaustive destructure above will have forced us to handle. Reject
    // the turn (hard, not a release-compiled-out `debug_assert`) so an
    // unbounded line never reaches disk.
    turn_encoded_len(turn) <= RECORD_TURN_MEASURE_CAP
}

/// Build a non-fatal `runs.record_turns` rejection: `ok: false`, nothing
/// appended, with the machine-readable `dropped` reason the agent treats
/// as "stop sending for this run".
fn record_turns_dropped(run_id: &str, reason: &str) -> Result<Value, String> {
    serde_json::to_value(car_proto::RunRecordTurnsResponse {
        run_id: run_id.to_string(),
        base_index: 0,
        count: 0,
        ok: false,
        dropped: Some(reason.to_string()),
    })
    .map_err(|e| e.to_string())
}

/// `runs.record_turns {run_id, turns}` — WS-only batch append of
/// client-narrated turns (feedback-agent A2UI/runs plan, U1).
///
/// The turn source is an out-of-pipeline agent whose work happens inside
/// its own subprocess (e.g. a `claude -p` resolver), invisible to the
/// proposal-path recorder. It builds full `RunTurn`s itself and pushes
/// them here in batches; the daemon appends them through the SAME
/// [`ServerState::record_run_turns`] the proposal recorder uses, so the
/// index re-stamp (FIX 6), JSONL persist, and `runs.trace.event` fanout
/// are byte-identical — this handler never re-implements locking, append,
/// or fanout.
///
/// Authorization (KTD): write access is the OWNING-agent binding only. The
/// read path's host-token gate must NOT apply here — a host-token or
/// unbound session writing another agent's turns is trace forgery. An
/// unknown run AND an unauthorized run collapse to the same uniform
/// `dropped: "run_not_found"` (mirroring `handle_runs_subscribe`'s FIX 3
/// not-found), so the response is never an existence/owner oracle.
///
/// Daemon-owned size invariants (R8a): per-field truncation, an aggregate
/// per-turn byte cap ([`enforce_turn_byte_cap`]), a batch-size cap (hard
/// error above), and a per-run turn ceiling (`run_turn_limit` — a hard
/// cap: no batch is accepted that would take the run past the ceiling).
/// Run existence/terminality is pre-checked to return a distinguishable
/// `ok: false` + reason rather than `record_run_turns`'s silent
/// zero-count drop; the benign TOCTOU (a run going terminal between the
/// check and the append) still drops silently inside `record_run_turns`,
/// matching existing recorder semantics.
async fn handle_runs_record_turns(
    req: &JsonRpcMessage,
    session: &crate::session::ClientSession,
    state: &Arc<ServerState>,
) -> Result<Value, String> {
    // Parse leniently: the wire `RunTurn` may omit `verifier_verdict`
    // (client-narrated turns default it to `not_run`) and carries a
    // client-supplied `index` we ignore. We deserialize each turn from a
    // generic object so we can inject the verdict default before the typed
    // `RunTurn` deserialize, without adding a serde default to the shared
    // proto type (the proposal recorder always sets the field).
    let raw_run_id = req
        .params
        .get("run_id")
        .and_then(Value::as_str)
        .map(str::to_string);
    let run_id = match raw_run_id {
        Some(id) if !id.trim().is_empty() => id,
        _ => return Err("runs.record_turns requires { run_id, turns: [RunTurn] }".to_string()),
    };

    let raw_turns = match req.params.get("turns").and_then(Value::as_array) {
        Some(arr) => arr,
        None => {
            return Err(
                "runs.record_turns requires { run_id, turns: [RunTurn] }: `turns` must be an array"
                    .to_string(),
            )
        }
    };
    if raw_turns.is_empty() {
        return Err(
            "runs.record_turns requires a non-empty `turns` array".to_string(),
        );
    }
    if raw_turns.len() > RECORD_TURNS_MAX_BATCH {
        return Err(format!(
            "runs.record_turns batch too large: {} turns exceeds the per-call cap of {}",
            raw_turns.len(),
            RECORD_TURNS_MAX_BATCH
        ));
    }

    // Decode each wire turn into a typed RunTurn, defaulting an absent
    // `verifier_verdict` to `not_run` and truncating its oversized string
    // fields to the daemon cap (R8a) — the daemon does not trust client
    // truncation for data that lands on disk.
    let mut turns: Vec<car_proto::RunTurn> = Vec::with_capacity(raw_turns.len());
    for (i, raw) in raw_turns.iter().enumerate() {
        let mut obj = raw.clone();
        match obj.as_object_mut() {
            Some(map) => {
                // The daemon owns the index (re-stamped under the `runs`
                // lock in `record_run_turns`), so a client-narrated turn's
                // `index` is genuinely IGNORED — the protocol doc says so.
                // ADV-6: overwrite it UNCONDITIONALLY with a placeholder `0`
                // (not `entry().or_insert`), so a present-but-invalid value
                // (`-1`, `"5"`, `1.5`) can't fail the typed `usize` decode and
                // reject the whole batch. The placeholder is re-stamped to the
                // live append position on append regardless.
                map.insert("index".to_string(), Value::Number(0.into()));
                map.entry("verifier_verdict")
                    .or_insert_with(|| Value::String("not_run".to_string()));
                // Truncate the heavy free-text / blob fields before the
                // typed decode so the bounded values are what gets stored.
                if let Some(p) = map.get_mut("prompt") {
                    truncate_turn_value(p);
                }
                if let Some(o) = map.get_mut("output") {
                    truncate_turn_value(o);
                }
                if let Some(params) = map.get_mut("parameters") {
                    truncate_turn_value(params);
                }
            }
            None => {
                return Err(format!(
                    "runs.record_turns requires {{ run_id, turns: [RunTurn] }}: \
                     turn {i} is not an object"
                ));
            }
        }
        let mut turn: car_proto::RunTurn = serde_json::from_value(obj).map_err(|e| {
            format!("runs.record_turns requires {{ run_id, turns: [RunTurn] }}: turn {i}: {e}")
        })?;
        // Aggregate cap (R8a): the per-field pass bounds each leaf, not the
        // sum of leaves — bound the whole encoded turn before it is stored.
        // ADV-5: a turn that is STILL over cap after the replace+field-cap
        // pass (a serialize failure, or a future free-form field) is REJECTED,
        // not waved through. Drop the whole batch with a structured reason and
        // log it server-side — an unbounded line must never reach disk.
        if !enforce_turn_byte_cap(&mut turn) {
            tracing::warn!(
                run_id = %run_id,
                turn = i,
                "runs.record_turns: turn could not be bounded under the per-turn byte cap; dropping batch"
            );
            return record_turns_dropped(&run_id, "turn_too_large");
        }
        turns.push(turn);
    }

    // Resolve the run's owning agent: in-memory registry (live run) first,
    // disk fallback for a run whose RunMeta isn't in this process — the
    // `handle_runs_subscribe` pattern. Unknown run → uniform not-found.
    //
    // ADV-2: read the run HEADER only — `(agent_id, terminal, turn_count)` —
    // not the whole `RunMeta`. The previous `run_meta` clone copied the
    // entire `turns` buffer (full prompts + CLI output) under the global
    // `runs` lock on every batch RPC; the handler needs only those three
    // facts.
    let header = state.run_header(&run_id).await;
    let owning_agent = match &header {
        Some((agent_id, _terminal, _len)) => agent_id.clone(),
        None => match state.run_store.agent_for_run(&run_id) {
            Some(a) => a,
            // FIX 3 uniform not-found: an unknown run and an unauthorized
            // run are indistinguishable to the caller.
            None => return record_turns_dropped(&run_id, "run_not_found"),
        },
    };

    // WRITE authz (KTD): owning-agent binding ONLY. A host-token or unbound
    // session must NOT be able to write another agent's turns — that is
    // forgery, the inverse of the read path's host-token allowance. An
    // unauthorized write collapses to the SAME `run_not_found` as an
    // unknown run (no existence/owner oracle); logged server-side only.
    let bound = session.agent_id.lock().await.clone();
    if bound.as_deref() != Some(owning_agent.as_str()) {
        tracing::debug!(
            run_id = %run_id,
            owning_agent = %owning_agent,
            client_id = %session.client_id,
            "runs.record_turns denied (not the owning agent); returning uniform not-found"
        );
        return record_turns_dropped(&run_id, "run_not_found");
    }

    // Pre-check terminality and the ceiling as a FAST PATH only. These read
    // the lock-free `run_header` snapshot, so a write against an obviously
    // closed / already-over-ceiling run skips the decode+append. They are
    // NOT authoritative: the dispatcher spawns a task per frame, so pipelined
    // batches can each pass this snapshot before any of them appends (ADV-1).
    // The authoritative terminal+ceiling enforcement is under the `runs` lock
    // inside `record_run_turns`, whose `RecordRunTurnsOutcome` we map below.
    if let Some((_agent_id, terminal, len)) = &header {
        if *terminal {
            return record_turns_dropped(&run_id, "run_terminal");
        }
        // Per-run turn ceiling (R8a) fast path: a runaway-loop backstop and a
        // TRUE hard cap — refuse any batch that would take the run PAST the
        // ceiling, whole, and tell the agent to stop sending. (Healthy
        // cycles are tens of turns; losing the straddling batch of a
        // 2000-turn runaway is correct, not data loss.) The under-lock check
        // in `record_run_turns` is the one that actually bounds the run; this
        // only short-circuits the common already-over case.
        if *len + turns.len() > RECORD_TURNS_RUN_CEILING {
            return record_turns_dropped(&run_id, "run_turn_limit");
        }
    }

    // Append through the shared path — index re-stamp, JSONL persist,
    // `runs.trace.event` fanout, AND the authoritative under-lock ceiling all
    // happen inside `record_run_turns` under its own lock discipline. It
    // returns a `RecordRunTurnsOutcome` that distinguishes a healthy append
    // from a ceiling refusal and from an unknown/terminal run (ADV-1) — we
    // must NOT let an under-lock ceiling refusal masquerade as `run_terminal`.
    let count = turns.len();
    let records: Vec<car_proto::RunRecord> =
        turns.into_iter().map(car_proto::RunRecord::Turn).collect();
    let new_total = match state.record_run_turns(&run_id, records).await {
        crate::session::RecordRunTurnsOutcome::Appended { new_total } => new_total,
        crate::session::RecordRunTurnsOutcome::RefusedCeiling => {
            // A pipelined batch crossed the ceiling under the lock even
            // though our fast-path snapshot was sub-ceiling — the runaway
            // backstop fired authoritatively. Tell the agent to stop.
            return record_turns_dropped(&run_id, "run_turn_limit");
        }
        crate::session::RecordRunTurnsOutcome::UnknownOrTerminal => {
            // `record_run_turns` appended nothing because the run is unknown
            // or went terminal. We resolved the run above, so the benign
            // cause is the TOCTOU: the run went terminal between our
            // fast-path check and the append. Surface a non-fatal drop.
            return record_turns_dropped(&run_id, "run_terminal");
        }
    };

    serde_json::to_value(car_proto::RunRecordTurnsResponse {
        run_id,
        base_index: new_total - count,
        count,
        ok: true,
        dropped: None,
    })
    .map_err(|e| e.to_string())
}

/// Authorize the calling connection to read/subscribe a run owned by
/// `agent_id` (R16 / KTD10 / Parslee-ai/car#254). A connection is
/// entitled when:
///
/// 1. It **owns** the agent — its `session.auth {agent_id}` binding
///    matches the run's owning `agent_id` (the supervised agent reading
///    its own trace), or
/// 2. It holds the **host-management role** — it authenticated via
///    `session.auth { host_token }` with the per-launch host token
///    (`ClientSession::is_host`).
///
/// What this gate buys, and its bound (be honest):
///
/// - It distinguishes agent-bound sessions, so one supervised agent's
///   connection cannot read a *different* agent's runs by guessing ids.
/// - Host-role is gated by the **host token**, NOT by `host.subscribe`
///   membership. This is the #254 fix: `host.subscribe` has no authz, so
///   keying on `is_subscribed` let ANY authenticated connection
///   self-elevate and read every agent's traces. The host token is read
///   only from the `0600` `host-token` file (never served over
///   `GET /auth-token`), so a different local user — or a client that
///   scraped the auth token off the HTTP endpoint — cannot obtain it.
/// - It is therefore a real confidentiality boundary against *another
///   local user* (multi-user / CI / remote daemon). It is NOT an
///   isolation boundary against a *malicious same-user process*: that
///   process can read the `0600` host-token (and the `0600` run files)
///   directly. Closing same-user isolation would need DPAPI/Keychain or
///   process-cred (SO_PEERCRED) auth — tracked separately, out of scope.
///
/// NOTE: lower-sensitivity host *metadata* — the agent roster, approvals,
/// and host events delivered by `host.subscribe` / `host.agents` /
/// `host.approvals` — is intentionally still available to any
/// authenticated connection (the local UI consumes it). Only run-trace
/// *content* (prompts, CLI output) requires host-role. Tightening that
/// metadata surface is a separate decision from #254.
async fn authorize_run_access(
    session: &crate::session::ClientSession,
    state: &Arc<ServerState>,
    owning_agent_id: &str,
) -> Result<(), String> {
    // 1. The connection owns the agent (its own run).
    if let Some(bound) = session.agent_id.lock().await.clone() {
        if bound == owning_agent_id {
            return Ok(());
        }
    }
    // 2. The connection authenticated as the host-management client by
    //    presenting the per-launch host token (Parslee-ai/car#254).
    //    NOTE: this is deliberately NOT `host.is_subscribed(...)` — that
    //    check let any authenticated connection self-elevate just by
    //    calling `host.subscribe` (which has no authz), reading every
    //    agent's run traces. host.subscribe still works for host *events*;
    //    it just no longer grants the cross-agent run-trace read.
    if session.is_host.load(std::sync::atomic::Ordering::Acquire) {
        return Ok(());
    }
    // Do NOT name the owning `agent_id` in the error (FIX 3): a caller-
    // supplied or resolved id is not a transparent key, and echoing it back
    // turns a rejection into an existence/owner oracle. Log it server-side
    // for diagnosis; keep the wire message agent-agnostic.
    tracing::debug!(
        owning_agent = %owning_agent_id,
        client_id = %session.client_id,
        "run access denied: connection neither owns the agent nor is the host client"
    );
    Err("not authorized to access this run: this connection neither owns the \
         owning agent (via session.auth) nor is the host management client"
        .to_string())
}

/// `runs.subscribe {run_id}` — register for the run's live
/// `runs.trace.event` stream and return a snapshot + cursor (U4).
///
/// Authorizes the caller against the run's owning `agent_id` (R16), then
/// atomically snapshots the run's turns and registers the subscriber
/// under the `runs` lock (invariant #1, in [`ServerState::subscribe_run`]).
///
/// FIX 3 (existence/owner oracle): an unknown `run_id` and an
/// exists-but-unauthorized `run_id` must be INDISTINGUISHABLE. Both return
/// the same uniform not-found marker (`{ run_id, not_found: true }`,
/// mirroring [`handle_runs_get_trace`]) — never an error frame that varies
/// by case, and never the owning `agent_id`. Otherwise an unentitled caller
/// could probe which `run_id`s exist (and learn the owning agent) by
/// telling "unknown" apart from "not authorized".
async fn handle_runs_subscribe(
    req: &JsonRpcMessage,
    session: &crate::session::ClientSession,
    state: &Arc<ServerState>,
) -> Result<Value, String> {
    let params: car_proto::RunSubscribeRequest = serde_json::from_value(req.params.clone())
        .map_err(|e| format!("runs.subscribe requires {{ run_id }}: {e}"))?;

    // Uniform not-found marker shared by the unknown-run and unauthorized
    // cases so the two are indistinguishable to the caller (FIX 3).
    let not_found = || {
        serde_json::to_value(serde_json::json!({
            "run_id": params.run_id,
            "not_found": true,
        }))
        .map_err(|e| e.to_string())
    };

    // Resolve the run's owning agent_id. Prefer the in-memory registry
    // (live run); fall back to the disk store (a run that exists but whose
    // RunMeta isn't in this process). Unknown run → uniform not-found.
    let owning_agent = match state.run_meta(&params.run_id).await {
        Some(meta) => meta.agent_id,
        None => match state.run_store.agent_for_run(&params.run_id) {
            Some(a) => a,
            None => return not_found(),
        },
    };

    // R16/KTD10: gate before snapshotting. An unauthorized caller gets the
    // SAME uniform not-found as an unknown run — no distinguishable outcome,
    // no leaked owning `agent_id` (logged server-side only).
    if authorize_run_access(session, state, &owning_agent)
        .await
        .is_err()
    {
        tracing::debug!(
            run_id = %params.run_id,
            owning_agent = %owning_agent,
            client_id = %session.client_id,
            "runs.subscribe denied (unauthorized); returning uniform not-found"
        );
        return not_found();
    }

    let snapshot = match state
        .subscribe_run(&params.run_id, &session.client_id, session.channel.clone())
        .await
    {
        Some(s) => s,
        // Raced away between resolution and subscribe — still uniform.
        None => return not_found(),
    };

    serde_json::to_value(snapshot).map_err(|e| e.to_string())
}

/// `runs.unsubscribe {run_id}` — drop this connection's live run-trace
/// subscription for `run_id` (U4). Idempotent: returns `removed: false`
/// if there was nothing to remove. No authorization gate is needed —
/// removing your own subscription leaks nothing.
async fn handle_runs_unsubscribe(
    req: &JsonRpcMessage,
    session: &crate::session::ClientSession,
    state: &Arc<ServerState>,
) -> Result<Value, String> {
    let params: car_proto::RunUnsubscribeRequest = serde_json::from_value(req.params.clone())
        .map_err(|e| format!("runs.unsubscribe requires {{ run_id }}: {e}"))?;
    let removed = state
        .unsubscribe_run(&params.run_id, &session.client_id)
        .await;
    serde_json::to_value(car_proto::RunUnsubscribeResponse {
        run_id: params.run_id,
        removed,
    })
    .map_err(|e| e.to_string())
}

/// `runs.list {agent_id}` — list an agent's runs newest-first for replay
/// (U5). Reads the disk store ([`RunStore::list_runs`]), so it works
/// across daemon restart / `client_id` churn — `agent_id` is the durable
/// key, not the connection.
///
/// R16/KTD10: the `agent_id` is **not** a transparent key. We authorize
/// the caller for it FIRST — it must own the agent (`session.auth
/// {agent_id}`) or be the CarHost host-client — so an unentitled caller
/// can't enumerate other agents' run lists. An authorized caller for an
/// agent with no runs gets an empty list (the empty state, R13).
async fn handle_runs_list(
    req: &JsonRpcMessage,
    session: &crate::session::ClientSession,
    state: &Arc<ServerState>,
) -> Result<Value, String> {
    let params: car_proto::RunListRequest = serde_json::from_value(req.params.clone())
        .map_err(|e| format!("runs.list requires {{ agent_id }}: {e}"))?;

    // Gate before reading — the param is an authorization subject, not a
    // lookup key. An unentitled caller is rejected, never served a list.
    authorize_run_access(session, state, &params.agent_id).await?;

    let runs = state.run_store.list_runs(&params.agent_id);
    serde_json::to_value(serde_json::json!({
        "agent_id": params.agent_id,
        "runs": runs,
    }))
    .map_err(|e| e.to_string())
}

/// `runs.get_trace {run_id, cursor?}` — fetch a run's full ordered trace
/// from the disk store for replay (U5). Like `runs.subscribe`, this reads
/// persisted records ([`RunStore::get_run_trace`]), so a completed run
/// replays identically to what the live stream delivered, and it works on
/// a freshly-restarted daemon with empty memory.
///
/// R16/KTD10: resolve the run's owning `agent_id` from disk and authorize
/// the caller against it before serving any record — a `run_id` is not a
/// transparent key. An unknown `run_id` (no file on disk) returns a clear
/// not-found marker (`{ run_id, not_found: true }`), not an error frame,
/// so a UI can distinguish "no such run" from a transport failure.
///
/// A `timeout`/`Incomplete` run is **not** an error: its persisted trail
/// (the turns recorded so far plus the terminal `Ended`/`Incomplete`
/// marker) is returned, so the dashboard renders the partial run.
async fn handle_runs_get_trace(
    req: &JsonRpcMessage,
    session: &crate::session::ClientSession,
    state: &Arc<ServerState>,
) -> Result<Value, String> {
    let params: car_proto::RunGetTraceRequest = serde_json::from_value(req.params.clone())
        .map_err(|e| format!("runs.get_trace requires {{ run_id, cursor? }}: {e}"))?;

    // Resolve the owning agent from disk (the durable key path — works
    // after a restart). An unknown run_id has no owner: not-found, and we
    // never reach the store read. We don't reveal whether the id exists to
    // an unauthorized caller — authorization is checked against the
    // resolved owner, and resolution failure is the same not-found either
    // way.
    let Some(owning_agent) = state.run_store.agent_for_run(&params.run_id) else {
        return serde_json::to_value(serde_json::json!({
            "run_id": params.run_id,
            "not_found": true,
        }))
        .map_err(|e| e.to_string());
    };

    // R16/KTD10: gate before serving any record.
    authorize_run_access(session, state, &owning_agent).await?;

    // Load the full ordered trail from disk. `agent_for_run` just resolved
    // the file, so this is the cheaper keyed read.
    let records = state
        .run_store
        .get_run_trace_for(&owning_agent, &params.run_id)
        .unwrap_or_default();

    // Optional paging: a non-zero cursor starts the returned slice at that
    // index into the ordered record stream. The response echoes the
    // applied cursor so the client knows the offset. An out-of-range
    // cursor yields an empty slice (the client has the whole run already).
    let cursor = params.cursor.unwrap_or(0);
    let paged: Vec<car_proto::RunRecord> = if cursor == 0 {
        records
    } else {
        records.into_iter().skip(cursor).collect()
    };

    serde_json::to_value(serde_json::json!({
        "run_id": params.run_id,
        "agent_id": owning_agent,
        "records": paged,
        "cursor": cursor,
    }))
    .map_err(|e| e.to_string())
}

/// Update the per-session replan config. Wire shape mirrors the
/// FFI's positional `set_replan_config` arguments — the engine
/// crate's `ReplanConfig` struct doesn't derive Serialize, so we
/// reconstruct it from a flat object here.
async fn handle_replan_set_config(
    msg: &JsonRpcMessage,
    session: &crate::session::ClientSession,
) -> Result<Value, String> {
    let max_replans = msg
        .params
        .get("max_replans")
        .and_then(|v| v.as_u64())
        .unwrap_or(0) as u32;
    let delay_ms = msg
        .params
        .get("delay_ms")
        .and_then(|v| v.as_u64())
        .unwrap_or(0);
    let verify_before_execute = msg
        .params
        .get("verify_before_execute")
        .and_then(|v| v.as_bool())
        .unwrap_or(true);
    let cfg = car_engine::ReplanConfig {
        max_replans,
        delay_ms,
        verify_before_execute,
    };
    session.runtime.set_replan_config(cfg).await;
    Ok(Value::Null)
}

async fn handle_skills_list(
    msg: &JsonRpcMessage,
    session: &crate::session::ClientSession,
) -> Result<Value, String> {
    let domain = msg.params.get("domain").and_then(|v| v.as_str());
    let engine = session.memgine.lock().await;
    let skills: Vec<serde_json::Value> = engine
        .graph
        .inner
        .node_indices()
        .filter_map(|nix| {
            let node = engine.graph.inner.node_weight(nix)?;
            if node.kind != car_memgine::MemKind::Skill {
                return None;
            }
            let meta = car_memgine::SkillMeta::from_node(node)?;
            if let Some(d) = domain {
                match &meta.scope {
                    car_memgine::SkillScope::Global => {}
                    car_memgine::SkillScope::Domain(sd) if sd == d => {}
                    _ => return None,
                }
            }
            Some(serde_json::to_value(&meta).unwrap_or_default())
        })
        .collect();
    serde_json::to_value(&skills).map_err(|e| e.to_string())
}

#[derive(serde::Deserialize)]
struct SecretParams {
    #[serde(default)]
    service: Option<String>,
    key: String,
    #[serde(default)]
    value: Option<String>,
}

fn handle_secret_put(req: &JsonRpcMessage) -> Result<Value, String> {
    let p: SecretParams = serde_json::from_value(req.params.clone()).map_err(|e| e.to_string())?;
    let value = p.value.ok_or_else(|| "missing 'value'".to_string())?;
    car_ffi_common::secrets::put(p.service.as_deref(), &p.key, &value)
}

fn handle_secret_get(req: &JsonRpcMessage) -> Result<Value, String> {
    let p: SecretParams = serde_json::from_value(req.params.clone()).map_err(|e| e.to_string())?;
    car_ffi_common::secrets::get(p.service.as_deref(), &p.key)
}

fn handle_secret_delete(req: &JsonRpcMessage) -> Result<Value, String> {
    let p: SecretParams = serde_json::from_value(req.params.clone()).map_err(|e| e.to_string())?;
    car_ffi_common::secrets::delete(p.service.as_deref(), &p.key)
}

fn handle_secret_status(req: &JsonRpcMessage) -> Result<Value, String> {
    let p: SecretParams = serde_json::from_value(req.params.clone()).map_err(|e| e.to_string())?;
    car_ffi_common::secrets::status(p.service.as_deref(), &p.key)
}

#[derive(serde::Deserialize)]
struct PermParams {
    domain: String,
    #[serde(default)]
    target_bundle_id: Option<String>,
}

fn handle_perm_status(req: &JsonRpcMessage) -> Result<Value, String> {
    let p: PermParams = serde_json::from_value(req.params.clone()).map_err(|e| e.to_string())?;
    car_ffi_common::permissions::status(&p.domain, p.target_bundle_id.as_deref())
}

fn handle_perm_request(req: &JsonRpcMessage) -> Result<Value, String> {
    let p: PermParams = serde_json::from_value(req.params.clone()).map_err(|e| e.to_string())?;
    car_ffi_common::permissions::request(&p.domain, p.target_bundle_id.as_deref())
}

fn handle_perm_explain(req: &JsonRpcMessage) -> Result<Value, String> {
    let p: PermParams = serde_json::from_value(req.params.clone()).map_err(|e| e.to_string())?;
    car_ffi_common::permissions::explain(&p.domain, p.target_bundle_id.as_deref())
}

fn handle_calendar_events(req: &JsonRpcMessage) -> Result<Value, String> {
    #[derive(serde::Deserialize)]
    struct P {
        start: String,
        end: String,
        #[serde(default)]
        calendar_ids: Vec<String>,
    }
    let p: P = serde_json::from_value(req.params.clone()).map_err(|e| e.to_string())?;
    let start = chrono::DateTime::parse_from_rfc3339(&p.start)
        .map_err(|e| format!("parse start: {}", e))?
        .with_timezone(&chrono::Utc);
    let end = chrono::DateTime::parse_from_rfc3339(&p.end)
        .map_err(|e| format!("parse end: {}", e))?
        .with_timezone(&chrono::Utc);
    car_ffi_common::integrations::calendar_events(start, end, &p.calendar_ids)
}

fn handle_calendar_create_event(req: &JsonRpcMessage) -> Result<Value, String> {
    let raw = req.params.to_string();
    car_ffi_common::integrations::calendar_create_event(&raw)
}

fn handle_calendar_update_event(req: &JsonRpcMessage) -> Result<Value, String> {
    let raw = req.params.to_string();
    car_ffi_common::integrations::calendar_update_event(&raw)
}

fn handle_calendar_delete_event(req: &JsonRpcMessage) -> Result<Value, String> {
    #[derive(serde::Deserialize)]
    struct P {
        event_id: String,
    }
    let p: P = serde_json::from_value(req.params.clone()).map_err(|e| e.to_string())?;
    car_ffi_common::integrations::calendar_delete_event(&p.event_id)
}

fn handle_contacts_find(req: &JsonRpcMessage) -> Result<Value, String> {
    #[derive(serde::Deserialize)]
    struct P {
        query: String,
        #[serde(default = "default_limit")]
        limit: usize,
        #[serde(default)]
        container_ids: Vec<String>,
    }
    fn default_limit() -> usize {
        50
    }
    let p: P = serde_json::from_value(req.params.clone()).map_err(|e| e.to_string())?;
    car_ffi_common::integrations::contacts_list(&p.query, &p.container_ids, p.limit)
}

fn handle_mail_inbox(req: &JsonRpcMessage) -> Result<Value, String> {
    #[derive(serde::Deserialize, Default)]
    struct P {
        #[serde(default)]
        account_ids: Vec<String>,
    }
    let p: P = serde_json::from_value(req.params.clone()).unwrap_or_default();
    car_ffi_common::integrations::mail_inbox(&p.account_ids)
}

fn handle_mail_send(req: &JsonRpcMessage) -> Result<Value, String> {
    let raw = req.params.to_string();
    car_ffi_common::integrations::mail_send(&raw)
}

fn handle_messages_chats(req: &JsonRpcMessage) -> Result<Value, String> {
    #[derive(serde::Deserialize)]
    struct P {
        #[serde(default = "default_limit")]
        limit: usize,
    }
    fn default_limit() -> usize {
        50
    }
    let p: P = serde_json::from_value(req.params.clone()).unwrap_or(P { limit: 50 });
    car_ffi_common::integrations::messages_chats(p.limit)
}

fn handle_messages_send(req: &JsonRpcMessage) -> Result<Value, String> {
    let raw = req.params.to_string();
    car_ffi_common::integrations::messages_send(&raw)
}

fn handle_notes_find(req: &JsonRpcMessage) -> Result<Value, String> {
    #[derive(serde::Deserialize)]
    struct P {
        query: String,
        #[serde(default = "default_limit")]
        limit: usize,
    }
    fn default_limit() -> usize {
        50
    }
    let p: P = serde_json::from_value(req.params.clone()).map_err(|e| e.to_string())?;
    car_ffi_common::integrations::notes_find(&p.query, p.limit)
}

fn handle_reminders_items(req: &JsonRpcMessage) -> Result<Value, String> {
    #[derive(serde::Deserialize)]
    struct P {
        #[serde(default = "default_limit")]
        limit: usize,
    }
    fn default_limit() -> usize {
        50
    }
    let p: P = serde_json::from_value(req.params.clone()).unwrap_or(P { limit: 50 });
    car_ffi_common::integrations::reminders_items(p.limit)
}

fn handle_bookmarks_list(req: &JsonRpcMessage) -> Result<Value, String> {
    #[derive(serde::Deserialize)]
    struct P {
        #[serde(default = "default_limit")]
        limit: usize,
    }
    fn default_limit() -> usize {
        100
    }
    let p: P = serde_json::from_value(req.params.clone()).unwrap_or(P { limit: 100 });
    car_ffi_common::integrations::bookmarks_list(p.limit)
}

fn handle_health_sleep(req: &JsonRpcMessage) -> Result<Value, String> {
    #[derive(serde::Deserialize)]
    struct P {
        start: String,
        end: String,
    }
    let p: P = serde_json::from_value(req.params.clone()).map_err(|e| e.to_string())?;
    let s = chrono::DateTime::parse_from_rfc3339(&p.start)
        .map_err(|e| format!("parse start: {}", e))?
        .with_timezone(&chrono::Utc);
    let e = chrono::DateTime::parse_from_rfc3339(&p.end)
        .map_err(|e| format!("parse end: {}", e))?
        .with_timezone(&chrono::Utc);
    car_ffi_common::health::sleep_windows(s, e)
}

fn handle_health_workouts(req: &JsonRpcMessage) -> Result<Value, String> {
    #[derive(serde::Deserialize)]
    struct P {
        start: String,
        end: String,
    }
    let p: P = serde_json::from_value(req.params.clone()).map_err(|e| e.to_string())?;
    let s = chrono::DateTime::parse_from_rfc3339(&p.start)
        .map_err(|e| format!("parse start: {}", e))?
        .with_timezone(&chrono::Utc);
    let e = chrono::DateTime::parse_from_rfc3339(&p.end)
        .map_err(|e| format!("parse end: {}", e))?
        .with_timezone(&chrono::Utc);
    car_ffi_common::health::workouts(s, e)
}

fn handle_health_activity(req: &JsonRpcMessage) -> Result<Value, String> {
    #[derive(serde::Deserialize)]
    struct P {
        start: String,
        end: String,
    }
    let p: P = serde_json::from_value(req.params.clone()).map_err(|e| e.to_string())?;
    let s = chrono::NaiveDate::parse_from_str(&p.start, "%Y-%m-%d")
        .map_err(|e| format!("parse start: {}", e))?;
    let e = chrono::NaiveDate::parse_from_str(&p.end, "%Y-%m-%d")
        .map_err(|e| format!("parse end: {}", e))?;
    car_ffi_common::health::activity(s, e)
}

async fn handle_browser_close(session: &crate::session::ClientSession) -> Result<Value, String> {
    let closed = session.browser.close().await?;
    Ok(serde_json::json!({"closed": closed}))
}

async fn handle_browser_run(
    req: &JsonRpcMessage,
    session: &crate::session::ClientSession,
) -> Result<Value, String> {
    #[derive(serde::Deserialize)]
    struct BrowserRunParams {
        /// Inline JSON string (CLI-compatible), OR the structured object.
        script: Value,
        #[serde(default)]
        width: Option<u32>,
        #[serde(default)]
        height: Option<u32>,
        /// When true, launches a visible Chromium window for interactive
        /// flows (first-time auth, 2FA, supervised runs). Only honored on
        /// the call that first launches the browser session — subsequent
        /// calls reuse the existing browser regardless.
        #[serde(default)]
        headed: Option<bool>,
        /// Extra Chromium command-line flags appended verbatim at
        /// launch (#112). Honoured only on the launch call.
        #[serde(default)]
        extra_args: Option<Vec<String>>,
    }
    let params: BrowserRunParams =
        serde_json::from_value(req.params.clone()).map_err(|e| e.to_string())?;

    // Accept either a JSON string OR a structured object under `script`.
    let script_json = match params.script {
        Value::String(s) => s,
        other => other.to_string(),
    };

    let browser_session = session
        .browser
        .get_or_launch(car_ffi_common::browser::BrowserLaunchOptions {
            width: params.width.unwrap_or(1280),
            height: params.height.unwrap_or(720),
            headless: !params.headed.unwrap_or(false),
            extra_args: params.extra_args.unwrap_or_default(),
        })
        .await?;

    let trace_json = browser_session.run(&script_json).await?;
    serde_json::from_str(&trace_json).map_err(|e| e.to_string())
}

// ---------------------------------------------------------------------------
// Voice streaming JSON-RPC methods
//
// Events are pushed back to the originating client as JSON-RPC notifications:
//   { "jsonrpc": "2.0", "method": "voice.event",
//     "params": { "session_id": "...", "event": {...} } }
//
// The session registry is process-wide (ServerState.voice_sessions); per-call
// WsVoiceEventSink instances bind each session to its originating WS so a
// client only ever sees events for sessions it started.
// ---------------------------------------------------------------------------

#[derive(Deserialize)]
struct VoiceStartParams {
    session_id: String,
    audio_source: Value,
    #[serde(default)]
    options: Option<Value>,
}

async fn handle_voice_transcribe_stream_start(
    req: &JsonRpcMessage,
    state: &Arc<ServerState>,
    session: &Arc<crate::session::ClientSession>,
) -> Result<Value, String> {
    let params: VoiceStartParams =
        serde_json::from_value(req.params.clone()).map_err(|e| e.to_string())?;
    let audio_source_json =
        serde_json::to_string(&params.audio_source).map_err(|e| e.to_string())?;
    let options_json = params
        .options
        .as_ref()
        .map(|v| serde_json::to_string(v).map_err(|e| e.to_string()))
        .transpose()?;
    let sink: Arc<dyn car_voice::VoiceEventSink> = Arc::new(crate::session::WsVoiceEventSink {
        channel: session.channel.clone(),
    });
    let json = car_ffi_common::voice::transcribe_stream_start(
        &params.session_id,
        &audio_source_json,
        options_json.as_deref(),
        state.voice_sessions.clone(),
        sink,
    )
    .await?;
    serde_json::from_str(&json).map_err(|e| e.to_string())
}

#[derive(Deserialize)]
struct VoiceStopParams {
    session_id: String,
}

async fn handle_voice_transcribe_stream_stop(
    req: &JsonRpcMessage,
    state: &Arc<ServerState>,
) -> Result<Value, String> {
    let params: VoiceStopParams =
        serde_json::from_value(req.params.clone()).map_err(|e| e.to_string())?;
    let json = car_ffi_common::voice::transcribe_stream_stop(
        &params.session_id,
        state.voice_sessions.clone(),
    )
    .await?;
    serde_json::from_str(&json).map_err(|e| e.to_string())
}

#[derive(Deserialize)]
struct VoicePushParams {
    session_id: String,
    /// Base64-encoded 16-bit signed PCM frame. JSON-RPC is text, so binary
    /// audio frames have to be encoded; clients in WS-binary contexts that
    /// want to skip the round trip can call the FFI directly.
    pcm_b64: String,
}

async fn handle_voice_transcribe_stream_push(
    req: &JsonRpcMessage,
    state: &Arc<ServerState>,
) -> Result<Value, String> {
    use base64::Engine;
    let params: VoicePushParams =
        serde_json::from_value(req.params.clone()).map_err(|e| e.to_string())?;
    let pcm = base64::engine::general_purpose::STANDARD
        .decode(&params.pcm_b64)
        .map_err(|e| format!("invalid pcm_b64: {}", e))?;
    let json = car_ffi_common::voice::transcribe_stream_push(
        &params.session_id,
        &pcm,
        state.voice_sessions.clone(),
    )
    .await?;
    serde_json::from_str(&json).map_err(|e| e.to_string())
}

fn handle_voice_sessions_list(state: &Arc<ServerState>) -> Value {
    let json = car_ffi_common::voice::list_voice_sessions(state.voice_sessions.clone());
    serde_json::from_str(&json).unwrap_or(Value::Null)
}

#[derive(Deserialize)]
struct VoiceTtsStreamStartParams {
    /// Caller-chosen opaque id for this stream. Used both as the
    /// session_id wrapped onto each `voice.event` notification AND as
    /// the key for `voice.tts_stream.cancel`.
    stream_id: String,
    /// Text to synthesize. Splitting into multiple synth calls (for
    /// long-form narration) is the caller's responsibility.
    text: String,
    /// Optional [`car_ffi_common::voice::TtsStreamOptions`] as a raw
    /// JSON value (provider, voice_id, binary_frames).
    #[serde(default)]
    options: Option<Value>,
}

async fn handle_voice_tts_stream_start(
    req: &JsonRpcMessage,
    session: &Arc<crate::session::ClientSession>,
) -> Result<Value, String> {
    let params: VoiceTtsStreamStartParams =
        serde_json::from_value(req.params.clone()).map_err(|e| e.to_string())?;
    let opts_str = params
        .options
        .as_ref()
        .map(|v| v.to_string())
        .filter(|s| !s.is_empty());
    let sink: Arc<dyn car_voice::VoiceEventSink> = Arc::new(crate::session::WsVoiceEventSink {
        channel: session.channel.clone(),
    });
    let json = car_ffi_common::voice::tts_stream_start(
        &params.stream_id,
        &params.text,
        opts_str.as_deref(),
        sink,
    )
    .await?;
    serde_json::from_str(&json).map_err(|e| e.to_string())
}

#[derive(Deserialize)]
struct VoiceTtsStreamCancelParams {
    stream_id: String,
}

async fn handle_voice_tts_stream_cancel(req: &JsonRpcMessage) -> Result<Value, String> {
    let params: VoiceTtsStreamCancelParams =
        serde_json::from_value(req.params.clone()).map_err(|e| e.to_string())?;
    let json = car_ffi_common::voice::tts_stream_cancel(&params.stream_id).await?;
    serde_json::from_str(&json).map_err(|e| e.to_string())
}

fn handle_voice_tts_stream_list() -> Value {
    let json = car_ffi_common::voice::list_tts_streams();
    serde_json::from_str(&json).unwrap_or(Value::Null)
}

async fn handle_voice_dispatch_turn(
    req: &JsonRpcMessage,
    state: &Arc<ServerState>,
    session: &Arc<crate::session::ClientSession>,
) -> Result<Value, String> {
    let req_value = req.params.clone();
    let request: crate::voice_turn::DispatchVoiceTurnRequest =
        serde_json::from_value(req_value).map_err(|e| e.to_string())?;
    let engine = get_inference_engine(state).clone();
    let sink: Arc<dyn car_voice::VoiceEventSink> = Arc::new(crate::session::WsVoiceEventSink {
        channel: session.channel.clone(),
    });
    let resp = crate::voice_turn::dispatch(engine, request, sink).await?;
    serde_json::to_value(resp).map_err(|e| e.to_string())
}

async fn handle_voice_cancel_turn() -> Result<Value, String> {
    crate::voice_turn::cancel().await;
    Ok(serde_json::json!({"cancelled": true}))
}

async fn handle_voice_prewarm_turn(state: &Arc<ServerState>) -> Result<Value, String> {
    let engine = get_inference_engine(state).clone();
    crate::voice_turn::prewarm(engine).await;
    Ok(serde_json::json!({"prewarmed": true}))
}

// ---------------------------------------------------------------------------
// Inference runner over WebSocket — closes Parslee-ai/car-releases#24
//
// Bidirectional protocol shape:
//   1. Client → server: `inference.register_runner` (no params). The
//      session that calls this becomes the host for delegated models.
//   2. Server → client: `inference.runner.invoke` notification with
//      {call_id, request} when CAR needs to dispatch a delegated turn.
//   3. Client → server: `inference.runner.event` with {call_id, event}
//      for each chunk; `inference.runner.complete` with {call_id, result}
//      on success; `inference.runner.fail` with {call_id, error} on
//      failure.
//
// The server-side data is process-wide because only one inference
// runner can be registered at a time (matches the FFI bindings'
// constraint). The per-call mailboxes live in dedicated DashMaps.
// ---------------------------------------------------------------------------

fn ws_runner_session() -> &'static std::sync::RwLock<Option<Arc<crate::session::WsChannel>>> {
    static SLOT: std::sync::OnceLock<std::sync::RwLock<Option<Arc<crate::session::WsChannel>>>> =
        std::sync::OnceLock::new();
    SLOT.get_or_init(|| std::sync::RwLock::new(None))
}

fn ws_runner_calls() -> &'static dashmap::DashMap<String, car_inference::EventEmitter> {
    static MAP: std::sync::OnceLock<dashmap::DashMap<String, car_inference::EventEmitter>> =
        std::sync::OnceLock::new();
    MAP.get_or_init(dashmap::DashMap::new)
}

fn ws_runner_completions() -> &'static dashmap::DashMap<
    String,
    tokio::sync::oneshot::Sender<std::result::Result<car_inference::RunnerResult, String>>,
> {
    static MAP: std::sync::OnceLock<
        dashmap::DashMap<
            String,
            tokio::sync::oneshot::Sender<std::result::Result<car_inference::RunnerResult, String>>,
        >,
    > = std::sync::OnceLock::new();
    MAP.get_or_init(dashmap::DashMap::new)
}

struct WsInferenceRunner;

#[async_trait::async_trait]
impl car_inference::InferenceRunner for WsInferenceRunner {
    async fn run(
        &self,
        request: car_inference::tasks::generate::GenerateRequest,
        emitter: car_inference::EventEmitter,
    ) -> std::result::Result<car_inference::RunnerResult, car_inference::RunnerError> {
        let channel = ws_runner_session()
            .read()
            .map_err(|e| {
                car_inference::RunnerError::Failed(format!("ws runner slot poisoned: {e}"))
            })?
            .clone()
            .ok_or_else(|| {
                car_inference::RunnerError::Declined(
                    "no WebSocket inference runner registered — call inference.register_runner first"
                        .into(),
                )
            })?;

        let call_id = uuid::Uuid::new_v4().to_string();
        let request_json = serde_json::to_value(&request)
            .map_err(|e| car_inference::RunnerError::Failed(e.to_string()))?;
        let (tx, rx) = tokio::sync::oneshot::channel();
        ws_runner_calls().insert(call_id.clone(), emitter);
        ws_runner_completions().insert(call_id.clone(), tx);

        // Fire the invoke notification.
        use futures::SinkExt;
        let notification = serde_json::json!({
            "jsonrpc": "2.0",
            "method": "inference.runner.invoke",
            "params": {
                "call_id": call_id,
                "request": request_json,
            },
        });
        let text = serde_json::to_string(&notification)
            .map_err(|e| car_inference::RunnerError::Failed(e.to_string()))?;
        let _ = channel
            .write
            .lock()
            .await
            .send(tokio_tungstenite::tungstenite::Message::Text(text.into()))
            .await;

        let result = rx.await.map_err(|_| {
            car_inference::RunnerError::Failed("runner completion channel dropped".into())
        })?;
        ws_runner_calls().remove(&call_id);
        result.map_err(car_inference::RunnerError::Failed)
    }
}

async fn handle_inference_register_runner(
    session: &Arc<crate::session::ClientSession>,
) -> Result<Value, String> {
    let mut guard = ws_runner_session()
        .write()
        .map_err(|e| format!("ws runner slot poisoned: {e}"))?;
    *guard = Some(session.channel.clone());
    drop(guard);
    car_inference::set_inference_runner(Some(Arc::new(WsInferenceRunner)));
    Ok(serde_json::json!({"registered": true}))
}

#[derive(serde::Deserialize)]
struct InferenceRunnerEventParams {
    call_id: String,
    event: Value,
}

async fn handle_inference_runner_event(req: &JsonRpcMessage) -> Result<Value, String> {
    let params: InferenceRunnerEventParams =
        serde_json::from_value(req.params.clone()).map_err(|e| e.to_string())?;
    let stream_event = match parse_runner_event_value(&params.event) {
        Some(e) => e,
        None => return Err("unrecognised runner event shape".into()),
    };
    if let Some(entry) = ws_runner_calls().get(&params.call_id) {
        let emitter = entry.value().clone();
        tokio::spawn(async move { emitter.emit(stream_event).await });
    }
    Ok(serde_json::json!({"emitted": true}))
}

#[derive(serde::Deserialize)]
struct InferenceRunnerCompleteParams {
    call_id: String,
    result: Value,
}

async fn handle_inference_runner_complete(req: &JsonRpcMessage) -> Result<Value, String> {
    let params: InferenceRunnerCompleteParams =
        serde_json::from_value(req.params.clone()).map_err(|e| e.to_string())?;
    let result: std::result::Result<car_inference::RunnerResult, String> =
        serde_json::from_value(params.result)
            .map_err(|e| format!("invalid RunnerResult JSON: {e}"));
    if let Some((_, tx)) = ws_runner_completions().remove(&params.call_id) {
        let _ = tx.send(result);
    }
    Ok(serde_json::json!({"completed": true}))
}

#[derive(serde::Deserialize)]
struct InferenceRunnerFailParams {
    call_id: String,
    error: String,
}

async fn handle_inference_runner_fail(req: &JsonRpcMessage) -> Result<Value, String> {
    let params: InferenceRunnerFailParams =
        serde_json::from_value(req.params.clone()).map_err(|e| e.to_string())?;
    if let Some((_, tx)) = ws_runner_completions().remove(&params.call_id) {
        let _ = tx.send(Err(params.error));
    }
    Ok(serde_json::json!({"failed": true}))
}

fn parse_runner_event_value(v: &Value) -> Option<car_inference::StreamEvent> {
    let ty = v.get("type").and_then(|t| t.as_str())?;
    match ty {
        "text" => Some(car_inference::StreamEvent::TextDelta(
            v.get("data")?.as_str()?.to_string(),
        )),
        "tool_start" => Some(car_inference::StreamEvent::ToolCallStart {
            name: v.get("name")?.as_str()?.to_string(),
            index: v.get("index")?.as_u64()? as usize,
            id: v.get("id").and_then(|i| i.as_str()).map(str::to_string),
        }),
        "tool_delta" => Some(car_inference::StreamEvent::ToolCallDelta {
            index: v.get("index")?.as_u64()? as usize,
            arguments_delta: v.get("data")?.as_str()?.to_string(),
        }),
        "usage" => Some(car_inference::StreamEvent::Usage {
            input_tokens: v.get("input_tokens")?.as_u64()?,
            output_tokens: v.get("output_tokens")?.as_u64()?,
        }),
        "done" => Some(car_inference::StreamEvent::Done {
            text: v.get("text")?.as_str()?.to_string(),
            tool_calls: v
                .get("tool_calls")
                .and_then(|tc| serde_json::from_value(tc.clone()).ok())
                .unwrap_or_default(),
        }),
        _ => None,
    }
}

#[derive(Deserialize)]
struct EnrollSpeakerParams {
    label: String,
    audio: Value,
}

async fn handle_enroll_speaker(req: &JsonRpcMessage) -> Result<Value, String> {
    let params: EnrollSpeakerParams =
        serde_json::from_value(req.params.clone()).map_err(|e| e.to_string())?;
    let audio_json = serde_json::to_string(&params.audio).map_err(|e| e.to_string())?;
    let json = car_ffi_common::voice::enroll_speaker(&params.label, &audio_json).await?;
    serde_json::from_str(&json).map_err(|e| e.to_string())
}

#[derive(Deserialize)]
struct RemoveEnrollmentParams {
    label: String,
}

fn handle_remove_enrollment(req: &JsonRpcMessage) -> Result<Value, String> {
    let params: RemoveEnrollmentParams =
        serde_json::from_value(req.params.clone()).map_err(|e| e.to_string())?;
    let json = car_ffi_common::voice::remove_enrollment(&params.label)?;
    serde_json::from_str(&json).map_err(|e| e.to_string())
}

#[derive(Deserialize)]
struct WorkflowRunParams {
    workflow: Value,
}

async fn handle_workflow_run(
    req: &JsonRpcMessage,
    session: &Arc<crate::session::ClientSession>,
) -> Result<Value, String> {
    let params: WorkflowRunParams =
        serde_json::from_value(req.params.clone()).map_err(|e| e.to_string())?;
    let workflow_json = serde_json::to_string(&params.workflow).map_err(|e| e.to_string())?;
    let runner: Arc<dyn car_multi::AgentRunner> = Arc::new(WsAgentRunner {
        channel: session.channel.clone(),
        host: session.host.clone(),
        client_id: session.client_id.clone(),
    });
    let json = car_ffi_common::workflow::run_workflow(&workflow_json, runner).await?;
    let result: Value = serde_json::from_str(&json).map_err(|e| e.to_string())?;
    // If the run parked at an approval gate, persist the checkpoint durably so
    // it survives a daemon restart and can be resumed by run_id.
    persist_if_paused(&result)?;
    Ok(result)
}

/// Re-arm workflow runs orphaned by a crash between an approval `claim` and its
/// `complete`. Call **once at daemon startup**, before serving connections, so a
/// restart mid-approval doesn't bury paused runs. Best-effort: logs and returns
/// on any error rather than failing boot.
pub fn recover_workflow_checkpoints() {
    let dir = match workflow_runs_dir() {
        Ok(d) => d,
        Err(e) => {
            tracing::debug!(error = %e, "no workflow checkpoint dir; skipping recovery");
            return;
        }
    };
    match car_workflow::CheckpointStore::open(&dir).and_then(|s| s.recover_orphaned()) {
        Ok(0) => {}
        Ok(n) => tracing::info!(rearmed = n, "recovered orphaned workflow checkpoints"),
        Err(e) => tracing::warn!(error = %e, "workflow checkpoint recovery failed"),
    }
}

/// Directory holding durable workflow checkpoints (`~/.car/workflow-runs`).
fn workflow_runs_dir() -> Result<std::path::PathBuf, String> {
    let home = std::env::var("HOME")
        .or_else(|_| std::env::var("USERPROFILE"))
        .map_err(|_| "cannot resolve home directory for workflow checkpoints".to_string())?;
    Ok(std::path::PathBuf::from(home)
        .join(".car")
        .join("workflow-runs"))
}

/// If `result` is a paused WorkflowResult, save its checkpoint to the store.
fn persist_if_paused(result: &Value) -> Result<(), String> {
    if result.get("status").and_then(|s| s.as_str()) != Some("paused") {
        return Ok(());
    }
    let Some(paused_value) = result.get("paused") else {
        return Err("paused result missing checkpoint".to_string());
    };
    let paused: car_workflow::PausedWorkflow =
        serde_json::from_value(paused_value.clone()).map_err(|e| e.to_string())?;
    let store = car_workflow::CheckpointStore::open(workflow_runs_dir()?).map_err(|e| e.to_string())?;
    store.save(&paused).map_err(|e| e.to_string())
}

#[derive(Deserialize)]
struct WorkflowResumeParams {
    run_id: String,
    #[serde(default)]
    input: Value,
}

async fn handle_workflow_resume(
    req: &JsonRpcMessage,
    session: &Arc<crate::session::ClientSession>,
) -> Result<Value, String> {
    let params: WorkflowResumeParams =
        serde_json::from_value(req.params.clone()).map_err(|e| e.to_string())?;

    let store =
        car_workflow::CheckpointStore::open(workflow_runs_dir()?).map_err(|e| e.to_string())?;
    // Atomic claim: a duplicate/racing resume of the same run gets nothing,
    // so a side-effecting downstream stage never runs twice.
    let paused = store
        .claim(&params.run_id)
        .map_err(|e| e.to_string())?
        .ok_or_else(|| {
            format!(
                "no paused workflow run '{}' (already resumed or unknown)",
                params.run_id
            )
        })?;
    let paused_json = serde_json::to_string(&paused).map_err(|e| e.to_string())?;
    let input_json = if params.input.is_null() {
        "{}".to_string()
    } else {
        serde_json::to_string(&params.input).map_err(|e| e.to_string())?
    };

    let runner: Arc<dyn car_multi::AgentRunner> = Arc::new(WsAgentRunner {
        channel: session.channel.clone(),
        host: session.host.clone(),
        client_id: session.client_id.clone(),
    });

    let json = car_ffi_common::workflow::resume_workflow(&paused_json, &input_json, runner).await;
    let result = match json {
        Ok(j) => serde_json::from_str::<Value>(&j).map_err(|e| e.to_string())?,
        Err(e) => {
            // Resume failed (e.g. invalid input) — release the in-flight marker
            // so the run can be resumed again with corrected input.
            let _ = store.save(&paused);
            let _ = store.complete(&params.run_id);
            return Err(e);
        }
    };
    // Re-paused at another gate → persist the fresh checkpoint; either way drop
    // the in-flight marker from the claim.
    persist_if_paused(&result)?;
    store.complete(&params.run_id).map_err(|e| e.to_string())?;
    Ok(result)
}

#[derive(Deserialize)]
struct BuilderBuildParams {
    goal: String,
    #[serde(default)]
    existing: Value,
    #[serde(default = "default_builder_attempts")]
    max_attempts: u32,
}

fn default_builder_attempts() -> u32 {
    3
}

/// `builder.build` — natural language → validated workflow. Runs on the daemon so
/// the catalog is authoritative: tools come from this session's registered tool
/// schemas and models from the inference registry, making the builder's
/// tool-existence cross-check meaningful.
async fn handle_builder_build(
    req: &JsonRpcMessage,
    state: &Arc<ServerState>,
    session: &Arc<crate::session::ClientSession>,
) -> Result<Value, String> {
    let params: BuilderBuildParams =
        serde_json::from_value(req.params.clone()).map_err(|e| e.to_string())?;
    if params.goal.trim().is_empty() {
        return Err("missing 'goal'".to_string());
    }

    let engine = get_inference_engine(state).clone();

    let tools: Vec<car_builder::ToolInfo> = session
        .runtime
        .registry
        .schemas()
        .await
        .into_iter()
        .map(|s| car_builder::ToolInfo {
            name: s.name,
            description: s.description,
        })
        .collect();
    let models: Vec<String> = engine
        .list_models_unified()
        .into_iter()
        .map(|m| m.id)
        .collect();
    let catalog = car_builder::ToolCatalog {
        tools,
        models,
        agents: Vec::new(),
    };

    let existing = if params.existing.is_null() {
        None
    } else {
        serde_json::from_value::<car_workflow::Workflow>(params.existing.clone()).ok()
    };

    let build_req = car_builder::BuildRequest {
        goal: params.goal,
        catalog,
        existing,
        feedback: None,
        max_attempts: params.max_attempts,
    };

    let result = car_builder::build_workflow(
        |prompt: String| {
            let engine = engine.clone();
            async move {
                let greq = car_inference::GenerateRequest {
                    prompt,
                    model: None,
                    params: car_inference::GenerateParams {
                        temperature: 0.2,
                        max_tokens: 4096,
                        ..Default::default()
                    },
                    context: None,
                    tools: None,
                    images: None,
                    messages: None,
                    cache_control: false,
                    response_format: None,
                    intent: None,
                };
                engine
                    .generate_tracked(greq)
                    .await
                    .map(|r| r.text)
                    .map_err(|e| e.to_string())
            }
        },
        &build_req,
    )
    .await;

    Ok(serde_json::json!({
        "valid": result.valid,
        "workflow": result.workflow,
        "issues": result.issues,
        "warnings": result.warnings,
        "attempts": result.attempts,
    }))
}

#[derive(Deserialize)]
struct WorkflowVerifyParams {
    workflow: Value,
}

fn handle_workflow_verify(req: &JsonRpcMessage) -> Result<Value, String> {
    let params: WorkflowVerifyParams =
        serde_json::from_value(req.params.clone()).map_err(|e| e.to_string())?;
    let workflow_json = serde_json::to_string(&params.workflow).map_err(|e| e.to_string())?;
    let json = car_ffi_common::workflow::verify_workflow(&workflow_json)?;
    serde_json::from_str(&json).map_err(|e| e.to_string())
}

// ---------------------------------------------------------------------------
// Meeting JSON-RPC methods
// ---------------------------------------------------------------------------

async fn handle_meeting_start(
    req: &JsonRpcMessage,
    state: &Arc<ServerState>,
    session: &Arc<crate::session::ClientSession>,
) -> Result<Value, String> {
    // We need the meeting id BEFORE handing the upstream sink to
    // start_meeting so the WsMemgineIngestSink stamps transcripts with
    // the correct `meeting/<id>/<source>` speaker. Parse the request
    // here, mint an id if none was provided, and pass the same id
    // through to start_meeting via the request JSON.
    let mut req_value = req.params.clone();
    let meeting_id = req_value
        .get("id")
        .and_then(|v| v.as_str())
        .map(str::to_string)
        .unwrap_or_else(|| uuid::Uuid::new_v4().simple().to_string());
    if let Some(map) = req_value.as_object_mut() {
        map.insert("id".into(), Value::String(meeting_id.clone()));
    }
    let request_json = serde_json::to_string(&req_value).map_err(|e| e.to_string())?;

    let ws_upstream: Arc<dyn car_voice::VoiceEventSink> =
        Arc::new(crate::session::WsVoiceEventSink {
            channel: session.channel.clone(),
        });

    // Wrap the WS upstream with a memgine-ingest fanout that uses the
    // tokio::sync::Mutex-wrapped session memgine. We pass `None` for
    // the FFI-common `start_meeting` memgine arg to avoid the
    // sync-mutex contract there — ingest happens here instead.
    let upstream: Arc<dyn car_voice::VoiceEventSink> =
        Arc::new(crate::session::WsMemgineIngestSink {
            meeting_id,
            engine: session.memgine.clone(),
            upstream: ws_upstream,
        });

    let cwd = std::env::current_dir().ok();
    let json = crate::meeting::start_meeting(
        &request_json,
        state.meetings.clone(),
        state.voice_sessions.clone(),
        upstream,
        None,
        cwd,
    )
    .await?;
    serde_json::from_str(&json).map_err(|e| e.to_string())
}

#[derive(Deserialize)]
struct MeetingStopParams {
    meeting_id: String,
    #[serde(default = "default_summarize")]
    summarize: bool,
}

fn default_summarize() -> bool {
    true
}

async fn handle_meeting_stop(
    req: &JsonRpcMessage,
    state: &Arc<ServerState>,
    _session: &Arc<crate::session::ClientSession>,
) -> Result<Value, String> {
    let params: MeetingStopParams =
        serde_json::from_value(req.params.clone()).map_err(|e| e.to_string())?;
    let inference = if params.summarize {
        Some(state.inference.get().cloned()).flatten()
    } else {
        None
    };
    let json = crate::meeting::stop_meeting(
        &params.meeting_id,
        params.summarize,
        state.meetings.clone(),
        state.voice_sessions.clone(),
        inference,
    )
    .await?;
    serde_json::from_str(&json).map_err(|e| e.to_string())
}

#[derive(Deserialize, Default)]
struct MeetingListParams {
    #[serde(default)]
    root: Option<std::path::PathBuf>,
}

fn handle_meeting_list(req: &JsonRpcMessage) -> Result<Value, String> {
    let params: MeetingListParams = serde_json::from_value(req.params.clone()).unwrap_or_default();
    let cwd = std::env::current_dir().ok();
    let json = crate::meeting::list_meetings(params.root, cwd)?;
    serde_json::from_str(&json).map_err(|e| e.to_string())
}

#[derive(Deserialize)]
struct MeetingGetParams {
    meeting_id: String,
    #[serde(default)]
    root: Option<std::path::PathBuf>,
}

fn handle_meeting_get(req: &JsonRpcMessage) -> Result<Value, String> {
    let params: MeetingGetParams =
        serde_json::from_value(req.params.clone()).map_err(|e| e.to_string())?;
    let cwd = std::env::current_dir().ok();
    let json = crate::meeting::get_meeting(&params.meeting_id, params.root, cwd)?;
    serde_json::from_str(&json).map_err(|e| e.to_string())
}

// ---------------------------------------------------------------------------
// Agent registry — file-based cross-process discovery (#111)
// ---------------------------------------------------------------------------

#[derive(Deserialize, Default)]
struct RegistryRegisterParams {
    /// Caller serializes their AgentEntry as a JSON value; we
    /// re-serialize it so the ffi-common helper can validate the
    /// shape with the same parser used by the bindings.
    entry: Value,
    #[serde(default)]
    registry_path: Option<std::path::PathBuf>,
}

fn handle_registry_register(req: &JsonRpcMessage) -> Result<Value, String> {
    let params: RegistryRegisterParams =
        serde_json::from_value(req.params.clone()).map_err(|e| e.to_string())?;
    let entry_json = serde_json::to_string(&params.entry).map_err(|e| e.to_string())?;
    car_ffi_common::registry::register_agent(&entry_json, params.registry_path)?;
    Ok(Value::Null)
}

#[derive(Deserialize, Default)]
struct RegistryNameParams {
    name: String,
    #[serde(default)]
    registry_path: Option<std::path::PathBuf>,
}

fn handle_registry_heartbeat(req: &JsonRpcMessage) -> Result<Value, String> {
    let params: RegistryNameParams =
        serde_json::from_value(req.params.clone()).map_err(|e| e.to_string())?;
    let json = car_ffi_common::registry::agent_heartbeat(&params.name, params.registry_path)?;
    serde_json::from_str(&json).map_err(|e| e.to_string())
}

fn handle_registry_unregister(req: &JsonRpcMessage) -> Result<Value, String> {
    let params: RegistryNameParams =
        serde_json::from_value(req.params.clone()).map_err(|e| e.to_string())?;
    car_ffi_common::registry::unregister_agent(&params.name, params.registry_path)?;
    Ok(Value::Null)
}

#[derive(Deserialize, Default)]
struct RegistryListParams {
    #[serde(default)]
    registry_path: Option<std::path::PathBuf>,
}

fn handle_registry_list(req: &JsonRpcMessage) -> Result<Value, String> {
    let params: RegistryListParams = serde_json::from_value(req.params.clone()).unwrap_or_default();
    let json = car_ffi_common::registry::list_agents(params.registry_path)?;
    serde_json::from_str(&json).map_err(|e| e.to_string())
}

#[derive(Deserialize, Default)]
struct RegistryReapParams {
    /// Heartbeats older than this many seconds are reaped. Default
    /// 60 — two missed 20s heartbeats trigger removal.
    #[serde(default = "default_reap_age")]
    max_age_secs: u64,
    #[serde(default)]
    registry_path: Option<std::path::PathBuf>,
}

fn default_reap_age() -> u64 {
    60
}

fn handle_registry_reap(req: &JsonRpcMessage) -> Result<Value, String> {
    let params: RegistryReapParams = serde_json::from_value(req.params.clone()).unwrap_or_default();
    let json =
        car_ffi_common::registry::reap_stale_agents(params.max_age_secs, params.registry_path)?;
    serde_json::from_str(&json).map_err(|e| e.to_string())
}

// ---------------------------------------------------------------------------
// car-a2a server lifecycle (mirrors NAPI startA2aServer / stopA2aServer /
// a2aServerStatus and PyO3 start_a2a_server / stop_a2a_server /
// a2a_server_status — closes the binding gap noted in #126).
// ---------------------------------------------------------------------------

async fn handle_a2a_start(
    req: &JsonRpcMessage,
    session: &crate::session::ClientSession,
) -> Result<Value, String> {
    let params_json = serde_json::to_string(&req.params).map_err(|e| e.to_string())?;
    // Always hand the session's runtime through. start_a2a uses it
    // only when `share_session_runtime: true` is set in params;
    // otherwise it falls back to the legacy fresh-Runtime + agent_basics
    // path. Passing it unconditionally keeps the FFI layer ignorant of
    // the flag's plumbing.
    let json = crate::a2a::start_a2a(&params_json, Some(session.runtime.clone())).await?;
    serde_json::from_str(&json).map_err(|e| e.to_string())
}

fn handle_a2a_stop() -> Result<Value, String> {
    let json = crate::a2a::stop_a2a()?;
    serde_json::from_str(&json).map_err(|e| e.to_string())
}

fn handle_a2a_status() -> Result<Value, String> {
    let json = crate::a2a::a2a_status()?;
    serde_json::from_str(&json).map_err(|e| e.to_string())
}

#[derive(Deserialize)]
#[serde(rename_all = "camelCase")]
struct A2aSendParams {
    endpoint: String,
    message: car_a2a::Message,
    #[serde(default)]
    blocking: bool,
    #[serde(default = "default_true")]
    ingest_a2ui: bool,
    #[serde(default)]
    route_auth: Option<A2aRouteAuth>,
    #[serde(default)]
    allow_untrusted_endpoint: bool,
}

fn default_true() -> bool {
    true
}

/// In-core A2A dispatcher entry point. Forwards the JSON-RPC method
/// + params to the lazy-initialized [`car_a2a::A2aDispatcher`] held
/// on `ServerState`. Closes Parslee-ai/car-releases#28.
///
/// Streaming methods (`message/stream`, `tasks/resubscribe` and their
/// PascalCase aliases) return `MethodNotFound` from the dispatcher's
/// transport-neutral surface — the standalone `start_a2a_listener`
/// HTTP path serves SSE for those, but the in-core WS surface is
/// JSON-RPC only. Same trade as the dispatcher itself.
async fn handle_a2a_dispatch(
    method: &str,
    req: &JsonRpcMessage,
    state: &Arc<ServerState>,
) -> Result<Value, String> {
    let dispatcher = state.a2a_dispatcher().await;
    dispatcher
        .dispatch(method, req.params.clone())
        .await
        .map_err(|e| e.to_string())
}

async fn handle_a2a_send(req: &JsonRpcMessage, state: &Arc<ServerState>) -> Result<Value, String> {
    let params: A2aSendParams =
        serde_json::from_value(req.params.clone()).map_err(|e| e.to_string())?;
    let endpoint = trusted_route_endpoint(
        Some(params.endpoint.clone()),
        params.allow_untrusted_endpoint,
    )
    .ok_or_else(|| {
        "`a2a.send` endpoint must be loopback unless allowUntrustedEndpoint is true".to_string()
    })?;
    let client = match params.route_auth.clone() {
        Some(auth) => {
            car_a2a::A2aClient::new(endpoint.clone()).with_auth(client_auth_from_route_auth(auth))
        }
        None => car_a2a::A2aClient::new(endpoint.clone()),
    };
    let result = client
        .send_message(params.message, params.blocking)
        .await
        .map_err(|e| e.to_string())?;
    let result_value = serde_json::to_value(&result).map_err(|e| e.to_string())?;
    let mut applied = Vec::new();
    if params.ingest_a2ui {
        state
            .a2ui
            .validate_payload(&result_value)
            .map_err(|e| e.to_string())?;
        let routed_endpoint = Some(endpoint.clone());
        for envelope in car_a2ui::envelopes_from_value(&result_value).map_err(|e| e.to_string())? {
            let owner = car_a2ui::owner_from_value(&result_value).map(|owner| {
                if owner.endpoint.is_none() {
                    owner.with_endpoint(routed_endpoint.clone())
                } else {
                    owner
                }
            });
            applied.push(
                apply_a2ui_envelope(state, envelope, owner, params.route_auth.clone()).await?,
            );
        }
    }
    Ok(serde_json::json!({
        "result": result,
        "a2ui": {
            "applied": applied,
        }
    }))
}

// ---------------------------------------------------------------------------
// macOS automation — AppleScript + Shortcuts (car-automation), Vision OCR
// (car-vision). Mirrors NAPI runApplescript / listShortcuts / runShortcut /
// visionOcr and PyO3 run_applescript / list_shortcuts / run_shortcut /
// vision_ocr.
// ---------------------------------------------------------------------------

async fn handle_run_applescript(req: &JsonRpcMessage) -> Result<Value, String> {
    let args_json = serde_json::to_string(&req.params).map_err(|e| e.to_string())?;
    let json = car_ffi_common::automation::run_applescript(&args_json).await?;
    serde_json::from_str(&json).map_err(|e| e.to_string())
}

async fn handle_list_shortcuts(req: &JsonRpcMessage) -> Result<Value, String> {
    let args_json = serde_json::to_string(&req.params).map_err(|e| e.to_string())?;
    let json = car_ffi_common::automation::list_shortcuts(&args_json).await?;
    serde_json::from_str(&json).map_err(|e| e.to_string())
}

async fn handle_run_shortcut(req: &JsonRpcMessage) -> Result<Value, String> {
    let args_json = serde_json::to_string(&req.params).map_err(|e| e.to_string())?;
    let json = car_ffi_common::automation::run_shortcut(&args_json).await?;
    serde_json::from_str(&json).map_err(|e| e.to_string())
}

async fn handle_local_notification(req: &JsonRpcMessage) -> Result<Value, String> {
    let args_json = serde_json::to_string(&req.params).map_err(|e| e.to_string())?;
    let json = car_ffi_common::notifications::local(&args_json).await?;
    serde_json::from_str(&json).map_err(|e| e.to_string())
}

async fn handle_vision_ocr(req: &JsonRpcMessage) -> Result<Value, String> {
    let args_json = serde_json::to_string(&req.params).map_err(|e| e.to_string())?;
    let json = car_ffi_common::vision::ocr(&args_json).await?;
    serde_json::from_str(&json).map_err(|e| e.to_string())
}

// ---------------------------------------------------------------------------
// Lifecycle-managed agents (car_registry::supervisor) — Parslee-ai/car-releases#27
// ---------------------------------------------------------------------------

async fn handle_agents_list(state: &Arc<ServerState>) -> Result<Value, String> {
    // Observe-only mode (Parslee-ai/car-releases#44): a second
    // car-server on the host can't take the supervisor lock, so it
    // can't drive `Supervisor::list` — but it can still answer
    // `agents.list` by reading the on-disk manifest directly. The
    // `attached` decoration is local to whichever daemon the caller
    // is talking to, so observer-mode entries return `attached:
    // false` (this daemon hasn't received `session.auth` from those
    // children; the primary one has).
    let agents = match state.observer_manifest_path() {
        Some(p) => car_registry::supervisor::Supervisor::list_from_manifest(p)
            .map_err(|e| e.to_string())?,
        None => {
            let supervisor = state.supervisor()?;
            supervisor.list().await
        }
    };
    // Decorate each entry with `attached` + `session_id` so operators
    // see whether the supervised process has actually called
    // `session.auth { agent_id }` and bound a WS connection (#169) —
    // the lifecycle status (`Running`, etc.) only reports the
    // process-level view, which can't tell "alive but never
    // attached" from "alive and attached".
    let attached = state.attached_agents.lock().await.clone();
    let mut decorated: Vec<Value> = Vec::with_capacity(agents.len());
    for a in agents {
        let mut v = serde_json::to_value(&a).map_err(|e| e.to_string())?;
        let session_id = attached.get(&a.spec.id).cloned();
        if let Some(map) = v.as_object_mut() {
            map.insert("attached".to_string(), Value::Bool(session_id.is_some()));
            if let Some(sid) = session_id {
                map.insert("session_id".to_string(), Value::String(sid));
            }
        }
        decorated.push(v);
    }
    Ok(Value::Array(decorated))
}

async fn handle_agents_upsert(
    req: &JsonRpcMessage,
    state: &Arc<ServerState>,
) -> Result<Value, String> {
    let mut params = req.params.clone();
    // Optional `interpreter` sugar (#171). When present, the
    // supervisor resolves the bare program name (`"node"`,
    // `"python"`, …) against `$PATH` and writes the absolute path
    // into `command` *before* validation. This keeps the strict
    // no-PATH-lookup rule at upsert time while letting callers
    // stop hand-coding `/opt/homebrew/bin/node` into every
    // agents.json entry. Resolution happens once; subsequent PATH
    // changes do not silently rewire the binding.
    if let Some(name) = params
        .get("interpreter")
        .and_then(|v| v.as_str())
        .map(str::to_string)
    {
        let resolved =
            car_registry::supervisor::resolve_interpreter(&name).map_err(|e| e.to_string())?;
        params["command"] = Value::String(resolved.to_string_lossy().into_owned());
    }
    let spec: car_registry::supervisor::AgentSpec =
        serde_json::from_value(params).map_err(|e| e.to_string())?;
    let supervisor = state.supervisor()?;
    let agent = supervisor.upsert(spec).await.map_err(|e| e.to_string())?;
    serde_json::to_value(agent).map_err(|e| e.to_string())
}

/// `agents.install` — install a contributed-agent manifest
/// (Parslee-ai/car#182 phase 3). Caller passes the parsed
/// `AgentManifest` JSON; the daemon runs install-time validation
/// (`car_min_version`, capability negotiation against the daemon's
/// own advertisement) and adopts the manifest. Returns
/// `{ report, agent? }` where `agent` is the spawnable
/// `ManagedAgent` for `external_process` transports and absent for
/// `pure_data` / health_url-only manifests.
///
/// The host capability advertisement comes from
/// `HostCapabilities::daemon_default(car_version)` — operators that
/// want a tighter advertisement go through a future config phase;
/// this MVP uses the runtime's natural surface.
async fn handle_agents_install(
    req: &JsonRpcMessage,
    state: &Arc<ServerState>,
) -> Result<Value, String> {
    let manifest: car_registry::manifest::AgentManifest =
        serde_json::from_value(req.params.clone()).map_err(|e| e.to_string())?;
    let host = car_registry::install::HostCapabilities::daemon_default(env!("CARGO_PKG_VERSION"));
    let supervisor = state.supervisor()?;
    let (report, managed) = supervisor
        .install_manifest(manifest, &host)
        .await
        .map_err(|e| e.to_string())?;
    Ok(serde_json::json!({
        "report": {
            "missingOptional": report
                .missing_optional
                .iter()
                .map(|(ns, feat)| serde_json::json!({ "namespace": ns, "feature": feat }))
                .collect::<Vec<_>>(),
        },
        "agent": managed,
    }))
}

async fn handle_agents_health(state: &Arc<ServerState>) -> Result<Value, String> {
    // Observe-only mode (Parslee-ai/car-releases#44) — see
    // `handle_agents_list` for the rationale. The health view is a
    // pure function of each entry's `command` plus the on-disk
    // sandbox rules, so reading from the manifest is equivalent to
    // calling the live supervisor's `health()`.
    let entries = match state.observer_manifest_path() {
        Some(p) => car_registry::supervisor::Supervisor::health_from_manifest(p)
            .map_err(|e| e.to_string())?,
        None => {
            let supervisor = state.supervisor()?;
            supervisor.health().await
        }
    };
    serde_json::to_value(entries).map_err(|e| e.to_string())
}

fn extract_agent_id(req: &JsonRpcMessage) -> Result<String, String> {
    req.params
        .get("id")
        .and_then(Value::as_str)
        .map(str::to_string)
        .ok_or_else(|| "missing required `id` parameter".to_string())
}

async fn handle_agents_remove(
    req: &JsonRpcMessage,
    state: &Arc<ServerState>,
) -> Result<Value, String> {
    let id = extract_agent_id(req)?;
    let supervisor = state.supervisor()?;
    let removed = supervisor.remove(&id).await.map_err(|e| e.to_string())?;
    Ok(serde_json::json!({ "removed": removed }))
}

async fn handle_agents_start(
    req: &JsonRpcMessage,
    state: &Arc<ServerState>,
) -> Result<Value, String> {
    let id = extract_agent_id(req)?;
    let supervisor = state.supervisor()?;
    let agent = supervisor.start(&id).await.map_err(|e| e.to_string())?;
    serde_json::to_value(agent).map_err(|e| e.to_string())
}

async fn handle_agents_stop(
    req: &JsonRpcMessage,
    state: &Arc<ServerState>,
) -> Result<Value, String> {
    let id = extract_agent_id(req)?;
    let signal: car_registry::supervisor::StopSignal = req
        .params
        .get("signal")
        .map(|v| serde_json::from_value(v.clone()))
        .transpose()
        .map_err(|e| e.to_string())?
        .unwrap_or_default();
    let supervisor = state.supervisor()?;
    let agent = supervisor
        .stop(&id, signal)
        .await
        .map_err(|e| e.to_string())?;
    serde_json::to_value(agent).map_err(|e| e.to_string())
}

async fn handle_agents_restart(
    req: &JsonRpcMessage,
    state: &Arc<ServerState>,
) -> Result<Value, String> {
    let id = extract_agent_id(req)?;
    let supervisor = state.supervisor()?;
    let agent = supervisor.restart(&id).await.map_err(|e| e.to_string())?;
    serde_json::to_value(agent).map_err(|e| e.to_string())
}

async fn handle_agents_tail_log(
    req: &JsonRpcMessage,
    state: &Arc<ServerState>,
) -> Result<Value, String> {
    let id = extract_agent_id(req)?;
    let n = req.params.get("n").and_then(Value::as_u64).unwrap_or(100) as usize;
    let supervisor = state.supervisor()?;
    let lines = supervisor
        .tail_log(&id, n)
        .await
        .map_err(|e| e.to_string())?;
    Ok(serde_json::json!({ "lines": lines }))
}

// ---------------------------------------------------------------------------
// External-agent detection (Phase 1 of docs/proposals/external-agent-detection.md)
//
// Discovery surface for agentic CLIs the user has already installed and
// authenticated (Claude Code, Codex, Gemini). Read-only — no invocation
// path yet; agents.invoke_external lands in Phase 2 alongside the JSON
// stdio adapter. The cache lives in car_ffi_common::external_agents so
// the in-process FFI singletons share the same snapshot.
// ---------------------------------------------------------------------------

async fn handle_agents_list_external(req: &JsonRpcMessage) -> Result<Value, String> {
    let include_health = req
        .params
        .get("include_health")
        .and_then(Value::as_bool)
        .unwrap_or(false);
    let json = car_ffi_common::external_agents::list(include_health).await?;
    serde_json::from_str(&json).map_err(|e| e.to_string())
}

async fn handle_agents_detect_external(req: &JsonRpcMessage) -> Result<Value, String> {
    let include_health = req
        .params
        .get("include_health")
        .and_then(Value::as_bool)
        .unwrap_or(false);
    let json = car_ffi_common::external_agents::detect(include_health).await?;
    serde_json::from_str(&json).map_err(|e| e.to_string())
}

/// Per-task invocation of an external CLI agent. Required params:
/// `id` (adapter, e.g. `"claude-code"`) and `task` (the prompt).
/// Optional: `cwd`, `allowed_tools`, `max_turns`, `timeout_secs`.
///
/// Phase 2 stage 3 ships with `claude-code` only. Other adapter
/// ids return `is_error: true` with a structured `error` so hosts
/// can surface the gap without a separate error code.
///
/// Phase 2 stage 4a (governance): every invocation appends a
/// structured audit record to `~/.car/external-agents.jsonl`. The
/// record captures id, task, options, result, and the full
/// `tool_uses` list the assistant emitted — so even though the
/// agent executes its built-in tools in-process (which we can't
/// gate via stream-json), there's a complete after-the-fact audit
/// trail. Full policy gating (proposing each tool_use to CAR's
/// validator + getting a yes/no) requires the MCP server route in
/// stage 4b.
async fn handle_agents_invoke_external(
    req: &JsonRpcMessage,
    state: &Arc<ServerState>,
    host_session: &Arc<crate::session::ClientSession>,
) -> Result<Value, String> {
    let id = req
        .params
        .get("id")
        .and_then(Value::as_str)
        .ok_or_else(|| "missing required `id` parameter".to_string())?
        .to_string();
    let task = req
        .params
        .get("task")
        .and_then(Value::as_str)
        .ok_or_else(|| "missing required `task` parameter".to_string())?
        .to_string();
    let stream = req
        .params
        .get("stream")
        .and_then(Value::as_bool)
        .unwrap_or(false);
    let session_id = req
        .params
        .get("session_id")
        .and_then(Value::as_str)
        .map(str::to_string)
        .unwrap_or_else(|| format!("ext-{}", uuid::Uuid::new_v4().simple()));

    // Build the options sub-object directly from req.params so
    // hosts can pass `cwd` / `allowed_tools` / `max_turns` /
    // `timeout_secs` as siblings of `id`/`task`. Strip the
    // dispatch + streaming fields so they don't pollute the
    // options serde.
    let mut options_value = req.params.clone();
    if let Some(obj) = options_value.as_object_mut() {
        obj.remove("id");
        obj.remove("task");
        obj.remove("stream");
        obj.remove("session_id");
        // Auto-fill `mcp_endpoint` from the bound MCP URL when the
        // caller didn't supply one. This is the load-bearing
        // wiring of MCP-4: external agents get CAR's tools (memory,
        // skills, verify) routed through the daemon's policy +
        // shared memgine without any per-call host configuration.
        // Callers who want to opt out can pass `"mcp_endpoint": ""`
        // (empty string) — the runner skips the temp-file write
        // when the value isn't a non-empty URL.
        let has_explicit_mcp = obj.contains_key("mcp_endpoint");
        if !has_explicit_mcp {
            if let Some(url) = state.mcp_url.get() {
                obj.insert("mcp_endpoint".to_string(), Value::String(url.clone()));
            }
        }
    }

    if !stream {
        // Legacy one-shot path. Unchanged shape for FFI consumers
        // and any caller that hasn't opted into streaming.
        let options_json = options_value.to_string();
        let json = car_ffi_common::external_agents::invoke(&id, &task, &options_json).await?;
        let result: Value = serde_json::from_str(&json).map_err(|e| e.to_string())?;
        append_external_agent_audit(&id, &task, &options_value, &result);
        return Ok(result);
    }

    // Streaming path. Returns an ack ({accepted, session_id})
    // immediately and streams `agents.chat.event` notifications
    // to the host's WS as the runner emits StreamEvents. Reuses
    // the chat_sessions routing infrastructure supervised agents
    // use — host UIs render both kinds through the same path.
    let opts: car_external_agents::InvokeOptions = serde_json::from_value(options_value.clone())
        .map_err(|e| format!("invalid options: {e}"))?;

    // Register the chat session BEFORE spawning so the host's
    // subscriber is correctly bound to this session_id by the
    // time the first event arrives. Reusing the supervised
    // agent chat infrastructure means `agents.chat.cancel`
    // routes to chat_sessions[session_id] and can find this
    // entry — though we don't currently honor cancel for
    // external invocations (the child process is killed only
    // on timeout or task drop; a future iteration can plumb a
    // CancellationToken through invoke_with_emitter).
    {
        // If a chat session is already registered for this id (the
        // typical proxy shape: host → agents.chat → supervised agent
        // → agents.invoke_external with the same session_id), DO NOT
        // overwrite it. The existing entry owns the routing to the
        // original host; clobbering it with our caller's client_id
        // would send streaming events to the proxying agent instead
        // of back to the host that issued agents.chat. Only register
        // a fresh entry when the slot is empty (a direct
        // host-to-invoke_external call without a prior agents.chat).
        let mut chats = state.chat_sessions.lock().await;
        chats.entry(session_id.clone()).or_insert_with(|| {
            let created_at = std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .map(|d| d.as_secs())
                .unwrap_or(0);
            crate::session::ChatSession {
                agent_id: id.clone(),
                host_client_id: host_session.client_id.clone(),
                created_at,
            }
        });
    }

    // Single drain task pulls StreamEvents off an unbounded
    // channel and serializes WS sends to the host. Per-event
    // tokio::spawn would let sends race (which token arrives
    // first depends on lock acquisition order). The channel
    // is unbounded because claude's event volume is bounded
    // by user turn count — typically <50 events per invocation.
    use tokio::sync::mpsc;
    let (tx, mut rx) = mpsc::unbounded_channel::<car_external_agents::StreamEvent>();

    let drain_state = state.clone();
    let drain_session_id = session_id.clone();
    let drain_agent_id = id.clone();
    tokio::spawn(async move {
        while let Some(event) = rx.recv().await {
            emit_external_chat_event(&drain_state, &drain_session_id, &drain_agent_id, event).await;
        }
    });

    let emitter_tx = tx.clone();
    let emitter: car_external_agents::StreamEventEmitter = Arc::new(move |event| {
        // Send failure (rx dropped) means the drain task has
        // exited — usually because the host disconnected. The
        // runner will keep going; let it finish so the audit
        // log captures the full result.
        let _ = emitter_tx.send(event);
    });

    // Run the invocation in a separate task so this handler
    // can return the ack right away. The runner's child-process
    // future owns the spawn lifetime; if the host disconnects
    // mid-stream, the runner still completes (its events fall
    // on the floor at the drain layer) so the audit log lands.
    let spawn_state = state.clone();
    let spawn_session_id = session_id.clone();
    let spawn_id = id.clone();
    let spawn_task = task.clone();
    let spawn_options = options_value.clone();
    tokio::spawn(async move {
        let outcome =
            car_external_agents::invoke_with_emitter(&spawn_id, &spawn_task, opts, Some(emitter))
                .await;
        drop(tx); // signal drain task to exit after queue empties

        // Synthesize a terminal agents.chat.event so the host's
        // bubble finalizes. The runner doesn't emit a "done" event
        // itself — the result is the aggregate InvokeResult. We
        // translate it here.
        let terminal_params: Value;
        let result_value: Value;
        match outcome {
            Ok(res) => {
                // Pack the metadata into `finish_reason` as a
                // human-readable summary so the host's existing
                // ChatEvent decoder surfaces it without a schema
                // change. Hosts that want structured data can
                // re-issue `agents.invoke_external` with
                // `stream: false` and read the InvokeResult.
                let mut parts: Vec<String> = Vec::new();
                if res.turns > 0 {
                    parts.push(format!(
                        "{} turn{}",
                        res.turns,
                        if res.turns == 1 { "" } else { "s" }
                    ));
                }
                if res.tool_calls > 0 {
                    parts.push(format!(
                        "{} tool{}",
                        res.tool_calls,
                        if res.tool_calls == 1 { "" } else { "s" }
                    ));
                }
                if res.duration_ms > 0 {
                    parts.push(format!("{:.1}s", res.duration_ms as f64 / 1000.0));
                }
                let summary = if parts.is_empty() {
                    "stop".to_string()
                } else {
                    parts.join(" · ")
                };
                if res.is_error {
                    terminal_params = serde_json::json!({
                        "session_id": spawn_session_id,
                        "agent_id": spawn_id,
                        "kind": "error",
                        "error": res.error.clone().unwrap_or_else(|| "external agent reported error".to_string()),
                    });
                } else {
                    terminal_params = serde_json::json!({
                        "session_id": spawn_session_id,
                        "agent_id": spawn_id,
                        "kind": "done",
                        "finish_reason": summary,
                    });
                }
                result_value = serde_json::to_value(&res).unwrap_or(Value::Null);
            }
            Err(e) => {
                let message = format!("{e}");
                terminal_params = serde_json::json!({
                    "session_id": spawn_session_id,
                    "agent_id": spawn_id,
                    "kind": "error",
                    "error": message.clone(),
                });
                result_value = serde_json::json!({ "is_error": true, "error": message });
            }
        }
        send_external_chat_frame(&spawn_state, &spawn_session_id, terminal_params).await;
        spawn_state
            .chat_sessions
            .lock()
            .await
            .remove(&spawn_session_id);
        append_external_agent_audit(&spawn_id, &spawn_task, &spawn_options, &result_value);
    });

    Ok(serde_json::json!({
        "accepted": true,
        "session_id": session_id,
    }))
}

/// Translate one [`StreamEvent`] from the running external CLI
/// into an `agents.chat.event` notification on the originating
/// host's WS. Same wire shape supervised agents emit, so host
/// UIs render both kinds with one decoder.
///
/// Mapping:
/// - `Assistant` events with `text` content blocks → `kind: "token"`
///   per text block. Each block carries the full text the
///   assistant emitted in that turn (claude doesn't expose
///   word-level deltas via stream-json — it emits per-turn or
///   per-content-block chunks).
/// - `Assistant` events with `tool_use` blocks → `kind: "tool_call"`
///   per block (tool name in `detail`).
/// - `System` / `User` / `Result` / others → dropped (Result's
///   metadata is folded into the terminal `done` event the
///   outer task emits when the invocation finishes).
async fn emit_external_chat_event(
    state: &Arc<ServerState>,
    session_id: &str,
    agent_id: &str,
    event: car_external_agents::StreamEvent,
) {
    use car_external_agents::StreamEvent;
    match event {
        StreamEvent::Assistant(a) => {
            if let Some(content) = a.message.get("content").and_then(|v| v.as_array()) {
                for block in content {
                    let block_type = block.get("type").and_then(|v| v.as_str()).unwrap_or("");
                    match block_type {
                        "text" => {
                            if let Some(text) = block.get("text").and_then(|v| v.as_str()) {
                                if !text.is_empty() {
                                    let params = serde_json::json!({
                                        "session_id": session_id,
                                        "agent_id": agent_id,
                                        "kind": "token",
                                        "delta": text,
                                    });
                                    send_external_chat_frame(state, session_id, params).await;
                                }
                            }
                        }
                        "tool_use" => {
                            let name = block
                                .get("name")
                                .and_then(|v| v.as_str())
                                .unwrap_or("(unknown tool)");
                            let params = serde_json::json!({
                                "session_id": session_id,
                                "agent_id": agent_id,
                                "kind": "tool_call",
                                "detail": name,
                            });
                            send_external_chat_frame(state, session_id, params).await;
                        }
                        _ => {}
                    }
                }
            }
        }
        _ => {
            // System (session id init), User (tool result echo),
            // Result (final aggregate — folded into terminal
            // `done` event by the outer task), RateLimitEvent,
            // Other: not surfaced to host.
        }
    }
}

/// Send a single `agents.chat.event` notification to the host
/// session bound by `session_id`. Best-effort: a missing route
/// or a closed WS is silently dropped, the runner continues so
/// the audit log lands.
async fn send_external_chat_frame(state: &Arc<ServerState>, session_id: &str, params: Value) {
    use futures::SinkExt;
    use tokio_tungstenite::tungstenite::Message;

    let host_client_id = state
        .chat_sessions
        .lock()
        .await
        .get(session_id)
        .map(|s| s.host_client_id.clone());
    let Some(host_client_id) = host_client_id else {
        return;
    };
    let host_channel = {
        let sessions = state.sessions.lock().await;
        sessions.get(&host_client_id).map(|s| s.channel.clone())
    };
    let Some(channel) = host_channel else {
        return;
    };
    let frame = serde_json::json!({
        "jsonrpc": "2.0",
        "method": "agents.chat.event",
        "params": params,
    });
    if let Ok(text) = serde_json::to_string(&frame) {
        let _ = channel
            .write
            .lock()
            .await
            .send(Message::Text(text.into()))
            .await;
    }
}

/// Append one JSONL audit record to `~/.car/external-agents.jsonl`.
/// Best-effort: a failure to open the journal must NOT fail the
/// invocation; the in-memory result already returned is the
/// authoritative answer. Logs at warn level when the write fails so
/// operators notice repeated failures.
fn append_external_agent_audit(id: &str, task: &str, options: &Value, result: &Value) {
    use std::io::Write;
    let car_dir = match std::env::var_os("HOME").map(std::path::PathBuf::from) {
        Some(home) => home.join(".car"),
        None => return,
    };
    if std::fs::create_dir_all(&car_dir).is_err() {
        return;
    }
    let path = car_dir.join("external-agents.jsonl");
    let record = serde_json::json!({
        "ts": chrono::Utc::now().to_rfc3339(),
        "adapter_id": id,
        "task": task,
        "options": options,
        "result": result,
    });
    let line = match serde_json::to_string(&record) {
        Ok(s) => s,
        Err(_) => return,
    };
    if let Ok(mut f) = std::fs::OpenOptions::new()
        .create(true)
        .append(true)
        .open(&path)
    {
        let _ = writeln!(f, "{}", line);
    } else {
        tracing::warn!(
            path = %path.display(),
            "failed to append external-agent audit record"
        );
    }
}

/// Ground-truth health check. Optional `id` param picks one tool;
/// without it, every detected adapter is checked. `force: true`
/// bypasses the 30s per-tool TTL cache. Replaces the Phase 1
/// credential-file shape heuristic as the load-bearing signal for
/// "is this tool ready to invoke."
async fn handle_agents_health_external(req: &JsonRpcMessage) -> Result<Value, String> {
    let force = req
        .params
        .get("force")
        .and_then(Value::as_bool)
        .unwrap_or(false);
    if let Some(id) = req.params.get("id").and_then(Value::as_str) {
        let json = car_ffi_common::external_agents::health_one(id, force).await?;
        serde_json::from_str(&json).map_err(|e| e.to_string())
    } else {
        let json = car_ffi_common::external_agents::health(force).await?;
        serde_json::from_str(&json).map_err(|e| e.to_string())
    }
}

// ---------------------------------------------------------------------------
// agents.chat — unified chat surface (docs/proposals/agent-chat-surface.md)
// ---------------------------------------------------------------------------
//
// Host calls `agents.chat { agent_id, prompt, session_id?, stream? }`.
// The server looks up the target agent's attached WS connection,
// reverse-calls `agent.chat { session_id, prompt, context }` on it
// (same pattern as `tools.execute`), and returns once the agent acks.
// The agent then streams `agent.chat.event` notifications back, which
// the dispatcher intercepts (see `try_forward_agent_chat_event`) and
// rewrites as `agents.chat.event` notifications on the originating
// host's channel.

/// Timeout the server waits for the agent to ack `agent.chat`. The
/// streamed tokens come later as separate notifications and have no
/// bearing on this — this is just "did the agent receive the prompt
/// and accept it." Five seconds is generous for a local IPC ack.
const AGENT_CHAT_ACK_TIMEOUT_SECS: u64 = 5;

/// `agents.chat` — host issues a chat turn to a named agent. Returns
/// `{ accepted: true, session_id }` once the agent acks; streamed
/// tokens arrive on the host's channel as `agents.chat.event`
/// notifications keyed by the same `session_id`.
async fn handle_agents_chat(
    req: &JsonRpcMessage,
    state: &Arc<ServerState>,
    host_session: &Arc<crate::session::ClientSession>,
) -> Result<Value, String> {
    use futures::SinkExt;
    use tokio::sync::oneshot;
    use tokio_tungstenite::tungstenite::Message;

    let agent_id = req
        .params
        .get("agent_id")
        .and_then(Value::as_str)
        .ok_or_else(|| "`agents.chat` requires `agent_id`".to_string())?
        .to_string();
    let prompt = req
        .params
        .get("prompt")
        .and_then(Value::as_str)
        .ok_or_else(|| "`agents.chat` requires `prompt`".to_string())?
        .to_string();
    let session_id = req
        .params
        .get("session_id")
        .and_then(Value::as_str)
        .map(str::to_string)
        .unwrap_or_else(|| format!("chat-{}", uuid::Uuid::new_v4().simple()));
    let stream = req
        .params
        .get("stream")
        .and_then(Value::as_bool)
        .unwrap_or(true);
    let voice_input = req
        .params
        .get("voice_input")
        .and_then(Value::as_bool)
        .unwrap_or(false);

    // Resolve the agent's attached WS channel via `attached_agents` →
    // `sessions` → `channel`. Both lookups must hit; a missing entry on
    // either side means the agent is registered in `agents.json` but
    // hasn't `session.auth`'d (or has disconnected), so refuse with a
    // structured error rather than silently parking the chat.
    let agent_client_id = state
        .attached_agents
        .lock()
        .await
        .get(&agent_id)
        .cloned()
        .ok_or_else(|| {
            format!(
                "agent `{}` is not attached to this daemon — supervisor may have it stopped, or it hasn't called session.auth yet",
                agent_id
            )
        })?;
    let agent_channel = {
        let sessions = state.sessions.lock().await;
        sessions
            .get(&agent_client_id)
            .map(|s| s.channel.clone())
            .ok_or_else(|| {
                format!(
                    "agent `{}` client_id `{}` not found in session registry (raced with disconnect)",
                    agent_id, agent_client_id
                )
            })?
    };

    // Register the chat session BEFORE sending the reverse call so any
    // `agent.chat.event` notifications the agent sends as part of
    // accepting the chat (e.g. an immediate `token` delta) route
    // correctly. Indexed by session_id so the notification interceptor
    // can locate the originating host without scanning.
    {
        let created_at = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .map(|d| d.as_secs())
            .unwrap_or(0);
        state.chat_sessions.lock().await.insert(
            session_id.clone(),
            crate::session::ChatSession {
                agent_id: agent_id.clone(),
                host_client_id: host_session.client_id.clone(),
                created_at,
            },
        );
    }

    // Reverse-callback: register a oneshot for the ack, send the
    // `agent.chat` JSON-RPC request on the agent's channel, await up
    // to AGENT_CHAT_ACK_TIMEOUT_SECS. Uses the same `pending` map the
    // tool-callback path uses (`WsToolExecutor`) — the dispatcher's
    // response demuxer at the top of `run_dispatch` already routes
    // `result` / `error` frames keyed by request id back through it.
    let request_id = agent_channel.next_request_id();
    let (tx, rx) = oneshot::channel();
    agent_channel
        .pending
        .lock()
        .await
        .insert(request_id.clone(), tx);

    let rpc_request = serde_json::json!({
        "jsonrpc": "2.0",
        "method": "agent.chat",
        "params": {
            "session_id": session_id,
            "prompt": prompt,
            "stream": stream,
            "context": {
                "host_client_id": host_session.client_id,
                "voice_input": voice_input,
            },
        },
        "id": request_id,
    });
    let msg = Message::Text(
        serde_json::to_string(&rpc_request)
            .map_err(|e| e.to_string())?
            .into(),
    );
    if let Err(e) = agent_channel.write.lock().await.send(msg).await {
        // Send failed — drop the pending waiter and the chat session
        // entry so a retry can take a fresh session_id without
        // colliding.
        agent_channel.pending.lock().await.remove(&request_id);
        state.chat_sessions.lock().await.remove(&session_id);
        return Err(format!(
            "failed to deliver agent.chat to `{}`: {}",
            agent_id, e
        ));
    }

    // Await the agent's ack. The dispatcher's response demuxer routes
    // the result/error back via the oneshot. Timeout means the agent
    // is alive but unresponsive — clean up routing state and surface a
    // structured error so the host UI doesn't hang.
    let ack = match tokio::time::timeout(
        std::time::Duration::from_secs(AGENT_CHAT_ACK_TIMEOUT_SECS),
        rx,
    )
    .await
    {
        Ok(Ok(resp)) => resp,
        Ok(Err(_)) => {
            // Channel closed — agent disconnected mid-call.
            state.chat_sessions.lock().await.remove(&session_id);
            return Err(format!(
                "agent `{}` disconnected before acking agents.chat",
                agent_id
            ));
        }
        Err(_) => {
            // Timeout — agent didn't respond in time. Don't keep the
            // chat session around: any later events from the agent
            // would route to a host that already returned an error.
            agent_channel.pending.lock().await.remove(&request_id);
            state.chat_sessions.lock().await.remove(&session_id);
            return Err(format!(
                "agent `{}` did not ack agents.chat within {}s",
                agent_id, AGENT_CHAT_ACK_TIMEOUT_SECS
            ));
        }
    };

    if let Some(err) = ack.error {
        // Agent explicitly rejected — drop the session and propagate.
        state.chat_sessions.lock().await.remove(&session_id);
        return Err(format!("agent `{}` rejected chat: {}", agent_id, err));
    }

    Ok(serde_json::json!({
        "accepted": true,
        "session_id": session_id,
    }))
}

/// `agents.chat.cancel` — host aborts an in-flight chat. Forwards
/// `agent.chat.cancel` to the bound agent so the agent can short-
/// circuit its inference stream + free upstream resources
/// (`inference.stream.cancel`). The chat session is dropped from
/// routing state immediately whether or not the agent acks the cancel
/// — further `agent.chat.event` notifications for this session_id
/// fall on the floor by design.
async fn handle_agents_chat_cancel(
    req: &JsonRpcMessage,
    state: &Arc<ServerState>,
) -> Result<Value, String> {
    use futures::SinkExt;
    use tokio_tungstenite::tungstenite::Message;

    let session_id = req
        .params
        .get("session_id")
        .and_then(Value::as_str)
        .ok_or_else(|| "`agents.chat.cancel` requires `session_id`".to_string())?
        .to_string();

    let chat = state.chat_sessions.lock().await.remove(&session_id);
    let chat = match chat {
        Some(c) => c,
        None => {
            // Already cancelled or never existed — idempotent.
            return Ok(serde_json::json!({ "cancelled": false, "reason": "unknown session_id" }));
        }
    };

    // Best-effort fire-and-forget to the agent. We've already removed
    // the routing entry, so no need to await any agent response.
    let agent_client_id = state
        .attached_agents
        .lock()
        .await
        .get(&chat.agent_id)
        .cloned();
    if let Some(client_id) = agent_client_id {
        let channel_opt = {
            let sessions = state.sessions.lock().await;
            sessions.get(&client_id).map(|s| s.channel.clone())
        };
        if let Some(channel) = channel_opt {
            let notification = serde_json::json!({
                "jsonrpc": "2.0",
                "method": "agent.chat.cancel",
                "params": { "session_id": session_id },
            });
            if let Ok(text) = serde_json::to_string(&notification) {
                let _ = channel
                    .write
                    .lock()
                    .await
                    .send(Message::Text(text.into()))
                    .await;
            }
        }
    }

    Ok(serde_json::json!({ "cancelled": true, "session_id": session_id }))
}

/// Forward an `agent.chat.event` notification from an agent's
/// connection to the originating host's connection, rewritten as an
/// `agents.chat.event` notification. Returns `true` if the inbound
/// frame was a chat-event we routed (so the dispatcher can `continue`
/// past the normal method dispatch and skip the wasted "unknown
/// method" response), `false` otherwise.
///
/// Terminal events (`kind: "done"` / `"error"`) also drop the routing
/// entry from `state.chat_sessions` so a later stray notification can
/// be rejected as orphaned without leaking memory.
pub(crate) async fn try_forward_agent_chat_event(
    parsed: &JsonRpcMessage,
    state: &Arc<ServerState>,
) -> bool {
    use futures::SinkExt;
    use tokio_tungstenite::tungstenite::Message;

    // Notification predicate: method is `agent.chat.event`, id is
    // missing/null (per JSON-RPC, notifications have no id), and
    // params carry a session_id.
    let Some(method) = parsed.method.as_deref() else {
        return false;
    };
    if method != "agent.chat.event" {
        return false;
    }
    if !parsed.id.is_null() {
        // Has an id → it's a request, not a notification. Let the
        // normal dispatcher handle it (and reply with method-not-found).
        return false;
    }
    let Some(session_id) = parsed.params.get("session_id").and_then(Value::as_str) else {
        return false;
    };
    let session_id = session_id.to_string();

    // Look up the routing entry. If gone (cancelled, agent dropped,
    // disconnect cleanup), drop the event silently — late frames from
    // a respawned agent for a stale session are not the host's
    // problem.
    let chat = state.chat_sessions.lock().await.get(&session_id).cloned();
    let Some(chat) = chat else {
        return true; // recognized the method, but routing has dropped — consumed.
    };

    // Pull the kind early so terminal-event cleanup runs even if the
    // host's send fails. Agents may omit `kind` and signal terminal
    // state via `finish_reason` / `error` instead (car#222) — derive
    // it from the frame shape so both the cleanup below AND the host
    // see a correct, host-protocol-compliant kind. The old code
    // defaulted a `finish_reason`-only "done" frame to "token", so
    // terminal cleanup never ran and the host (which requires `kind`)
    // dropped every frame silently.
    let kind = parsed
        .params
        .get("kind")
        .and_then(Value::as_str)
        .map(str::to_string)
        .unwrap_or_else(|| {
            if parsed.params.get("error").is_some() {
                "error".to_string()
            } else if parsed.params.get("finish_reason").is_some() {
                "done".to_string()
            } else {
                "token".to_string()
            }
        });

    // Forward to the host. Rewrites the method name to the host-facing
    // form and attaches `agent_id` so the host doesn't have to remember
    // which agent owns each session.
    let host_channel = {
        let sessions = state.sessions.lock().await;
        sessions
            .get(&chat.host_client_id)
            .map(|s| s.channel.clone())
    };
    if let Some(channel) = host_channel {
        let mut params = parsed.params.clone();
        if let Some(obj) = params.as_object_mut() {
            obj.insert("agent_id".to_string(), Value::String(chat.agent_id.clone()));
            // host-protocol.md requires a top-level `kind` on every
            // agents.chat.event. Agents that omit it (signalling via
            // finish_reason/error) were dropped wholesale by the host
            // decoder — normalize here so the contract holds. car#222.
            obj.entry("kind")
                .or_insert_with(|| Value::String(kind.clone()));
        }
        let forward = serde_json::json!({
            "jsonrpc": "2.0",
            "method": "agents.chat.event",
            "params": params,
        });
        if let Ok(text) = serde_json::to_string(&forward) {
            let send_result = channel
                .write
                .lock()
                .await
                .send(Message::Text(text.into()))
                .await;
            if let Err(e) = send_result {
                tracing::warn!(
                    session_id = %session_id,
                    agent_id = %chat.agent_id,
                    host_client_id = %chat.host_client_id,
                    kind = %kind,
                    error = %e,
                    "agent.chat.event forward to host failed at the WS send step"
                );
            }
        }
    } else {
        // Host disconnected mid-stream — chat_sessions still holds
        // the routing entry but the originating client_id no longer
        // resolves to a session. Pre-#233 this was silent and the
        // operator had no way to tell whether the event was dropped
        // here or never emitted by the agent. Log + drop the
        // routing entry so subsequent stray events are no-ops.
        tracing::warn!(
            session_id = %session_id,
            agent_id = %chat.agent_id,
            host_client_id = %chat.host_client_id,
            kind = %kind,
            "agent.chat.event from supervised agent had no host channel \
             (host disconnected since `agents.chat`); dropping routing entry"
        );
        state.chat_sessions.lock().await.remove(&session_id);
        return true;
    }

    // Terminal-kind cleanup. The host_channel branch above already
    // forwarded the terminal event; we just remove the routing entry
    // here so subsequent stray frames are no-ops.
    if matches!(kind.as_str(), "done" | "error") {
        state.chat_sessions.lock().await.remove(&session_id);
    }

    true
}

#[cfg(test)]
mod fd_leak_regression {
    //! car#209 regression: an abrupt transport error must still run
    //! the connection cleanup. Before the fix, `let msg = msg?;`
    //! propagated the read error out of `run_dispatch`, skipping
    //! `remove_session`, so `state.sessions` (holding the
    //! `Arc<ClientSession>` -> `Arc<WsChannel>` -> socket FD) leaked
    //! forever on every peer reset / crash-loop disconnect.
    use super::run_dispatch;
    use futures::SinkExt;
    use std::sync::Arc;
    use tokio_tungstenite::tungstenite::{Error as WsError, Message};

    #[tokio::test]
    async fn abrupt_read_error_still_runs_session_cleanup() {
        let tmp = tempfile::TempDir::new().unwrap();
        let state = Arc::new(crate::session::ServerState::standalone(
            tmp.path().to_path_buf(),
        ));

        // Read stream that immediately yields a transport error (peer
        // reset), then ends -- the exact shape of an ungraceful
        // client disconnect.
        let read = futures::stream::iter(vec![Err::<Message, WsError>(
            WsError::ConnectionClosed,
        )]);
        let write: crate::session::WsSink = Box::pin(
            futures::sink::drain().sink_map_err(|_| WsError::ConnectionClosed),
        );

        let result =
            run_dispatch(read, write, "test-peer".to_string(), state.clone()).await;
        assert!(
            result.is_ok(),
            "run_dispatch must return Ok after cleanup, got {result:?}"
        );

        // The session (and its channel/FD) must be gone -- cleanup
        // ran despite the abrupt error.
        assert!(
            state.sessions.lock().await.is_empty(),
            "state.sessions must be empty after an abrupt disconnect (car#209)"
        );
    }
}

#[cfg(test)]
mod a2ui_action_delivery {
    //! Parslee-ai/car-releases#58: a ClientAction must reach A2UI
    //! subscribers on the `a2ui.event` channel — the same one that
    //! already carries surface updates — so an agent that created a
    //! surface receives the button click it would otherwise never see.
    use super::{handle_a2ui_action, JsonRpcMessage};
    use crate::session::{ServerState, WsChannel, WsSink};
    use futures::{SinkExt, StreamExt};
    use std::collections::HashMap;
    use std::sync::atomic::AtomicU64;
    use std::sync::Arc;
    use tokio::sync::Mutex;
    use tokio_tungstenite::tungstenite::{Error as WsError, Message};

    #[tokio::test]
    async fn client_action_broadcasts_to_a2ui_subscribers() {
        let tmp = tempfile::TempDir::new().unwrap();
        let state = Arc::new(ServerState::standalone(tmp.path().to_path_buf()));

        // Capturing subscriber channel: a futures mpsc whose sink half is
        // erased into the WsSink the server writes to, receiver kept here.
        let (tx, mut rx) = futures::channel::mpsc::unbounded::<Message>();
        let sink: WsSink = Box::pin(tx.sink_map_err(|_| WsError::ConnectionClosed));
        let channel = Arc::new(WsChannel {
            write: Mutex::new(sink),
            pending: Mutex::new(HashMap::new()),
            next_id: AtomicU64::new(0),
        });
        state
            .a2ui_subscribers
            .lock()
            .await
            .insert("test-sub".to_string(), channel);

        // Send the action under the `action` key (not `name`) to also
        // exercise the serde alias — the web renderer forwards
        // `@a2ui/web_core`'s object verbatim.
        let req: JsonRpcMessage = serde_json::from_value(serde_json::json!({
            "jsonrpc": "2.0",
            "method": "a2ui.action",
            "id": 1,
            "params": {
                "action": "trader:pause",
                "surfaceId": "surf-1",
                "sourceComponentId": "b1",
                "timestamp": "2026-06-03T00:00:00Z"
            }
        }))
        .unwrap();

        let out = handle_a2ui_action(&req, &state).await;
        assert!(out.is_ok(), "handle_a2ui_action failed: {out:?}");

        let msg = rx.next().await.expect("subscriber received no frame");
        let text = match msg {
            Message::Text(t) => t.to_string(),
            other => panic!("expected text frame, got {other:?}"),
        };
        let v: serde_json::Value = serde_json::from_str(&text).unwrap();
        assert_eq!(v["method"], "a2ui.event");
        assert_eq!(v["params"]["kind"], "a2ui.action");
        // Alias resolved `action` -> `name`.
        assert_eq!(
            v["params"]["result"]["action"]["name"], "trader:pause",
            "ClientAction.name should accept the `action` alias"
        );
        assert_eq!(v["params"]["result"]["surfaceId"], "surf-1");
    }
}