gonzalo_mcp/

lib.rs

//! MCP server exposing the gonzalo code graph to agents (EPIC D).
//!
//! [`GonzaloMcp`] implements rmcp's [`ServerHandler`] over a
//! [`Service`](gonzalo_server::Service): agents spawn the `gonzalo-mcp` binary
//! (stdio) and call tools that answer from the local store. It exposes a
//! view-independent `status` and `views` tools plus two families of code-graph
//! query, each taking a `(repo, view_id)` view selector:
//!
//! - **Discovery** — `views` lists the indexed `(repo, view_id)` pairs and
//!   `status` reports how many exist. A selector naming no indexed view is a
//!   tool *error* that lists the real ones, never an empty result: the two are
//!   otherwise indistinguishable, and an agent reads `[]` as "nothing calls
//!   this" rather than "you asked the wrong question" (#210).
//! - **Per-symbol** — `search`/`node`/`callers`/`callees`/`impact`/`explore`
//!   answer questions about a name the caller already has, and `diff` compares
//!   two views.
//! - **Whole-view** — `overview`/`top`/`list`/`unreferenced` answer questions
//!   *about the graph* (what is here, what is heavily referenced, what is
//!   ambiguous, what nothing calls) so a caller can orient without knowing a
//!   symbol name first. `unreferenced` is explicitly heuristic; its tool
//!   description carries the caveats.
//!
//! The tool logic lives in plain methods ([`GonzaloMcp::tools`],
//! [`GonzaloMcp::dispatch`]) so it is unit-testable without an rmcp
//! [`RequestContext`]; the trait methods are thin adapters over them.

use gonzalo_core::{
    Body, DeleteResult, Identity, KeyPrefix, Meta, PutResult, Record, RecordKey, RecordKind,
    Revision,
};
use gonzalo_graph::{Ranking, SymbolFilter, SymbolKind};
use gonzalo_server::Service;
use rmcp::handler::server::ServerHandler;
use rmcp::model::{
    CallToolRequestMethod, CallToolRequestParams, CallToolResult, Content, ListToolsResult,
    PaginatedRequestParams, ServerCapabilities, ServerInfo, Tool,
};
use rmcp::service::{RequestContext, RoleServer};
use serde::Serialize;
use serde_json::{Map, Value};
use std::sync::Arc;

/// Most defining paths listed beside a call-graph answer. Past this the list is
/// noise rather than a lead, and `definition_count` alone carries the warning.
const MAX_DEFINED_IN: usize = 10;

/// An MCP server backed by a gonzalo [`Service`].
#[derive(Clone)]
pub struct GonzaloMcp {
    service: Service,
    root: String,
    /// Whether the record write tools (`record_put`, `record_delete`) are
    /// exposed. Off by default: reading a store is one thing, mutating it is
    /// another, and that decision belongs to whoever runs the server rather
    /// than to the agent talking to it (#197).
    writes: bool,
}

impl GonzaloMcp {
    /// Build a server over `service`, reporting `root` (the store location) in
    /// `status`.
    pub fn new(service: Service, root: impl Into<String>) -> Self {
        Self {
            service,
            root: root.into(),
            writes: false,
        }
    }

    /// Expose the record write tools. The advertised tool list changes with it,
    /// so an agent never sees a tool it would not be allowed to call.
    pub fn with_writes(mut self, writes: bool) -> Self {
        self.writes = writes;
        self
    }

    /// Whether this server exposes the record write tools.
    pub fn writes_enabled(&self) -> bool {
        self.writes
    }

    /// The backing service (used by the D1b graph tools).
    pub fn service(&self) -> &Service {
        &self.service
    }

    /// The tools this server advertises: a view-independent `status`, the
    /// per-symbol code-graph queries, and the whole-view aggregates — all the
    /// graph tools taking a `(repo, view_id)` view selector.
    pub fn tools(writes: bool) -> Vec<Tool> {
        let mut tools = vec![
            Tool::new(
                "status",
                "Report that the gonzalo-mcp server is up and its configured store root.",
                status_schema(),
            ),
            Tool::new(
                "search",
                "Find where a symbol `name` is defined in a view. Returns located definitions.",
                view_query_schema(),
            ),
            Tool::new(
                "node",
                "Inspect a symbol: its definitions, its callers, and its callees in a view.",
                view_query_schema(),
            ),
            Tool::new(
                "callers",
                "The enclosing functions that call `name` in a view, in `callers`. This is a raw \
                 name match, so read `definition_count` with it: above 1 means `name` is defined \
                 in several places and the list merges callers of all of them, which is not the \
                 same question you asked. `defined_in` names those files while there are few \
                 enough to be useful. A count of 0 means the name is defined nowhere in the view, \
                 which is what makes an empty list readable. `callers` is functions only; a call \
                 sitting in a module-level declaration (a Zod schema, a tRPC handler) appears in \
                 `module_callers` instead, so the two questions stay separable.",
                view_query_schema(),
            ),
            Tool::new(
                "callees",
                "The names called from within `name` in a view, in `callees`. Like `callers` this \
                 is a raw name match, so `definition_count` above 1 means several functions share \
                 the name and the list merges what all of them call. `defined_in` names those \
                 files while the count is small.",
                view_query_schema(),
            ),
            Tool::new(
                "impact",
                "Every symbol transitively affected if `name` changes (the caller closure). Only \
                 call edges that resolve to a specific definition are followed, so the walk does \
                 not merge unrelated code that happens to share an identifier; edges that cannot \
                 be attributed are counted in `ambiguous_edges` rather than traversed, and a \
                 non-zero count means the true set may be larger. A function handed to a call \
                 rather than called (`register(handler)`) is a real dependency but too \
                 over-inclusive to walk, so it is counted in `value_edges` instead — non-zero \
                 there means go look. Each result carries the path defining it. `truncated` means \
                 the walk stopped at `max_depth` with frontier left, so completeness is unknown. Note that only calls are edges: `impact` on a \
                 struct, trait, or type is empty because type usage is not recorded, which means \
                 \"not applicable\" rather than \"nothing depends on it\".",
                impact_schema(),
            ),
            Tool::new(
                "explore",
                "List references to `name` in a view (with their paths), for navigating outward.",
                view_query_schema(),
            ),
            Tool::new(
                "diff",
                "Structural diff between two views of a repo: symbols and references added/removed \
                 going from `view_a` to `view_b`.",
                diff_schema(),
            ),
            Tool::new(
                "overview",
                "Summarize a whole view without needing a symbol name: file/symbol/reference \
                 counts, a breakdown by kind and language, and the largest files. Start here when \
                 orienting in an unfamiliar repo.",
                overview_schema(),
            ),
            Tool::new(
                "top",
                "Rank a view's symbols: `fan_in` (most referenced), `fan_out` (calls the most \
                 names), or `definitions` (defined in the most places — a score above 1 means the \
                 name is ambiguous and traversals through it are unreliable).",
                top_schema(),
            ),
            Tool::new(
                "list",
                "Enumerate a view's symbols, optionally filtered by path prefix, kind, and name \
                 substring. Answers \"what is in this crate\" rather than \"where is this name\".",
                list_schema(),
            ),
            Tool::new(
                "views",
                "List every indexed view as (repo, view_id) with its file count and the commit it \
                 was indexed at. Call this first: `repo` and `view_id` must match a view produced \
                 by `gonzalo index`, and this is the only way to discover the valid values. \
                 Compare `base_commit` against the checkout's HEAD to spot a stale view.",
                status_schema(),
            ),
            Tool::new(
                "unreferenced",
                "Symbols with no inbound reference — dead-code CANDIDATES, not dead code. This is \
                 a heuristic over a name-matched graph and it does produce false positives. A \
                 function used only as a value (higher-order usage, e.g. `map_err(be)`) now counts \
                 as referenced and is no longer reported wrongly, but an unused name is still \
                 hidden by any same-named symbol that is used. References from tests and from the \
                 symbol itself do count, so test-only and recursive-only functions are never \
                 reported. Confirm every hit against the source before acting on it.",
                unreferenced_schema(),
            ),
            Tool::new(
                "record_get",
                "Read one record by (namespace, collection, id). Returns the whole record: body, \
                 kind, revision and provenance. A key that holds nothing — or holds a deleted \
                 record — is an error naming the key, never an empty result.",
                record_key_schema(),
            ),
            Tool::new(
                "record_list",
                "List record keys in a namespace, or in one collection of it. Deleted records are \
                 not listed. Returns keys only; read one with `record_get`.",
                record_list_schema(),
            ),
        ];
        if writes {
            tools.push(Tool::new(
                "record_put",
                "Write a record at (namespace, collection, id). Without `expected_revision` this \
                 creates: it succeeds only if the key holds nothing, so it cannot silently \
                 overwrite. To update, pass the `revision` that `record_get` returned. A write the \
                 store refuses because the key moved on comes back as `committed: false` with the \
                 record that is actually there — an outcome to act on, not an error.",
                record_put_schema(),
            ));
            tools.push(Tool::new(
                "record_delete",
                "Delete the record at (namespace, collection, id). The delete replicates: it \
                 leaves a tombstone that sync carries to other stores, so it is not local. \
                 Deleting a key that holds nothing succeeds and changes nothing.",
                record_key_schema(),
            ));
        }
        tools
    }

    /// The `status` payload: server health, the configured store root, and how
    /// many views are indexed.
    ///
    /// The view count is the point (#210): `status` is the tool an agent reaches
    /// for to self-check, and reporting only `ok` meant a server pointed at an
    /// empty or wrong store looked perfectly healthy.
    pub async fn status_json(&self) -> serde_json::Value {
        match self.service.graph_views().await {
            Ok(views) => serde_json::json!({
                "status": "ok",
                "root": self.root,
                "views": views.len(),
            }),
            Err(e) => serde_json::json!({
                "status": "degraded",
                "root": self.root,
                "error": e.to_string(),
            }),
        }
    }

