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;

    while let Some(msg) = read.next().await {
        let msg = msg?;
        if msg.is_text() {
            let text = msg.to_text()?;
            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?;
                    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;
                    }
                }
            }

            // 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) or $XDG_RUNTIME_DIR/ai.parslee.car/auth-token (Linux) \
                         as the first frame on this connection",
                    );
                    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);
                            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;
                tokio::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,
                        "session.init" => handle_session_init(&parsed, &session).await,
                        "host.subscribe" => handle_host_subscribe(&session).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).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
                        }
                        "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,
                        "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,
                        "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.upgrades" => handle_models_upgrades(&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),
                        "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.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.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).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_close() {
            info!("Client {} disconnected", client_id);
            break;
        }
    }

    session.host.unsubscribe(&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 (60s). 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) -> 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,
    })
    .map_err(|e| e.to_string())
}

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;
    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;
    }
    serde_json::to_value(
        session
            .host
            .create_approval(Some(&session.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> {
    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 }))
}

/// 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()],
    };
    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,
) -> 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,
    };
    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())?;
    let tools: std::collections::HashSet<String> =
        session.runtime.tools.read().await.keys().cloned().collect();
    let result = car_verify::verify(&vr.proposal, Some(&vr.initial_state), Some(&tools), 30);
    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)
}

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 = car_multi::SharedInfra::new();

    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 = car_multi::SharedInfra::new();

    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 = car_multi::SharedInfra::new();

    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 = car_multi::SharedInfra::new();

    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 = car_multi::SharedInfra::new();

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

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

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

async fn handle_models_pull(msg: &JsonRpcMessage, state: &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")?;
    let engine = get_inference_engine(state);
    let path = engine.pull_model(name).await.map_err(|e| e.to_string())?;
    Ok(serde_json::json!({"path": path.display().to_string()}))
}

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

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

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

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?;
    serde_json::from_str(&json).map_err(|e| e.to_string())
}

#[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> {
    let supervisor = state.supervisor()?;
    let agents = 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> {
    let supervisor = state.supervisor()?;
    let entries = 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>,
) -> Result<Value, String> {
    let id = req
        .params
        .get("id")
        .and_then(Value::as_str)
        .ok_or_else(|| "missing required `id` parameter".to_string())?;
    let task = req
        .params
        .get("task")
        .and_then(Value::as_str)
        .ok_or_else(|| "missing required `task` parameter".to_string())?;
    // 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 two
    // dispatch 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");
        // 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()));
            }
        }
    }
    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);
    Ok(result)
}

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