agentmux_srv/server/app_api/

mcp.rs

// Copyright 2025-2026, AgentMux Corp.
// SPDX-License-Identifier: Apache-2.0

//! MuxAPI handlers for the v1 standalone MCP Server primitive.
//! mcp.list / mcp.get / mcp.upsert / mcp.delete / mcp.bind / mcp.unbind
//!
//! Plus the Connectors-level catalog (mcp.catalog.*): global servers only,
//! window-scoped — no `agent_id`, no `check_s1` (mirrors bundle.* auth
//! shape, since Connectors has no agent connection context to gate on).
//!
//! `db_mcp_servers` is authoritatively `identity_store` as of Phase 2 of
//! SPEC_DURABLE_BINDINGS_2026_09_10.md (§5.1/§5.4) — see `skill.rs`'s
//! module doc comment for the full `catalog`/`mstore` split; this file
//! mirrors it exactly for MCP servers.

use super::*;

pub fn register(engine: &Arc<WshRpcEngine>, state: &AppState) {
    register_mcp_list(engine, state);
    register_mcp_get(engine, state);
    register_mcp_upsert(engine, state);
    register_mcp_delete(engine, state);
    register_mcp_bind(engine, state);
    register_mcp_unbind(engine, state);
    register_mcp_probe(engine, state);
    register_mcp_catalog_list(engine, state);
    register_mcp_catalog_upsert(engine, state);
    register_mcp_catalog_delete(engine, state);
    register_mcp_catalog_probe(engine, state);
    register_mcp_catalog_set_reach(engine, state);
    register_mcp_observed_list(engine, state);
    register_mcp_catalog_bind(engine, state);
    register_mcp_catalog_list_for_agent(engine, state);
    register_mcp_catalog_unbind(engine, state);
}

fn register_mcp_list(engine: &Arc<WshRpcEngine>, state: &AppState) {
    let mstore = state.mstore.clone();
    let identity_store = state.identity_store.clone();
    engine.register_typed(
        COMMAND_MCP_LIST,
        move |req: McpAgentScopeData, ctx| {
            let mstore = mstore.clone();
            let identity_store = identity_store.clone();
            async move {
                check_s1(&mstore, &ctx, &req.agent_id)?;
                let servers = mstore.mcp_server_list(&identity_store, &req.agent_id)
                    .map_err(|e| format!("mcp.list: {e}"))?;
                Ok(servers)
            }
        },
    );
}

fn register_mcp_get(engine: &Arc<WshRpcEngine>, state: &AppState) {
    let mstore = state.mstore.clone();
    let identity_store = state.identity_store.clone();
    engine.register_typed(
        COMMAND_MCP_GET,
        move |req: McpAgentItemData, ctx| {
            let mstore = mstore.clone();
            let identity_store = identity_store.clone();
            async move {
                check_s1(&mstore, &ctx, &req.agent_id)?;
                if !mstore.mcp_server_is_accessible_to(&identity_store, &req.agent_id, &req.id)
                    .map_err(|e| format!("mcp.get: {e}"))?
                {
                    return Err("FORBIDDEN: MCP server not accessible to this agent".to_string());
                }
                let server = identity_store.mcp_server_get(&req.id)
                    .map_err(|e| format!("mcp.get: {e}"))?;
                Ok(server)
            }
        },
    );
}