    /// Dispatch a tool call by name, independent of the rmcp transport so it can
    /// be unit-tested. Bad arguments and store errors surface as a tool error
    /// (`CallToolResult::error`); an unknown tool is a `method_not_found` error.
    pub async fn dispatch(
        &self,
        name: &str,
        arguments: Option<Map<String, Value>>,
    ) -> Result<CallToolResult, rmcp::ErrorData> {
        // `status` takes no view selector.
        if name == "status" {
            return Ok(CallToolResult::success(vec![Content::text(
                self.status_json().await.to_string(),
            )]));
        }

        // `views` is the discovery tool — it takes no selector by definition.
        if name == "views" {
            return self.result(self.service.graph_views().await);
        }

        // `diff` selects two views instead of one view + a name.
        if name == "diff" {
            let (repo, view_a, view_b) = match diff_args(&arguments) {
                Ok(t) => t,
                Err(msg) => return Ok(tool_error(msg)),
            };
            for view in [&view_a, &view_b] {
                if let Some(err) = self.unknown_view_error(&repo, view).await {
                    return Ok(err);
                }
            }
            return self.result(self.service.graph_diff(&repo, &view_a, &view_b).await);
        }

        // The aggregate tools select a view but take no symbol name.
        if matches!(name, "overview" | "top" | "list" | "unreferenced") {
            let (repo, view) = match selector_args(&arguments) {
                Ok(t) => t,
                Err(msg) => return Ok(tool_error(msg)),
            };
            if let Some(err) = self.unknown_view_error(&repo, &view).await {
                return Ok(err);
            }
            return self.aggregate(name, &repo, &view, &arguments).await;
        }

        // The record tools address a store key rather than a view (#197). The
        // write pair is dispatched only when this server was started with
        // writes enabled; otherwise it falls through to `method_not_found`,
        // which is what an unadvertised tool should be.
        if matches!(name, "record_get" | "record_list")
            || (self.writes && matches!(name, "record_put" | "record_delete"))
        {
            return self.record_tool(name, &arguments).await;
        }

        // An unknown tool is `method_not_found` regardless of arguments — decided
        // before parsing the view selector so it isn't masked by a missing-arg
        // error.
        if !matches!(
            name,
            "search" | "node" | "callers" | "callees" | "impact" | "explore"
        ) {
            return Err(rmcp::ErrorData::method_not_found::<CallToolRequestMethod>());
        }

        // Every graph tool selects a view by (repo, view_id) and a `name`.
        let (repo, view, sym) = match view_args(&arguments) {
            Ok(t) => t,
            Err(msg) => return Ok(tool_error(msg)),
        };

        // An unresolvable selector is an error, never an empty result (#210):
        // otherwise a typo in `view_id` reads as "nothing calls this".
        if let Some(err) = self.unknown_view_error(&repo, &view).await {
            return Ok(err);
        }

        match name {
            "search" => self.result(self.service.graph_definitions(&repo, &view, &sym).await),
            "callers" | "callees" => self.call_graph(&repo, &view, &sym, name).await,
            "impact" => {
                let max_depth = match usize_arg(&arguments, "max_depth", 0) {
                    Ok(0) => None,
                    Ok(n) => Some(n),
                    Err(msg) => return Ok(tool_error(msg)),
                };
                self.result(
                    self.service
                        .graph_impact(&repo, &view, &sym, max_depth)
                        .await,
                )
            }
            "explore" => self.result(self.service.graph_references_to(&repo, &view, &sym).await),
            "node" => self.node(&repo, &view, &sym).await,
            // Unreachable: the known-tool guard above already returned for any
            // other name.
            _ => Err(rmcp::ErrorData::method_not_found::<CallToolRequestMethod>()),
        }
    }

    /// The record tools: read, list, write and delete records by key (#197).
    ///
    /// These answer from the `Store` rather than the code graph, so they work
    /// against whatever the server was pointed at — a local fs root or a running
    /// daemon.
    async fn record_tool(
        &self,
        name: &str,
        arguments: &Option<Map<String, Value>>,
    ) -> Result<CallToolResult, rmcp::ErrorData> {
        match name {
            "record_list" => {
                let namespace = match str_arg(arguments, "namespace") {
                    Ok(ns) => ns,
                    Err(msg) => return Ok(tool_error(msg)),
                };
                let collection = arguments
                    .as_ref()
                    .and_then(|m| m.get("collection"))
                    .and_then(Value::as_str)
                    .map(str::to_string);
                let prefix = KeyPrefix {
                    namespace: Some(namespace),
                    collection,
                };
                match self.service.list(&prefix).await {
                    // Keys as strings: `namespace/collection/id` is the form the
                    // rest of the surface prints, and it round-trips into the
                    // other record tools' three arguments.
                    Ok(keys) => {
                        let keys: Vec<String> = keys.iter().map(RecordKey::to_string).collect();
                        self.result(Ok(serde_json::json!({ "keys": keys })))
                    }
                    Err(e) => Ok(tool_error(e.to_string())),
                }
            }
            "record_get" => {
                let key = match record_key_args(arguments) {
                    Ok(k) => k,
                    Err(msg) => return Ok(tool_error(msg)),
                };
                match self.service.get(&key).await {
                    // Absent is an error, not an empty answer, for the same
                    // reason an unknown view is (#210): an agent reads nothing
                    // as "there is nothing there" rather than "you asked wrong".
                    Ok(None) => Ok(tool_error(format!("no record at {key}"))),
                    Ok(Some(record)) => self.result(Ok(record)),
                    Err(e) => Ok(tool_error(e.to_string())),
                }
            }
            "record_put" => {
                let key = match record_key_args(arguments) {
                    Ok(k) => k,
                    Err(msg) => return Ok(tool_error(msg)),
                };
                let body = match str_arg(arguments, "body") {
                    Ok(b) => b,
                    Err(msg) => return Ok(tool_error(msg)),
                };
                let kind = match kind_arg(arguments) {
                    Ok(k) => k,
                    Err(msg) => return Ok(tool_error(msg)),
                };
                let expected = match expected_revision_arg(arguments) {
                    Ok(rev) => rev,
                    Err(msg) => return Ok(tool_error(msg)),
                };
                let meta = Meta::new(Identity::new("gonzalo-mcp"), "mcp");
                let record = match &expected {
                    // An update follows the revision the caller read, so the
                    // parent chain stays intact; a create starts a new one.
                    Some(rev) => Record {
                        parent: Some(rev.clone()),
                        ..Record::create(key.clone(), kind, Body::Inline(body.into_bytes()), meta)
                    },
                    None => {
                        Record::create(key.clone(), kind, Body::Inline(body.into_bytes()), meta)
                    }
                };
                match self.service.put(record, expected).await {
                    // A conflict is a normal OCC outcome, so it is data the
                    // agent can act on — the current record comes back with it.
                    Ok(PutResult::Committed(revision)) => self.result(Ok(serde_json::json!({
                        "committed": true,
                        "key": key.to_string(),
                        "revision": revision,
                    }))),
                    Ok(PutResult::Conflict(conflict)) => self.result(Ok(serde_json::json!({
                        "committed": false,
                        "key": key.to_string(),
                        "reason": "the key moved on since the revision this write expected",
                        "current": conflict.current,
                    }))),
                    Err(e) => Ok(tool_error(e.to_string())),
                }
            }
            "record_delete" => {
                let key = match record_key_args(arguments) {
                    Ok(k) => k,
                    Err(msg) => return Ok(tool_error(msg)),
                };
                match self.service.delete_as(&key, None, None).await {
                    Ok(DeleteResult::Deleted) => self.result(Ok(serde_json::json!({
                        "deleted": true,
                        "key": key.to_string(),
                    }))),
                    Ok(DeleteResult::Conflict(conflict)) => self.result(Ok(serde_json::json!({
                        "deleted": false,
                        "key": key.to_string(),
                        "current": conflict.current,
                    }))),
                    Err(e) => Ok(tool_error(e.to_string())),
                }
            }
            // Unreachable: the caller matched the name before dispatching here.
            _ => Err(rmcp::ErrorData::method_not_found::<CallToolRequestMethod>()),
        }
    }

    /// A tool error when `(repo, view_id)` names no indexed view, or `None` when
    /// the selector resolves.
    ///
    /// The message names the unresolved selector *and* lists the views that do
    /// exist, so a caller that guessed wrong can correct itself in one round
    /// trip instead of concluding the code is not there (#210).
    async fn unknown_view_error(&self, repo: &str, view: &str) -> Option<CallToolResult> {
        match self.service.view_exists(repo, view).await {
            Ok(true) => None,
            Ok(false) => {
                let known = match self.service.graph_views().await {
                    Ok(views) if !views.is_empty() => views
                        .iter()
                        .map(|v| format!("{}/{}", v.repo, v.view_id))
                        .collect::<Vec<_>>()
                        .join(", "),
                    Ok(_) => "none — run `gonzalo index` first".to_string(),
                    Err(e) => format!("<could not list views: {e}>"),
                };
                Some(tool_error(format!(
                    "no indexed view '{repo}/{view}'. This is a selector error, not an empty \
                     result. Indexed views: {known}. Call `views` to list them."
                )))
            }
            // A store failure is reported as itself rather than as "no view".
            Err(e) => Some(tool_error(e.to_string())),
        }
    }

    /// Turn a service query outcome into a tool result: JSON on success, a tool
    /// error carrying the message on failure.
    fn result<T: Serialize>(
        &self,
        outcome: gonzalo_core::Result<T>,
    ) -> Result<CallToolResult, rmcp::ErrorData> {
        match outcome {
            Ok(value) => Ok(success_json(&value)),
            Err(e) => Ok(tool_error(e.to_string())),
        }
    }

