feat(cli): castle mesh command + shared-config API + NATS integration tests

- castle mesh status|nodes|config (list/get/set) — hits the local castle-api;
  surfaces the shared-config write path (authority-gated)
- API: GET/PUT /mesh/config/{key:path}, GET /mesh/config; nats_client gains
  list_shared_config
- tests: NATS integration suite against a real broker via a docker fixture
  (peer discovery, presence, graceful-offline, shared config, secrets-off-wire);
  plus HTTP-layer /mesh/config tests. Closes the runtime-coverage gap.
- AGENTS.md §8 updated for NATS + the mesh CLI

100 api + 211 core tests pass.

AGENTS.md

@@ -271,11 +271,21 @@ a service public unless it authenticates or is meant to be open.
 
 ## 8. Mesh — multi-node coordination (opt-in)
 
-Disabled by default. Enable via env on `castle-api`:
+Runs on **NATS JetStream** (`castle-nats`, TLS + token). Enable via env on
-`CASTLE_API_MQTT_ENABLED=true` (+ `_MQTT_HOST`/`_PORT`), `CASTLE_API_MDNS_ENABLED=true`.
+`castle-api`: `CASTLE_API_NATS_ENABLED=true`, `CASTLE_API_NATS_URL=tls://castle-nats.<domain>:4222`,
-Nodes then advertise/discover over MQTT (Mosquitto, `castle-mqtt`) + mDNS; remote
+`CASTLE_API_NATS_TOKEN=${secret:NATS_TOKEN}`. Each node publishes its
-deployments surface as `manager: none` **reference** kinds. Inspect:
+(secret-stripped) registry to a JetStream **KV** bucket, renews a **presence**
-`GET /mesh/status`, `GET /nodes`. Modules: `castle_api.mesh`, `.mqtt_client`, `.mdns`.
+key, and watches for peers; remote deployments surface as `manager: none`
+**reference** kinds. A static **`role`** (`authority`|`follower`, in `castle.yaml`)
+gates who may write the shared-config bucket. A consumed cross-node service
+(`requires: - ref: X` satisfied by a peer) is routed by the gateway with a
+presence-gated circuit-breaker.
+
+Inspect + drive from the CLI: **`castle mesh status`** / **`castle mesh nodes`** /
+**`castle mesh config list|get|set`** (or `GET /mesh/status`, `/nodes`,
+`/mesh/config`). Modules: `castle_api.nats_client`, `.mesh`, `.mesh_gateway`,
+`.mdns`; secrets via `core` `secret_backends` (file default, OpenBao opt-in).
+→ Full history + operations: **`docs/fleet-mesh-plan.md`**.
 
 ---
 

castle-api/src/castle_api/nats_client.py

@@ -239,6 +239,15 @@ class CastleNATSClient:
             raise RuntimeError("config bucket not available")
         await self._config_kv.put(key, value.encode())
 