fn register_mcp_upsert(engine: &Arc<WshRpcEngine>, state: &AppState) {
    let mstore = state.mstore.clone();
    let identity_store = state.identity_store.clone();
    let broker = state.broker.clone();
    engine.register_typed(
        COMMAND_MCP_UPSERT,
        move |req: CommandMcpUpsertData, ctx| {
            let mstore = mstore.clone();
            let identity_store = identity_store.clone();
            let broker = broker.clone();
            async move {
                check_s1(&mstore, &ctx, &req.agent_id)?;

                match serde_json::from_str::<serde_json::Value>(&req.config) {
                    Ok(serde_json::Value::Object(_)) => {}
                    Ok(_) => return Err("mcp.upsert: config must be a JSON object".to_string()),
                    Err(_) => return Err("mcp.upsert: config must be valid JSON".to_string()),
                }
                if req.name == "agentmux" {
                    return Err("FORBIDDEN: 'agentmux' is a reserved MCP server name".to_string());
                }

                let now = now_ms();
                // created_at is preserved across updates (ON CONFLICT keeps the
                // original), so the returned struct never reports a timestamp the
                // DB won't persist. New rows default to `now`.
                let (id, created_at) = if req.id.is_empty() {
                    (uuid::Uuid::new_v4().to_string(), now)
                } else {
                    if !mstore.mcp_server_is_bound_to(&req.agent_id, &req.id)
                        .map_err(|e| format!("mcp.upsert: {e}"))?
                    {
                        return Err("FORBIDDEN: MCP server not bound to this agent".to_string());
                    }
                    let existing = identity_store.mcp_server_get(&req.id)
                        .map_err(|e| format!("mcp.upsert: {e}"))?;
                    match existing {
                        Some(s) if s.is_global => {
                            return Err("FORBIDDEN: cannot mutate a global MCP server".to_string());
                        }
                        Some(s) => (req.id.clone(), s.created_at),
                        None => (req.id.clone(), now),
                    }
                };

                let server = crate::backend::storage::McpServer {
                    id: id.clone(),
                    name: req.name,
                    transport: req.transport,
                    config: req.config,
                    is_global: false,
                    created_at,
                    updated_at: now,
                };
                // No longer a single transaction across the catalog insert and
                // the ref-bind (Phase 2 of SPEC_DURABLE_BINDINGS_2026_09_10.md
                // §5.4, accepted interim gap) — see
                // `Store::managed_upsert_unique`'s doc comment.
                mstore.mcp_server_upsert_unique(&identity_store, &req.agent_id, &server, req.id.is_empty())
                    .map_err(|e| format!("mcp.upsert: {e}"))?;
                broker.publish_event::<crate::backend::mps_events::McpChanged>(&(), vec![], 0);
                Ok(server)
            }
        },
    );
}

fn register_mcp_delete(engine: &Arc<WshRpcEngine>, state: &AppState) {
    let mstore = state.mstore.clone();
    let identity_store = state.identity_store.clone();
    let broker = state.broker.clone();
    engine.register_typed(
        COMMAND_MCP_DELETE,
        move |req: McpAgentItemData, ctx| {
            let mstore = mstore.clone();
            let identity_store = identity_store.clone();
            let broker = broker.clone();
            async move {
                check_s1(&mstore, &ctx, &req.agent_id)?;
                if !mstore.mcp_server_is_bound_to(&req.agent_id, &req.id)
                    .map_err(|e| format!("mcp.delete: {e}"))?
                {
                    return Err("FORBIDDEN: MCP server not bound to this agent".to_string());
                }
                if let Some(existing) = identity_store.mcp_server_get(&req.id)
                    .map_err(|e| format!("mcp.delete: {e}"))?
                {
                    if existing.is_global {
                        return Err("FORBIDDEN: cannot delete a global MCP server".to_string());
                    }
                }
                let deleted = mstore.mcp_server_delete(&identity_store, &req.id)
                    .map_err(|e| format!("mcp.delete: {e}"))?;
                if deleted {
                    broker.publish_event::<crate::backend::mps_events::McpChanged>(&(), vec![], 0);
                }
                Ok(McpDeleteResult { deleted })
            }
        },
    );
}