    /// Dispatch the view-wide aggregate tools, which take a `(repo, view_id)`
    /// selector plus their own optional arguments rather than a symbol name.
    async fn aggregate(
        &self,
        name: &str,
        repo: &str,
        view: &str,
        arguments: &Option<Map<String, Value>>,
    ) -> Result<CallToolResult, rmcp::ErrorData> {
        match name {
            "overview" => {
                let largest = match usize_arg(arguments, "largest", DEFAULT_TOP_LIMIT) {
                    Ok(n) => n,
                    Err(msg) => return Ok(tool_error(msg)),
                };
                self.result(self.service.graph_overview(repo, view, largest).await)
            }
            "top" => {
                let ranking = match ranking_arg(arguments) {
                    Ok(r) => r,
                    Err(msg) => return Ok(tool_error(msg)),
                };
                let limit = match usize_arg(arguments, "limit", DEFAULT_TOP_LIMIT) {
                    Ok(n) => n,
                    Err(msg) => return Ok(tool_error(msg)),
                };
                self.result(self.service.graph_top(repo, view, ranking, limit).await)
            }
            "list" => {
                let filter = match filter_args(arguments) {
                    Ok(f) => f,
                    Err(msg) => return Ok(tool_error(msg)),
                };
                let limit = match usize_arg(arguments, "limit", DEFAULT_LIST_LIMIT) {
                    Ok(n) => n,
                    Err(msg) => return Ok(tool_error(msg)),
                };
                self.result(self.service.graph_list(repo, view, &filter, limit).await)
            }
            "unreferenced" => {
                let filter = match filter_args(arguments) {
                    Ok(f) => f,
                    Err(msg) => return Ok(tool_error(msg)),
                };
                let exclude_tests = match bool_arg(arguments, "exclude_tests", true) {
                    Ok(b) => b,
                    Err(msg) => return Ok(tool_error(msg)),
                };
                let limit = match usize_arg(arguments, "limit", DEFAULT_LIST_LIMIT) {
                    Ok(n) => n,
                    Err(msg) => return Ok(tool_error(msg)),
                };
                self.result(
                    self.service
                        .graph_unreferenced(repo, view, &filter, exclude_tests, limit)
                        .await,
                )
            }
            // Unreachable: the caller already matched on these four names.
            _ => Err(rmcp::ErrorData::method_not_found::<CallToolRequestMethod>()),
        }
    }

    /// `callers`/`callees`, answered with how many definitions the queried name
    /// has.
    ///
    /// These are raw name matches: a name defined in several places returns the
    /// merged answer for all of them, and a bare list says nothing about that.
    /// It is the point where a caller is most likely to read a heuristic result
    /// as a precise one, so the count travels with the answer (#249). The
    /// defining paths come too while there are few enough to be a lead rather
    /// than noise.
    async fn call_graph(
        &self,
        repo: &str,
        view: &str,
        sym: &str,
        tool: &str,
    ) -> Result<CallToolResult, rmcp::ErrorData> {
        // `callers` splits by what the caller *is*: a function, or a
        // module-level binding such as a Zod schema or a tRPC handler. Kept as
        // two fields rather than one list of objects so `callers` means exactly
        // what it meant before — functions that call this (#268).
        let (names, module_names) = if tool == "callers" {
            match self.service.graph_callers_scoped(repo, view, sym).await {
                Ok(scoped) => {
                    let (functions, modules): (Vec<_>, Vec<_>) = scoped
                        .into_iter()
                        .partition(|(_, scope)| scope.is_function());
                    (
                        functions.into_iter().map(|(name, _)| name).collect(),
                        Some(
                            modules
                                .into_iter()
                                .map(|(name, _)| name)
                                .collect::<Vec<_>>(),
                        ),
                    )
                }
                Err(e) => return Ok(tool_error(e.to_string())),
            }
        } else {
            match self.service.graph_callees(repo, view, sym).await {
                Ok(names) => (names, None),
                Err(e) => return Ok(tool_error(e.to_string())),
            }
        };
        let defs = match self.service.graph_definitions(repo, view, sym).await {
            Ok(d) => d,
            Err(e) => return Ok(tool_error(e.to_string())),
        };

        let mut paths: Vec<&str> = defs.iter().map(|d| d.path.as_str()).collect();
        paths.sort_unstable();
        paths.dedup();

        let mut payload = serde_json::json!({
            tool: names,
            "definition_count": defs.len(),
        });
        if let Some(module_names) = module_names {
            payload["module_callers"] = serde_json::json!(module_names);
        }
        if paths.len() <= MAX_DEFINED_IN {
            payload["defined_in"] = serde_json::json!(paths);
        }
        Ok(success_json(&payload))
    }

    /// The `node` aggregate: definitions + callers + callees for one symbol.
    async fn node(
        &self,
        repo: &str,
        view: &str,
        sym: &str,
    ) -> Result<CallToolResult, rmcp::ErrorData> {
        let defs = match self.service.graph_definitions(repo, view, sym).await {
            Ok(d) => d,
            Err(e) => return Ok(tool_error(e.to_string())),
        };
        let callers = match self.service.graph_callers_of(repo, view, sym).await {
            Ok(c) => c,
            Err(e) => return Ok(tool_error(e.to_string())),
        };
        let callees = match self.service.graph_callees(repo, view, sym).await {
            Ok(c) => c,
            Err(e) => return Ok(tool_error(e.to_string())),
        };
        let payload = serde_json::json!({
            "definitions": defs,
            "callers": callers,
            "callees": callees,
        });
        Ok(success_json(&payload))
    }
}

/// Input schema for the view queries: `repo`, `view_id`, and `name` — all
/// required strings.
fn view_query_schema() -> Arc<Map<String, Value>> {
    let schema = serde_json::json!({
        "type": "object",
        "properties": {
            "repo": {
                "type": "string",
                "description": "repository of an INDEXED view, e.g. acme/widgets — must match one \
                                reported by `views`, not an arbitrary name"
            },
            "view_id": {
                "type": "string",
                "description": "view id of an INDEXED view, e.g. main — must match one reported \
                                by `views`"
            },
            "name": { "type": "string", "description": "symbol name to query" }
        },
        "required": ["repo", "view_id", "name"],
        "additionalProperties": false
    });
    Arc::new(schema.as_object().expect("object schema").clone())
}

/// Input schema for `diff`: `repo`, `view_a`, and `view_b` — all required.
fn diff_schema() -> Arc<Map<String, Value>> {
    let schema = serde_json::json!({
        "type": "object",
        "properties": {
            "repo": {
                "type": "string",
                "description": "repository of two INDEXED views — must match `views` output"
            },
            "view_a": { "type": "string", "description": "the base view id (must be indexed)" },
            "view_b": {
                "type": "string",
                "description": "the view id to compare against the base (must be indexed)"
            }
        },
        "required": ["repo", "view_a", "view_b"],
        "additionalProperties": false
    });
    Arc::new(schema.as_object().expect("object schema").clone())
}

/// Default number of entries returned by `overview.largest_files` and `top`.
const DEFAULT_TOP_LIMIT: usize = 20;
/// Default number of symbols returned by `list`.
const DEFAULT_LIST_LIMIT: usize = 100;

/// Input schema for `overview`: a view selector plus an optional cap on the
/// `largest_files` listing.
fn overview_schema() -> Arc<Map<String, Value>> {
    let schema = serde_json::json!({
        "type": "object",
        "properties": {
            "repo": {
                "type": "string",
                "description": "repository of an INDEXED view, e.g. acme/widgets — must match one \
                                reported by `views`, not an arbitrary name"
            },
            "view_id": {
                "type": "string",
                "description": "view id of an INDEXED view, e.g. main — must match one reported \
                                by `views`"
            },
            "largest": {
                "type": "integer",
                "minimum": 0,
                "description": "how many of the largest files to list (default 20); the counts \
                                themselves are never truncated"
            }
        },
        "required": ["repo", "view_id"],
        "additionalProperties": false
    });
    Arc::new(schema.as_object().expect("object schema").clone())
}

/// Input schema for `top`: a view selector, the required ranking, and a limit.
fn top_schema() -> Arc<Map<String, Value>> {
    let schema = serde_json::json!({
        "type": "object",
        "properties": {
            "repo": {
                "type": "string",
                "description": "repository of an INDEXED view, e.g. acme/widgets — must match one \
                                reported by `views`, not an arbitrary name"
            },
            "view_id": {
                "type": "string",
                "description": "view id of an INDEXED view, e.g. main — must match one reported \
                                by `views`"
            },
            "by": {
                "type": "string",
                "enum": ["fan_in", "fan_out", "definitions"],
                "description": "fan_in = most referenced; fan_out = calls the most distinct \
                                names; definitions = defined in the most paths (>1 is ambiguous)"
            },
            "limit": {
                "type": "integer",
                "minimum": 0,
                "description": "maximum entries to return (default 20)"
            }
        },
        "required": ["repo", "view_id", "by"],
        "additionalProperties": false
    });
    Arc::new(schema.as_object().expect("object schema").clone())
}

/// Input schema for `list`: a view selector plus conjunctive filters.
fn list_schema() -> Arc<Map<String, Value>> {
    let schema = serde_json::json!({
        "type": "object",
        "properties": {
            "repo": {
                "type": "string",
                "description": "repository of an INDEXED view, e.g. acme/widgets — must match one \
                                reported by `views`, not an arbitrary name"
            },
            "view_id": {
                "type": "string",
                "description": "view id of an INDEXED view, e.g. main — must match one reported \
                                by `views`"
            },
            "path_prefix": {
                "type": "string",
                "description": "only symbols whose path starts with this (scopes to a crate or \
                                directory)"
            },
            "kind": {
                "type": "string",
                "enum": [
                    "function", "struct", "enum", "trait", "impl", "module",
                    "const", "static", "type_alias", "class", "interface"
                ],
                "description": "only symbols of this kind"
            },
            "name_contains": {
                "type": "string",
                "description": "only symbols whose name contains this substring"
            },
            "limit": {
                "type": "integer",
                "minimum": 0,
                "description": "maximum symbols to return (default 100)"
            }
        },
        "required": ["repo", "view_id"],
        "additionalProperties": false
    });
    Arc::new(schema.as_object().expect("object schema").clone())
}

