car-mcp 0.49.0

//! Transport-agnostic JSON-RPC dispatch.
//!
//! [`Server::handle`] takes a parsed [`Request`] and returns
//! `Option<Response>` (None for notifications). Pure function: no
//! I/O, no transport coupling. The stdio binary and the daemon's
//! HTTP-streamable endpoint both call this same entry point.

use std::collections::HashMap;
use std::path::PathBuf;
use std::sync::Arc;

use car_ir::ActionProposal;
use car_memgine::note_store::Note;
use car_memgine::MemgineEngine;
use serde::{Deserialize, Serialize};
use serde_json::{json, Value};
use tokio::sync::Mutex;

use crate::error_codes::{
    INTERNAL as E_INTERNAL, INVALID_PARAMS as E_INVALID_PARAMS,
    INVALID_REQUEST as E_INVALID_REQUEST, METHOD_NOT_FOUND as E_METHOD_NOT_FOUND,
};
use crate::schemas::{cached_prompt_schemas, cached_tool_schemas};
use crate::{PROTOCOL_VERSION, SERVER_NAME, SUPPORTED_VERSIONS};

/// JSON-RPC 2.0 request as it arrives on the wire.
#[derive(Debug, Deserialize)]
pub struct Request {
    pub jsonrpc: String,
    #[serde(default)]
    pub id: Option<Value>,
    pub method: String,
    #[serde(default)]
    pub params: Value,
}

/// JSON-RPC 2.0 response.
#[derive(Debug, Serialize)]
pub struct Response {
    pub jsonrpc: &'static str,
    pub id: Value,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub result: Option<Value>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub error: Option<ErrorObj>,
}

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

pub fn ok(id: Value, result: Value) -> Response {
    Response {
        jsonrpc: "2.0",
        id,
        result: Some(result),
        error: None,
    }
}

pub fn err(id: Value, code: i32, message: impl Into<String>) -> Response {
    Response {
        jsonrpc: "2.0",
        id,
        result: None,
        error: Some(ErrorObj {
            code,
            message: message.into(),
        }),
    }
}

/// Typed tool-dispatch errors. The variant determines the JSON-RPC
/// error code returned to the client.
#[derive(Debug)]
pub enum ToolError {
    InvalidParams(String),
    Internal(String),
    UnknownTool(String),
}

impl ToolError {
    pub fn code(&self) -> i32 {
        match self {
            ToolError::InvalidParams(_) => E_INVALID_PARAMS,
            ToolError::Internal(_) => E_INTERNAL,
            ToolError::UnknownTool(_) => E_METHOD_NOT_FOUND,
        }
    }
    pub fn message(&self) -> &str {
        match self {
            ToolError::InvalidParams(m) | ToolError::Internal(m) | ToolError::UnknownTool(m) => m,
        }
    }

    /// Whether this is a *tool execution* error rather than a *protocol* error.
    ///
    /// MCP separates the two, and the difference is who sees the failure. A
    /// protocol error is a JSON-RPC `error`: the client handles it, and the
    /// model is typically never told. A tool execution error is a normal
    /// `result` carrying `isError: true`, so the failure text lands in the
    /// conversation where the model can read it and correct itself.
    ///
    /// The spec's split (2025-06-18, "Error Handling") is by cause, not by
    /// severity:
    ///
    /// - protocol — unknown tools, invalid arguments
    /// - execution — API failures, business logic errors
    ///
    /// So a tool that never ran because it does not exist, or was called with
    /// arguments that do not typecheck, is a protocol error. A tool that RAN
    /// and failed is an execution error, however badly it failed.
    ///
    /// `memory_add_fact` is the case that motivated this: a store write that
    /// fails returns `Internal`, and as a JSON-RPC error the model was told
    /// nothing — it believed the fact was remembered, because the only signal
    /// went to the client. See car#972 §5.
    pub fn is_execution_error(&self) -> bool {
        match self {
            ToolError::Internal(_) => true,
            ToolError::InvalidParams(_) | ToolError::UnknownTool(_) => false,
        }
    }
}

/// A `tools/call` result reporting that the tool ran and failed.
///
/// Shape is a normal result, not a JSON-RPC error — see
/// [`ToolError::is_execution_error`].
fn tool_execution_error(message: &str) -> Value {
    json!({
        "content": [{ "type": "text", "text": message }],
        "isError": true,
    })
}

fn missing(field: &str) -> ToolError {
    ToolError::InvalidParams(format!("missing {}", field))
}

fn from_tool_value<T: serde::de::DeserializeOwned>(
    value: &Value,
    label: &str,
) -> Result<T, ToolError> {
    serde_json::from_value(value.clone())
        .map_err(|e| ToolError::InvalidParams(format!("{label}: {e}")))
}

fn to_json_text<T: Serialize>(value: &T) -> Result<String, ToolError> {
    serde_json::to_string(value).map_err(|e| ToolError::Internal(e.to_string()))
}

/// The most probe states `equivalent` will accept in one call.
///
/// Each state costs two full simulations, and the loop runs synchronously on
/// the `tools/call` worker — on the daemon's shared HTTP endpoint an unbounded
/// list is caller-controlled work that stalls a tokio worker for its duration.
/// The cap is far above any sampling a caller would hand-write; it only
/// forecloses the pathological list.
pub(crate) const MAX_TEST_STATES: usize = 256;

/// How many resources one `resources/list` page carries.
const RESOURCE_PAGE_SIZE: usize = 100;

/// How many completion values one `completion/complete` response carries.
///
/// The spec caps a completion result at 100 values. Past that the server
/// reports the honest `total` and sets `hasMore`, rather than understating how
/// many matches exist.
const MAX_COMPLETION_VALUES: usize = 100;

/// The values the `car_context` prompt's `mode` argument accepts.
///
/// This list is the completion side of the `match mode` in
/// [`Server::prompts_get`]. Keep the two together: a mode this list offers
/// that `prompts_get` then rejects is a completion that hands the client a
/// `-32602`. `every_completed_mode_is_accepted_by_prompts_get` is the guard.
/// (`prompts_get` also tolerates `""` as an alias for `full`; an empty string
/// is not a value worth *offering*, so it is not listed here.)
const CONTEXT_MODES: &[&str] = &["full", "fast"];

/// Encode a resume key as an opaque cursor: hex of `"v1:" + uri`.
///
/// MCP requires the cursor be opaque to the client, so it must not read as a
/// URI or as an offset that a client might be tempted to compute. Hex-of-bytes
/// keeps it ASCII-safe on every transport; the `v1:` tag leaves room to change
/// the resume key later without mistaking an old cursor for a new one. Rolled
/// by hand rather than pulled in — the workspace has no base64 or hex crate,
/// and ten lines does not justify adding one.
fn encode_cursor(uri: &str) -> String {
    let mut out = String::with_capacity((uri.len() + 3) * 2);
    for b in format!("v1:{}", uri).bytes() {
        out.push_str(&format!("{:02x}", b));
    }
    out
}

/// Inverse of [`encode_cursor`]. Anything that does not decode is a client
/// error, not a server error: `-32602`, per the spec's pagination section.
fn decode_cursor(cursor: &str) -> Result<String, ToolError> {
    let bad = || ToolError::InvalidParams(format!("invalid cursor: {}", cursor));
    if !cursor.len().is_multiple_of(2) {
        return Err(bad());
    }
    let mut bytes = Vec::with_capacity(cursor.len() / 2);
    for pair in cursor.as_bytes().chunks(2) {
        let hex = std::str::from_utf8(pair).map_err(|_| bad())?;
        bytes.push(u8::from_str_radix(hex, 16).map_err(|_| bad())?);
    }
    let decoded = String::from_utf8(bytes).map_err(|_| bad())?;
    decoded
        .strip_prefix("v1:")
        .map(str::to_string)
        .ok_or_else(bad)
}

/// A client-to-server notification, recognized by method name.
///
/// JSON-RPC 2.0 forbids a response to a message with no `id`, so every variant
/// here — including [`Notification::Unknown`] — is answered with silence. The
/// point of naming them is that the server knows what it was told and can say
/// so in the log, instead of one undifferentiated "notification received".
#[derive(Debug, PartialEq)]
enum Notification {
    /// `notifications/initialized` — the handshake is complete.
    Initialized,
    /// `notifications/cancelled` — the client abandoned `request_id`.
    Cancelled {
        request_id: Option<Value>,
        reason: Option<String>,
    },
    /// `notifications/progress` — a progress report against a token.
    Progress { token: Option<Value> },
    /// `notifications/roots/list_changed` — the client's roots moved.
    RootsListChanged,
    /// Anything else. Still answered with silence, per JSON-RPC 2.0.
    Unknown,
}

/// Classify a notification by method name. Pure: no I/O, no state.
fn classify_notification(method: &str, params: &Value) -> Notification {
    match method {
        "notifications/initialized" | "initialized" => Notification::Initialized,
        "notifications/cancelled" => Notification::Cancelled {
            request_id: params.get("requestId").cloned(),
            reason: params
                .get("reason")
                .and_then(|v| v.as_str())
                .map(str::to_string),
        },
        "notifications/progress" => Notification::Progress {
            token: params.get("progressToken").cloned(),
        },
        "notifications/roots/list_changed" => Notification::RootsListChanged,
        _ => Notification::Unknown,
    }
}

/// Pick the protocol version to answer `initialize` with.
///
/// The spec's rule: reply with the revision the client asked for when this
/// server supports it, otherwise with the server's own latest and let the
/// client decide whether to proceed or disconnect. Absent, non-string, and
/// unsupported all take that same fallback — in particular a client that sends
/// no `protocolVersion` is answered exactly as it was before this existed,
/// which is [`PROTOCOL_VERSION`] unconditionally.
///
/// `find` returns the entry out of `supported`, never the client's own bytes,
/// so the reply is a `&'static str` this server chose and can never echo a
/// client-controlled string back onto the wire.
///
/// Split from [`negotiate_version`] only so the echo path is testable against
/// more than one supported revision: with a single-entry list, echo and
/// fallback produce the same string and prove nothing (car#972 §3).
fn negotiate_version_in(supported: &[&'static str], params: &Value) -> &'static str {
    params
        .get("protocolVersion")
        .and_then(Value::as_str)
        .and_then(|want| supported.iter().copied().find(|v| *v == want))
        .unwrap_or(PROTOCOL_VERSION)
}

/// [`negotiate_version_in`] against the real [`SUPPORTED_VERSIONS`].
fn negotiate_version(params: &Value) -> &'static str {
    negotiate_version_in(SUPPORTED_VERSIONS, params)
}

/// A tool implementation supplied by the embedding transport.
///
/// The built-in tools are `match` arms inside [`Server::tools_call`]; this is
/// how a transport adds one the other transport must not advertise. The daemon
/// can serve tools that need a live `Runtime` — the assistant, the scheduler,
/// external agents (car#972 §6) — while the stdio binary, which has none of
/// that, keeps advertising only what it can actually run.
///
/// The signature is deliberately the one every built-in `tool_*` already has:
/// `Result<String, ToolError>`. A registered tool therefore inherits the error
/// split for free — [`ToolError::Internal`] comes back as a result carrying
/// `isError: true` so the model can read the failure, while `InvalidParams`
/// and `UnknownTool` stay JSON-RPC protocol errors. One error convention, not
/// two.
#[async_trait::async_trait]
pub trait ToolHandler: Send + Sync + 'static {
    /// Run the tool against its `arguments` object and return the text result.
    /// `args` is `Value::Null` when the caller sent no arguments.
    async fn call(&self, args: Value) -> Result<String, ToolError>;
}

/// Why [`Server::register_tool`] refused a registration.
///
/// Typed rather than a bare `String` because a caller registers at startup and
/// has to decide what to do about it: a collision with a built-in is a
/// programming error worth failing loudly on, a missing annotation is a schema
/// to go fix.
#[derive(Debug)]
pub enum RegisterError {
    /// The schema carries no `name` string.
    MissingName,
    /// The schema is missing one or more of the four `annotations` hints.
    /// Carries the tool name.
    MissingAnnotations(String),
    /// The name is one of the built-ins. Refused rather than shadowing it, so
    /// a transport cannot quietly replace `memory_delete`.
    CollidesWithBuiltIn(String),
    /// This `Server` already has a tool by that name.
    AlreadyRegistered(String),
}

impl std::fmt::Display for RegisterError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            RegisterError::MissingName => write!(f, "tool schema has no \"name\""),
            RegisterError::MissingAnnotations(name) => write!(
                f,
                "tool schema for {name} must carry all four annotation hints \
                 (readOnlyHint, destructiveHint, idempotentHint, openWorldHint)"
            ),
            RegisterError::CollidesWithBuiltIn(name) => {
                write!(f, "{name} is a built-in tool and cannot be replaced")
            }
            RegisterError::AlreadyRegistered(name) => {
                write!(f, "{name} is already registered on this server")
            }
        }
    }
}

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

/// The hints every advertised tool must classify itself with — the same set
/// `every_advertised_tool_carries_all_four_annotations` enforces over the
/// built-ins. A registered tool goes through the same gate, or this seam would
/// be a hole straight through it.
const REQUIRED_ANNOTATIONS: [&str; 4] = [
    "readOnlyHint",
    "destructiveHint",
    "idempotentHint",
    "openWorldHint",
];