fn register_mcp_bind(engine: &Arc<WshRpcEngine>, state: &AppState) {
    let mstore = state.mstore.clone();
    let identity_store = state.identity_store.clone();
    let broker = state.broker.clone();
    engine.register_typed(
        COMMAND_MCP_BIND,
        move |req: McpAgentBindingData, ctx| {
            let mstore = mstore.clone();
            let identity_store = identity_store.clone();
            let broker = broker.clone();
            async move {
                check_s1(&mstore, &ctx, &req.agent_id)?;
                // Only global servers (or ones already bound) may be bound, so an
                // agent can't escalate to read another agent's server config.
                match identity_store.mcp_server_get(&req.mcp_id)
                    .map_err(|e| format!("mcp.bind: {e}"))?
                {
                    None => return Err("mcp.bind: MCP server not found".to_string()),
                    Some(s) if !s.is_global => {
                        if !mstore.mcp_server_is_bound_to(&req.agent_id, &req.mcp_id)
                            .map_err(|e| format!("mcp.bind: {e}"))?
                        {
                            return Err("FORBIDDEN: can only bind global MCP servers".to_string());
                        }
                    }
                    Some(_) => {}
                }
                mstore.mcp_server_bind(&identity_store, &req.agent_id, &req.mcp_id)
                    .map_err(|e| format!("mcp.bind: {e}"))?;
                // An agent binding a server to itself over its own authenticated
                // connection should reach an already-open Stash MCP Servers tab
                // for that agent too — same reactivity as the catalog-tier bind.
                // reagentx P2 on PR #2329.
                broker.publish_event::<crate::backend::mps_events::McpChanged>(&(), vec![], 0);
                Ok(McpBindResult { bound: true })
            }
        },
    );
}

fn register_mcp_unbind(engine: &Arc<WshRpcEngine>, state: &AppState) {
    let mstore = state.mstore.clone();
    let broker = state.broker.clone();
    engine.register_typed(
        COMMAND_MCP_UNBIND,
        move |req: McpAgentBindingData, ctx| {
            let mstore = mstore.clone();
            let broker = broker.clone();
            async move {
                check_s1(&mstore, &ctx, &req.agent_id)?;
                let unbound = mstore.mcp_server_unbind(&req.agent_id, &req.mcp_id)
                    .map_err(|e| format!("mcp.unbind: {e}"))?;
                if unbound {
                    broker.publish_event::<crate::backend::mps_events::McpChanged>(&(), vec![], 0);
                }
                Ok(McpUnbindResult { unbound })
            }
        },
    );
}

/// Health/prerequisite probe for one of this agent's own or bound-global
/// servers (SPEC_MCP_INTEGRATION_PARITY_ABLETON_PILOT_2026_07_08.md §4.4).
/// Opens a short-lived MCP connection and reports whether the server
/// actually speaks the protocol — distinct from `mcp.upsert`'s "is this
/// valid JSON" check, which tells you nothing about reachability.
fn register_mcp_probe(engine: &Arc<WshRpcEngine>, state: &AppState) {
    let mstore = state.mstore.clone();
    let identity_store = state.identity_store.clone();
    engine.register_typed(
        COMMAND_MCP_PROBE,
        move |req: McpAgentItemData, ctx| {
            let mstore = mstore.clone();
            let identity_store = identity_store.clone();
            async move {
                check_s1(&mstore, &ctx, &req.agent_id)?;
                if !mstore.mcp_server_is_accessible_to(&identity_store, &req.agent_id, &req.id)
                    .map_err(|e| format!("mcp.probe: {e}"))?
                {
                    return Err("FORBIDDEN: MCP server not accessible to this agent".to_string());
                }
                let server = identity_store.mcp_server_get(&req.id)
                    .map_err(|e| format!("mcp.probe: {e}"))?
                    .ok_or_else(|| "mcp.probe: MCP server not found".to_string())?;
                let result = crate::backend::mcp_probe::probe(&server.transport, &server.config).await;
                Ok(result)
            }
        },
    );
}

fn register_mcp_catalog_list(engine: &Arc<WshRpcEngine>, state: &AppState) {
    let mstore = state.mstore.clone();
    let identity_store = state.identity_store.clone();
    engine.register_typed(
        COMMAND_MCP_CATALOG_LIST,
        move |_req: McpCatalogListData, _ctx| {
            let mstore = mstore.clone();
            let identity_store = identity_store.clone();
            async move {
                let servers = mstore.mcp_server_list_global(&identity_store)
                    .map_err(|e| format!("mcp.catalog.list: {e}"))?;
                Ok(servers)
            }
        },
    );
}