/// Input schema for `unreferenced`: the `list` filters plus the test-scope
/// toggle.
fn unreferenced_schema() -> Arc<Map<String, Value>> {
    let schema = serde_json::json!({
        "type": "object",
        "properties": {
            "repo": {
                "type": "string",
                "description": "repository of an INDEXED view, e.g. acme/widgets — must match one \
                                reported by `views`, not an arbitrary name"
            },
            "view_id": {
                "type": "string",
                "description": "view id of an INDEXED view, e.g. main — must match one reported \
                                by `views`"
            },
            "path_prefix": {
                "type": "string",
                "description": "only symbols whose path starts with this (scopes to a crate or \
                                directory)"
            },
            "kind": {
                "type": "string",
                "enum": [
                    "function", "struct", "enum", "trait", "impl", "module",
                    "const", "static", "type_alias", "class", "interface"
                ],
                "description": "only symbols of this kind; `function` is usually what you want"
            },
            "name_contains": {
                "type": "string",
                "description": "only symbols whose name contains this substring"
            },
            "exclude_tests": {
                "type": "boolean",
                "description": "drop symbols inside a `mod tests`/`mod test` block or under a \
                                tests/ directory (default true); without it the result is mostly \
                                test helpers"
            },
            "limit": {
                "type": "integer",
                "minimum": 0,
                "description": "maximum candidates to return (default 100)"
            }
        },
        "required": ["repo", "view_id"],
        "additionalProperties": false
    });
    Arc::new(schema.as_object().expect("object schema").clone())
}

/// Input schema for `impact`: the per-symbol selector plus an optional depth cap.
fn impact_schema() -> Arc<Map<String, Value>> {
    let schema = serde_json::json!({
        "type": "object",
        "properties": {
            "repo": {
                "type": "string",
                "description": "repository of an INDEXED view, e.g. acme/widgets — must match one \
                                reported by `views`, not an arbitrary name"
            },
            "view_id": {
                "type": "string",
                "description": "view id of an INDEXED view, e.g. main — must match one reported \
                                by `views`"
            },
            "name": { "type": "string", "description": "symbol name to seed the closure from" },
            "max_depth": {
                "type": "integer",
                "minimum": 0,
                "description": "stop after this many hops (0 or omitted = unbounded); the result \
                                reports `truncated` when a cap stopped it early"
            }
        },
        "required": ["repo", "view_id", "name"],
        "additionalProperties": false
    });
    Arc::new(schema.as_object().expect("object schema").clone())
}

/// A minimal object schema for the argument-free `status` tool.
fn status_schema() -> Arc<Map<String, Value>> {
    let schema = serde_json::json!({ "type": "object", "additionalProperties": false });
    Arc::new(schema.as_object().expect("object schema").clone())
}

/// Input schema for the tools that address one record: the three key parts.
fn record_key_schema() -> Arc<Map<String, Value>> {
    let schema = serde_json::json!({
        "type": "object",
        "properties": {
            "namespace": { "type": "string", "description": "record namespace, e.g. memory" },
            "collection": { "type": "string", "description": "collection within the namespace, e.g. topics" },
            "id": { "type": "string", "description": "record id within the collection" }
        },
        "required": ["namespace", "collection", "id"],
        "additionalProperties": false
    });
    Arc::new(schema.as_object().expect("object schema").clone())
}

/// Input schema for `record_list`: a namespace, optionally narrowed to one
/// collection.
fn record_list_schema() -> Arc<Map<String, Value>> {
    let schema = serde_json::json!({
        "type": "object",
        "properties": {
            "namespace": { "type": "string", "description": "record namespace to list" },
            "collection": {
                "type": "string",
                "description": "optional collection to narrow the listing to"
            }
        },
        "required": ["namespace"],
        "additionalProperties": false
    });
    Arc::new(schema.as_object().expect("object schema").clone())
}

/// Input schema for `record_put`: the key, the body, the kind, and the
/// optimistic-concurrency revision.
fn record_put_schema() -> Arc<Map<String, Value>> {
    let schema = serde_json::json!({
        "type": "object",
        "properties": {
            "namespace": { "type": "string", "description": "record namespace" },
            "collection": { "type": "string", "description": "collection within the namespace" },
            "id": { "type": "string", "description": "record id within the collection" },
            "body": { "type": "string", "description": "the record body, stored verbatim" },
            "kind": {
                "type": "string",
                "description": "record kind, exactly as `record_get` reports it. Defaults to \
                                Topic. `Tombstone` is not writable — delete with `record_delete`.",
                "enum": [
                    "MemoryTier", "Topic", "Session", "Checkpoint", "Ticket", "TicketEvent",
                    "GraphManifest", "Person", "IdentityBinding", "RoleGrant", "ChannelConfig",
                    "LinkToken", "AuditEntry"
                ]
            },
            "expected_revision": {
                "type": "object",
                "description": "the revision this write expects to replace — pass back the \
                                `revision` object exactly as `record_get` returned it. Omit to \
                                create a record that does not exist yet."
            }
        },
        "required": ["namespace", "collection", "id", "body"],
        "additionalProperties": false
    });
    Arc::new(schema.as_object().expect("object schema").clone())
}

/// The `(namespace, collection, id)` a record tool addresses.
fn record_key_args(arguments: &Option<Map<String, Value>>) -> Result<RecordKey, String> {
    Ok(RecordKey::new(
        str_arg(arguments, "namespace")?,
        str_arg(arguments, "collection")?,
        str_arg(arguments, "id")?,
    ))
}

/// The `kind` a `record_put` writes, defaulting to `topic`.
///
/// An unknown kind is refused rather than defaulted: silently storing a record
/// under the wrong kind would change how it merges during sync, which is not
/// something to guess at on the caller's behalf.
fn kind_arg(arguments: &Option<Map<String, Value>>) -> Result<RecordKind, String> {
    let Some(raw) = arguments
        .as_ref()
        .and_then(|m| m.get("kind"))
        .and_then(Value::as_str)
    else {
        return Ok(RecordKind::Topic);
    };
    serde_json::from_value(Value::String(raw.to_string()))
        .map_err(|_| format!("unknown record kind '{raw}'"))
}

/// The optional `expected_revision` a `record_put` carries, as `record_get`
/// reported it. Absent means "create": the write applies only to an empty key.
fn expected_revision_arg(
    arguments: &Option<Map<String, Value>>,
) -> Result<Option<Revision>, String> {
    let Some(raw) = arguments.as_ref().and_then(|m| m.get("expected_revision")) else {
        return Ok(None);
    };
    if raw.is_null() {
        return Ok(None);
    }
    serde_json::from_value(raw.clone())
        .map(Some)
        .map_err(|e| format!("expected_revision is not a revision: {e}"))
}

/// Extract a required string argument by key.
fn str_arg(arguments: &Option<Map<String, Value>>, key: &str) -> Result<String, String> {
    arguments
        .as_ref()
        .and_then(|m| m.get(key))
        .and_then(Value::as_str)
        .map(str::to_string)
        .ok_or_else(|| format!("missing required string argument '{key}'"))
}

/// Extract the required `(repo, view_id, name)` view selector from tool args.
fn view_args(arguments: &Option<Map<String, Value>>) -> Result<(String, String, String), String> {
    Ok((
        str_arg(arguments, "repo")?,
        str_arg(arguments, "view_id")?,
        str_arg(arguments, "name")?,
    ))
}

/// Extract the required `(repo, view_id)` selector, for tools that address a
/// whole view rather than a symbol in it.
fn selector_args(arguments: &Option<Map<String, Value>>) -> Result<(String, String), String> {
    Ok((str_arg(arguments, "repo")?, str_arg(arguments, "view_id")?))
}

/// Extract an optional non-negative integer argument, or `default` if absent.
fn usize_arg(
    arguments: &Option<Map<String, Value>>,
    key: &str,
    default: usize,
) -> Result<usize, String> {
    match arguments.as_ref().and_then(|m| m.get(key)) {
        None | Some(Value::Null) => Ok(default),
        Some(v) => v
            .as_u64()
            .map(|n| n as usize)
            .ok_or_else(|| format!("argument '{key}' must be a non-negative integer, got {v}")),
    }
}

/// Extract an optional boolean argument, or `default` if absent.
fn bool_arg(
    arguments: &Option<Map<String, Value>>,
    key: &str,
    default: bool,
) -> Result<bool, String> {
    match arguments.as_ref().and_then(|m| m.get(key)) {
        None | Some(Value::Null) => Ok(default),
        Some(Value::Bool(b)) => Ok(*b),
        Some(v) => Err(format!("argument '{key}' must be a boolean, got {v}")),
    }
}

/// Extract the required `by` ranking for `top`.
fn ranking_arg(arguments: &Option<Map<String, Value>>) -> Result<Ranking, String> {
    let raw = str_arg(arguments, "by")?;
    match raw.as_str() {
        "fan_in" => Ok(Ranking::FanIn),
        "fan_out" => Ok(Ranking::FanOut),
        "definitions" => Ok(Ranking::Definitions),
        other => Err(format!(
            "unknown ranking '{other}': expected one of fan_in, fan_out, definitions"
        )),
    }
}

/// Build the `list` filter from its optional arguments.
fn filter_args(arguments: &Option<Map<String, Value>>) -> Result<SymbolFilter, String> {
    let mut filter = SymbolFilter::default();
    if let Some(prefix) = opt_str_arg(arguments, "path_prefix")? {
        filter = filter.path_prefix(prefix);
    }
    if let Some(needle) = opt_str_arg(arguments, "name_contains")? {
        filter = filter.name_contains(needle);
    }
    if let Some(raw) = opt_str_arg(arguments, "kind")? {
        filter = filter.kind(parse_kind(&raw)?);
    }
    Ok(filter)
}