/// MCP server state.
///
/// The memgine is held behind `Arc<Mutex<...>>` so the daemon can
/// share its in-process engine across both the MCP HTTP endpoint
/// and the existing WS dispatcher — facts ingested via MCP show up
/// in WS-served queries and vice versa.
///
/// ## Durability
///
/// `store` is the path to the user's note store
/// ([`car_memgine::note_store`]). When set, the server loads it at
/// construction and appends to it on every `memory_add_fact`.
///
/// Before car#972 §1 there was no such field, and the stdio binary ran on a
/// throwaway graph: a fact an editor plugin remembered was gone the moment
/// stdin closed, and `memory_query` never saw anything the user had actually
/// remembered through `car do`. Both halves matter, and the read half is the
/// bigger one — a memory tool that cannot see the user's memory is worse than
/// useless, because it answers "nothing found" with authority.
///
/// The daemon leaves `store` as `None`: it shares a live engine that it
/// persists itself, and a second writer to the same file would silently drop
/// one side's appends.
///
/// `notes` mirrors the store so an append can be written without re-deriving
/// it from the graph. **Lock order is `notes` before `memgine`** — only
/// [`Server::tool_add_fact`] takes both, but keep it that way if another path
/// ever needs them.
pub struct Server {
    memgine: Arc<Mutex<MemgineEngine>>,
    /// The durable note store, mirrored in memory. Empty and unsaved when
    /// `store` is `None`.
    notes: Mutex<Vec<Note>>,
    /// Where `notes` is persisted, or `None` for an ephemeral server.
    store: Option<PathBuf>,
    /// Transport-registered tools, by name. See [`Server::register_tool`].
    handlers: HashMap<String, Arc<dyn ToolHandler>>,
    /// The advertised tool list, or `None` for "the built-ins, unmodified".
    ///
    /// `None` rather than an eager clone so a server with nothing registered —
    /// both transports today — costs nothing to build and `tools/list` keeps
    /// borrowing the process-wide `OnceLock` cache. The list materializes only
    /// on the first [`Server::register_tool`].
    advertised: Option<Vec<Value>>,
}

impl Default for Server {
    fn default() -> Self {
        Self::new()
    }
}

impl Server {
    /// Construct with a fresh in-memory memgine and NO durable store.
    ///
    /// Used by tests. The stdio binary wants [`Server::with_store`] — an
    /// ephemeral server is exactly the bug car#972 §1 describes.
    pub fn new() -> Self {
        Self::with_memgine(Arc::new(Mutex::new(MemgineEngine::new(None))))
    }

    /// Construct backed by the note store at `path`, loading whatever the user
    /// has already remembered so `memory_query` can see it, and appending to
    /// it on every `memory_add_fact`.
    ///
    /// A missing store is the first run. A store that exists but cannot be
    /// parsed is reported to the caller rather than silently treated as empty:
    /// this server is about to start appending to that file, and overwriting
    /// someone's memory without a word is not a recovery strategy.
    pub fn with_store(path: PathBuf) -> Result<Self, String> {
        let notes = car_memgine::note_store::load_checked(&path)?;
        let mut engine = MemgineEngine::new(None);
        car_memgine::note_store::ingest_all(&mut engine, &notes);
        tracing::info!(
            path = %path.display(),
            notes = notes.len(),
            "MCP server opened the durable note store"
        );
        Ok(Self {
            memgine: Arc::new(Mutex::new(engine)),
            notes: Mutex::new(notes),
            store: Some(path),
            handlers: HashMap::new(),
            advertised: None,
        })
    }

    /// Construct with a caller-supplied memgine handle. Used by the
    /// daemon (`car-server-core`) so the same `MemgineEngine` backs
    /// the MCP endpoint and the WS dispatcher — facts and skills
    /// stay unified. The `Arc<Mutex<...>>` shape mirrors the
    /// daemon's existing memgine wrapper exactly so callers don't
    /// need an adapter layer.
    pub fn with_memgine(memgine: Arc<Mutex<MemgineEngine>>) -> Self {
        Self {
            memgine,
            notes: Mutex::new(Vec::new()),
            store: None,
            handlers: HashMap::new(),
            advertised: None,
        }
    }

    /// Advertise `schema` as a tool of this server alone, dispatching calls to
    /// it through `handler`.
    ///
    /// The tool set used to be process-global: `tools/list` returned
    /// `cached_tool_schemas()` and `tools/call` matched a fixed list, so two
    /// `Server`s in one process could not differ (car#972 §6). Registration is
    /// per-`Server`, so the daemon can carry a tool the stdio binary must not
    /// advertise, while a `Server` with nothing registered advertises exactly
    /// the built-ins, byte for byte.
    ///
    /// Refused when the schema has no name, when the name is a built-in
    /// (rejected rather than silently shadowing it), when this server already
    /// has that name, or when the schema does not carry all four annotation
    /// hints — the same classification gate every built-in passes.
    ///
    /// `&mut self` because registration belongs to startup, before the `Server`
    /// is wrapped in the `Arc` a transport serves from.
    pub fn register_tool(
        &mut self,
        schema: Value,
        handler: Arc<dyn ToolHandler>,
    ) -> Result<(), RegisterError> {
        let name = schema
            .get("name")
            .and_then(|v| v.as_str())
            .ok_or(RegisterError::MissingName)?
            .to_string();

        // Checked against the schema list rather than the `tools/call` match
        // arms so the two cannot drift apart.
        if cached_tool_schemas()
            .iter()
            .any(|t| t["name"].as_str() == Some(name.as_str()))
        {
            return Err(RegisterError::CollidesWithBuiltIn(name));
        }
        if self.handlers.contains_key(&name) {
            return Err(RegisterError::AlreadyRegistered(name));
        }
        let classified = schema
            .get("annotations")
            .and_then(|a| a.as_object())
            .is_some_and(|a| {
                REQUIRED_ANNOTATIONS
                    .iter()
                    .all(|hint| a.get(*hint).is_some_and(Value::is_boolean))
            });
        if !classified {
            return Err(RegisterError::MissingAnnotations(name));
        }

        // Push order is the advertised order: built-ins, then registration
        // order, so `tools/list` is deterministic across runs.
        self.advertised
            .get_or_insert_with(|| cached_tool_schemas().clone())
            .push(schema);
        self.handlers.insert(name, handler);
        Ok(())
    }

    /// What `tools/list` advertises: the built-ins until something is
    /// registered, the built-ins plus the registrations after.
    fn advertised_tools(&self) -> &[Value] {
        self.advertised
            .as_deref()
            .unwrap_or_else(|| cached_tool_schemas().as_slice())
    }

    /// Dispatch one JSON-RPC request. Returns `None` for
    /// notifications (no `id` field). The caller is responsible for
    /// transporting the response back to the client.
    pub async fn handle(&self, req: Request) -> Option<Response> {
        let id = match req.id.clone() {
            Some(id) => id,
            // A message with no `id` is a notification, and JSON-RPC 2.0
            // forbids answering one — so every branch below returns `None`,
            // unrecognized methods included. Recognizing them buys a log line
            // that names what arrived, not a reply.
            //
            // Cancellation deserves a word, because "acknowledged and dropped"
            // is a deliberate choice rather than an oversight. This server
            // executes each `tools/call` synchronously inside `handle` and
            // keeps no registry of in-flight requests: by the time a
            // `notifications/cancelled` can be read off the transport, the
            // request it names has already completed and its response has
            // already been written. There is nothing left to stop. Revisit
            // this the moment a long-running or streaming tool lands
            // (car#972 §6) — that is when an in-flight registry starts to
            // mean something.
            None => {
                match classify_notification(&req.method, &req.params) {
                    Notification::Initialized => {
                        tracing::debug!("client completed the initialize handshake");
                    }
                    Notification::Cancelled { request_id, reason } => {
                        tracing::debug!(
                            request_id = ?request_id,
                            reason = ?reason,
                            "cancellation for an already-completed request; nothing to stop"
                        );
                    }
                    Notification::Progress { token } => {
                        tracing::debug!(token = ?token, "client progress notification");
                    }
                    Notification::RootsListChanged => {
                        tracing::debug!("client roots list changed");
                    }
                    Notification::Unknown => {
                        tracing::debug!(method = %req.method, "unrecognized notification");
                    }
                }
                return None;
            }
        };

        if req.jsonrpc != "2.0" {
            return Some(err(id, E_INVALID_REQUEST, "jsonrpc must be \"2.0\""));
        }

        match req.method.as_str() {
            "initialize" => Some(ok(
                id,
                json!({
                    "protocolVersion": negotiate_version(&req.params),
                    "capabilities": {
                        "tools": {},
                        "resources": { "subscribe": false, "listChanged": false },
                        "prompts": { "listChanged": false },
                        // `completions` was introduced in the 2025-03-26
                        // revision, and this server still negotiates
                        // 2024-11-05 ([`PROTOCOL_VERSION`]). Declaring it
                        // anyway is deliberate: capabilities are an unordered
                        // bag, a 2024-11-05 client ignores a key it does not
                        // know, and hosts routinely probe `completion/complete`
                        // off this flag regardless of the negotiated revision.
                        // Withholding it would make a method we serve
                        // undiscoverable; it costs nothing to advertise, and it
                        // needs no edit when [`SUPPORTED_VERSIONS`] grows.
                        "completions": {},
                    },
                    "serverInfo": { "name": SERVER_NAME, "version": env!("CARGO_PKG_VERSION") },
                }),
            )),
            "ping" => Some(ok(id, json!({}))),
            "tools/list" => Some(ok(id, json!({ "tools": self.advertised_tools() }))),
            "tools/call" => Some(match self.tools_call(&req.params).await {
                Ok(v) => ok(id, v),
                // A tool that ran and failed is reported IN the result, so the
                // model sees it. Only protocol errors take the error channel.
                Err(e) if e.is_execution_error() => ok(id, tool_execution_error(e.message())),
                Err(e) => err(id, e.code(), e.message()),
            }),
            "resources/list" => Some(match self.resources_list(&req.params).await {
                Ok(v) => ok(id, v),
                Err(e) => err(id, e.code(), e.message()),
            }),
            "resources/read" => Some(match self.resources_read(&req.params).await {
                Ok(v) => ok(id, v),
                Err(e) => err(id, e.code(), e.message()),
            }),
            "prompts/list" => Some(ok(id, json!({ "prompts": cached_prompt_schemas() }))),
            "prompts/get" => Some(match self.prompts_get(&req.params).await {
                Ok(v) => ok(id, v),
                Err(e) => err(id, e.code(), e.message()),
            }),
            "completion/complete" => Some(match self.completion_complete(&req.params).await {
                Ok(v) => ok(id, v),
                Err(e) => err(id, e.code(), e.message()),
            }),
            other => Some(err(
                id,
                E_METHOD_NOT_FOUND,
                format!("method not found: {}", other),
            )),
        }
    }

    async fn tools_call(&self, params: &Value) -> Result<Value, ToolError> {
        let name = params
            .get("name")
            .and_then(|v| v.as_str())
            .ok_or_else(|| missing("name"))?;
        let args = params.get("arguments").cloned().unwrap_or(Value::Null);

        let text = match name {
            "memory_add_fact" => self.tool_add_fact(&args).await?,
            "memory_query" => self.tool_query(&args).await?,
            "memory_update_status" => self.tool_memory_update_status(&args).await?,
            "memory_save_knowledge" => self.tool_memory_save_knowledge(&args).await?,
            "memory_save_procedural" => self.tool_memory_save_procedural(&args).await?,
            "memory_delete" => self.tool_memory_delete(&args).await?,
            "memory_intervene" => self.tool_memory_intervene(&args).await?,
            "memory_evaluate" => self.tool_memory_evaluate(&args).await?,
            "skill_find" => self.tool_skill_find(&args).await?,
            "skill_ingest" => self.tool_skill_ingest(&args).await?,
            "skill_list" => self.tool_skill_list(&args).await?,
            "verify" => self.tool_verify(&args)?,
            "simulate" => self.tool_simulate(&args)?,
            "equivalent" => self.tool_equivalent(&args)?,
            "optimize" => self.tool_optimize(&args)?,
            "policy_check" => self.tool_policy_check(&args)?,
            // A name the built-ins do not claim may still belong to this
            // transport. An unregistered one stays a protocol error, exactly as
            // before — the tool never ran.
            other => match self.handlers.get(other) {
                Some(handler) => handler.call(args).await?,
                None => return Err(ToolError::UnknownTool(format!("unknown tool: {}", other))),
            },
        };

        Ok(json!({
            "content": [{ "type": "text", "text": text }],
            "isError": false,
        }))
    }