fn register_mcp_catalog_upsert(engine: &Arc<WshRpcEngine>, state: &AppState) {
    let identity_store = state.identity_store.clone();
    let broker = state.broker.clone();
    engine.register_typed(
        COMMAND_MCP_CATALOG_UPSERT,
        move |req: CommandMcpCatalogUpsertData, _ctx| {
            let identity_store = identity_store.clone();
            let broker = broker.clone();
            async move {

                match serde_json::from_str::<serde_json::Value>(&req.config) {
                    Ok(serde_json::Value::Object(_)) => {}
                    Ok(_) => return Err("mcp.catalog.upsert: config must be a JSON object".to_string()),
                    Err(_) => return Err("mcp.catalog.upsert: config must be valid JSON".to_string()),
                }
                if req.name == "agentmux" {
                    return Err("FORBIDDEN: 'agentmux' is a reserved MCP server name".to_string());
                }

                let now = now_ms();
                let (id, created_at) = if req.id.is_empty() {
                    (uuid::Uuid::new_v4().to_string(), now)
                } else {
                    let existing = identity_store.mcp_server_get(&req.id)
                        .map_err(|e| format!("mcp.catalog.upsert: {e}"))?;
                    match existing {
                        // No agent_id/check_s1 here to verify ownership of a private
                        // server, so the catalog surface may only create new global
                        // rows or edit ones that are ALREADY global — never promote
                        // a private server into the catalog by supplying its id.
                        Some(s) if !s.is_global => {
                            return Err("FORBIDDEN: cannot promote a private MCP server via the catalog".to_string());
                        }
                        Some(s) => (req.id.clone(), s.created_at),
                        None => (req.id.clone(), now),
                    }
                };

                let server = crate::backend::storage::McpServer {
                    id: id.clone(),
                    name: req.name,
                    transport: req.transport,
                    config: req.config,
                    is_global: true,
                    created_at,
                    updated_at: now,
                };
                // Global-scoped uniqueness (not mcp_server_upsert_unique's
                // per-agent check): agent_config.rs merges servers into a
                // JSON object keyed by name, so two same-named global
                // servers would silently clobber each other's config for
                // every agent that has either bound. Reagent P1 on #1948.
                // Pure single-table op — called directly on identity_store.
                identity_store.mcp_server_upsert_unique_global(&server)
                    .map_err(|e| format!("mcp.catalog.upsert: {e}"))?;
                broker.publish_event::<crate::backend::mps_events::McpChanged>(&(), vec![], 0);
                Ok(server)
            }
        },
    );
}

// Catalog-tier sibling of register_mcp_bind (above) — same DB write and
// "only global servers may be bound" safety check, but no check_s1: the
// Connectors' WebSocket never authenticates as an agent (see this file's
// module doc comment), so a check_s1-gated bind can never be satisfied from
// that caller. mcp.bind itself is left untouched — it may still be the
// intended surface for an agent binding a server to itself over its own
// authenticated connection, and loosening its gate would silently widen
// what any authenticated agent connection can do (bind arbitrary servers to
// *other* agent ids), which nobody asked for.
// See docs/reports/REPORT_ARMORY_SKILLS_MARKDOWN_AND_BIND_BUG_2026_07_27.md.
fn register_mcp_catalog_bind(engine: &Arc<WshRpcEngine>, state: &AppState) {
    let mstore = state.mstore.clone();
    let identity_store = state.identity_store.clone();
    let broker = state.broker.clone();
    engine.register_typed(
        COMMAND_MCP_CATALOG_BIND,
        move |req: McpAgentBindingData, _ctx| {
            let mstore = mstore.clone();
            let identity_store = identity_store.clone();
            let broker = broker.clone();
            async move {
                // Only global servers (or ones already bound) may be bound, so
                // the catalog surface can't be used to bootstrap read access
                // to another agent's private server config — same rule as
                // mcp.bind.
                match identity_store.mcp_server_get(&req.mcp_id)
                    .map_err(|e| format!("mcp.catalog.bind: {e}"))?
                {
                    None => return Err("mcp.catalog.bind: MCP server not found".to_string()),
                    Some(s) if !s.is_global => {
                        if !mstore.mcp_server_is_bound_to(&req.agent_id, &req.mcp_id)
                            .map_err(|e| format!("mcp.catalog.bind: {e}"))?
                        {
                            return Err("FORBIDDEN: can only bind global MCP servers".to_string());
                        }
                    }
                    Some(_) => {}
                }
                mstore.mcp_server_bind(&identity_store, &req.agent_id, &req.mcp_id)
                    .map_err(|e| format!("mcp.catalog.bind: {e}"))?;
                // Lets any other open Stash/Connectors view for this agent pick up
                // the new binding without a manual refresh — mcp.bind (the
                // check_s1 agent-self-service path) intentionally left alone.
                broker.publish_event::<crate::backend::mps_events::McpChanged>(&(), vec![], 0);
                Ok(McpBindResult { bound: true })
            }
        },
    );
}