/// Extract an optional string argument, erroring if present but not a string.
fn opt_str_arg(
    arguments: &Option<Map<String, Value>>,
    key: &str,
) -> Result<Option<String>, String> {
    match arguments.as_ref().and_then(|m| m.get(key)) {
        None | Some(Value::Null) => Ok(None),
        Some(Value::String(s)) => Ok(Some(s.clone())),
        Some(v) => Err(format!("argument '{key}' must be a string, got {v}")),
    }
}

/// Parse a [`SymbolKind`] from its lowercase wire name.
fn parse_kind(raw: &str) -> Result<SymbolKind, String> {
    [
        SymbolKind::Function,
        SymbolKind::Struct,
        SymbolKind::Enum,
        SymbolKind::Trait,
        SymbolKind::Impl,
        SymbolKind::Module,
        SymbolKind::Const,
        SymbolKind::Static,
        SymbolKind::TypeAlias,
        SymbolKind::Class,
        SymbolKind::Interface,
    ]
    .into_iter()
    .find(|k| k.as_str() == raw)
    .ok_or_else(|| format!("unknown kind '{raw}'"))
}

/// Extract the required `(repo, view_a, view_b)` selector for `diff`.
fn diff_args(arguments: &Option<Map<String, Value>>) -> Result<(String, String, String), String> {
    Ok((
        str_arg(arguments, "repo")?,
        str_arg(arguments, "view_a")?,
        str_arg(arguments, "view_b")?,
    ))
}

/// A tool error carrying `msg` as text (`isError = true`).
fn tool_error(msg: impl Into<String>) -> CallToolResult {
    CallToolResult::error(vec![Content::text(msg.into())])
}

/// A successful tool result whose single text block is `value` as JSON.
fn success_json<T: Serialize>(value: &T) -> CallToolResult {
    match serde_json::to_string(value) {
        Ok(json) => CallToolResult::success(vec![Content::text(json)]),
        Err(e) => tool_error(format!("failed to serialize result: {e}")),
    }
}

impl ServerHandler for GonzaloMcp {
    fn get_info(&self) -> ServerInfo {
        ServerInfo::new(ServerCapabilities::builder().enable_tools().build())
            .with_instructions("gonzalo-mcp: code-graph queries over a gonzalo view")
    }

    async fn list_tools(
        &self,
        _request: Option<PaginatedRequestParams>,
        _context: RequestContext<RoleServer>,
    ) -> Result<ListToolsResult, rmcp::ErrorData> {
        Ok(ListToolsResult::with_all_items(Self::tools(self.writes)))
    }