    async fn tool_add_fact(&self, args: &Value) -> Result<String, ToolError> {
        let subject = args
            .get("subject")
            .and_then(|v| v.as_str())
            .ok_or_else(|| missing("subject"))?;
        let body = args
            .get("body")
            .and_then(|v| v.as_str())
            .ok_or_else(|| missing("body"))?;
        // This tool's schema speaks the `constraint`/`pattern` dialect, which
        // predates `NoteKind`. Translate rather than reimplement: a
        // "constraint" that round-trips as a plain fact silently demotes a
        // standing rule to something recall only sometimes surfaces.
        let kind = car_memgine::note_store::NoteKind::from_constraint_dialect(
            args.get("kind").and_then(|v| v.as_str()),
        );
        let note = Note {
            subject: subject.to_string(),
            body: body.to_string(),
            kind,
        };

        // Lock order: notes, then memgine. See the struct docs.
        let mut notes = self.notes.lock().await;

        // Re-read immediately before appending so a concurrent writer's notes
        // are not clobbered. This narrows the lost-update window; it does not
        // close it (see the `note_store` module docs on concurrency).
        if let Some(path) = &self.store {
            let on_disk = car_memgine::note_store::load(path);
            if on_disk.len() > notes.len() {
                let mut engine = self.memgine.lock().await;
                for (idx, n) in on_disk.iter().enumerate().skip(notes.len()) {
                    car_memgine::note_store::ingest(&mut engine, idx, n);
                }
                *notes = on_disk;
            }
        }

        let idx = notes.len();
        notes.push(note.clone());
        if let Some(path) = &self.store {
            // A write failure must NOT leave the caller believing the fact is
            // durable, so it fails the tool call rather than being logged and
            // swallowed — the in-memory graph is rolled back to match.
            if let Err(e) = car_memgine::note_store::save(path, &notes) {
                notes.pop();
                return Err(ToolError::Internal(format!(
                    "fact not remembered — could not write the memory store: {e}"
                )));
            }
        }

        let mut engine = self.memgine.lock().await;
        car_memgine::note_store::ingest(&mut engine, idx, &note);
        Ok(format!(
            "fact remembered id=assistant-note-{} total={} durable={}",
            idx,
            engine.valid_fact_count(),
            self.store.is_some()
        ))
    }

    /// `policy_check` — see [`car_policy::tool_gate`] for what it evaluates and
    /// why an allow is not always a pass.
    ///
    /// The working directory is the process's own: a host launches this server
    /// in the project it is operating on, which is the same convention `car do`
    /// uses for `.car/policies/`. Deliberately NOT a caller-supplied path — a
    /// tool that reads policy from wherever it is told is a tool that can be
    /// pointed at an empty directory to manufacture an allow.
    fn tool_policy_check(&self, args: &Value) -> Result<String, ToolError> {
        let tool = args
            .get("tool")
            .and_then(|v| v.as_str())
            .ok_or_else(|| missing("tool"))?;
        let params = args.get("params").cloned().unwrap_or_else(|| json!({}));
        let cwd = std::env::current_dir().unwrap_or_else(|_| std::path::PathBuf::from("."));
        to_json_text(&car_policy::tool_gate::check(tool, &params, &cwd))
    }

    async fn tool_query(&self, args: &Value) -> Result<String, ToolError> {
        let query = args
            .get("query")
            .and_then(|v| v.as_str())
            .ok_or_else(|| missing("query"))?;
        let k = args.get("k").and_then(|v| v.as_u64()).unwrap_or(5) as usize;
        let engine = self.memgine.lock().await;
        let seeds = engine.graph.find_seeds(query, 5);
        let hits = if !seeds.is_empty() {
            engine.graph.retrieve(&seeds, 3, k, 0.6, 0.05)
        } else {
            vec![]
        };
        let out: Vec<Value> = hits
            .iter()
            .filter_map(|hit| {
                let node = engine.graph.inner.node_weight(hit.node_ix)?;
                Some(json!({
                    "subject": node.key,
                    "body": node.value,
                    "activation": hit.activation,
                }))
            })
            .collect();
        serde_json::to_string(&out).map_err(|e| ToolError::Internal(e.to_string()))
    }

    async fn tool_memory_update_status(&self, args: &Value) -> Result<String, ToolError> {
        let body = args
            .get("body")
            .and_then(|v| v.as_str())
            .ok_or_else(|| missing("body"))?;
        let tenant_id = args
            .get("tenant_id")
            .and_then(|v| v.as_str())
            .map(str::to_string);
        let mut engine = self.memgine.lock().await;
        let status = engine.update_proactive_status(body, tenant_id);
        to_json_text(&status)
    }

    async fn tool_memory_save_knowledge(&self, args: &Value) -> Result<String, ToolError> {
        let save: car_memgine::ProactiveMemorySave =
            from_tool_value(args, "memory_save_knowledge")?;
        let mut engine = self.memgine.lock().await;
        let saved = engine.save_proactive_knowledge(save);
        to_json_text(&saved)
    }

    async fn tool_memory_save_procedural(&self, args: &Value) -> Result<String, ToolError> {
        let save: car_memgine::ProactiveMemorySave =
            from_tool_value(args, "memory_save_procedural")?;
        let mut engine = self.memgine.lock().await;
        let saved = engine.save_proactive_procedural(save);
        to_json_text(&saved)
    }

    async fn tool_memory_delete(&self, args: &Value) -> Result<String, ToolError> {
        let id = args
            .get("id")
            .and_then(|v| v.as_str())
            .ok_or_else(|| missing("id"))?;
        let mut engine = self.memgine.lock().await;
        let deleted = engine.delete_proactive_memory(id);
        to_json_text(&deleted)
    }

    async fn tool_memory_intervene(&self, args: &Value) -> Result<String, ToolError> {
        let request: car_memgine::ProactiveMemoryRequest =
            from_tool_value(args, "memory_intervene")?;
        let mut engine = self.memgine.lock().await;
        let decision = engine.proactive_intervention(&request);
        to_json_text(&decision)
    }

    async fn tool_memory_evaluate(&self, args: &Value) -> Result<String, ToolError> {
        let request: car_memgine::ProactiveEvaluationRequest =
            from_tool_value(args, "memory_evaluate")?;
        let engine = self.memgine.lock().await;
        let report = engine.evaluate_proactive_memory(&request);
        to_json_text(&report)
    }

    async fn tool_skill_find(&self, args: &Value) -> Result<String, ToolError> {
        let persona = args.get("persona").and_then(|v| v.as_str()).unwrap_or("");
        let url = args.get("url").and_then(|v| v.as_str()).unwrap_or("");
        let task = args
            .get("task")
            .and_then(|v| v.as_str())
            .ok_or_else(|| missing("task"))?;
        let k = args.get("k").and_then(|v| v.as_u64()).unwrap_or(3) as usize;
        let engine = self.memgine.lock().await;
        let results = engine.find_skill(persona, url, task, k);
        let out: Vec<Value> = results
            .iter()
            .map(|(meta, score)| json!({ "skill": meta, "score": score }))
            .collect();
        serde_json::to_string(&out).map_err(|e| ToolError::Internal(e.to_string()))
    }

    async fn tool_skill_ingest(&self, args: &Value) -> Result<String, ToolError> {
        let name = args
            .get("name")
            .and_then(|v| v.as_str())
            .ok_or_else(|| missing("name"))?;
        let code = args
            .get("code")
            .and_then(|v| v.as_str())
            .ok_or_else(|| missing("code"))?;
        let platform = args
            .get("platform")
            .and_then(|v| v.as_str())
            .unwrap_or("unknown");
        let persona = args.get("persona").and_then(|v| v.as_str()).unwrap_or("");
        let url_pattern = args
            .get("url_pattern")
            .and_then(|v| v.as_str())
            .unwrap_or("");
        let description = args
            .get("description")
            .and_then(|v| v.as_str())
            .unwrap_or("");
        let supersedes = args.get("supersedes").and_then(|v| v.as_str());
        let keywords: Vec<String> = args
            .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 = self.memgine.lock().await;
        engine.ingest_skill(
            name,
            code,
            platform,
            trigger,
            description,
            supersedes,
            vec![],
            vec![],
        );
        Ok(format!("skill ingested: {}", name))
    }