/// Catalog-tier sibling of `mcp.list` (above) — GLOBAL SERVERS ONLY, each
/// annotated with `bound_to_agent` for the caller-supplied `agent_id`, no
/// `check_s1`. AgentStashModal's MCP Servers tab (the per-agent Stash view)
/// runs over the dashboard's connection, which is never agent-authenticated,
/// so `mcp.list`'s gate can never be satisfied from that caller — the exact
/// same reasoning as `mcp.catalog.bind` above. Deliberately does NOT reuse
/// `mcp.list`'s full computation (global + this agent's own private
/// servers): with no `check_s1`, `agent_id` is unverified, and a private
/// server's `config` can carry secrets — returning it for an arbitrary
/// caller-chosen `agent_id` would be an IDOR into every agent's private MCP
/// config. See `Store::mcp_server_list_global_for_agent`'s doc comment and
/// docs/reports/REPORT_ARMORY_ARCHITECTURE_AND_NAMING_REVIEW_2026_07_23.md §2.2.
fn register_mcp_catalog_list_for_agent(engine: &Arc<WshRpcEngine>, state: &AppState) {
    let mstore = state.mstore.clone();
    let identity_store = state.identity_store.clone();
    engine.register_typed(
        COMMAND_MCP_CATALOG_LIST_FOR_AGENT,
        move |req: McpAgentScopeData, _ctx| {
            let mstore = mstore.clone();
            let identity_store = identity_store.clone();
            async move {
                let servers = mstore.mcp_server_list_global_for_agent(&identity_store, &req.agent_id)
                    .map_err(|e| format!("mcp.catalog.list_for_agent: {e}"))?;
                Ok(servers)
            }
        },
    );
}

/// Catalog-tier sibling of `mcp.unbind` (above) — same DB write, no
/// `check_s1`. Restricted to global rows only, matching the guard
/// `register_mcp_catalog_bind`/`register_mcp_catalog_delete` already apply
/// in this file: any window connection can sever any agent's binding to a
/// *global* server (an intentional, pre-existing trust boundary — bind and
/// delete already have the same or larger blast radius), but a private
/// row's own binding must not be touchable via this no-`check_s1` surface,
/// even though `mcp_id` is never exposed by any no-`check_s1` command
/// today — defense in depth over relying solely on UUID secrecy.
/// reagentx P1 on PR #2329 (round 2).
fn register_mcp_catalog_unbind(engine: &Arc<WshRpcEngine>, state: &AppState) {
    let mstore = state.mstore.clone();
    let identity_store = state.identity_store.clone();
    let broker = state.broker.clone();
    engine.register_typed(
        COMMAND_MCP_CATALOG_UNBIND,
        move |req: McpAgentBindingData, _ctx| {
            let mstore = mstore.clone();
            let identity_store = identity_store.clone();
            let broker = broker.clone();
            async move {
                match identity_store.mcp_server_get(&req.mcp_id)
                    .map_err(|e| format!("mcp.catalog.unbind: {e}"))?
                {
                    None => return Err("mcp.catalog.unbind: MCP server not found".to_string()),
                    Some(s) if !s.is_global => {
                        return Err("FORBIDDEN: can only unbind global MCP servers".to_string());
                    }
                    Some(_) => {}
                }
                let unbound = mstore.mcp_server_unbind(&req.agent_id, &req.mcp_id)
                    .map_err(|e| format!("mcp.catalog.unbind: {e}"))?;
                if unbound {
                    broker.publish_event::<crate::backend::mps_events::McpChanged>(&(), vec![], 0);
                }
                Ok(McpUnbindResult { unbound })
            }
        },
    );
}