+    async def list_shared_config(self) -> list[str]:
+        """All shared-config keys (empty if none/unavailable)."""
+        if self._config_kv is None:
+            return []
+        try:
+            return sorted(await self._config_kv.keys())
+        except Exception:
+            return []  # empty bucket raises NoKeysError in nats-py
+
     async def _config_watch_loop(self) -> None:
         """Watch shared config; announce changes so followers can reconcile.
 

castle-api/src/castle_api/nodes.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 from fastapi import APIRouter, HTTPException, Request, status
+from pydantic import BaseModel
 
 from castle_api.config import get_registry, settings
 from castle_api.mesh import mesh_state
@@ -11,6 +12,42 @@ from castle_api.models import DeploymentSummary, MeshStatus, NodeDetail, NodeSum
 router = APIRouter(tags=["nodes"])
 
 
+class ConfigValue(BaseModel):
+    value: str
+
+
+@router.get("/mesh/config")
+async def list_mesh_config(request: Request) -> dict:
+    """List shared-config keys + this node's role (only the authority may write)."""
+    client = getattr(request.app.state, "nats_client", None)
+    if client is None:
+        return {"keys": [], "role": get_registry().node.role}
+    return {"keys": await client.list_shared_config(), "role": client.role}
+
+
+@router.get("/mesh/config/{key:path}")
+async def get_mesh_config(key: str, request: Request) -> dict:
+    """Read a shared-config value."""
+    client = getattr(request.app.state, "nats_client", None)
+    value = await client.get_shared_config(key) if client else None
+    if value is None:
+        raise HTTPException(status.HTTP_404_NOT_FOUND, f"config key '{key}' not set")
+    return {"key": key, "value": value}
+
+
+@router.put("/mesh/config/{key:path}")
+async def set_mesh_config(key: str, body: ConfigValue, request: Request) -> dict:
+    """Write a shared-config value (authority only)."""
+    client = getattr(request.app.state, "nats_client", None)
+    if client is None:
+        raise HTTPException(status.HTTP_503_SERVICE_UNAVAILABLE, "mesh not enabled")
+    try:
+        await client.put_shared_config(key, body.value)
+    except PermissionError as exc:
+        raise HTTPException(status.HTTP_403_FORBIDDEN, str(exc)) from exc
+    return {"key": key, "ok": True}
+
+
 def _local_node_summary(registry: object) -> NodeSummary:
     """Build a NodeSummary for the local node from the registry."""
     return NodeSummary(

castle-api/tests/conftest.py

@@ -1,5 +1,9 @@
 """Test fixtures for castle-api."""
 
+import socket
+import subprocess
+import time
+import urllib.request
 from collections.abc import Generator
 from pathlib import Path
 
@@ -17,6 +21,48 @@ from castle_core.registry import (
 )
 
 
+def _free_port() -> int:
+    s = socket.socket()
+    s.bind(("", 0))
+    port = s.getsockname()[1]
+    s.close()
+    return port
+
+
+def _docker_available() -> bool:
+    try:
+        return subprocess.run(
+            ["docker", "info"], capture_output=True, timeout=5
+        ).returncode == 0
+    except Exception:
+        return False
+
+
+@pytest.fixture
+def nats_url() -> Generator[str, None, None]:
+    """A throwaway NATS+JetStream broker in docker (fresh per test for clean
+    buckets). Skips if docker is unavailable."""
+    if not _docker_available():
+        pytest.skip("docker unavailable — skipping NATS integration tests")
+    cport, mport = _free_port(), _free_port()
+    name = f"castle-test-nats-{cport}"
+    subprocess.run(
+        ["docker", "run", "-d", "--rm", "--name", name,
+         "-p", f"{cport}:4222", "-p", f"{mport}:8222", "nats:2", "-js", "-m", "8222"],
+        check=True, capture_output=True,
+    )
+    try:
+        for _ in range(50):  # wait for readiness
+            try:
+                urllib.request.urlopen(f"http://127.0.0.1:{mport}/healthz", timeout=1)
+                break
+            except Exception:
+                time.sleep(0.2)
+        yield f"nats://127.0.0.1:{cport}"
+    finally:
+        subprocess.run(["docker", "rm", "-f", name], capture_output=True)
+
+
 def _write_castle_config(root: Path, config: dict) -> None:
     """Scatter a nested castle config dict into the directory-per-resource layout."""
     globals_data = {k: v for k, v in config.items() if k in ("gateway", "repo")}

castle-api/tests/test_mesh_config_api.py

@@ -0,0 +1,26 @@
+"""HTTP-layer tests for the /mesh/config endpoints (mesh-disabled paths).
+
+The write/read-through-the-client behavior is covered by test_nats_integration;
+here we pin the endpoint wiring when no mesh client is attached.
+"""
+
+from __future__ import annotations
+
+from fastapi.testclient import TestClient
+
+
+def test_list_config_reports_role_when_mesh_disabled(client: TestClient) -> None:
+    r = client.get("/mesh/config")
+    assert r.status_code == 200
+    body = r.json()
+    assert body["keys"] == []
+    assert body["role"] == "follower"  # test-node has no explicit role
+
+
+def test_get_missing_config_is_404(client: TestClient) -> None:
+    assert client.get("/mesh/config/does/not/exist").status_code == 404
+
+
+def test_write_without_mesh_is_503(client: TestClient) -> None:
+    r = client.put("/mesh/config/fleet/motd", json={"value": "x"})
+    assert r.status_code == 503

castle-api/tests/test_nats_integration.py

@@ -0,0 +1,117 @@
+"""Integration tests for CastleNATSClient against a real NATS/JetStream broker.
+
+Exercises the runtime the unit tests can't: connect, KV publish, peer discovery
+via watch, presence, graceful-offline, and shared-config. Requires docker (the
+`nats_url` fixture); skipped otherwise.
+"""
+
+from __future__ import annotations
+
+import asyncio
+
+import pytest
+from castle_core.registry import Deployment, NodeConfig, NodeRegistry
+
+import castle_api.nats_client as ncmod
+from castle_api.mesh import mesh_state
+from castle_api.nats_client import CastleNATSClient
+
+
+@pytest.fixture(autouse=True)
+def _isolate(monkeypatch):
+    """Stub the gateway-regen side effect (so tests never touch the host's real
+    Caddyfile) and clear the shared mesh_state singleton around each test."""
+
+    async def _stub(*_a, **_k):
+        return False
+
+    monkeypatch.setattr(ncmod, "refresh_remote_routes", _stub)
+    mesh_state._nodes.clear()
+    yield
+    mesh_state._nodes.clear()
+
+
+def _reg(host: str, deployed=None, role: str = "follower") -> NodeRegistry:
+    return NodeRegistry(
+        node=NodeConfig(hostname=host, role=role), deployed=deployed or {}
+    )
+
+
+def _widget_reg(host: str) -> NodeRegistry:
+    w = Deployment(
+        manager="systemd", launcher="python", run_cmd=[], name="widget",
+        kind="service", port=9099, subdomain="widget",
+    )
+    return _reg(host, {NodeRegistry.key("service", "widget"): w})
+
+
+def test_publish_peer_discovery_and_offline(nats_url: str) -> None:
+    async def run() -> None:
+        a = CastleNATSClient("node-a", _reg("node-a"), servers=nats_url)
+        b = CastleNATSClient("node-b", _widget_reg("node-b"), servers=nats_url)
+        await a.start()
+        await b.start()
+        await asyncio.sleep(1.0)  # let watches propagate
+
+        nodes = mesh_state.all_nodes(include_stale=True)
+        assert "node-b" in nodes, "node-a should discover node-b via the KV watch"
+        assert nodes["node-b"].registry.get("service", "widget") is not None
+
+        # Graceful stop deletes b's key -> the DELETE watch marks it offline.
+        await b.stop()
+        await asyncio.sleep(1.0)
+        nb = mesh_state.get_node("node-b")
+        assert nb is None or not nb.online
+
+        await a.stop()
+
+    asyncio.run(run())
+
+
+def test_presence_key_written(nats_url: str) -> None:
+    async def run() -> None:
+        c = CastleNATSClient("solo", _reg("solo"), servers=nats_url)
+        await c.start()
+        keys = await c._presence_kv.keys()
+        assert "solo" in keys
+        await c.stop()
+
+    asyncio.run(run())
+
+
+def test_shared_config_authority_write_read(nats_url: str) -> None:
+    async def run() -> None:
+        auth = CastleNATSClient("auth", _reg("auth", role="authority"), servers=nats_url)
+        await auth.start()
+        await auth.put_shared_config("fleet/motd", "hello")
+        assert await auth.get_shared_config("fleet/motd") == "hello"
+        assert "fleet/motd" in await auth.list_shared_config()
+        await auth.stop()
+
+    asyncio.run(run())
+
+
+def test_secrets_never_on_the_wire(nats_url: str) -> None:
+    """The registry a peer receives must carry no env/run_cmd."""
+    async def run() -> None:
+        secretful = Deployment(
+            manager="systemd", launcher="python",
+            run_cmd=["uv", "run", "svc"], env={"API_KEY": "s3cr3t"},
+            name="svc", kind="service", port=9001, subdomain="svc",
+        )
+        a = CastleNATSClient("wa", _reg("wa"), servers=nats_url)
+        b = CastleNATSClient(
+            "wb", _reg("wb", {NodeRegistry.key("service", "svc"): secretful}),
+            servers=nats_url,
+        )
+        await a.start()
+        await b.start()
+        await asyncio.sleep(1.0)
+        svc = mesh_state.get_node("wb").registry.get("service", "svc")
+        assert svc is not None
+        assert not svc.env, "env must not cross the wire"
+        assert not svc.run_cmd, "run_cmd must not cross the wire"
+        await a.stop()
+        await b.stop()
+
+    asyncio.run(run())

cli/src/castle_cli/commands/mesh.py

@@ -0,0 +1,105 @@
+"""castle mesh — inspect the NATS mesh and manage shared config.
+
+The mesh lives in the running castle-api (it holds the live peer state), so this
+command talks to the local API over HTTP rather than reading files.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import urllib.error
+import urllib.request
+
+
+def _api_base() -> str:
+    port = None
+    try:
+        from castle_core.config import load_config
+
+        config = load_config()
+        dep = next((d for _k, d in config.deployments_named("castle-api")), None)
+        internal = getattr(getattr(getattr(dep, "expose", None), "http", None), "internal", None)
+        port = getattr(internal, "port", None)
+    except Exception:
+        pass
+    return f"http://localhost:{port or 9020}"
+
+
+def _get(path: str):
+    with urllib.request.urlopen(_api_base() + path, timeout=5) as r:  # noqa: S310
+        return json.load(r)
+
+
+def _put(path: str, body: dict):
+    req = urllib.request.Request(  # noqa: S310
+        _api_base() + path,
+        data=json.dumps(body).encode(),
+        method="PUT",
+        headers={"Content-Type": "application/json"},
+    )
+    with urllib.request.urlopen(req, timeout=5) as r:  # noqa: S310
+        return json.load(r)
+
+
+def run_mesh(args: argparse.Namespace) -> int:
+    sub = getattr(args, "mesh_command", None) or "status"
+    try:
+        if sub == "status":
+            return _status()
+        if sub == "nodes":
+            return _nodes()
+        if sub == "config":
+            return _config(args)
+    except urllib.error.HTTPError as e:
+        detail = e.read().decode(errors="replace")
+        print(f"error: HTTP {e.code} — {detail}")
+        return 1
+    except urllib.error.URLError as e:
+        print(f"castle-api not reachable ({e.reason}). Is it running + mesh enabled?")
+        return 1
+    print(f"unknown mesh command: {sub}")
+    return 2
+
+
+def _status() -> int:
+    s = _get("/mesh/status")
+    on = "connected" if s.get("connected") else "disconnected"
+    print(f"mesh: {'enabled' if s.get('enabled') else 'disabled'} ({on})")
+    print(f"  transport: {s.get('nats_url')}")
+    print(f"  peers ({s.get('peer_count', 0)}): {', '.join(s.get('peers', [])) or '—'}")
+    return 0
+
+
+def _nodes() -> int:
+    nodes = _get("/nodes")
+    print(f"{'NODE':<14}{'STATUS':<10}{'DEPLOYED':<10}LOCAL")
+    for n in nodes:
+        if n.get("online"):
+            status = "online"
+        else:
+            status = "stale" if n.get("is_stale") else "offline"
+        local = "yes" if n.get("is_local") else ""
+        print(f"{n['hostname']:<14}{status:<10}{n.get('deployed_count', 0):<10}{local}")
+    return 0
+
+
+def _config(args: argparse.Namespace) -> int:
+    cmd = getattr(args, "mesh_config_command", None) or "list"
+    if cmd == "list":
+        data = _get("/mesh/config")
+        print(f"shared config (this node role: {data.get('role')}):")
+        for k in data.get("keys", []):
+            print(f"  {k}")
+        if not data.get("keys"):
+            print("  (none)")
+        return 0
+    if cmd == "get":
+        print(_get(f"/mesh/config/{args.key}").get("value", ""))
+        return 0
+    if cmd == "set":
+        _put(f"/mesh/config/{args.key}", {"value": args.value})
+        print(f"set {args.key}")
+        return 0
+    print(f"unknown config command: {cmd}")
+    return 2

cli/src/castle_cli/main.py

@@ -168,6 +168,20 @@ def build_parser() -> argparse.ArgumentParser:
     gw_sub = gw.add_subparsers(dest="gateway_command")
     gw_sub.add_parser("status", help="Show gateway status + routes (the default)")
 
+    # Mesh — inspect nodes + manage authority-written shared config.
+    mesh = subparsers.add_parser("mesh", help="Inspect the mesh + shared config")
+    mesh_sub = mesh.add_subparsers(dest="mesh_command")
+    mesh_sub.add_parser("status", help="Mesh coordination status (the default)")
+    mesh_sub.add_parser("nodes", help="List mesh nodes (local + remote)")
+    mc = mesh_sub.add_parser("config", help="Shared config (only the authority writes)")
+    mc_sub = mc.add_subparsers(dest="mesh_config_command")
+    mc_sub.add_parser("list", help="List shared-config keys")
+    mc_get = mc_sub.add_parser("get", help="Get a shared-config value")
+    mc_get.add_argument("key")
+    mc_set = mc_sub.add_parser("set", help="Set a shared-config value (authority only)")
+    mc_set.add_argument("key")
+    mc_set.add_argument("value")
+
     # TLS material for raw-TCP services (cert cut from the gateway wildcard).
     tls = subparsers.add_parser(
         "tls", help="Manage castle-materialized TLS certs for raw-TCP services"
@@ -320,6 +334,10 @@ def main() -> int:
         from castle_cli.commands.gateway import run_gateway
 
         return run_gateway(args)
+    if cmd == "mesh":
+        from castle_cli.commands.mesh import run_mesh
+
+        return run_mesh(args)
     if cmd == "tls":
         from castle_cli.commands.tls import run_tls