    async fn call_tool(
        &self,
        request: CallToolRequestParams,
        _context: RequestContext<RoleServer>,
    ) -> Result<CallToolResult, rmcp::ErrorData> {
        self.dispatch(request.name.as_ref(), request.arguments)
            .await
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use gonzalo_core::{
        BlobStore, Identity, Manifest, Meta, PutResult, Record, RecordKind, Revision, Store,
    };
    use gonzalo_graph::build_rust;
    use gonzalo_store_fs::FsStore;
    use std::collections::BTreeMap;

    fn server(root: &str) -> GonzaloMcp {
        let fs = Arc::new(FsStore::new(tempfile::tempdir().unwrap().keep()));
        GonzaloMcp::new(Service::new(fs.clone(), fs), root)
    }

    /// A server whose store already holds view `r`/`main` with two slices.
    async fn seeded_server() -> GonzaloMcp {
        seeded_with(&[
            ("lib.rs", "fn helper() {}"),
            ("main.rs", "fn main() { helper(); }"),
        ])
        .await
    }

    /// A server whose store holds view `r`/`main` assembled from `slices`.
    /// Like [`seeded_with`], for a language other than Rust.
    async fn seeded_with_language(
        language: gonzalo_graph::Language,
        slices: &[(&str, &str)],
    ) -> GonzaloMcp {
        seeded_from(slices, |src| gonzalo_graph::build(language, src)).await
    }

    async fn seeded_with(slices: &[(&str, &str)]) -> GonzaloMcp {
        seeded_from(slices, build_rust).await
    }

    async fn seeded_from(
        slices: &[(&str, &str)],
        parse: impl Fn(&str) -> gonzalo_graph::CodeGraph,
    ) -> GonzaloMcp {
        let fs = Arc::new(FsStore::new(tempfile::tempdir().unwrap().keep()));
        let mut manifest = Manifest::new();
        for &(path, src) in slices {
            let hash = fs.put_blob(&parse(src).to_slice_bytes()).await.unwrap();
            manifest.insert(path, hash);
        }
        let body = manifest.to_body();
        let record = Record {
            revision: Revision::initial(body.bytes()),
            parent: None,
            body,
            kind: RecordKind::GraphManifest,
            meta: Meta {
                author: Identity::new("tester"),
                origin_system: "test".into(),
                created: 0,
                updated: 0,
                labels: BTreeMap::new(),
            },
            links: Vec::new(),
            key: Manifest::key("r", "main"),
            ancestors: Vec::new(),
            deleted_at: None,
            deleted_blob: None,
            deleted_kind: None,
        };
        assert!(matches!(
            fs.put(record, None).await.unwrap(),
            PutResult::Committed(_)
        ));
        GonzaloMcp::new(Service::new(fs.clone(), fs), "test-root")
    }

    fn args(repo: &str, view: &str, name: &str) -> Option<Map<String, Value>> {
        Some(
            serde_json::json!({ "repo": repo, "view_id": view, "name": name })
                .as_object()
                .unwrap()
                .clone(),
        )
    }

    /// The text of a result's first content block (via the serialized result).
    fn result_text(result: &CallToolResult) -> String {
        let v = serde_json::to_value(result).unwrap();
        v["content"][0]["text"]
            .as_str()
            .expect("text content")
            .to_string()
    }

    #[test]
    fn advertises_status_and_the_graph_tools() {
        let names: Vec<String> = GonzaloMcp::tools(false)
            .iter()
            .map(|t| t.name.to_string())
            .collect();
        assert_eq!(
            names,
            vec![
                "status",
                "search",
                "node",
                "callers",
                "callees",
                "impact",
                "explore",
                "diff",
                "overview",
                "top",
                "list",
                "views",
                "unreferenced",
                "record_get",
                "record_list",
            ]
        );
    }

    // ---- aggregate / structural tools (#214) ------------------------------

    /// `(repo, view_id)` plus whatever extra arguments a tool takes.
    fn view_args_with(repo: &str, view: &str, extra: Value) -> Option<Map<String, Value>> {
        let mut map = serde_json::json!({ "repo": repo, "view_id": view })
            .as_object()
            .unwrap()
            .clone();
        for (k, v) in extra.as_object().expect("object") {
            map.insert(k.clone(), v.clone());
        }
        Some(map)
    }

    async fn call(tool: &str, extra: Value) -> Value {
        let s = seeded_server().await;
        let result = s
            .dispatch(tool, view_args_with("r", "main", extra))
            .await
            .unwrap();
        assert_ne!(result.is_error, Some(true), "tool {tool} errored");
        serde_json::from_str(&result_text(&result)).unwrap()
    }

    #[tokio::test]
    async fn overview_tool_reports_the_shape_of_the_view() {
        // The seeded view is lib.rs (`helper`) + main.rs (`main` calling helper).
        let v = call("overview", serde_json::json!({})).await;
        assert_eq!(v["files"], 2);
        assert_eq!(v["symbols"], 2);
        assert_eq!(v["references"], 1);
        assert_eq!(v["by_kind"]["function"], 2);
        assert_eq!(v["by_language"]["rust"], 2);
        assert_eq!(v["largest_files"].as_array().unwrap().len(), 2);
    }

    #[tokio::test]
    async fn overview_bounds_the_largest_files_listing() {
        let v = call("overview", serde_json::json!({ "largest": 1 })).await;
        assert_eq!(v["largest_files"].as_array().unwrap().len(), 1);
        assert_eq!(v["files"], 2, "the count itself is not truncated");
    }

    #[tokio::test]
    async fn top_tool_ranks_by_fan_in() {
        let v = call("top", serde_json::json!({ "by": "fan_in" })).await;
        let helper = v["items"]
            .as_array()
            .unwrap()
            .iter()
            .find(|r| r["name"] == "helper")
            .expect("helper is referenced once");
        assert_eq!(helper["score"], 1);
        assert_eq!(helper["paths"], serde_json::json!(["lib.rs"]));
    }

    #[tokio::test]
    async fn top_tool_ranks_by_definitions_for_the_ambiguity_report() {
        let v = call("top", serde_json::json!({ "by": "definitions" })).await;
        // Nothing is ambiguous in this view: every name has exactly one home.
        assert!(
            v["items"]
                .as_array()
                .unwrap()
                .iter()
                .all(|r| r["score"] == 1)
        );
    }

    #[tokio::test]
    async fn top_tool_reports_truncation() {
        // Two names are defined in the seeded view (`helper`, `main`), so a
        // limit of 1 must report that something was left out.
        let v = call(
            "top",
            serde_json::json!({ "by": "definitions", "limit": 1 }),
        )
        .await;
        assert_eq!(v["items"].as_array().unwrap().len(), 1);
        assert_eq!(v["total"], 2);
        assert_eq!(v["truncated"], true);
    }

    #[tokio::test]
    async fn top_tool_rejects_an_unknown_ranking() {
        let s = seeded_server().await;
        let result = s
            .dispatch(
                "top",
                view_args_with("r", "main", serde_json::json!({ "by": "sideways" })),
            )
            .await
            .unwrap();
        assert_eq!(result.is_error, Some(true));
        assert!(result_text(&result).contains("sideways"));
    }

    #[tokio::test]
    async fn top_tool_requires_the_ranking_argument() {
        let s = seeded_server().await;
        let result = s
            .dispatch("top", view_args_with("r", "main", serde_json::json!({})))
            .await
            .unwrap();
        assert_eq!(result.is_error, Some(true));
        assert!(result_text(&result).contains("by"));
    }

    #[tokio::test]
    async fn list_tool_filters_by_kind() {
        let v = call("list", serde_json::json!({ "kind": "function" })).await;
        assert_eq!(v["total"], 2);
        assert!(
            v["items"]
                .as_array()
                .unwrap()
                .iter()
                .all(|l| l["item"]["kind"] == "function")
        );
    }

    #[tokio::test]
    async fn list_tool_filters_by_path_prefix() {
        let v = call("list", serde_json::json!({ "path_prefix": "lib" })).await;
        assert_eq!(v["total"], 1);
        assert_eq!(v["items"][0]["path"], "lib.rs");
    }

    #[tokio::test]
    async fn list_tool_reports_truncation() {
        let v = call("list", serde_json::json!({ "limit": 1 })).await;
        assert_eq!(v["items"].as_array().unwrap().len(), 1);
        assert_eq!(v["truncated"], true);
        assert_eq!(v["total"], 2);
    }

    #[tokio::test]
    async fn list_tool_rejects_an_unknown_kind() {
        let s = seeded_server().await;
        let result = s
            .dispatch(
                "list",
                view_args_with("r", "main", serde_json::json!({ "kind": "gizmo" })),
            )
            .await
            .unwrap();
        assert_eq!(result.is_error, Some(true));
        assert!(result_text(&result).contains("gizmo"));
    }

    // ---- unknown view vs empty result (#210) -------------------------------

    /// Every tool that takes a view selector, with valid arguments except the
    /// view id.
    const VIEW_SCOPED: &[&str] = &[
        "search",
        "node",
        "callers",
        "callees",
        "impact",
        "explore",
        "overview",
        "top",
        "list",
        "unreferenced",
    ];

    fn args_for(tool: &str, repo: &str, view: &str) -> Option<Map<String, Value>> {
        let mut m = serde_json::json!({ "repo": repo, "view_id": view })
            .as_object()
            .unwrap()
            .clone();
        // Per-tool required extras.
        if matches!(
            tool,
            "search" | "node" | "callers" | "callees" | "impact" | "explore"
        ) {
            m.insert("name".into(), Value::String("helper".into()));
        }
        if tool == "top" {
            m.insert("by".into(), Value::String("fan_in".into()));
        }
        Some(m)
    }

    #[tokio::test]
    async fn an_unknown_view_id_is_an_error_not_an_empty_result() {
        let s = seeded_server().await;
        for tool in VIEW_SCOPED {
            let result = s.dispatch(tool, args_for(tool, "r", "mian")).await.unwrap();
            assert_eq!(result.is_error, Some(true), "{tool} must reject 'mian'");
            let text = result_text(&result);
            assert!(
                text.contains("r/mian"),
                "{tool} must name the selector: {text}"
            );
        }
    }

    #[tokio::test]
    async fn an_unknown_repo_is_an_error_not_an_empty_result() {
        let s = seeded_server().await;
        for tool in VIEW_SCOPED {
            let result = s
                .dispatch(tool, args_for(tool, "does-not-exist", "main"))
                .await
                .unwrap();
            assert_eq!(result.is_error, Some(true), "{tool} must reject the repo");
        }
    }

    #[tokio::test]
    async fn an_unknown_view_error_lists_the_views_that_exist() {
        let s = seeded_server().await;
        let result = s
            .dispatch("search", args_for("search", "r", "mian"))
            .await
            .unwrap();
        let text = result_text(&result);
        assert!(
            text.contains("r/main"),
            "must point at the real view: {text}"
        );
        assert!(
            text.contains("views"),
            "must name the discovery tool: {text}"
        );
    }

    #[tokio::test]
    async fn an_absent_symbol_in_a_valid_view_is_still_an_empty_result() {
        // The other half of #210: the two cases must be distinguishable, so a
        // genuine miss must NOT become an error.
        let s = seeded_server().await;
        let result = s
            .dispatch("search", args_for("search", "r", "main"))
            .await
            .unwrap();
        assert_ne!(result.is_error, Some(true));

        let result = s
            .dispatch(
                "callers",
                Some(
                    serde_json::json!({ "repo": "r", "view_id": "main", "name": "nonexistent" })
                        .as_object()
                        .unwrap()
                        .clone(),
                ),
            )
            .await
            .unwrap();
        assert_ne!(result.is_error, Some(true), "a real miss is not an error");
        let v: Value = serde_json::from_str(&result_text(&result)).unwrap();
        assert_eq!(v["callers"], serde_json::json!([]));
        // And the answer says the name is defined nowhere, which is what makes
        // the empty list readable (#249).
        assert_eq!(v["definition_count"], 0);
    }

    #[tokio::test]
    async fn diff_rejects_an_unknown_view_on_either_side() {
        let s = seeded_server().await;
        for (a, b) in [("main", "nope"), ("nope", "main")] {
            let result = s
                .dispatch(
                    "diff",
                    Some(
                        serde_json::json!({ "repo": "r", "view_a": a, "view_b": b })
                            .as_object()
                            .unwrap()
                            .clone(),
                    ),
                )
                .await
                .unwrap();
            assert_eq!(result.is_error, Some(true), "diff {a}->{b}");
            assert!(result_text(&result).contains("r/nope"));
        }
    }

    #[tokio::test]
    async fn views_tool_lists_indexed_views() {
        let s = seeded_server().await;
        let result = s.dispatch("views", None).await.unwrap();
        assert_ne!(result.is_error, Some(true));
        let v: Value = serde_json::from_str(&result_text(&result)).unwrap();
        assert_eq!(v.as_array().unwrap().len(), 1);
        assert_eq!(v[0]["repo"], "r");
        assert_eq!(v[0]["view_id"], "main");
        assert_eq!(v[0]["files"], 2, "lib.rs + main.rs");
    }

    #[tokio::test]
    async fn views_tool_is_empty_on_a_fresh_store() {
        let s = server("test-root");
        let result = s.dispatch("views", None).await.unwrap();
        let v: Value = serde_json::from_str(&result_text(&result)).unwrap();
        assert!(v.as_array().unwrap().is_empty());
    }

    #[tokio::test]
    async fn status_reports_the_number_of_indexed_views() {
        let s = seeded_server().await;
        let v: Value =
            serde_json::from_str(&result_text(&s.dispatch("status", None).await.unwrap())).unwrap();
        assert_eq!(v["status"], "ok");
        assert_eq!(v["views"], 1, "an empty store must be distinguishable");

        let empty = server("test-root");
        let v: Value =
            serde_json::from_str(&result_text(&empty.dispatch("status", None).await.unwrap()))
                .unwrap();
        assert_eq!(v["views"], 0);
    }

    #[tokio::test]
    async fn aggregate_tools_require_a_view_selector() {
        let s = seeded_server().await;
        for tool in ["overview", "top", "list", "unreferenced"] {
            let result = s.dispatch(tool, None).await.unwrap();
            assert_eq!(result.is_error, Some(true), "{tool} should reject no args");
            assert!(result_text(&result).contains("repo"));
        }
    }

    /// A view with a used function, an unused one, and a `mod tests` block.
    const DEAD_SRC: &str = "fn helper() {}\n\
                            fn main() { helper(); }\n\
                            fn orphan() {}\n\
                            #[cfg(test)]\n\
                            mod tests {\n    \
                                fn t_only() {}\n\
                            }\n";

    async fn call_on(server: &GonzaloMcp, tool: &str, extra: Value) -> Value {
        let result = server
            .dispatch(tool, view_args_with("r", "main", extra))
            .await
            .unwrap();
        assert_ne!(result.is_error, Some(true), "tool {tool} errored");
        serde_json::from_str(&result_text(&result)).unwrap()
    }

    fn item_names(v: &Value) -> Vec<String> {
        v["items"]
            .as_array()
            .unwrap()
            .iter()
            .map(|l| l["item"]["name"].as_str().unwrap().to_string())
            .collect()
    }

    #[tokio::test]
    async fn unreferenced_tool_reports_only_uncalled_symbols() {
        let v = call("unreferenced", serde_json::json!({})).await;
        // The seeded view is `helper` (called from main) + `main` (called by
        // nothing), so only `main` is a candidate.
        assert_eq!(item_names(&v), vec!["main"]);
        assert_eq!(v["items"][0]["path"], "main.rs");
    }

    #[tokio::test]
    async fn unreferenced_tool_excludes_test_scopes_by_default() {
        let s = seeded_with(&[("lib.rs", DEAD_SRC)]).await;
        let v = call_on(&s, "unreferenced", serde_json::json!({})).await;
        let names = item_names(&v);
        assert!(names.contains(&"orphan".to_string()));
        assert!(!names.contains(&"t_only".to_string()));
    }

    #[tokio::test]
    async fn unreferenced_tool_can_include_test_scopes() {
        let s = seeded_with(&[("lib.rs", DEAD_SRC)]).await;
        let v = call_on(
            &s,
            "unreferenced",
            serde_json::json!({ "exclude_tests": false }),
        )
        .await;
        assert!(item_names(&v).contains(&"t_only".to_string()));
    }

    #[tokio::test]
    async fn unreferenced_tool_applies_the_symbol_filter() {
        let s = seeded_with(&[("lib.rs", DEAD_SRC)]).await;
        let v = call_on(&s, "unreferenced", serde_json::json!({ "kind": "struct" })).await;
        assert_eq!(v["total"], 0);
    }

    #[tokio::test]
    async fn unreferenced_tool_reports_truncation() {
        let s = seeded_with(&[("lib.rs", "fn a() {} fn b() {} fn c() {}")]).await;
        let v = call_on(&s, "unreferenced", serde_json::json!({ "limit": 2 })).await;
        assert_eq!(v["items"].as_array().unwrap().len(), 2);
        assert_eq!(v["truncated"], true);
        assert_eq!(v["total"], 3);
    }

    #[tokio::test]
    async fn unreferenced_tool_rejects_a_non_boolean_exclude_tests() {
        let s = seeded_server().await;
        let result = s
            .dispatch(
                "unreferenced",
                view_args_with("r", "main", serde_json::json!({ "exclude_tests": "yes" })),
            )
            .await
            .unwrap();
        assert_eq!(result.is_error, Some(true));
        assert!(result_text(&result).contains("exclude_tests"));
    }

    #[test]
    fn unreferenced_tool_description_states_the_heuristic_blind_spot() {
        let tool = GonzaloMcp::tools(false)
            .into_iter()
            .find(|t| t.name == "unreferenced")
            .expect("unreferenced is advertised");
        let description = tool.description.as_deref().unwrap_or_default();
        assert!(
            description.contains("heuristic"),
            "must not present candidates as proof: {description}"
        );
        assert!(
            description.contains("higher-order"),
            "must name the higher-order-usage blind spot: {description}"
        );
    }

    #[tokio::test]
    async fn diff_tool_reports_changes_between_two_views() {
        // Seed two views of `r` sharing the store, differing by one symbol.
        let fs = Arc::new(FsStore::new(tempfile::tempdir().unwrap().keep()));
        for (view, src) in [
            ("v1", "fn keep() {}\nfn gone() {}"),
            ("v2", "fn keep() {}\nfn fresh() {}"),
        ] {
            let hash = fs
                .put_blob(&build_rust(src).to_slice_bytes())
                .await
                .unwrap();
            let mut manifest = Manifest::new();
            manifest.insert("lib.rs", hash);
            let body = manifest.to_body();
            let record = Record {
                revision: Revision::initial(body.bytes()),
                parent: None,
                body,
                kind: RecordKind::GraphManifest,
                meta: Meta {
                    author: Identity::new("t"),
                    origin_system: "t".into(),
                    created: 0,
                    updated: 0,
                    labels: BTreeMap::new(),
                },
                links: Vec::new(),
                key: Manifest::key("r", view),
                ancestors: Vec::new(),
                deleted_at: None,
                deleted_blob: None,
                deleted_kind: None,
            };
            assert!(matches!(
                fs.put(record, None).await.unwrap(),
                PutResult::Committed(_)
            ));
        }
        let mcp = GonzaloMcp::new(Service::new(fs.clone(), fs), "test-root");

        let diff_args = Some(
            serde_json::json!({ "repo": "r", "view_a": "v1", "view_b": "v2" })
                .as_object()
                .unwrap()
                .clone(),
        );
        let result = mcp.dispatch("diff", diff_args).await.unwrap();
        let diff: Value = serde_json::from_str(&result_text(&result)).unwrap();
        let added: Vec<&str> = diff["added_symbols"]
            .as_array()
            .unwrap()
            .iter()
            .map(|s| s["item"]["name"].as_str().unwrap())
            .collect();
        assert!(added.contains(&"fresh"));
        let removed: Vec<&str> = diff["removed_symbols"]
            .as_array()
            .unwrap()
            .iter()
            .map(|s| s["item"]["name"].as_str().unwrap())
            .collect();
        assert!(removed.contains(&"gone"));
    }

    #[tokio::test]
    async fn status_reports_ok_and_configured_root() {
        let s = server("/tmp/some-root");
        let payload = s.status_json().await;
        assert_eq!(payload["status"], "ok");
        assert_eq!(payload["root"], "/tmp/some-root");
    }

    #[tokio::test]
    async fn search_returns_located_definitions() {
        let s = seeded_server().await;
        let result = s
            .dispatch("search", args("r", "main", "helper"))
            .await
            .unwrap();
        assert_ne!(result.is_error, Some(true));
        let defs: Value = serde_json::from_str(&result_text(&result)).unwrap();
        assert_eq!(defs[0]["path"], "lib.rs");
        assert_eq!(defs[0]["item"]["name"], "helper");
    }

    #[tokio::test]
    async fn callees_returns_names() {
        let s = seeded_server().await;
        let callees = s
            .dispatch("callees", args("r", "main", "main"))
            .await
            .unwrap();
        let v: Value = serde_json::from_str(&result_text(&callees)).unwrap();
        assert_eq!(v["callees"], serde_json::json!(["helper"]));
        assert_eq!(v["definition_count"], 1, "`main` is defined once");
    }

    // ---- resolution-gated impact (#207) ------------------------------------

    #[tokio::test]
    async fn impact_returns_a_report_with_paths() {
        let s = seeded_server().await;
        let v: Value = serde_json::from_str(&result_text(
            &s.dispatch("impact", args("r", "main", "helper"))
                .await
                .unwrap(),
        ))
        .unwrap();
        // Not a bare name list any more: every node carries its defining path,
        // and the honesty fields ride alongside.
        assert_eq!(v["reached"][0]["name"], "main");
        assert_eq!(v["reached"][0]["path"], "main.rs");
        assert_eq!(v["ambiguous_edges"], 0);
        assert_eq!(v["truncated"], false);
    }

    #[tokio::test]
    async fn impact_does_not_merge_subgraphs_through_a_shared_name() {
        // `helper` is defined in both files; nothing else is shared. The old
        // name-matched closure reached `top_b` from `leaf_a` (#207).
        let s = seeded_with(&[
            (
                "a.rs",
                "fn leaf_a() {}\nfn helper() { leaf_a(); }\nfn top_a() { helper(); }",
            ),
            (
                "b.rs",
                "fn leaf_b() {}\nfn helper() { leaf_b(); }\nfn top_b() { helper(); }",
            ),
        ])
        .await;
        let v: Value = serde_json::from_str(&result_text(
            &s.dispatch("impact", args("r", "main", "leaf_a"))
                .await
                .unwrap(),
        ))
        .unwrap();
        let names: Vec<&str> = v["reached"]
            .as_array()
            .unwrap()
            .iter()
            .map(|n| n["name"].as_str().unwrap())
            .collect();
        assert!(names.contains(&"top_a"), "{names:?}");
        assert!(!names.contains(&"top_b"), "must not cross files: {names:?}");
    }

    #[tokio::test]
    async fn impact_accepts_a_max_depth() {
        let s = seeded_with(&[("a.rs", "fn l() {}\nfn m() { l(); }\nfn t() { m(); }")]).await;
        let mut a = args("r", "main", "l").unwrap();
        a.insert("max_depth".into(), serde_json::json!(1));
        let v: Value =
            serde_json::from_str(&result_text(&s.dispatch("impact", Some(a)).await.unwrap()))
                .unwrap();
        assert_eq!(v["reached"].as_array().unwrap().len(), 1);
        assert_eq!(v["reached"][0]["name"], "m");
        assert_eq!(v["truncated"], true);
    }

    #[tokio::test]
    async fn impact_rejects_a_non_integer_max_depth() {
        let s = seeded_server().await;
        let mut a = args("r", "main", "helper").unwrap();
        a.insert("max_depth".into(), serde_json::json!("deep"));
        let result = s.dispatch("impact", Some(a)).await.unwrap();
        assert_eq!(result.is_error, Some(true));
        assert!(result_text(&result).contains("max_depth"));
    }

    #[tokio::test]
    async fn node_aggregates_definitions_callers_and_callees() {
        let s = seeded_server().await;
        let result = s
            .dispatch("node", args("r", "main", "helper"))
            .await
            .unwrap();
        let node: Value = serde_json::from_str(&result_text(&result)).unwrap();
        assert_eq!(node["definitions"][0]["path"], "lib.rs");
        assert_eq!(node["callers"], serde_json::json!(["main"]));
        assert_eq!(node["callees"], serde_json::json!([]));
    }

    #[tokio::test]
    async fn missing_arguments_are_a_tool_error() {
        let s = seeded_server().await;
        // No `name` provided.
        let bad = Some(
            serde_json::json!({ "repo": "r", "view_id": "main" })
                .as_object()
                .unwrap()
                .clone(),
        );
        let result = s.dispatch("search", bad).await.unwrap();
        assert_eq!(result.is_error, Some(true));
        assert!(result_text(&result).contains("name"));
    }

    #[tokio::test]
    async fn dispatch_unknown_tool_is_method_not_found() {
        let s = server("/r");
        assert!(s.dispatch("no_such_tool", None).await.is_err());
    }

    // ---- ambiguity signal on callers/callees (#249) -----------------------

    /// Two files define `foo`; a third calls it. Asking for callers of `foo`
    /// merges callers of both, with nothing in the answer saying so.
    async fn overloaded() -> GonzaloMcp {
        seeded_with(&[
            ("a.rs", "fn foo() {}"),
            ("b.rs", "fn foo() {}"),
            ("c.rs", "fn caller() { foo(); }"),
        ])
        .await
    }

    #[tokio::test]
    async fn callers_reports_how_many_definitions_the_name_has() {
        let s = overloaded().await;
        let v = call_on(&s, "callers", serde_json::json!({ "name": "foo" })).await;
        assert_eq!(v["callers"], serde_json::json!(["caller"]));
        assert_eq!(
            v["definition_count"], 2,
            "the result merges two symbols: {v}"
        );
        assert_eq!(v["defined_in"], serde_json::json!(["a.rs", "b.rs"]));
    }

    #[tokio::test]
    async fn callees_reports_how_many_definitions_the_name_has() {
        let s = overloaded().await;
        let v = call_on(&s, "callees", serde_json::json!({ "name": "foo" })).await;
        assert_eq!(v["definition_count"], 2, "{v}");
    }

    #[tokio::test]
    async fn an_unambiguous_name_reports_one_definition() {
        // Not zero, and not absent: the field always answers.
        let v = call("callers", serde_json::json!({ "name": "helper" })).await;
        assert_eq!(v["callers"], serde_json::json!(["main"]));
        assert_eq!(v["definition_count"], 1, "{v}");
    }

    #[tokio::test]
    async fn a_name_defined_nowhere_reports_zero_definitions() {
        let v = call("callers", serde_json::json!({ "name": "absent" })).await;
        assert_eq!(v["callers"], serde_json::json!([]));
        assert_eq!(v["definition_count"], 0, "{v}");
    }

    #[tokio::test]
    async fn the_defining_paths_are_dropped_once_the_name_is_hopeless() {
        // Past a handful the list stops being a lead and becomes noise; the
        // count alone already says "do not read this as precise".
        let slices: Vec<(String, String)> = (0..MAX_DEFINED_IN + 2)
            .map(|i| (format!("f{i}.rs"), "fn foo() {}".to_string()))
            .collect();
        let borrowed: Vec<(&str, &str)> = slices
            .iter()
            .map(|(p, src)| (p.as_str(), src.as_str()))
            .collect();
        let s = seeded_with(&borrowed).await;

        let v = call_on(&s, "callers", serde_json::json!({ "name": "foo" })).await;
        assert_eq!(v["definition_count"], MAX_DEFINED_IN + 2, "{v}");
        assert!(v["defined_in"].is_null(), "paths dropped: {v}");
    }

    // ---- module-level callers (#268) ---------------------------------------

    #[tokio::test]
    async fn callers_separates_functions_from_module_level_declarations() {
        // `callers` must keep meaning what it meant — functions that call this.
        // The module-level ones are real dependencies but a different question,
        // so they travel in their own field (#268).
        let s = seeded_with_language(
            gonzalo_graph::Language::TypeScript,
            &[(
                "app.ts",
                "const schema = z.object({});\nconst run = () => { z.object({}); };",
            )],
        )
        .await;
        let v = call_on(&s, "callers", serde_json::json!({ "name": "object" })).await;
        assert_eq!(v["callers"], serde_json::json!(["run"]));
        assert_eq!(v["module_callers"], serde_json::json!(["schema"]));
    }

    #[tokio::test]
    async fn callees_carries_no_module_field() {
        // The split is meaningful for callers only: `callees` asks what one
        // named thing calls, and its scope is whatever that thing is.
        let v = call("callees", serde_json::json!({ "name": "main" })).await;
        assert_eq!(v["callees"], serde_json::json!(["helper"]));
        assert!(v["module_callers"].is_null(), "{v}");
    }

    // ---- record tools (#197) ----------------------------------------------

    /// A server over an empty store, with the store handle so a test can seed
    /// records directly and assert what the tools did to them.
    fn record_server(writes: bool) -> (GonzaloMcp, Arc<FsStore>) {
        let fs = Arc::new(FsStore::new(tempfile::tempdir().unwrap().keep()));
        let server =
            GonzaloMcp::new(Service::new(fs.clone(), fs.clone()), "test-root").with_writes(writes);
        (server, fs)
    }

    fn topic(key: RecordKey, payload: &[u8]) -> Record {
        let body = Body::Inline(payload.to_vec());
        Record {
            revision: Revision::initial(body.bytes()),
            parent: None,
            body,
            kind: RecordKind::Topic,
            meta: Meta {
                author: Identity::new("tester"),
                origin_system: "test".into(),
                created: 0,
                updated: 0,
                labels: BTreeMap::new(),
            },
            links: Vec::new(),
            key,
            ancestors: Vec::new(),
            deleted_at: None,
            deleted_blob: None,
            deleted_kind: None,
        }
    }

    /// Seed a record, asserting the write committed rather than conflicted.
    async fn put_ok(fs: &FsStore, record: Record) -> Revision {
        match fs.put(record, None).await.unwrap() {
            PutResult::Committed(rev) => rev,
            PutResult::Conflict(c) => panic!("unexpected conflict: {c:?}"),
        }
    }

    fn record_args(extra: Value) -> Option<Map<String, Value>> {
        Some(extra.as_object().expect("object").clone())
    }

    #[test]
    fn write_tools_are_advertised_only_when_writes_are_enabled() {
        // An agent must never see a tool it cannot call, so the advertised list
        // is the mode: read-only by default, writes on request (#197).
        let read_only: Vec<String> = GonzaloMcp::tools(false)
            .iter()
            .map(|t| t.name.to_string())
            .collect();
        assert!(read_only.contains(&"record_get".to_string()));
        assert!(read_only.contains(&"record_list".to_string()));
        assert!(!read_only.contains(&"record_put".to_string()));
        assert!(!read_only.contains(&"record_delete".to_string()));

        let writable: Vec<String> = GonzaloMcp::tools(true)
            .iter()
            .map(|t| t.name.to_string())
            .collect();
        for tool in ["record_get", "record_list", "record_put", "record_delete"] {
            assert!(writable.contains(&tool.to_string()), "missing {tool}");
        }
    }

    #[tokio::test]
    async fn record_get_returns_the_stored_record() {
        let (server, fs) = record_server(false);
        let key = RecordKey::new("memory", "topics", "rust");
        put_ok(&fs, topic(key.clone(), b"{\"note\":1}")).await;

        let result = server
            .dispatch(
                "record_get",
                record_args(
                    serde_json::json!({"namespace": "memory", "collection": "topics", "id": "rust"}),
                ),
            )
            .await
            .unwrap();
        assert_ne!(result.is_error, Some(true));
        let v: Value = serde_json::from_str(&result_text(&result)).unwrap();
        assert_eq!(v["key"]["id"], "rust");
        assert_eq!(v["kind"], "Topic");
    }

    #[tokio::test]
    async fn record_get_of_an_absent_key_is_an_error_not_an_empty_result() {
        // Same rule the view selector follows (#210): an agent reads an empty
        // answer as "there is nothing there", which is not what a typo means.
        let (server, _fs) = record_server(false);
        let result = server
            .dispatch(
                "record_get",
                record_args(
                    serde_json::json!({"namespace": "memory", "collection": "topics", "id": "nope"}),
                ),
            )
            .await
            .unwrap();
        assert_eq!(result.is_error, Some(true));
        assert!(result_text(&result).contains("memory/topics/nope"));
    }

    #[tokio::test]
    async fn record_list_lists_a_collection_and_hides_tombstones() {
        let (server, fs) = record_server(false);
        let kept = RecordKey::new("memory", "topics", "kept");
        let gone = RecordKey::new("memory", "topics", "gone");
        put_ok(&fs, topic(kept.clone(), b"a")).await;
        let rev = put_ok(&fs, topic(gone.clone(), b"b")).await;
        assert_eq!(
            fs.delete(&gone, Some(rev)).await.unwrap(),
            DeleteResult::Deleted
        );

        let result = server
            .dispatch(
                "record_list",
                record_args(serde_json::json!({"namespace": "memory", "collection": "topics"})),
            )
            .await
            .unwrap();
        let v: Value = serde_json::from_str(&result_text(&result)).unwrap();
        assert_eq!(v["keys"], serde_json::json!(["memory/topics/kept"]));
    }

    #[tokio::test]
    async fn write_tools_are_unavailable_in_read_only_mode() {
        let (server, _fs) = record_server(false);
        for tool in ["record_put", "record_delete"] {
            let err = server
                .dispatch(
                    tool,
                    record_args(serde_json::json!({
                        "namespace": "memory", "collection": "topics", "id": "x",
                        "kind": "Topic", "body": "{}"
                    })),
                )
                .await;
            assert!(err.is_err(), "{tool} must not be callable read-only");
        }
    }

    #[tokio::test]
    async fn record_put_creates_then_conflicts_without_the_current_revision() {
        let (server, fs) = record_server(true);
        let result = server
            .dispatch(
                "record_put",
                record_args(serde_json::json!({
                    "namespace": "memory", "collection": "topics", "id": "new",
                    "kind": "Topic", "body": "hello"
                })),
            )
            .await
            .unwrap();
        assert_ne!(result.is_error, Some(true), "{}", result_text(&result));
        let v: Value = serde_json::from_str(&result_text(&result)).unwrap();
        assert_eq!(v["committed"], true);

        let key = RecordKey::new("memory", "topics", "new");
        let stored = fs.get(&key).await.unwrap().expect("stored");
        assert_eq!(stored.body.bytes(), b"hello");

        // A second create over a live record is a conflict, reported as data
        // rather than an error: it is a normal OCC outcome the agent can act on.
        let again = server
            .dispatch(
                "record_put",
                record_args(serde_json::json!({
                    "namespace": "memory", "collection": "topics", "id": "new",
                    "kind": "Topic", "body": "clobber"
                })),
            )
            .await
            .unwrap();
        let v: Value = serde_json::from_str(&result_text(&again)).unwrap();
        assert_eq!(v["committed"], false);
        assert_eq!(fs.get(&key).await.unwrap().unwrap().body.bytes(), b"hello");
    }

    #[tokio::test]
    async fn record_delete_removes_the_record() {
        let (server, fs) = record_server(true);
        let key = RecordKey::new("memory", "topics", "bye");
        put_ok(&fs, topic(key.clone(), b"a")).await;

        let result = server
            .dispatch(
                "record_delete",
                record_args(
                    serde_json::json!({"namespace": "memory", "collection": "topics", "id": "bye"}),
                ),
            )
            .await
            .unwrap();
        assert_ne!(result.is_error, Some(true), "{}", result_text(&result));
        assert_eq!(fs.get(&key).await.unwrap(), None);
        // The tombstone is still there for replication.
        assert!(fs.get_raw(&key).await.unwrap().unwrap().is_tombstone());
    }
}