fn register_mcp_catalog_delete(engine: &Arc<WshRpcEngine>, state: &AppState) {
    let mstore = state.mstore.clone();
    let identity_store = state.identity_store.clone();
    let broker = state.broker.clone();
    engine.register_typed(
        COMMAND_MCP_CATALOG_DELETE,
        move |req: McpCatalogItemData, _ctx| {
            let mstore = mstore.clone();
            let identity_store = identity_store.clone();
            let broker = broker.clone();
            async move {
                if let Some(existing) = identity_store.mcp_server_get(&req.id)
                    .map_err(|e| format!("mcp.catalog.delete: {e}"))?
                {
                    if !existing.is_global {
                        return Err("FORBIDDEN: cannot delete a private MCP server via the catalog".to_string());
                    }
                }
                let deleted = mstore.mcp_server_delete(&identity_store, &req.id)
                    .map_err(|e| format!("mcp.catalog.delete: {e}"))?;
                if deleted {
                    broker.publish_event::<crate::backend::mps_events::McpChanged>(&(), vec![], 0);
                }
                Ok(McpDeleteResult { deleted })
            }
        },
    );
}

/// Who a global catalog server reaches beyond its binds (every agent,
/// container agents too). Global rows only, like the rest of `mcp.catalog.*`.
/// The switches live in the identity store beside the catalog, so they
/// survive an upgrade (SPEC_CONNECTORS_MCP_WHAT_AGENTS_RUN_2026_10_10.md §2.1).
fn register_mcp_catalog_set_reach(engine: &Arc<WshRpcEngine>, state: &AppState) {
    let identity_store = state.identity_store.clone();
    let broker = state.broker.clone();
    engine.register_typed(
        COMMAND_MCP_CATALOG_SET_REACH,
        move |req: CommandMcpCatalogSetReachData, _ctx| {
            let identity_store = identity_store.clone();
            let broker = broker.clone();
            async move {
                match identity_store.mcp_server_get(&req.id).map_err(|e| format!("mcp.catalog.set_reach: {e}"))? {
                    Some(s) if s.is_global => {}
                    Some(_) => return Err("FORBIDDEN: a private MCP server has no catalog reach".to_string()),
                    None => return Err(format!("mcp.catalog.set_reach: no MCP server {}", req.id)),
                }
                let reach = crate::backend::storage::McpServerReach {
                    all_agents: req.all_agents,
                    in_containers: req.in_containers,
                };
                identity_store
                    .mcp_server_set_reach(&req.id, reach, now_ms())
                    .map_err(|e| format!("mcp.catalog.set_reach: {e}"))?;
                broker.publish_event::<crate::backend::mps_events::McpChanged>(&(), vec![], 0);
                Ok(reach)
            }
        },
    );
}

/// The MCP servers each running agent's CLI reports having
/// (`backend::mcp_observed`), with AgentMux's own sources named: `builtin`
/// for the `agentmux` server, `catalog` for a catalog server of that name.
/// Names and statuses only; a pane that no longer exists is dropped.
fn register_mcp_observed_list(engine: &Arc<WshRpcEngine>, state: &AppState) {
    let mstore = state.mstore.clone();
    let identity_store = state.identity_store.clone();
    engine.register_typed(
        COMMAND_MCP_OBSERVED_LIST,
        move |_req: McpObservedListData, _ctx| {
            let mstore = mstore.clone();
            let identity_store = identity_store.clone();
            async move {
                let catalog: std::collections::HashSet<String> = mstore
                    .mcp_server_list_global(&identity_store)
                    .map_err(|e| format!("mcp.observed.list: {e}"))?
                    .into_iter()
                    .map(|item| item.server.name)
                    .collect();
                Ok(observed_agents(&mstore, &identity_store, &catalog))
            }
        },
    );
}