    async fn tool_skill_list(&self, args: &Value) -> Result<String, ToolError> {
        let domain = args.get("domain").and_then(|v| v.as_str());
        let engine = self.memgine.lock().await;
        let skills: Vec<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,
                    }
                }
                serde_json::to_value(&meta).ok()
            })
            .collect();
        serde_json::to_string(&skills).map_err(|e| ToolError::Internal(e.to_string()))
    }

    /// Walk the graph once and return every exposed resource as
    /// `(uri, descriptor)`, sorted and deduplicated by URI.
    ///
    /// Extracted from [`Server::resources_list`] so that the two callers that
    /// need this enumeration — the paginated list and `completion/complete`
    /// for `ref/resource` — cannot drift apart. A URI a client can complete
    /// but not list, or the reverse, is worse than no completion at all.
    async fn collect_resources(&self) -> Vec<(String, Value)> {
        let engine = self.memgine.lock().await;
        let mut all: Vec<(String, Value)> = Vec::new();
        for nix in engine.graph.inner.node_indices() {
            let Some(node) = engine.graph.inner.node_weight(nix) else {
                continue;
            };
            match node.kind {
                car_memgine::MemKind::Fact => {
                    let Some(fid) = node.fact_id.as_deref() else {
                        continue;
                    };
                    let uri = format!("car://memory/fact/{}", fid);
                    all.push((
                        uri.clone(),
                        json!({
                            "uri": uri,
                            "name": node.key,
                            "description": if node.is_constraint { "CAR constraint" } else { "CAR fact" },
                            "mimeType": "text/plain",
                        }),
                    ));
                }
                car_memgine::MemKind::Skill => {
                    let uri = format!("car://memory/skill/{}", node.key);
                    all.push((
                        uri.clone(),
                        json!({
                            "uri": uri,
                            "name": node.key,
                            "description": "CAR skill",
                            "mimeType": "application/json",
                        }),
                    ));
                }
                _ => {}
            }
        }
        drop(engine);

        all.sort_unstable_by(|a, b| a.0.cmp(&b.0));
        all.dedup_by(|a, b| a.0 == b.0);
        all
    }

    /// Enumerate graph-backed resources: facts as
    /// `car://memory/fact/{id}` and skills as
    /// `car://memory/skill/{name}`. Invalidated/superseded nodes are
    /// skipped.
    ///
    /// Paginated, per car#972 §4. The page key is the resource URI: the list
    /// is sorted and deduplicated by URI, so the ordering is total and
    /// strictly increasing, and "resume after this URI" is enough to place a
    /// cursor. That makes the cursor **stateless** — it carries its own resume
    /// key, needs no session table, and survives a server restart or a
    /// different process serving the next page, which matters because the
    /// daemon's HTTP endpoint does not pin a client to one server instance.
    ///
    /// Known limitation: the graph walk is still O(nodes) per page. Pagination
    /// bounds the *response*, not the scan. Bounding the scan needs a URI
    /// index in `car-memgine` and is deliberately out of this slice.
    async fn resources_list(&self, params: &Value) -> Result<Value, ToolError> {
        let after = match params.get("cursor") {
            Some(Value::Null) | None => None,
            Some(Value::String(c)) => Some(decode_cursor(c)?),
            Some(_) => return Err(ToolError::InvalidParams("cursor must be a string".into())),
        };

        let all = self.collect_resources().await;

        let start = match after.as_deref() {
            // `partition_point` over a sorted key set: the first index whose
            // URI is strictly greater than the cursor. A cursor naming a
            // resource that has since been deleted still resumes correctly.
            Some(cursor) => all.partition_point(|(uri, _)| uri.as_str() <= cursor),
            None => 0,
        };

        let end = start.saturating_add(RESOURCE_PAGE_SIZE).min(all.len());
        let page: Vec<Value> = all[start..end].iter().map(|(_, v)| v.clone()).collect();

        // Omit `nextCursor` entirely on the last page — its absence is how a
        // client knows it is done.
        let next = if end < all.len() {
            all.get(end - 1).map(|(uri, _)| encode_cursor(uri))
        } else {
            None
        };

        Ok(match next {
            Some(c) => json!({ "resources": page, "nextCursor": c }),
            None => json!({ "resources": page }),
        })
    }

    async fn resources_read(&self, params: &Value) -> Result<Value, ToolError> {
        let uri = params
            .get("uri")
            .and_then(|v| v.as_str())
            .ok_or_else(|| missing("uri"))?;

        let engine = self.memgine.lock().await;

        if let Some(fid) = uri.strip_prefix("car://memory/fact/") {
            for nix in engine.graph.inner.node_indices() {
                let Some(node) = engine.graph.inner.node_weight(nix) else {
                    continue;
                };
                if node.kind != car_memgine::MemKind::Fact {
                    continue;
                }
                if node.fact_id.as_deref() == Some(fid) {
                    return Ok(json!({
                        "contents": [{
                            "uri": uri,
                            "mimeType": "text/plain",
                            "text": format!("{}\n\n{}", node.key, node.value),
                        }],
                    }));
                }
            }
            return Err(ToolError::InvalidParams(format!("fact not found: {}", fid)));
        }

        if let Some(name) = uri.strip_prefix("car://memory/skill/") {
            for nix in engine.graph.inner.node_indices() {
                let Some(node) = engine.graph.inner.node_weight(nix) else {
                    continue;
                };
                if node.kind != car_memgine::MemKind::Skill {
                    continue;
                }
                if node.key == name {
                    let meta = car_memgine::SkillMeta::from_node(node);
                    let body =
                        serde_json::to_string_pretty(&meta).unwrap_or_else(|_| node.value.clone());
                    return Ok(json!({
                        "contents": [{
                            "uri": uri,
                            "mimeType": "application/json",
                            "text": body,
                        }],
                    }));
                }
            }
            return Err(ToolError::InvalidParams(format!(
                "skill not found: {}",
                name
            )));
        }

        Err(ToolError::InvalidParams(format!(
            "unsupported uri scheme: {}",
            uri
        )))
    }

    /// Render CAR's four-layer context assembly as an MCP prompt so
    /// hosts can inject it as the preamble to their own model calls.
    async fn prompts_get(&self, params: &Value) -> Result<Value, ToolError> {
        let name = params
            .get("name")
            .and_then(|v| v.as_str())
            .ok_or_else(|| missing("name"))?;
        if name != "car_context" {
            return Err(ToolError::InvalidParams(format!(
                "unknown prompt: {}",
                name
            )));
        }
        let args = params.get("arguments").cloned().unwrap_or(Value::Null);
        let query = args
            .get("query")
            .and_then(|v| v.as_str())
            .ok_or_else(|| missing("arguments.query"))?;
        let mode = args.get("mode").and_then(|v| v.as_str()).unwrap_or("full");

        let mut engine = self.memgine.lock().await;
        let text = match mode {
            "fast" => engine.build_context_fast(query),
            "full" | "" => engine.build_context(query),
            other => return Err(ToolError::InvalidParams(format!("unknown mode: {}", other))),
        };

        Ok(json!({
            "description": "CAR four-layer context (identity → constraints → facts → conversation → environment → known-unknowns) assembled for the query.",
            "messages": [{
                "role": "user",
                "content": { "type": "text", "text": text },
            }],
        }))
    }

    /// Argument autocompletion for prompts and resources, per car#972 §4.
    ///
    /// This is the one §4 item that needs no server-to-client push: it is a
    /// plain request/response, so it works identically on stdio and on the
    /// daemon's stateless HTTP endpoint. `listChanged`, subscriptions and
    /// progress all need a channel back to the client; this does not.
    ///
    /// Two shapes of `ref` exist in the spec and both are answered here:
    ///
    /// - `ref/prompt` completes one argument of a named prompt. Only
    ///   `car_context` exists, and only its `mode` argument has a closed value
    ///   set ([`CONTEXT_MODES`]) — `query` is arbitrary task text.
    /// - `ref/resource` completes a resource URI. CAR exposes no URI
    ///   *templates* (`resources/templates/list` is not implemented, so there
    ///   is no `{variable}` to fill in), which leaves one meaningful reading:
    ///   "which of the URIs I would actually list start with this prefix?".
    ///   The candidates come from [`Server::collect_resources`], the same walk
    ///   `resources/list` pages over, so a completed URI is always a readable
    ///   one.
    ///
    /// The empty-vs-error split is the same shape car#984 drew for `tools/call`
    /// under car#972 §5: the error channel is for what the CALLER got wrong at
    /// the protocol level, not for a well-formed request with a thin answer.
    ///
    /// So a *malformed* request — no `ref`, no `argument`, an `argument` with
    /// no `name`, a non-object in either slot — never named anything to
    /// complete, and is a protocol error (`-32602`). A *well-formed* request
    /// naming something this server does not know — an unknown prompt, an
    /// unknown argument, an unfamiliar or absent `ref.type` — is answered with
    /// an empty completion: "nothing to suggest" is a legitimate answer to a
    /// legitimate question, and a host's argument picker should not raise an
    /// error dialog because the user tabbed into an unfamiliar field.
    async fn completion_complete(&self, params: &Value) -> Result<Value, ToolError> {
        let reference = match params.get("ref") {
            Some(Value::Object(o)) => o,
            Some(_) => return Err(ToolError::InvalidParams("ref must be an object".into())),
            None => return Err(missing("ref")),
        };
        let argument = match params.get("argument") {
            Some(Value::Object(o)) => o,
            Some(_) => {
                return Err(ToolError::InvalidParams(
                    "argument must be an object".into(),
                ))
            }
            None => return Err(missing("argument")),
        };
        let arg_name = argument
            .get("name")
            .and_then(|v| v.as_str())
            .ok_or_else(|| missing("argument.name"))?;
        // An absent `value` is a client asking for the whole list, not a
        // malformed request — every prefix starts with the empty string.
        let value = argument.get("value").and_then(|v| v.as_str()).unwrap_or("");

        let ref_type = reference.get("type").and_then(|v| v.as_str()).unwrap_or("");

        let matches: Vec<String> = match ref_type {
            "ref/prompt" => {
                let prompt = reference.get("name").and_then(|v| v.as_str()).unwrap_or("");
                match (prompt, arg_name) {
                    ("car_context", "mode") => CONTEXT_MODES
                        .iter()
                        .filter(|m| m.starts_with(value))
                        .map(|m| (*m).to_string())
                        .collect(),
                    // `query` is free-form task text. An empty list is the
                    // honest answer: this server has no corpus of past queries
                    // to suggest from, and inventing one would be worse than
                    // suggesting nothing.
                    _ => Vec::new(),
                }
            }
            "ref/resource" => {
                let prefix = reference.get("uri").and_then(|v| v.as_str()).unwrap_or("");
                self.collect_resources()
                    .await
                    .into_iter()
                    .map(|(uri, _)| uri)
                    .filter(|uri| uri.starts_with(prefix))
                    .collect()
            }
            // Includes a missing or non-string `type`: a shape we do not
            // recognize is answered, not rejected.
            _ => Vec::new(),
        };

        // `total` is the pre-truncation count and is always reported. A client
        // that sees 100 values needs to know whether that is all of them.
        let total = matches.len();
        let values: Vec<String> = matches.into_iter().take(MAX_COMPLETION_VALUES).collect();
        let has_more = total > values.len();
        Ok(json!({
            "completion": {
                "values": values,
                "total": total,
                "hasMore": has_more,
            },
        }))
    }

    fn tool_verify(&self, args: &Value) -> Result<String, ToolError> {
        let proposal_val = args.get("proposal").ok_or_else(|| missing("proposal"))?;
        let proposal: ActionProposal = serde_json::from_value(proposal_val.clone())
            .map_err(|e| ToolError::InvalidParams(format!("proposal: {}", e)))?;
        let max_actions = args
            .get("max_actions")
            .and_then(|v| v.as_u64())
            .unwrap_or(30) as usize;
        let result = car_verify::verify(&proposal, None, None, max_actions);
        serde_json::to_string(&json!({
            "valid": result.valid,
            "issues": result.issues.iter().map(|i| json!({
                "action_id": i.action_id,
                "severity": i.severity,
                "message": i.message,
                // Which kind of check produced the finding — "decision_procedure"
                // | "heuristic" | "sampled". Severity says how bad it would be;
                // this says how it was derived, so an MCP client can tell an
                // exact set-membership failure from the `count >= 3` loop rule
                // of thumb without recognising the message string. Same field
                // and same labels as the JSON-RPC and NAPI/PyO3 projections.
                "tier": i.tier.as_str(),
            })).collect::<Vec<_>>(),
            "simulated_state": result.simulated_state,
        }))
        .map_err(|e| ToolError::Internal(e.to_string()))
    }

    /// `simulate` — the state an executor would leave behind, predicted from
    /// the *declared* `expected_effects` alone. No tool runs.
    ///
    /// The gating cascade belongs to [`car_verify::simulate`], not to this
    /// wrapper: an action whose preconditions or state dependencies are
    /// unsatisfied contributes nothing and takes its dependents with it. What
    /// the tool adds is the wire shape; the caveats live in the schema's
    /// description, because that is the text the calling model actually reads
    /// — a declared effect is *assumed* to land, and `failure_behavior` is not
    /// modelled.
    fn tool_simulate(&self, args: &Value) -> Result<String, ToolError> {
        let proposal: ActionProposal = from_tool_value(
            args.get("proposal").ok_or_else(|| missing("proposal"))?,
            "proposal",
        )?;
        let initial_state: Option<HashMap<String, Value>> = match args.get("initial_state") {
            None | Some(Value::Null) => None,
            Some(v) => Some(from_tool_value(v, "initial_state")?),
        };
        to_json_text(&json!({
            "final_state": car_verify::simulate(&proposal, initial_state.as_ref()),
        }))
    }

    /// `equivalent` — do two proposals leave the same state behind?
    ///
    /// Sampled, and the result says so. `tier` carries the same `"sampled"`
    /// label `verify`'s issues use, so a client learns how the answer was
    /// derived without parsing the description. `states_tested` and
    /// `used_default_states` are the rest of that story: a `true` off the two
    /// trivial defaults is a far weaker claim than a `true` off states the
    /// caller chose, and only these fields separate them.
    ///
    /// The one input this tool will not take literally is an empty
    /// `test_states`: zero probes is a `true` backed by no evidence at all, so
    /// `[]` falls back to the defaults the schema already promises a caller
    /// who supplies none. An over-long list is rejected rather than run — see
    /// [`MAX_TEST_STATES`].
    fn tool_equivalent(&self, args: &Value) -> Result<String, ToolError> {
        let proposal_a: ActionProposal = from_tool_value(
            args.get("proposal_a")
                .ok_or_else(|| missing("proposal_a"))?,
            "proposal_a",
        )?;
        let proposal_b: ActionProposal = from_tool_value(
            args.get("proposal_b")
                .ok_or_else(|| missing("proposal_b"))?,
            "proposal_b",
        )?;
        let test_states: Option<Vec<HashMap<String, Value>>> = match args.get("test_states") {
            None | Some(Value::Null) => None,
            Some(v) => match from_tool_value::<Vec<HashMap<String, Value>>>(v, "test_states")? {
                // An explicitly-empty array is the one input that would make
                // this tool answer without probing anything: `equivalent`
                // loops over the supplied states, so zero states is an
                // unconditional `true`. The schema promises a caller who
                // supplies no states gets the two defaults, and a host that
                // fills an optional array with `[]` is taking that offer — so
                // `[]` takes the default path rather than manufacturing a
                // zero-evidence yes.
                v if v.is_empty() => None,
                v if v.len() > MAX_TEST_STATES => {
                    return Err(ToolError::InvalidParams(format!(
                        "test_states must hold at most {} states (got {})",
                        MAX_TEST_STATES,
                        v.len()
                    )));
                }
                v => Some(v),
            },
        };
        let equivalent = car_verify::equivalent(&proposal_a, &proposal_b, test_states.as_deref());
        to_json_text(&json!({
            "equivalent": equivalent,
            // How the answer was derived. `equivalent` probes states; it does
            // not decide the question, and a client reading only the bool
            // will over-trust a `true`.
            "tier": car_verify::EvidenceTier::Sampled.as_str(),
            // 2 is car_verify::equivalent's own default pair — empty, and {x:1, y:2}.
            "states_tested": test_states.as_ref().map_or(2, Vec::len),
            "used_default_states": test_states.is_none(),
        }))
    }

    /// `optimize` — drop every `state_dependency` naming a key nothing in the
    /// proposal writes, so the DAG builder can widen an execution level.
    ///
    /// A rewrite, not a check. The `pruned` list is diffed here rather than
    /// returned by [`car_verify::optimize`], which hands back only the new
    /// proposal: without it a caller sees a changed proposal and no account of
    /// what changed. It matters because a pruned dependency is exactly what
    /// `verify` would have reported as unavailable — the rewrite removes the
    /// finding, not the underlying gap.
    fn tool_optimize(&self, args: &Value) -> Result<String, ToolError> {
        let proposal: ActionProposal = from_tool_value(
            args.get("proposal").ok_or_else(|| missing("proposal"))?,
            "proposal",
        )?;
        let optimized = car_verify::optimize(&proposal);
        // `optimize` maps the action list one-for-one and in order, so zip
        // pairs each action with its own rewrite.
        let pruned: Vec<Value> = proposal
            .actions
            .iter()
            .zip(optimized.actions.iter())
            .filter_map(|(before, after)| {
                let removed: Vec<&String> = before
                    .state_dependencies
                    .iter()
                    .filter(|d| !after.state_dependencies.contains(d))
                    .collect();
                (!removed.is_empty()).then(|| {
                    json!({
                        "action_id": before.id,
                        "removed": removed,
                    })
                })
            })
            .collect();
        to_json_text(&json!({
            "proposal": optimized,
            "pruned": pruned,
        }))
    }
}

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

    fn make_request(method: &str, params: Value, id: u64) -> Request {
        Request {
            jsonrpc: "2.0".to_string(),
            id: Some(json!(id)),
            method: method.to_string(),
            params,
        }
    }

    /// Call `memory_add_fact` and return the tool's text result.
    async fn add_fact(server: &Server, subject: &str, body: &str, kind: Option<&str>) -> String {
        let mut args = json!({ "subject": subject, "body": body });
        if let Some(k) = kind {
            args["kind"] = json!(k);
        }
        let resp = server
            .handle(make_request(
                "tools/call",
                json!({ "name": "memory_add_fact", "arguments": args }),
                1,
            ))
            .await
            .expect("response");
        let result = resp.result.expect("tool result");
        result["content"][0]["text"]
            .as_str()
            .expect("text")
            .to_string()
    }

    // ---- car#972 §4a: resources/list pagination -------------------------

    /// Call `resources/list`, optionally with a cursor, and return the result.
    async fn list_resources(server: &Server, cursor: Option<&str>) -> Response {
        let params = match cursor {
            Some(c) => json!({ "cursor": c }),
            None => json!({}),
        };
        server
            .handle(make_request("resources/list", params, 1))
            .await
            .expect("resources/list is a request, not a notification")
    }

    /// The §4a regression: an unpaginated `resources/list` returns the whole
    /// graph in one response. On a real memory graph that is unbounded.
    #[tokio::test]
    async fn resources_list_paginates_and_the_cursor_resumes() {
        let server = Server::new();
        let total = 250;
        for i in 0..total {
            // Zero-padded so the ingest order and the URI order differ from
            // each other's obvious guess; the page walk must not depend on
            // insertion order.
            add_fact(&server, &format!("subject {:03}", i), "body", None).await;
        }

        let first = list_resources(&server, None).await;
        let result = first.result.expect("page 1");
        let page1 = result["resources"].as_array().expect("resources array");
        assert_eq!(
            page1.len(),
            RESOURCE_PAGE_SIZE,
            "page 1 must be capped at one page, not the whole graph"
        );
        let mut cursor = result["nextCursor"]
            .as_str()
            .expect("a full page must carry nextCursor")
            .to_string();

        let mut seen: Vec<String> = page1
            .iter()
            .map(|r| r["uri"].as_str().unwrap().to_string())
            .collect();
        let mut pages = 1;

        loop {
            let resp = list_resources(&server, Some(&cursor)).await;
            let result = resp.result.expect("page result");
            let page = result["resources"].as_array().expect("resources array");
            pages += 1;
            assert!(pages <= 10, "pagination did not terminate");
            for r in page {
                seen.push(r["uri"].as_str().unwrap().to_string());
            }
            match result.get("nextCursor") {
                Some(c) => cursor = c.as_str().expect("cursor is a string").to_string(),
                // The last page omits nextCursor entirely — that absence is
                // how the client knows it is done.
                None => {
                    assert!(
                        page.len() < RESOURCE_PAGE_SIZE || seen.len() == total,
                        "a final page should not be a full page unless the total divides evenly"
                    );
                    break;
                }
            }
        }

        assert_eq!(
            seen.len(),
            total,
            "the union of the pages lost or duplicated entries"
        );
        let mut sorted = seen.clone();
        sorted.sort();
        sorted.dedup();
        assert_eq!(sorted.len(), total, "duplicate URIs across pages");
        assert_eq!(
            seen, sorted,
            "pages must arrive in the total URI order the cursor is defined against"
        );
    }

    /// Per the spec's pagination section, an undecodable cursor is a client
    /// error: `-32602`, not an empty page and not a silent restart from the top.
    #[tokio::test]
    async fn resources_list_rejects_a_malformed_cursor() {
        let server = Server::new();
        add_fact(&server, "one", "body", None).await;

        for bad in ["not-a-cursor", "zz", "6a6a", ""] {
            let resp = list_resources(&server, Some(bad)).await;
            let e = resp
                .error
                .unwrap_or_else(|| panic!("cursor {:?} was accepted", bad));
            assert_eq!(e.code, E_INVALID_PARAMS, "cursor {:?}", bad);
        }

        // A non-string cursor is rejected the same way.
        let resp = server
            .handle(make_request("resources/list", json!({ "cursor": 3 }), 1))
            .await
            .expect("response");
        assert_eq!(resp.error.expect("error").code, E_INVALID_PARAMS);
    }

    /// MCP requires the cursor be opaque. A client must not be able to read a
    /// URI or compute an offset out of it, or it will start doing exactly that.
    #[tokio::test]
    async fn resources_list_cursor_is_opaque() {
        let server = Server::new();
        for i in 0..(RESOURCE_PAGE_SIZE + 5) {
            add_fact(&server, &format!("subject {:03}", i), "body", None).await;
        }
        let resp = list_resources(&server, None).await;
        let cursor = resp.result.expect("result")["nextCursor"]
            .as_str()
            .expect("nextCursor")
            .to_string();

        assert!(!cursor.contains("car://"), "cursor leaks the resource URI");
        assert!(
            cursor.parse::<u64>().is_err(),
            "cursor reads as a decimal offset: {}",
            cursor
        );
        // Opaque to the client, still self-describing to us.
        assert!(decode_cursor(&cursor)
            .unwrap()
            .starts_with("car://memory/fact/"));
    }

    // ---- car#972 §4b: notification recognition --------------------------

    /// §4b. The wire behavior is deliberately unchanged — JSON-RPC 2.0 forbids
    /// answering a message with no `id`, so silence is what MCP requires. What
    /// changes is that the server now knows what it was told: the classifier
    /// parses the cancellation's `requestId`/`reason` instead of logging one
    /// undifferentiated "notification received".
    #[tokio::test]
    async fn a_cancellation_notification_is_recognized_and_answers_nothing() {
        assert_eq!(
            classify_notification(
                "notifications/cancelled",
                &json!({ "requestId": 42, "reason": "user pressed escape" }),
            ),
            Notification::Cancelled {
                request_id: Some(json!(42)),
                reason: Some("user pressed escape".to_string()),
            }
        );

        // `reason` is optional per spec; its absence is not a parse failure.
        assert_eq!(
            classify_notification("notifications/cancelled", &json!({ "requestId": "abc" })),
            Notification::Cancelled {
                request_id: Some(json!("abc")),
                reason: None,
            }
        );

        assert_eq!(
            classify_notification("notifications/initialized", &json!({})),
            Notification::Initialized
        );
        assert_eq!(
            classify_notification("notifications/progress", &json!({ "progressToken": 7 })),
            Notification::Progress {
                token: Some(json!(7))
            }
        );
        assert_eq!(
            classify_notification("notifications/roots/list_changed", &json!({})),
            Notification::RootsListChanged
        );
        assert_eq!(
            classify_notification("notifications/nothing_we_know", &json!({})),
            Notification::Unknown
        );

        // Every one of them, unknown included, is answered with silence. An
        // unrecognized notification must NOT become -32601: a message with no
        // `id` gets no response at all.
        let server = Server::new();
        for method in [
            "notifications/cancelled",
            "notifications/initialized",
            "notifications/progress",
            "notifications/roots/list_changed",
            "notifications/nothing_we_know",
        ] {
            let req = Request {
                jsonrpc: "2.0".to_string(),
                id: None,
                method: method.to_string(),
                params: json!({ "requestId": 1 }),
            };
            assert!(
                server.handle(req).await.is_none(),
                "{} was answered; JSON-RPC 2.0 forbids replying to a notification",
                method
            );
        }
    }

    /// The regression this whole change exists for (car#972 §1): a fact
    /// remembered through MCP must survive the process.
    #[tokio::test]
    async fn a_remembered_fact_outlives_the_server() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("memory.json");

        let first = Server::with_store(path.clone()).unwrap();
        add_fact(&first, "deploy cadence", "we ship on Thursdays", None).await;
        drop(first);

        // A brand-new server — as if the host restarted the stdio binary.
        let second = Server::with_store(path.clone()).unwrap();
        let engine = second.memgine.lock().await;
        assert_eq!(
            engine.valid_fact_count(),
            1,
            "the fact did not survive the restart"
        );
    }

    /// The other half of §1, and the one that matters more: an MCP client must
    /// be able to READ what the user remembered elsewhere. A memory tool that
    /// cannot see the user's memory answers "nothing found" with authority.
    #[tokio::test]
    async fn a_fact_written_by_another_process_is_visible_to_query() {
        use car_memgine::note_store::{save, Note, NoteKind};
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("memory.json");
        save(
            &path,
            &[Note {
                subject: "pet".into(),
                body: "a corgi named Biscuit".into(),
                kind: NoteKind::Fact,
            }],
        )
        .unwrap();

        let server = Server::with_store(path).unwrap();
        let resp = server
            .handle(make_request(
                "tools/call",
                json!({ "name": "memory_query", "arguments": { "query": "pet" } }),
                1,
            ))
            .await
            .expect("response");
        let text = resp.result.expect("result")["content"][0]["text"]
            .as_str()
            .unwrap()
            .to_string();
        assert!(
            text.contains("Biscuit"),
            "query did not see the store: {text}"
        );
    }

    #[tokio::test]
    async fn a_constraint_survives_as_a_constraint() {
        // Round-tripping a standing rule as a plain fact would demote it out
        // of the always-included constraints layer, silently.
        use car_memgine::note_store::{load, NoteKind};
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("memory.json");

        let server = Server::with_store(path.clone()).unwrap();
        add_fact(&server, "style", "always use tabs", Some("constraint")).await;

        let notes = load(&path);
        assert_eq!(notes.len(), 1);
        assert_eq!(notes[0].kind, NoteKind::Preference);
        assert!(notes[0].kind.is_constraint());
    }

    #[tokio::test]
    async fn appends_do_not_clobber_the_existing_store() {
        use car_memgine::note_store::load;
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("memory.json");

        let server = Server::with_store(path.clone()).unwrap();
        add_fact(&server, "one", "first", None).await;
        add_fact(&server, "two", "second", None).await;

        let notes = load(&path);
        assert_eq!(notes.len(), 2);
        assert_eq!(notes[0].subject, "one");
        assert_eq!(notes[1].subject, "two");
    }

    #[tokio::test]
    async fn a_concurrent_writers_append_is_picked_up_not_overwritten() {
        // The store is lock-free, so the server re-reads immediately before
        // appending. This does not close the race, but it must at least not
        // discard a note another process wrote while this server was idle.
        use car_memgine::note_store::{load, save, Note, NoteKind};
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("memory.json");

        let server = Server::with_store(path.clone()).unwrap();
        add_fact(&server, "ours", "from mcp", None).await;

        // Another process (say `car do`) appends while we sit idle.
        let mut theirs = load(&path);
        theirs.push(Note {
            subject: "theirs".into(),
            body: "from the assistant".into(),
            kind: NoteKind::Fact,
        });
        save(&path, &theirs).unwrap();

        add_fact(&server, "ours-again", "second from mcp", None).await;

        let notes = load(&path);
        let subjects: Vec<&str> = notes.iter().map(|n| n.subject.as_str()).collect();
        assert_eq!(subjects, vec!["ours", "theirs", "ours-again"]);
    }

    #[tokio::test]
    async fn an_unreadable_store_refuses_rather_than_overwriting_it() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("memory.json");
        std::fs::write(&path, "{ this is not the store").unwrap();

        assert!(
            Server::with_store(path.clone()).is_err(),
            "started on a corrupt store, and would have overwritten it"
        );
        // And the file is untouched.
        assert_eq!(
            std::fs::read_to_string(&path).unwrap(),
            "{ this is not the store"
        );
    }

    #[tokio::test]
    async fn an_ephemeral_server_still_works_and_says_so() {
        // `Server::new()` keeps working for the daemon and tests; it just
        // reports that nothing is durable.
        let server = Server::new();
        let text = add_fact(&server, "s", "b", None).await;
        assert!(text.contains("durable=false"), "{text}");
    }

    /// The absent-`protocolVersion` regression guard: a client that sends no
    /// version must be answered exactly as it was before negotiation existed.
    #[tokio::test]
    async fn initialize_returns_protocol_version() {
        let server = Server::new();
        let resp = server
            .handle(make_request("initialize", json!({}), 1))
            .await
            .expect("response");
        let result = resp.result.unwrap();
        assert_eq!(result["protocolVersion"], PROTOCOL_VERSION);
        assert_eq!(result["serverInfo"]["name"], SERVER_NAME);
    }

    /// The one negotiation assertion that a constant-returner cannot pass:
    /// two supported revisions, each echoed back when asked for. Injected
    /// rather than run against [`SUPPORTED_VERSIONS`], which holds one entry
    /// today — this is the guard that makes appending a second entry safe.
    #[test]
    fn negotiate_echoes_a_supported_version_the_client_asked_for() {
        let supported = ["2024-11-05", "2025-06-18"];
        for want in supported {
            assert_eq!(
                negotiate_version_in(&supported, &json!({ "protocolVersion": want })),
                want,
                "asked for {want}"
            );
        }
    }

    /// Unknown, malformed, and absent all take the same fallback — the spec's
    /// "server does not support the requested version" behavior, which leaves
    /// the client to proceed or disconnect rather than erroring the handshake.
    #[test]
    fn negotiate_falls_back_for_unknown_malformed_and_absent() {
        for params in [
            json!({ "protocolVersion": "1999-01-01" }),
            json!({ "protocolVersion": 5 }),
            json!({}),
        ] {
            assert_eq!(negotiate_version(&params), PROTOCOL_VERSION, "{params}");
        }
    }

    /// End to end through `Server::handle`: the negotiated version reaches the
    /// wire, and nothing else in the `initialize` result moved.
    #[tokio::test]
    async fn initialize_negotiates_the_requested_version() {
        let server = Server::new();
        let params = json!({
            "protocolVersion": PROTOCOL_VERSION,
            "capabilities": {},
            "clientInfo": { "name": "c", "version": "0" },
        });
        let resp = server
            .handle(make_request("initialize", params, 1))
            .await
            .expect("response");
        let result = resp.result.unwrap();
        assert_eq!(result["protocolVersion"], PROTOCOL_VERSION);
        assert_eq!(result["capabilities"]["resources"]["subscribe"], false);
        assert_eq!(result["serverInfo"]["name"], SERVER_NAME);
    }

    /// An unsupported request is answered with a *result* carrying our latest,
    /// never a JSON-RPC error — the client decides what to do with it.
    #[tokio::test]
    async fn initialize_answers_an_unsupported_version_with_our_latest() {
        let server = Server::new();
        let resp = server
            .handle(make_request(
                "initialize",
                json!({ "protocolVersion": "1999-01-01" }),
                1,
            ))
            .await
            .expect("response");
        assert!(resp.error.is_none(), "{:?}", resp.error);
        assert_eq!(resp.result.unwrap()["protocolVersion"], PROTOCOL_VERSION);
    }

    #[tokio::test]
    async fn notification_returns_none() {
        let server = Server::new();
        let req = Request {
            jsonrpc: "2.0".to_string(),
            id: None,
            method: "notifications/initialized".to_string(),
            params: Value::Null,
        };
        assert!(server.handle(req).await.is_none());
    }

    /// car#972 §5. A tool that RAN and failed must come back as a normal result
    /// carrying `isError: true`, not as a JSON-RPC error — otherwise the failure
    /// goes to the client and the model is never told.
    ///
    /// The failure is forced the way it actually happens: the store cannot be
    /// written. A directory sitting where the store file belongs makes
    /// `fs::write` fail while construction still succeeds, so this exercises the
    /// real `tool_add_fact` write path rather than a stubbed error.
    #[tokio::test]
    async fn a_tool_that_ran_and_failed_reports_is_error_in_the_result() {
        let dir = tempfile::tempdir().expect("tempdir");
        let store = dir.path().join("notes.json");
        let server = Server::with_store(store.clone()).expect("store opens when absent");
        std::fs::create_dir(&store).expect("occupy the store path with a directory");

        let resp = server
            .handle(make_request(
                "tools/call",
                json!({
                    "name": "memory_add_fact",
                    "arguments": { "subject": "s", "body": "b" }
                }),
                1,
            ))
            .await
            .expect("response");

        assert!(
            resp.error.is_none(),
            "an execution failure must not take the JSON-RPC error channel: {:?}",
            resp.error
        );
        let result = resp.result.expect("result");
        assert_eq!(result["isError"], true);
        let text = result["content"][0]["text"].as_str().unwrap();
        assert!(
            text.contains("could not write the memory store"),
            "the model must be able to READ why it failed; got: {}",
            text
        );
    }

    /// The other half of the split: an unknown tool never ran, so it stays a
    /// protocol error. Without this, "make failures visible to the model" could
    /// be over-applied until every error became `isError` and clients lost the
    /// ability to distinguish a broken call from a failed one.
    #[tokio::test]
    async fn an_unknown_tool_is_still_a_protocol_error() {
        let server = Server::new();
        let resp = server
            .handle(make_request(
                "tools/call",
                json!({ "name": "no_such_tool", "arguments": {} }),
                1,
            ))
            .await
            .expect("response");
        assert!(resp.result.is_none(), "must not be reported as a result");
        assert_eq!(resp.error.expect("error").code, E_METHOD_NOT_FOUND);
    }

    /// Likewise invalid arguments — the tool never ran.
    #[tokio::test]
    async fn invalid_arguments_are_still_a_protocol_error() {
        let server = Server::new();
        let resp = server
            .handle(make_request(
                "tools/call",
                // `body` is required and absent.
                json!({ "name": "memory_add_fact", "arguments": { "subject": "s" } }),
                1,
            ))
            .await
            .expect("response");
        assert!(resp.result.is_none(), "must not be reported as a result");
        assert_eq!(resp.error.expect("error").code, E_INVALID_PARAMS);
    }

    #[tokio::test]
    async fn unknown_method_returns_method_not_found_error() {
        let server = Server::new();
        let resp = server
            .handle(make_request("bogus/method", json!({}), 1))
            .await
            .expect("response");
        let err = resp.error.unwrap();
        assert_eq!(err.code, E_METHOD_NOT_FOUND);
    }

    #[tokio::test]
    async fn tools_list_returns_the_advertised_surface() {
        // Asserts the SET, not the count. A count test fails just as loudly
        // when a tool goes missing but does not say which one, and it invites
        // being "fixed" by bumping the number.
        let server = Server::new();
        let resp = server
            .handle(make_request("tools/list", json!({}), 1))
            .await
            .expect("response");
        let result = resp.result.unwrap();
        let mut names: Vec<&str> = result["tools"]
            .as_array()
            .unwrap()
            .iter()
            .map(|t| t["name"].as_str().unwrap())
            .collect();
        names.sort_unstable();
        assert_eq!(
            names,
            vec![
                "equivalent",
                "memory_add_fact",
                "memory_delete",
                "memory_evaluate",
                "memory_intervene",
                "memory_query",
                "memory_save_knowledge",
                "memory_save_procedural",
                "memory_update_status",
                "optimize",
                "policy_check",
                "simulate",
                "skill_find",
                "skill_ingest",
                "skill_list",
                "verify",
            ]
        );
    }

    // ---- car#972 §6: the tool set is per-Server, not per-process --------

    /// A registered tool. `fail` makes it report an execution failure so the
    /// error split can be checked through the seam.
    struct TestTool {
        text: &'static str,
        fail: bool,
    }

    #[async_trait::async_trait]
    impl ToolHandler for TestTool {
        async fn call(&self, _args: Value) -> Result<String, ToolError> {
            if self.fail {
                Err(ToolError::Internal(self.text.to_string()))
            } else {
                Ok(self.text.to_string())
            }
        }
    }

    /// A well-formed schema for a registered tool — classified, as
    /// `register_tool` requires.
    fn test_schema(name: &str) -> Value {
        json!({
            "name": name,
            "description": "a test tool",
            "inputSchema": { "type": "object", "properties": {} },
            "annotations": {
                "readOnlyHint": true,
                "destructiveHint": false,
                "idempotentHint": true,
                "openWorldHint": false,
            },
        })
    }

    async fn tool_names(server: &Server) -> Vec<String> {
        let resp = server
            .handle(make_request("tools/list", json!({}), 1))
            .await
            .expect("response");
        resp.result.expect("result")["tools"]
            .as_array()
            .expect("array")
            .iter()
            .map(|t| t["name"].as_str().expect("name").to_string())
            .collect()
    }

    async fn call_tool(server: &Server, name: &str) -> Response {
        server
            .handle(make_request(
                "tools/call",
                json!({ "name": name, "arguments": {} }),
                1,
            ))
            .await
            .expect("response")
    }

    /// The compatibility half: making the surface per-server must not change
    /// what a server with nothing registered advertises. Compared as strings,
    /// so this is byte identity and not merely `Value` equality.
    #[tokio::test]
    async fn with_nothing_registered_the_surface_is_exactly_the_built_ins() {
        let server = Server::new();
        let resp = server
            .handle(make_request("tools/list", json!({}), 1))
            .await
            .expect("response");
        let result = resp.result.expect("result");
        assert_eq!(
            serde_json::to_string(&result["tools"]).unwrap(),
            serde_json::to_string(cached_tool_schemas()).unwrap(),
        );
    }

    #[tokio::test]
    async fn a_registered_tool_is_advertised_and_dispatched() {
        let mut server = Server::new();
        server
            .register_tool(
                test_schema("assistant_start"),
                Arc::new(TestTool {
                    text: "ran",
                    fail: false,
                }),
            )
            .expect("registers");

        let names = tool_names(&server).await;
        assert_eq!(names.len(), cached_tool_schemas().len() + 1);
        // Registrations come after the built-ins, in registration order.
        assert_eq!(names.last().map(String::as_str), Some("assistant_start"));

        let resp = call_tool(&server, "assistant_start").await;
        assert!(resp.error.is_none(), "{:?}", resp.error);
        let result = resp.result.expect("result");
        assert_eq!(result["content"][0]["text"], "ran");
        assert_eq!(result["isError"], false);
    }

    /// The regression car#972 §6 names: with a process-global tool list, two
    /// servers in one process could not advertise different surfaces, so the
    /// daemon could not carry a tool the stdio binary must not offer.
    #[tokio::test]
    async fn two_servers_in_one_process_advertise_different_tool_sets() {
        let mut daemon = Server::new();
        daemon
            .register_tool(
                test_schema("assistant_start"),
                Arc::new(TestTool {
                    text: "ran",
                    fail: false,
                }),
            )
            .expect("registers");
        let stdio = Server::new();

        let daemon_names = tool_names(&daemon).await;
        let stdio_names = tool_names(&stdio).await;

        assert_eq!(daemon_names.len(), cached_tool_schemas().len() + 1);
        assert_eq!(stdio_names.len(), cached_tool_schemas().len());
        assert!(daemon_names.iter().any(|n| n == "assistant_start"));
        assert!(!stdio_names.iter().any(|n| n == "assistant_start"));

        // And the one that did not register cannot be made to run it.
        let resp = call_tool(&stdio, "assistant_start").await;
        assert_eq!(resp.error.expect("error").code, E_METHOD_NOT_FOUND);
    }

    /// The server the **stdio binary** actually builds must not offer the
    /// assistant trio (car#972 §6).
    ///
    /// `car-mcp-server` is `car-mcp` + `car-telemetry`: no `Runtime`, no
    /// inference engine, no daemon state. The daemon registers those three; the
    /// counterpart assertion — that the daemon's server DOES advertise all
    /// three — lives with the registration, in
    /// `car_server_core::mcp_assistant`. Between them they pin both sides of
    /// "daemon only", which one test alone cannot.
    #[tokio::test]
    async fn the_stdio_server_offers_no_assistant_tools() {
        let dir = tempfile::tempdir().expect("tempdir");
        let server =
            Server::with_store(dir.path().join("notes.json")).expect("opens a fresh store");

        let names = tool_names(&server).await;
        for tool in ["assistant_start", "assistant_poll", "assistant_cancel"] {
            assert!(
                !names.iter().any(|n| n == tool),
                "{tool} must not be advertised on stdio: {names:?}"
            );
            // Never ran, so it is a protocol error and not an `isError` result.
            let resp = call_tool(&server, tool).await;
            assert_eq!(resp.error.expect("error").code, E_METHOD_NOT_FOUND);
        }
    }

    #[tokio::test]
    async fn registering_a_built_in_name_is_rejected() {
        let mut server = Server::new();
        let e = server
            .register_tool(
                test_schema("memory_query"),
                Arc::new(TestTool {
                    text: "hijacked",
                    fail: false,
                }),
            )
            .expect_err("a built-in cannot be shadowed");
        assert!(matches!(e, RegisterError::CollidesWithBuiltIn(ref n) if n == "memory_query"));

        assert_eq!(tool_names(&server).await.len(), cached_tool_schemas().len());
        // And the built-in still answers, rather than the would-be handler.
        let resp = server
            .handle(make_request(
                "tools/call",
                json!({ "name": "memory_query", "arguments": { "query": "anything" } }),
                1,
            ))
            .await
            .expect("response");
        let text = resp.result.expect("result")["content"][0]["text"]
            .as_str()
            .expect("text")
            .to_string();
        assert_ne!(text, "hijacked");
    }

    #[tokio::test]
    async fn registering_the_same_name_twice_is_rejected() {
        let mut server = Server::new();
        let tool = || {
            Arc::new(TestTool {
                text: "ran",
                fail: false,
            })
        };
        server
            .register_tool(test_schema("assistant_start"), tool())
            .expect("first registration");
        let e = server
            .register_tool(test_schema("assistant_start"), tool())
            .expect_err("second registration");
        assert!(matches!(e, RegisterError::AlreadyRegistered(ref n) if n == "assistant_start"));
        assert_eq!(
            tool_names(&server).await.len(),
            cached_tool_schemas().len() + 1,
            "a refused registration must not have been advertised"
        );
    }

    /// Without this, the seam would be a hole straight through
    /// `every_advertised_tool_carries_all_four_annotations`.
    #[tokio::test]
    async fn a_registered_tool_without_annotations_is_rejected() {
        let mut server = Server::new();
        let mut schema = test_schema("assistant_start");
        schema["annotations"]
            .as_object_mut()
            .expect("object")
            .remove("idempotentHint");
        let e = server
            .register_tool(
                schema,
                Arc::new(TestTool {
                    text: "ran",
                    fail: false,
                }),
            )
            .expect_err("an unclassified tool cannot be advertised");
        assert!(matches!(e, RegisterError::MissingAnnotations(ref n) if n == "assistant_start"));
    }

    /// A registered tool inherits the car#972 §5 error split rather than
    /// inventing a second convention: it ran and failed, so the model sees it.
    #[tokio::test]
    async fn a_registered_tool_that_fails_reports_an_execution_error() {
        let mut server = Server::new();
        server
            .register_tool(
                test_schema("assistant_start"),
                Arc::new(TestTool {
                    text: "the runtime refused",
                    fail: true,
                }),
            )
            .expect("registers");

        let resp = call_tool(&server, "assistant_start").await;
        assert!(
            resp.error.is_none(),
            "an execution failure must not take the JSON-RPC error channel: {:?}",
            resp.error
        );
        let result = resp.result.expect("result");
        assert_eq!(result["isError"], true);
        assert_eq!(result["content"][0]["text"], "the runtime refused");
    }

    #[test]
    fn every_advertised_tool_carries_all_four_annotations() {
        // The gate: a tool cannot be added to `tool_schemas()` without being
        // classified. Checks the key SET rather than "has annotations", so a
        // three-hint entry fails here instead of shipping a hint a host reads
        // as its spec default.
        for tool in cached_tool_schemas() {
            let name = tool["name"].as_str().expect("every tool has a name");
            let ann = tool["annotations"]
                .as_object()
                .unwrap_or_else(|| panic!("{name} advertises no annotations object"));
            let mut keys: Vec<&str> = ann.keys().map(String::as_str).collect();
            keys.sort_unstable();
            assert_eq!(
                keys,
                vec![
                    "destructiveHint",
                    "idempotentHint",
                    "openWorldHint",
                    "readOnlyHint",
                ],
                "{name}'s annotation keys"
            );
            for (hint, value) in ann {
                assert!(value.is_boolean(), "{name}.{hint} must be a bool");
            }
            // `destructiveHint` is meaningless when `readOnlyHint` is true and
            // its spec default is `true`, so a read-only tool that leaves it
            // set reads as destructive to a host that does not special-case
            // the combination.
            if ann["readOnlyHint"] == json!(true) {
                assert_eq!(
                    ann["destructiveHint"],
                    json!(false),
                    "{name} is read-only, so destructiveHint must be false"
                );
            }
        }
    }

    #[test]
    fn the_annotations_match_what_the_tools_actually_do() {
        // Pins the classifications a rename or a copy-pasted block would get
        // wrong. Each one is re-derivable from car-memgine:
        //   memory_intervene  -> engine.rs:947 calls record_proactive_injection,
        //                        which bumps proactive_injections (engine.rs:728)
        //   skill_ingest      -> engine.rs:1253 flips the superseded skill's
        //                        node to MemKind::SkillDeprecated
        //   memory_delete     -> engine.rs:844 removes the node outright
        let hint = |tool: &str, key: &str| -> bool {
            cached_tool_schemas()
                .iter()
                .find(|t| t["name"] == json!(tool))
                .unwrap_or_else(|| panic!("{tool} is not advertised"))["annotations"][key]
                .as_bool()
                .unwrap_or_else(|| panic!("{tool}.{key} is not a bool"))
        };

        assert!(hint("memory_query", "readOnlyHint"));
        assert!(hint("skill_find", "readOnlyHint"));
        assert!(hint("policy_check", "readOnlyHint"));
        assert!(hint("verify", "readOnlyHint"));

        assert!(hint("memory_delete", "destructiveHint"));
        // Reads like a query, is a write.
        assert!(!hint("memory_intervene", "readOnlyHint"));
        // Reads like an insert, can retire an existing skill.
        assert!(hint("skill_ingest", "destructiveHint"));
        // An overwrite of the single status slot, not an append.
        assert!(hint("memory_update_status", "destructiveHint"));
        // An append that mints a fresh id on collision — not an overwrite.
        assert!(!hint("memory_save_knowledge", "destructiveHint"));

        // Nothing here reaches outside the local memory graph, the local
        // policy files, or a pure in-process check.
        for tool in cached_tool_schemas() {
            let name = tool["name"].as_str().expect("name");
            assert_eq!(
                tool["annotations"]["openWorldHint"],
                json!(false),
                "{name} claims an open world — is that true, or a copy-paste?"
            );
        }
    }

    #[tokio::test]
    async fn add_fact_then_query_round_trips() {
        let server = Server::new();
        let _add = server
            .handle(make_request(
                "tools/call",
                json!({
                    "name": "memory_add_fact",
                    "arguments": { "subject": "color", "body": "the sky is blue" }
                }),
                1,
            ))
            .await
            .unwrap();
        let q = server
            .handle(make_request(
                "tools/call",
                json!({
                    "name": "memory_query",
                    "arguments": { "query": "color", "k": 5 }
                }),
                2,
            ))
            .await
            .unwrap();
        let text = q.result.unwrap()["content"][0]["text"]
            .as_str()
            .unwrap()
            .to_string();
        assert!(text.contains("color") || text.contains("sky"));
    }

    #[tokio::test]
    async fn proactive_memory_tools_round_trip() {
        let server = Server::new();
        let save = server
            .handle(make_request(
                "tools/call",
                json!({
                    "name": "memory_save_knowledge",
                    "arguments": {
                        "id": "mcp-policy",
                        "subject": "deployment policy",
                        "body": "Must verify before shipping",
                        "tags": ["policy"],
                        "is_constraint": true
                    }
                }),
                1,
            ))
            .await
            .unwrap();
        let save_text = save.result.unwrap()["content"][0]["text"]
            .as_str()
            .unwrap()
            .to_string();
        assert!(save_text.contains("mcp-policy"));

        let intervene = server
            .handle(make_request(
                "tools/call",
                json!({
                    "name": "memory_intervene",
                    "arguments": {
                        "query": "ship deployment",
                        "trigger": { "high_risk_action": true }
                    }
                }),
                2,
            ))
            .await
            .unwrap();
        let intervene_text = intervene.result.unwrap()["content"][0]["text"]
            .as_str()
            .unwrap()
            .to_string();
        assert!(intervene_text.contains("\"decision\":\"inject\""));
        assert!(intervene_text.contains("Must verify before shipping"));

        let eval = server
            .handle(make_request(
                "tools/call",
                json!({
                    "name": "memory_evaluate",
                    "arguments": {
                        "cases": [{
                            "id": "ship",
                            "request": {
                                "query": "ship deployment",
                                "trigger": { "high_risk_action": true }
                            },
                            "relevant_fact_ids": ["mcp-policy"]
                        }]
                    }
                }),
                3,
            ))
            .await
            .unwrap();
        let eval_text = eval.result.unwrap()["content"][0]["text"]
            .as_str()
            .unwrap()
            .to_string();
        assert!(eval_text.contains("\"true_positives\":1"));
    }

    /// Every issue the `verify` tool returns names the *kind* of check that
    /// produced it. Without the tier, an MCP client has to recognise the
    /// message string to tell the `count >= 3` loop rule of thumb from an
    /// exact registry lookup — which is the gap `EvidenceTier` closed on every
    /// other surface, and which this projection silently reopened once before.
    #[tokio::test]
    async fn verify_issues_carry_their_evidence_tier() {
        let server = Server::new();
        let resp = server
            .handle(make_request(
                "tools/call",
                json!({
                    "name": "verify",
                    "arguments": {
                        "proposal": {
                            "actions": [
                                { "id": "a1", "type": "tool_call", "tool": "poll" },
                                { "id": "a2", "type": "tool_call", "tool": "poll" },
                                { "id": "a3", "type": "tool_call", "tool": "poll" }
                            ]
                        }
                    }
                }),
                1,
            ))
            .await
            .expect("response");
        let text = resp.result.unwrap()["content"][0]["text"]
            .as_str()
            .unwrap()
            .to_string();
        let parsed: Value = serde_json::from_str(&text).expect("verify result is JSON");
        let issues = parsed["issues"].as_array().expect("issues array");
        assert!(
            !issues.is_empty(),
            "three identical calls trip loop detection"
        );
        for issue in issues {
            let tier = issue["tier"].as_str().expect("every issue carries a tier");
            assert!(
                ["decision_procedure", "heuristic", "sampled"].contains(&tier),
                "unexpected tier {tier} — the labels are shared with the FFI and \
                 JSON-RPC projections"
            );
        }
        // The repeated call is a proxy signal, not a decided property.
        assert!(issues
            .iter()
            .any(|i| i["tier"] == "heuristic"
                && i["message"].as_str().unwrap().contains("likely loop")));
    }

    // ---- car#972 §6: verify's siblings ---------------------------------

    /// Call a tool and parse its text blob back into JSON — every one of the
    /// four verification tools answers with a serialized object.
    async fn call_verification_tool(name: &str, arguments: Value) -> Value {
        let server = Server::new();
        let resp = server
            .handle(make_request(
                "tools/call",
                json!({ "name": name, "arguments": arguments }),
                1,
            ))
            .await
            .expect("response");
        let result = resp
            .result
            .unwrap_or_else(|| panic!("{name} returned no result"));
        assert_eq!(result["isError"], false, "{name} reported a failure");
        serde_json::from_str(result["content"][0]["text"].as_str().expect("text"))
            .unwrap_or_else(|e| panic!("{name}'s text blob is not JSON: {e}"))
    }

    #[tokio::test]
    async fn simulate_returns_the_state_the_declared_effects_imply() {
        let parsed = call_verification_tool(
            "simulate",
            json!({
                "proposal": {
                    "actions": [
                        { "id": "a1", "type": "state_write", "parameters": { "key": "x", "value": 10 } },
                        { "id": "a2", "type": "state_write", "parameters": { "key": "y", "value": 20 } }
                    ]
                },
                "initial_state": { "seed": true }
            }),
        )
        .await;
        assert_eq!(parsed["final_state"]["x"], json!(10));
        assert_eq!(parsed["final_state"]["y"], json!(20));
        // The initial state is carried, not discarded.
        assert_eq!(parsed["final_state"]["seed"], json!(true));
    }

    /// The property that makes `simulate` worth exposing separately from
    /// `verify`: an action the executor would reject contributes nothing, so
    /// the predicted state never claims an effect that could not have landed.
    #[tokio::test]
    async fn simulate_does_not_credit_an_action_whose_dependency_is_missing() {
        let parsed = call_verification_tool(
            "simulate",
            json!({
                "proposal": {
                    "actions": [{
                        "id": "a1",
                        "type": "tool_call",
                        "tool": "deploy",
                        "state_dependencies": ["nobody_writes_this"],
                        "expected_effects": { "deployed": true }
                    }]
                }
            }),
        )
        .await;
        assert!(
            parsed["final_state"].get("deployed").is_none(),
            "a blocked action must not contribute its declared effects: {}",
            parsed["final_state"]
        );
    }

    /// `equivalent` samples, and the result has to say so — a client that
    /// reads only the bool will treat a `true` off two trivial default states
    /// as a claim the proposals never diverge.
    #[tokio::test]
    async fn equivalent_reports_how_the_answer_was_derived() {
        let write = |id: &str, key: &str, value: i64| json!({ "id": id, "type": "state_write", "parameters": { "key": key, "value": value } });

        // Same writes, opposite order — indistinguishable on every sampled state.
        let same = call_verification_tool(
            "equivalent",
            json!({
                "proposal_a": { "actions": [write("a1", "x", 1), write("a2", "y", 2)] },
                "proposal_b": { "actions": [write("b1", "y", 2), write("b2", "x", 1)] },
            }),
        )
        .await;
        assert_eq!(same["equivalent"], json!(true));
        assert_eq!(same["tier"], json!("sampled"));
        assert_eq!(same["states_tested"], json!(2));
        assert_eq!(same["used_default_states"], json!(true));

        // A different value for the same key — a sampled state separates them.
        let differs = call_verification_tool(
            "equivalent",
            json!({
                "proposal_a": { "actions": [write("a1", "x", 1)] },
                "proposal_b": { "actions": [write("b1", "x", 99)] },
                "test_states": [{}, { "x": 0 }, { "unrelated": 7 }],
            }),
        )
        .await;
        assert_eq!(differs["equivalent"], json!(false));
        assert_eq!(differs["states_tested"], json!(3));
        assert_eq!(differs["used_default_states"], json!(false));
    }

    /// The one shape that could make `equivalent` answer off no evidence: an
    /// explicitly-empty `test_states` probes nothing, and a loop over zero
    /// states returns `true` unconditionally. A host filling an optional array
    /// with `[]` is asking for the default, so it gets the default — never a
    /// zero-probe yes on two proposals that provably differ.
    #[tokio::test]
    async fn an_empty_test_states_never_buys_a_zero_probe_true() {
        let write = |id: &str, key: &str, value: i64| json!({ "id": id, "type": "state_write", "parameters": { "key": key, "value": value } });
        let parsed = call_verification_tool(
            "equivalent",
            json!({
                "proposal_a": { "actions": [write("a1", "x", 1)] },
                "proposal_b": { "actions": [write("b1", "x", 2)] },
                "test_states": [],
            }),
        )
        .await;
        assert_eq!(
            parsed["equivalent"],
            json!(false),
            "x=1 and x=2 differ on the default states; [] must not vacuously agree"
        );
        assert_eq!(
            parsed["states_tested"],
            json!(2),
            "[] falls back to the two defaults, so the probe count is never 0"
        );
        assert_eq!(parsed["used_default_states"], json!(true));
    }

    /// Each probe state costs two full simulations on the `tools/call` worker,
    /// so the list is bounded and an over-long one is refused before any of it
    /// runs — a protocol error, because the tool never ran.
    #[tokio::test]
    async fn an_over_long_test_states_is_refused_rather_than_run() {
        let server = Server::new();
        let states: Vec<Value> = (0..=MAX_TEST_STATES).map(|i| json!({ "x": i })).collect();
        let resp = server
            .handle(make_request(
                "tools/call",
                json!({
                    "name": "equivalent",
                    "arguments": {
                        "proposal_a": { "actions": [] },
                        "proposal_b": { "actions": [] },
                        "test_states": states,
                    },
                }),
                1,
            ))
            .await
            .expect("response");
        assert!(resp.result.is_none(), "the tool must not have run");
        assert_eq!(resp.error.expect("error").code, E_INVALID_PARAMS);
    }

    #[tokio::test]
    async fn optimize_prunes_a_phantom_dependency_and_names_it() {
        let parsed = call_verification_tool(
            "optimize",
            json!({
                "proposal": {
                    "actions": [
                        { "id": "a1", "type": "state_write", "parameters": { "key": "x", "value": 1 } },
                        {
                            "id": "a2",
                            "type": "tool_call",
                            "tool": "report",
                            "state_dependencies": ["x", "nobody_writes_this"]
                        }
                    ]
                }
            }),
        )
        .await;
        let actions = parsed["proposal"]["actions"].as_array().expect("actions");
        assert_eq!(
            actions.len(),
            2,
            "optimize rewrites, it does not drop actions"
        );
        // The real dependency survives; the phantom does not.
        assert_eq!(actions[1]["state_dependencies"], json!(["x"]));
        assert_eq!(
            parsed["pruned"],
            json!([{ "action_id": "a2", "removed": ["nobody_writes_this"] }]),
            "the caller has to be told what the rewrite dropped"
        );
    }

    /// A proposal that does not deserialize means the tool never ran, so it is
    /// a protocol error — a JSON-RPC `-32602`, not an `isError` result. Same
    /// split `verify` and the registered-tool seam already honour.
    #[tokio::test]
    async fn a_malformed_proposal_is_a_protocol_error_on_every_sibling() {
        let server = Server::new();
        let cases = [
            ("simulate", json!({ "proposal": "not an object" })),
            (
                "equivalent",
                json!({ "proposal_a": { "actions": [] }, "proposal_b": 7 }),
            ),
            ("optimize", json!({ "proposal": ["not", "an", "object"] })),
        ];
        for (name, arguments) in cases {
            let resp = server
                .handle(make_request(
                    "tools/call",
                    json!({ "name": name, "arguments": arguments }),
                    1,
                ))
                .await
                .expect("response");
            assert!(
                resp.result.is_none(),
                "{name} must not answer a malformed proposal with a result"
            );
            assert_eq!(
                resp.error.expect("error").code,
                E_INVALID_PARAMS,
                "{name}'s error code"
            );
        }
    }

    /// Each sibling is a pure function of its arguments, so all four hints are
    /// the same and a host may auto-approve them the way it does `verify`.
    #[test]
    fn the_verification_siblings_are_advertised_as_pure_reads() {
        for name in ["simulate", "equivalent", "optimize"] {
            let ann = &cached_tool_schemas()
                .iter()
                .find(|t| t["name"] == json!(name))
                .unwrap_or_else(|| panic!("{name} is not advertised"))["annotations"];
            assert_eq!(ann["readOnlyHint"], json!(true), "{name}.readOnlyHint");
            assert_eq!(
                ann["destructiveHint"],
                json!(false),
                "{name}.destructiveHint"
            );
            assert_eq!(ann["idempotentHint"], json!(true), "{name}.idempotentHint");
            assert_eq!(ann["openWorldHint"], json!(false), "{name}.openWorldHint");
        }
    }

    #[tokio::test]
    async fn invalid_jsonrpc_version_rejected() {
        let server = Server::new();
        let req = Request {
            jsonrpc: "1.0".to_string(),
            id: Some(json!(1)),
            method: "ping".to_string(),
            params: Value::Null,
        };
        let resp = server.handle(req).await.unwrap();
        let err = resp.error.unwrap();
        assert_eq!(err.code, E_INVALID_REQUEST);
    }

    #[tokio::test]
    async fn prompts_get_unknown_returns_invalid_params() {
        let server = Server::new();
        let resp = server
            .handle(make_request(
                "prompts/get",
                json!({ "name": "does_not_exist", "arguments": { "query": "x" } }),
                1,
            ))
            .await
            .unwrap();
        let err = resp.error.unwrap();
        assert_eq!(err.code, E_INVALID_PARAMS);
    }

    // ---- car#972 §4: completion/complete ---------------------------------

    /// Call `completion/complete` and return the `completion` object.
    async fn complete(server: &Server, params: Value) -> Value {
        let resp = server
            .handle(make_request("completion/complete", params, 1))
            .await
            .expect("completion/complete is a request, not a notification");
        assert!(
            resp.error.is_none(),
            "expected a result, got error {:?}",
            resp.error
        );
        resp.result.expect("result")["completion"].clone()
    }

    /// Call `completion/complete` and return the JSON-RPC error code.
    async fn complete_err(server: &Server, params: Value) -> i32 {
        let resp = server
            .handle(make_request("completion/complete", params, 1))
            .await
            .expect("response");
        resp.error.expect("expected an error").code
    }

    fn values(completion: &Value) -> Vec<String> {
        completion["values"]
            .as_array()
            .expect("values")
            .iter()
            .map(|v| v.as_str().expect("value is a string").to_string())
            .collect()
    }

    /// The §4 regression: `completion/complete` was `method not found`, so a
    /// host's argument picker had nothing to populate itself from.
    #[tokio::test]
    async fn completion_completes_the_mode_argument() {
        let server = Server::new();

        // No `value` at all is a client asking for the whole list.
        let all = complete(
            &server,
            json!({
                "ref": { "type": "ref/prompt", "name": "car_context" },
                "argument": { "name": "mode" },
            }),
        )
        .await;
        let mut got = values(&all);
        got.sort();
        assert_eq!(got, vec!["fast".to_string(), "full".to_string()]);
        assert_eq!(all["total"], 2);
        assert_eq!(all["hasMore"], false);

        // Two characters, because "f" alone still matches both.
        let narrowed = complete(
            &server,
            json!({
                "ref": { "type": "ref/prompt", "name": "car_context" },
                "argument": { "name": "mode", "value": "fu" },
            }),
        )
        .await;
        assert_eq!(values(&narrowed), vec!["full".to_string()]);
        assert_eq!(narrowed["total"], 1);
        assert_eq!(narrowed["hasMore"], false);
    }

    /// `query` is arbitrary task text. An empty list is the honest answer —
    /// and it is a *result*, not an error.
    #[tokio::test]
    async fn completion_of_a_free_form_argument_is_empty() {
        let server = Server::new();
        let c = complete(
            &server,
            json!({
                "ref": { "type": "ref/prompt", "name": "car_context" },
                "argument": { "name": "query", "value": "how do I" },
            }),
        )
        .await;
        assert_eq!(values(&c), Vec::<String>::new());
        assert_eq!(c["total"], 0);
        assert_eq!(c["hasMore"], false);
    }

    /// Naming something we do not know is a well-formed question with an empty
    /// answer, not a protocol error (car#984's split).
    #[tokio::test]
    async fn completion_of_an_unknown_prompt_is_empty_not_an_error() {
        let server = Server::new();

        let unknown_prompt = complete(
            &server,
            json!({
                "ref": { "type": "ref/prompt", "name": "nope" },
                "argument": { "name": "mode", "value": "f" },
            }),
        )
        .await;
        assert_eq!(values(&unknown_prompt), Vec::<String>::new());
        assert_eq!(unknown_prompt["total"], 0);

        // An unrecognized ref type takes the same path...
        let unknown_ref = complete(
            &server,
            json!({
                "ref": { "type": "ref/something-new", "name": "car_context" },
                "argument": { "name": "mode" },
            }),
        )
        .await;
        assert_eq!(values(&unknown_ref), Vec::<String>::new());

        // ...as does a ref with no `type` at all...
        let no_type = complete(
            &server,
            json!({
                "ref": { "name": "car_context" },
                "argument": { "name": "mode" },
            }),
        )
        .await;
        assert_eq!(values(&no_type), Vec::<String>::new());

        // ...as does an argument name the prompt does not declare.
        let unknown_arg = complete(
            &server,
            json!({
                "ref": { "type": "ref/prompt", "name": "car_context" },
                "argument": { "name": "not_an_argument" },
            }),
        )
        .await;
        assert_eq!(values(&unknown_arg), Vec::<String>::new());
    }

    /// A completed URI must be one `resources/list` would have listed —
    /// that is what [`Server::collect_resources`] being shared buys.
    #[tokio::test]
    async fn completion_completes_resource_uris_by_prefix() {
        let server = Server::new();
        for i in 0..3 {
            add_fact(&server, &format!("subject {}", i), "body", None).await;
        }
        server
            .handle(make_request(
                "tools/call",
                json!({
                    "name": "skill_ingest",
                    "arguments": { "name": "a_skill", "code": "// noop" },
                }),
                1,
            ))
            .await
            .expect("response");

        let listed: Vec<String> = list_resources(&server, None).await.result.expect("result")
            ["resources"]
            .as_array()
            .expect("resources")
            .iter()
            .map(|r| r["uri"].as_str().expect("uri").to_string())
            .collect();
        let listed_facts: Vec<String> = listed
            .iter()
            .filter(|u| u.starts_with("car://memory/fact/"))
            .cloned()
            .collect();
        assert_eq!(listed_facts.len(), 3, "fixture should seed three facts");
        assert!(
            listed.iter().any(|u| u.starts_with("car://memory/skill/")),
            "fixture should seed a skill too, so the prefix has something to exclude"
        );

        let c = complete(
            &server,
            json!({
                "ref": { "type": "ref/resource", "uri": "car://memory/fact/" },
                "argument": { "name": "uri", "value": "" },
            }),
        )
        .await;
        assert_eq!(values(&c), listed_facts);
        assert_eq!(c["total"], 3);
        assert_eq!(c["hasMore"], false);
    }

    /// The cap truncates `values` but never `total` — a client that sees 100
    /// values has to be able to tell whether that is all of them.
    #[tokio::test]
    async fn completion_caps_values_and_reports_total() {
        let server = Server::new();
        let total = 250;
        for i in 0..total {
            add_fact(&server, &format!("subject {:03}", i), "body", None).await;
        }

        let c = complete(
            &server,
            json!({
                "ref": { "type": "ref/resource", "uri": "car://memory/fact/" },
                "argument": { "name": "uri" },
            }),
        )
        .await;
        assert_eq!(values(&c).len(), MAX_COMPLETION_VALUES);
        assert_eq!(c["total"], total);
        assert_eq!(c["hasMore"], true);
    }

    /// A request that never named anything to complete is malformed, and
    /// malformed is `-32602` — the other half of car#984's split.
    #[tokio::test]
    async fn completion_rejects_malformed_params() {
        let server = Server::new();
        let cases = [
            // no `ref`
            json!({ "argument": { "name": "mode" } }),
            // no `argument`
            json!({ "ref": { "type": "ref/prompt", "name": "car_context" } }),
            // `argument` with no `name`
            json!({
                "ref": { "type": "ref/prompt", "name": "car_context" },
                "argument": { "value": "fu" },
            }),
            // `ref` that is not an object
            json!({ "ref": "ref/prompt", "argument": { "name": "mode" } }),
            // `argument` that is not an object
            json!({ "ref": { "type": "ref/prompt", "name": "car_context" }, "argument": "mode" }),
            // nothing at all
            json!({}),
        ];
        for params in cases {
            assert_eq!(
                complete_err(&server, params.clone()).await,
                E_INVALID_PARAMS,
                "{params}"
            );
        }
    }

    #[tokio::test]
    async fn initialize_advertises_the_completions_capability() {
        let server = Server::new();
        let resp = server
            .handle(make_request("initialize", json!({}), 1))
            .await
            .unwrap();
        let caps = resp.result.expect("result")["capabilities"].clone();
        assert!(
            caps.get("completions").is_some(),
            "initialize must advertise `completions`, got {caps}"
        );
    }

    /// The drift guard for [`CONTEXT_MODES`]: every value the completion
    /// offers has to be a value `prompts/get` actually accepts. Without this,
    /// the two lists can diverge and the completion starts handing clients
    /// arguments that come back `-32602`.
    #[tokio::test]
    async fn every_completed_mode_is_accepted_by_prompts_get() {
        let server = Server::new();
        let c = complete(
            &server,
            json!({
                "ref": { "type": "ref/prompt", "name": "car_context" },
                "argument": { "name": "mode" },
            }),
        )
        .await;
        let offered = values(&c);
        assert!(!offered.is_empty(), "the completion must offer something");

        for mode in offered {
            let resp = server
                .handle(make_request(
                    "prompts/get",
                    json!({
                        "name": "car_context",
                        "arguments": { "query": "anything", "mode": mode },
                    }),
                    1,
                ))
                .await
                .expect("response");
            assert!(
                resp.error.is_none(),
                "prompts/get rejected the completed mode {mode:?}: {:?}",
                resp.error
            );
        }
    }
}