/// Where an observed server comes from: AgentMux's own sources by name
/// (its `agentmux` server, a catalog server), else what the CLI said.
fn observed_source(name: &str, cli_source: String, catalog: &std::collections::HashSet<String>) -> String {
    if name == "agentmux" {
        "builtin".to_string()
    } else if catalog.contains(name) {
        "catalog".to_string()
    } else {
        cli_source
    }
}

fn observed_agents(
    mstore: &crate::backend::storage::store::Store,
    identity_store: &crate::backend::storage::store::Store,
    catalog: &std::collections::HashSet<String>,
) -> Vec<McpObservedAgent> {
    // Panes whose block is gone: their reports and exit markers go. A block
    // the store couldn't read is kept: that isn't a gone block.
    crate::backend::mcp_observed::prune(|block_id| {
        !matches!(mstore.get::<crate::backend::obj::Block>(block_id), Ok(None))
    });
    let mut agents = Vec::new();
    for (block_id, observation) in crate::backend::mcp_observed::all() {
        let Some(label) = crate::backend::tower_sampler::block_label(mstore, &block_id) else {
            continue;
        };
        let agent_id = crate::identity::resolver::block_agent_id(mstore, &block_id);
        let missing = agent_id
            .as_deref()
            .map(|id| {
                let pane_mode = mstore
                    .get::<crate::backend::obj::Block>(&block_id)
                    .ok()
                    .flatten()
                    .map(|block| crate::backend::obj::meta_get_string(&block.meta, "agentMode", ""));
                let def = mstore.agent_def_get(id).ok().flatten();
                crate::backend::mcp_launch::missing_for_pane(
                    mstore,
                    identity_store,
                    id,
                    pane_mode.as_deref(),
                    def.as_ref().map(|d| d.agent_type.as_str()),
                    &observation.servers,
                )
            })
            .unwrap_or_default();
        let servers = observation
            .servers
            .into_iter()
            .map(|s| McpObservedServer {
                source: observed_source(&s.name, s.source, catalog),
                name: s.name,
                status: s.status,
                tools: s.tools,
            })
            .collect();
        agents.push(McpObservedAgent {
            agent_id,
            label: label.label,
            at_ms: observation.at_ms,
            servers,
            missing,
            block_id,
        });
    }
    agents.sort_by(|a, b| a.label.cmp(&b.label).then_with(|| a.block_id.cmp(&b.block_id)));
    agents
}

/// Health/prerequisite probe for a global catalog server, callable without
/// an agent context (mirrors mcp.catalog.*'s window-scoped, no-`check_s1`
/// shape) — this is what lets Connectors → MCP servers show a
/// connected/error status per row before any agent has bound the server.
/// See SPEC_MCP_INTEGRATION_PARITY_ABLETON_PILOT_2026_07_08.md §4.4.
fn register_mcp_catalog_probe(engine: &Arc<WshRpcEngine>, state: &AppState) {
    let identity_store = state.identity_store.clone();
    engine.register_typed(
        COMMAND_MCP_CATALOG_PROBE,
        move |req: McpCatalogItemData, _ctx| {
            let identity_store = identity_store.clone();
            async move {
                let server = identity_store.mcp_server_get(&req.id)
                    .map_err(|e| format!("mcp.catalog.probe: {e}"))?
                    .ok_or_else(|| "mcp.catalog.probe: MCP server not found".to_string())?;
                if !server.is_global {
                    return Err("FORBIDDEN: mcp.catalog.probe only probes global servers".to_string());
                }
                let result = crate::backend::mcp_probe::probe(&server.transport, &server.config).await;
                Ok(result)
            }
        },
    );
}

#[cfg(test)]
mod observed_tests {
    use super::observed_source;

    #[test]
    fn agentmux_and_catalog_servers_are_named_as_ours_the_rest_keep_the_clis_label() {
        let catalog: std::collections::HashSet<String> = ["git".to_string()].into();
        assert_eq!(observed_source("agentmux", "project".into(), &catalog), "builtin");
        assert_eq!(observed_source("git", "project".into(), &catalog), "catalog");
        assert_eq!(observed_source("claude.ai Gmail", "claudeai".into(), &catalog), "claudeai");
        assert_eq!(observed_source("mine", "project".into(), &catalog), "project");
    }
}