Merge pull request #3 from payneio/feat/fleet-mesh-nats-openbao

Fleet mesh: NATS transport + OpenBao secrets (Phases 0–2, 4-read)

app/src/components/MeshPanel.tsx

@@ -19,25 +19,23 @@ export function MeshPanel({ mesh }: MeshPanelProps) {
           <span
             className={cn(
               "inline-flex items-center gap-1.5 text-xs px-2 py-0.5 rounded-full",
-              mesh.mqtt_connected
+              mesh.connected
                 ? "bg-green-800/50 text-green-300"
                 : "bg-red-800/50 text-red-300",
             )}
           >
-            {mesh.mqtt_connected ? (
+            {mesh.connected ? (
               <Wifi size={10} />
             ) : (
               <WifiOff size={10} />
             )}
-            {mesh.mqtt_connected ? "connected" : "disconnected"}
+            {mesh.connected ? "connected" : "disconnected"}
           </span>
         </div>
 
         <div className="flex items-center gap-4 text-xs text-[var(--muted)]">
-          {mesh.mqtt_broker_host && (
+          {mesh.nats_url && (
-            <span className="font-mono">
+            <span className="font-mono">{mesh.nats_url}</span>
-              mqtt://{mesh.mqtt_broker_host}:{mesh.mqtt_broker_port}
-            </span>
           )}
           {mesh.mdns_enabled && (
             <span>mDNS active</span>

app/src/pages/SystemMap.tsx

@@ -399,8 +399,6 @@ interface NodeMeta {
   kind: string
   remote: boolean
   node: string | null
-  provides: string[]
-  capsConsumes: string[]
   reach: string | null
   exposable: boolean
   launchUrl?: string
@@ -882,8 +880,6 @@ export function SystemMapPage() {
         kind: n.kind,
         remote: false,
         node: null,
-        provides: n.provides ?? [],
-        capsConsumes: n.consumes ?? [],
         reach: n.reach,
         exposable: n.kind === "service" || n.kind === "static",
         launchUrl: n.kind === "reference" ? (n.base_url ?? undefined) : launchOf(n),
@@ -894,8 +890,6 @@ export function SystemMapPage() {
         kind: ri.md.kind,
         remote: true,
         node: ri.machine,
-        provides: [],
-        capsConsumes: [],
         reach: null,
         exposable: false,
         launchUrl: remoteLaunchOf(ri.md),
@@ -1193,8 +1187,6 @@ export function SystemMapPage() {
       kind: m.kind,
       remote: m.remote,
       node: m.node,
-      provides: m.provides,
-      capsConsumes: m.capsConsumes,
       launchUrl: m.launchUrl,
       reach: m.reach,
       exposable: m.exposable,
@@ -1422,8 +1414,6 @@ function InspectPanel({
     remote: boolean
     node: string | null
     launchUrl?: string
-    provides: string[]
-    capsConsumes: string[]
     consumes: Consume[]
     consumedBy: Dependent[]
   }
@@ -1488,20 +1478,6 @@ function InspectPanel({
             </div>
           </PanelSection>
         )}
-        {(info.provides.length > 0 || info.capsConsumes.length > 0) && (
-          <PanelSection title="Capabilities">
-            {info.provides.map((t) => (
-              <span key={`p-${t}`} className="mr-1 inline-block rounded bg-green-900/40 px-1 text-[9px] text-green-300" title="provides">
-                provides {t}
-              </span>
-            ))}
-            {info.capsConsumes.map((t) => (
-              <span key={`c-${t}`} className="mr-1 inline-block rounded bg-blue-900/40 px-1 text-[9px] text-blue-300" title="needs">
-                needs {t}
-              </span>
-            ))}
-          </PanelSection>
-        )}
         <PanelSection title={`Consumes (${info.consumes.length})`}>
           {info.consumes.length === 0 && <Empty />}
           {info.consumes.map((c) => (

app/src/types/index.ts

@@ -325,9 +325,8 @@ export interface NodeDetail extends NodeSummary {
 
 export interface MeshStatus {
   enabled: boolean
-  mqtt_connected: boolean
+  connected: boolean
-  mqtt_broker_host: string | null
+  nats_url: string | null
-  mqtt_broker_port: number | null
   mdns_enabled: boolean
   peer_count: number
   peers: string[]

castle-api/pyproject.toml

@@ -9,7 +9,7 @@ dependencies = [
     "pydantic-settings>=2.0.0",
     "httpx>=0.27.0",
     "castle-core",
-    "paho-mqtt>=2.0.0",
+    "nats-py>=2.9.0",
     "zeroconf>=0.131.0",
 ]
 

castle-api/src/castle_api/config.py

@@ -15,9 +15,8 @@ class Settings(BaseSettings):
     port: int = 9020
 
     # Mesh coordination (all off by default — single-node works without them)
-    mqtt_enabled: bool = False
+    nats_enabled: bool = False
-    mqtt_host: str = "localhost"
+    nats_url: str = "nats://localhost:4222"
-    mqtt_port: int = 1883
     mdns_enabled: bool = False
 
     model_config = {

castle-api/src/castle_api/main.py

@@ -56,25 +56,24 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
     poll_task = asyncio.create_task(health_poll_loop())
 
     # --- Mesh coordination (opt-in) ---
-    mqtt_client = None
+    nats_client = None
     mdns_service = None
 
-    if settings.mqtt_enabled:
+    if settings.nats_enabled:
         try:
-            from castle_api.mqtt_client import CastleMQTTClient
+            from castle_api.nats_client import CastleNATSClient
 
             registry = get_registry()
-            mqtt_client = CastleMQTTClient(
+            nats_client = CastleNATSClient(
                 local_hostname=registry.node.hostname,
                 local_registry=registry,
-                broker_host=settings.mqtt_host,
+                servers=settings.nats_url,
-                broker_port=settings.mqtt_port,
             )
-            await mqtt_client.start()
+            await nats_client.start()
-            app.state.mqtt_client = mqtt_client
+            app.state.nats_client = nats_client
         except Exception:
-            logger.exception("Failed to start MQTT client")
+            logger.exception("Failed to start NATS mesh client")
-            mqtt_client = None
+            nats_client = None
 
     if settings.mdns_enabled:
         try:
@@ -97,8 +96,8 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
     _shutting_down = True
     poll_task.cancel()
 
-    if mqtt_client:
+    if nats_client:
-        await mqtt_client.stop()
+        await nats_client.stop()
     if mdns_service:
         mdns_service.stop()
 

castle-api/src/castle_api/mdns.py

@@ -1,8 +1,7 @@
-"""mDNS discovery for castle mesh — zero-config LAN peer and broker discovery.
+"""mDNS discovery for castle mesh — zero-config LAN peer discovery.
 
-Advertises this node as _castle._tcp and browses for:
+Advertises this node as _castle._tcp and browses for _castle._tcp peers. This is
-  - _castle._tcp  — peer castle nodes
+the on-LAN convenience path; cross-network meshes use explicit NATS seed URLs.
-  - _mqtt._tcp    — MQTT broker address
 """
 
 from __future__ import annotations
@@ -15,11 +14,10 @@ from zeroconf import ServiceBrowser, ServiceInfo, ServiceStateChange, Zeroconf
 logger = logging.getLogger(__name__)
 
 CASTLE_SERVICE_TYPE = "_castle._tcp.local."
-MQTT_SERVICE_TYPE = "_mqtt._tcp.local."
 
 
 class CastleMDNS:
-    """Advertise this castle node and discover peers + MQTT broker via mDNS."""
+    """Advertise this castle node and discover peers via mDNS."""
 
     def __init__(
         self,
@@ -38,7 +36,6 @@ class CastleMDNS:
         self.peers: dict[
             str, dict
         ] = {}  # hostname -> {gateway_port, api_port, addresses}
-        self.mqtt_broker: dict | None = None  # {host, port} or None
 
     def _on_service_state_change(
         self,
@@ -55,8 +52,6 @@ class CastleMDNS:
 
             if service_type == CASTLE_SERVICE_TYPE:
                 self._handle_castle_peer(info)
-            elif service_type == MQTT_SERVICE_TYPE:
-                self._handle_mqtt_broker(info)
 
         elif state_change == ServiceStateChange.Removed:
             if service_type == CASTLE_SERVICE_TYPE:
@@ -89,20 +84,6 @@ class CastleMDNS:
         }
         logger.info("mDNS: discovered peer %s at %s", peer_hostname, addresses)
 
-    def _handle_mqtt_broker(self, info: ServiceInfo) -> None:
-        """Process a discovered MQTT broker."""
-        addresses = [
-            socket.inet_ntoa(addr) for addr in info.addresses if len(addr) == 4
-        ]
-        if addresses:
-            self.mqtt_broker = {
-                "host": addresses[0],
-                "port": info.port,
-            }
-            logger.info(
-                "mDNS: discovered MQTT broker at %s:%d", addresses[0], info.port
-            )
-
     def start(self) -> None:
         """Start advertising and browsing."""
         self._zeroconf = Zeroconf()
@@ -123,7 +104,7 @@ class CastleMDNS:
             "mDNS: advertising %s on port %d", self._hostname, self._gateway_port
         )
 
-        # Browse for peers and MQTT broker
+        # Browse for peer castle nodes
         self._browsers.append(
             ServiceBrowser(
                 self._zeroconf,
@@ -131,13 +112,6 @@ class CastleMDNS:
                 handlers=[self._on_service_state_change],
             )
         )
-        self._browsers.append(
-            ServiceBrowser(
-                self._zeroconf,
-                MQTT_SERVICE_TYPE,
-                handlers=[self._on_service_state_change],
-            )
-        )
 
     def stop(self) -> None:
         """Stop advertising and close."""

castle-api/src/castle_api/mesh_gateway.py

@@ -0,0 +1,55 @@
+"""Regenerate the gateway with cross-node (remote) routes from the live mesh.
+
+Local routes come from `castle apply` (static). Remote routes are dynamic — they
+appear/vanish as peers join/leave — so the API owns them: on a mesh change it
+re-renders the Caddyfile (same generator `apply` uses, plus remote routes for
+online peers) and reloads the gateway iff the content changed.
+
+Safety: with no cross-node `requires`, the output equals the local-only Caddyfile,
+so this is a verified no-op until this node actually consumes a peer service.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+import subprocess
+
+from castle_core.config import SPECS_DIR
+from castle_core.generators.caddyfile import generate_caddyfile_from_registry
+
+from castle_api.config import get_registry
+from castle_api.mesh import mesh_state
+
+logger = logging.getLogger(__name__)
+
+_GATEWAY_UNIT = "castle-castle-gateway.service"
+
+
+def _regenerate(reload: bool) -> bool:
+    try:
+        reg = get_registry()
+    except Exception:
+        return False
+    remotes = {h: n.registry for h, n in mesh_state.all_nodes().items()}
+    try:
+        content = generate_caddyfile_from_registry(reg, remotes)
+    except Exception:
+        logger.exception("mesh gateway: Caddyfile generation failed")
+        return False
+    path = SPECS_DIR / "Caddyfile"
+    old = path.read_text() if path.exists() else ""
+    if content == old:
+        return False
+    path.write_text(content)
+    logger.info("mesh gateway: Caddyfile updated with cross-node routes")
+    if reload:
+        subprocess.run(
+            ["systemctl", "--user", "reload", _GATEWAY_UNIT], check=False
+        )
+    return True
+
+
+async def refresh_remote_routes(reload: bool = True) -> bool:
+    """Async wrapper — runs the blocking regen off the event loop."""
+    return await asyncio.to_thread(_regenerate, reload)

castle-api/src/castle_api/mesh_wire.py

@@ -0,0 +1,105 @@
+"""Mesh wire format — (de)serialize a NodeRegistry for cross-node transport.
+
+Transport-agnostic (no MQTT/NATS imports). Only the fields needed for mesh
+routing are included; env vars, run_cmd, and castle_root are **excluded** to
+avoid leaking secrets — this invariant is load-bearing and must be preserved by
+any transport that carries this payload.
+"""
+
+from __future__ import annotations
+
+import json
+
+from castle_core.registry import (
+    Deployment,
+    NodeConfig,
+    NodeRegistry,
+)
+
+
+def registry_to_json(registry: NodeRegistry) -> str:
+    """Serialize a NodeRegistry to JSON (secret-stripped)."""
+    data: dict = {
+        "node": {
+            "hostname": registry.node.hostname,
+            "gateway_port": registry.node.gateway_port,
+            # acme domain — lets peers build launch URLs (<subdomain>.<gateway_domain>)
+            # for this node's exposed apps. Omitted when the node has no domain.
+            "gateway_domain": registry.node.gateway_domain,
+            # fleet role — so peers know which node is the config/secret authority.
+            "role": registry.node.role,
+            # routable host peers proxy to for this node's services.
+            "address": registry.node.address,
+        },
+        "deployed": {},
+    }
+
+    for _kind, name, comp in registry.all():
+        entry: dict = {
+            "manager": comp.manager,
+            "launcher": comp.launcher,
+            "kind": comp.kind,
+        }
+        if comp.stack:
+            entry["stack"] = comp.stack
+        if comp.description:
+            entry["description"] = comp.description
+        if comp.port is not None:
+            entry["port"] = comp.port
+        if comp.health_path:
+            entry["health_path"] = comp.health_path
+        if comp.subdomain:
+            entry["subdomain"] = comp.subdomain
+        if comp.schedule:
+            entry["schedule"] = comp.schedule
+        if comp.managed:
+            entry["managed"] = comp.managed
+        # Socket surface + external target — so a peer can resolve cross-node
+        # consumption endpoints (still no secrets: only ports/URLs).
+        if getattr(comp, "tcp_port", None) is not None:
+            entry["tcp_port"] = comp.tcp_port
+        if getattr(comp, "base_url", None):
+            entry["base_url"] = comp.base_url
+        # requires — deployment refs (no secrets), so peers can draw cross-node deps.
+        if getattr(comp, "requires", None):
+            entry["requires"] = comp.requires
+        data["deployed"][NodeRegistry.key(comp.kind, name)] = entry
+
+    return json.dumps(data)
+
+
+def json_to_registry(payload: str) -> NodeRegistry:
+    """Deserialize a NodeRegistry from a JSON payload."""
+    data = json.loads(payload)
+    node_data = data.get("node", {})
+    node = NodeConfig(
+        hostname=node_data.get("hostname", ""),
+        castle_root=node_data.get("castle_root"),
+        gateway_port=node_data.get("gateway_port", 9000),
+        gateway_domain=node_data.get("gateway_domain"),
+        role=node_data.get("role", "follower"),
+        address=node_data.get("address"),
+    )
+    deployed: dict[str, Deployment] = {}
+    for key, comp_data in data.get("deployed", {}).items():
+        key_kind, name = key.split("/", 1) if "/" in key else (None, key)
+        kind = comp_data.get("kind") or key_kind or "service"
+        deployed[NodeRegistry.key(kind, name)] = Deployment(
+            manager=comp_data.get("manager", "systemd"),
+            launcher=comp_data.get("launcher"),
+            run_cmd=comp_data.get("run_cmd", []),
+            env=comp_data.get("env", {}),
+            description=comp_data.get("description"),
+            name=name,
+            kind=kind,
+            stack=comp_data.get("stack"),
+            port=comp_data.get("port"),
+            health_path=comp_data.get("health_path"),
+            subdomain=comp_data.get("subdomain"),
+            schedule=comp_data.get("schedule"),
+            managed=comp_data.get("managed", False),
+            tcp_port=comp_data.get("tcp_port"),
+            base_url=comp_data.get("base_url"),
+            requires=comp_data.get("requires", []),
+        )
+    return NodeRegistry(node=node, deployed=deployed)

castle-api/src/castle_api/models.py

@@ -218,9 +218,8 @@ class MeshStatus(BaseModel):
     """Current state of the mesh coordination layer."""
 
     enabled: bool = False
-    mqtt_connected: bool = False
+    connected: bool = False
-    mqtt_broker_host: str | None = None
+    nats_url: str | None = None
-    mqtt_broker_port: int | None = None
     mdns_enabled: bool = False
     peer_count: int = 0
     peers: list[str] = []

castle-api/src/castle_api/mqtt_client.py

@@ -1,283 +0,0 @@
-"""MQTT client for inter-node mesh coordination.
-
-Topics:
-  castle/{hostname}/registry  — retained JSON NodeRegistry, published on connect
-  castle/{hostname}/status    — "online" (retained) / "offline" (LWT)
-
-On incoming messages from other hostnames, updates MeshStateManager.
-"""
-
-from __future__ import annotations
-
-import asyncio
-import json
-import logging
-
-import paho.mqtt.client as mqtt
-
-from castle_core.registry import (
-    Deployment,
-    NodeConfig,
-    NodeRegistry,
-)
-
-from castle_api.mesh import mesh_state
-from castle_api.stream import broadcast
-
-logger = logging.getLogger(__name__)
-
-
-def _registry_to_json(registry: NodeRegistry) -> str:
-    """Serialize a NodeRegistry to JSON for MQTT publishing.
-
-    Only includes fields needed for mesh routing — env vars, run_cmd,
-    and castle_root are excluded to avoid leaking secrets.
-    """
-    data: dict = {
-        "node": {
-            "hostname": registry.node.hostname,
-            "gateway_port": registry.node.gateway_port,
-            # acme domain — lets peers build launch URLs (<subdomain>.<gateway_domain>)
-            # for this node's exposed apps. Omitted when the node has no domain.
-            "gateway_domain": registry.node.gateway_domain,
-        },
-        "deployed": {},
-    }
-
-    for _kind, name, comp in registry.all():
-        entry: dict = {
-            "manager": comp.manager,
-            "launcher": comp.launcher,
-            "kind": comp.kind,
-        }
-        if comp.stack:
-            entry["stack"] = comp.stack
-        if comp.description:
-            entry["description"] = comp.description
-        if comp.port is not None:
-            entry["port"] = comp.port
-        if comp.health_path:
-            entry["health_path"] = comp.health_path
-        if comp.subdomain:
-            entry["subdomain"] = comp.subdomain
-        if comp.schedule:
-            entry["schedule"] = comp.schedule
-        if comp.managed:
-            entry["managed"] = comp.managed
-        # Socket surface + external target — so a peer can resolve cross-node
-        # consumption endpoints (still no secrets: only ports/URLs).
-        if getattr(comp, "tcp_port", None) is not None:
-            entry["tcp_port"] = comp.tcp_port
-        if getattr(comp, "base_url", None):
-            entry["base_url"] = comp.base_url
-        # requires — deployment refs (no secrets), so peers can draw cross-node deps.
-        if getattr(comp, "requires", None):
-            entry["requires"] = comp.requires
-        data["deployed"][NodeRegistry.key(comp.kind, name)] = entry
-
-    return json.dumps(data)
-
-
-def _json_to_registry(payload: str) -> NodeRegistry:
-    """Deserialize a NodeRegistry from MQTT JSON payload."""
-    data = json.loads(payload)
-    node_data = data.get("node", {})
-    node = NodeConfig(
-        hostname=node_data.get("hostname", ""),
-        castle_root=node_data.get("castle_root"),
-        gateway_port=node_data.get("gateway_port", 9000),
-        gateway_domain=node_data.get("gateway_domain"),
-    )
-    deployed: dict[str, Deployment] = {}
-    for key, comp_data in data.get("deployed", {}).items():
-        key_kind, name = key.split("/", 1) if "/" in key else (None, key)
-        kind = comp_data.get("kind") or key_kind or "service"
-        deployed[NodeRegistry.key(kind, name)] = Deployment(
-            manager=comp_data.get("manager", "systemd"),
-            launcher=comp_data.get("launcher"),
-            run_cmd=comp_data.get("run_cmd", []),
-            env=comp_data.get("env", {}),
-            description=comp_data.get("description"),
-            name=name,
-            kind=kind,
-            stack=comp_data.get("stack"),
-            port=comp_data.get("port"),
-            health_path=comp_data.get("health_path"),
-            subdomain=comp_data.get("subdomain"),
-            schedule=comp_data.get("schedule"),
-            managed=comp_data.get("managed", False),
-            tcp_port=comp_data.get("tcp_port"),
-            base_url=comp_data.get("base_url"),
-            requires=comp_data.get("requires", []),
-        )
-    return NodeRegistry(node=node, deployed=deployed)
-
-
-class CastleMQTTClient:
-    """Async wrapper around paho-mqtt for castle mesh coordination."""
-
-    def __init__(
-        self,
-        local_hostname: str,
-        local_registry: NodeRegistry,
-        broker_host: str = "localhost",
-        broker_port: int = 1883,
-        loop: asyncio.AbstractEventLoop | None = None,
-    ) -> None:
-        self._local_hostname = local_hostname
-        self._local_registry = local_registry
-        self._broker_host = broker_host
-        self._broker_port = broker_port
-        self._loop = loop or asyncio.get_event_loop()
-
-        self._client = mqtt.Client(
-            callback_api_version=mqtt.CallbackAPIVersion.VERSION2,
-            client_id=f"castle-{local_hostname}",
-            clean_session=True,
-        )
-
-        # LWT: if we disconnect unexpectedly, broker publishes "offline"
-        self._client.will_set(
-            f"castle/{local_hostname}/status",
-            payload="offline",
-            qos=1,
-            retain=True,
-        )
-
-        self._connected = False
-        self._client.on_connect = self._on_connect
-        self._client.on_disconnect = self._on_disconnect
-        self._client.on_message = self._on_message
-
-    @property
-    def connected(self) -> bool:
-        return self._connected
-
-    @property
-    def broker_host(self) -> str:
-        return self._broker_host
-
-    @property
-    def broker_port(self) -> int:
-        return self._broker_port
-
-    def _on_disconnect(
-        self,
-        client: mqtt.Client,
-        userdata: object,
-        flags: mqtt.DisconnectFlags,
-        rc: mqtt.ReasonCode,
-        properties: mqtt.Properties | None = None,
-    ) -> None:
-        self._connected = False
-        logger.info("Disconnected from MQTT broker (rc=%s)", rc)
-
-    def _on_connect(
-        self,
-        client: mqtt.Client,
-        userdata: object,
-        flags: mqtt.ConnectFlags,
-        rc: mqtt.ReasonCode,
-        properties: mqtt.Properties | None = None,
-    ) -> None:
-        """Called when connected to broker — publish our state and subscribe."""
-        if rc.is_failure:
-            logger.error("MQTT connect failed: %s", rc)
-            self._connected = False
-            return
-
-        self._connected = True
-        logger.info(
-            "Connected to MQTT broker at %s:%d", self._broker_host, self._broker_port
-        )
-
-        # Publish our status as online (retained)
-        client.publish(
-            f"castle/{self._local_hostname}/status",
-            payload="online",
-            qos=1,
-            retain=True,
-        )
-
-        # Publish our registry (retained)
-        self.publish_registry(self._local_registry)
-
-        # Subscribe to all castle nodes
-        client.subscribe("castle/+/registry", qos=1)
-        client.subscribe("castle/+/status", qos=1)
-
-    def _on_message(
-        self,
-        client: mqtt.Client,
-        userdata: object,
-        msg: mqtt.MQTTMessage,
-    ) -> None:
-        """Process incoming MQTT messages from other nodes."""
-        try:
-            parts = msg.topic.split("/")
-            if len(parts) != 3 or parts[0] != "castle":
-                return
-
-            hostname = parts[1]
-            msg_type = parts[2]
-
-            # Ignore our own messages
-            if hostname == self._local_hostname:
-                return
-
-            payload = msg.payload.decode()
-
-            if msg_type == "registry":
-                registry = _json_to_registry(payload)
-                mesh_state.update_node(hostname, registry)
-                # Notify SSE clients about mesh change
-                asyncio.run_coroutine_threadsafe(
-                    broadcast("mesh", {"event": "node_updated", "hostname": hostname}),
-                    self._loop,
-                )
-
-            elif msg_type == "status":
-                if payload == "offline":
-                    mesh_state.set_offline(hostname)
-                    asyncio.run_coroutine_threadsafe(
-                        broadcast(
-                            "mesh", {"event": "node_offline", "hostname": hostname}
-                        ),
-                        self._loop,
-                    )
-
-        except Exception:
-            logger.exception("Error processing MQTT message on %s", msg.topic)
-
-    def publish_registry(self, registry: NodeRegistry) -> None:
-        """Publish (or re-publish) our local registry."""
-        self._local_registry = registry
-        self._client.publish(
-            f"castle/{self._local_hostname}/registry",
-            payload=_registry_to_json(registry),
-            qos=1,
-            retain=True,
-        )
-
-    async def start(self) -> None:
-        """Connect to the broker and start the network loop."""
-        self._client.connect_async(self._broker_host, self._broker_port)
-        self._client.loop_start()
-        logger.info(
-            "MQTT client starting (broker=%s:%d)",
-            self._broker_host,
-            self._broker_port,
-        )
-
-    async def stop(self) -> None:
-        """Disconnect and stop the network loop."""
-        # Publish offline status before disconnecting
-        self._client.publish(
-            f"castle/{self._local_hostname}/status",
-            payload="offline",
-            qos=1,
-            retain=True,
-        )
-        self._client.loop_stop()
-        self._client.disconnect()
-        logger.info("MQTT client stopped")

castle-api/src/castle_api/nats_client.py

@@ -0,0 +1,261 @@
+"""NATS JetStream client for inter-node mesh coordination.
+
+Replaces the MQTT transport. State lives in a JetStream **KV bucket** rather than
+retained MQTT messages:
+
+  castle-registry  — key=<hostname>, value=secret-stripped NodeRegistry JSON
+
+Lifecycle:
+  * on connect: ensure the bucket, PUT our registry, seed local state from every
+    existing key, then watch the bucket for peer changes.
+  * heartbeat: re-PUT our registry every HEARTBEAT_SEC so peers refresh their
+    last-seen clock (crash liveness rides the existing stale-TTL).
+  * graceful stop: DELETE our key → peers get an immediate offline signal.
+
+Being asyncio-native, the watch callback calls ``broadcast`` directly — no
+cross-thread ``run_coroutine_threadsafe`` hop (which the paho client needed).
+"""
+
+from __future__ import annotations
+
+import asyncio
+import contextlib
+import logging
+
+import nats
+from nats.js.api import KeyValueConfig
+
+from castle_core.registry import NodeRegistry
+
+from castle_api.mesh import mesh_state
+from castle_api.mesh_gateway import refresh_remote_routes
+from castle_api.mesh_wire import json_to_registry, registry_to_json
+from castle_api.stream import broadcast
+
+logger = logging.getLogger(__name__)
+
+REGISTRY_BUCKET = "castle-registry"
+PRESENCE_BUCKET = "castle-presence"
+CONFIG_BUCKET = "castle-config"
+HEARTBEAT_SEC = 30.0
+PRUNE_SEC = 30.0
+PRESENCE_TTL = 90.0  # a node whose presence key expires within this is gone
+
+
+class CastleNATSClient:
+    """Async NATS/JetStream mesh client."""
+
+    def __init__(
+        self,
+        local_hostname: str,
+        local_registry: NodeRegistry,
+        servers: str | list[str] = "nats://localhost:4222",
+    ) -> None:
+        self._local_hostname = local_hostname
+        self._local_registry = local_registry
+        self._servers = servers
+        self._nc: nats.NATS | None = None
+        self._kv = None
+        self._presence_kv = None
+        self._config_kv = None
+        self._tasks: list[asyncio.Task] = []
+        self._last_json: dict[str, str] = {}
+        self._online: set[str] = set()
+
+    @property
+    def connected(self) -> bool:
+        return self._nc is not None and self._nc.is_connected
+
+    @property
+    def servers(self) -> str | list[str]:
+        return self._servers
+
+    @property
+    def role(self) -> str:
+        """This node's fleet role — 'authority' or 'follower'."""
+        return self._local_registry.node.role
+
+    async def start(self) -> None:
+        """Connect, publish our registry, seed state, and start watchers."""
+        self._nc = await nats.connect(
+            self._servers,
+            name=f"castle-{self._local_hostname}",
+            max_reconnect_attempts=-1,  # reconnect forever — nodes come and go
+        )
+        js = self._nc.jetstream()
+        self._kv = await self._ensure_bucket(js, REGISTRY_BUCKET, history=1)
+        # Presence: a short-TTL key each node renews; its expiry = the node is gone.
+        self._presence_kv = await self._ensure_bucket(
+            js, PRESENCE_BUCKET, history=1, ttl=PRESENCE_TTL
+        )
+        # Shared config: authority-written, followers watch + reconcile.
+        self._config_kv = await self._ensure_bucket(js, CONFIG_BUCKET, history=5)
+
+        await self.publish_registry(self._local_registry)
+        await self._presence_kv.put(self._local_hostname, b"online")
+        await self._seed_existing()
+        await refresh_remote_routes()  # establish any cross-node routes on startup
+
+        self._tasks = [
+            asyncio.create_task(self._watch_loop()),
+            asyncio.create_task(self._config_watch_loop()),
+            asyncio.create_task(self._heartbeat_loop()),
+            asyncio.create_task(self._prune_loop()),
+        ]
+        logger.info(
+            "NATS mesh client started (servers=%s, role=%s)",
+            self._servers,
+            self.role,
+        )
+
+    @staticmethod
+    async def _ensure_bucket(js, bucket: str, *, history: int = 1, ttl=None):
+        """Bind an existing KV bucket or create it."""
+        try:
+            return await js.key_value(bucket)
+        except Exception:
+            return await js.create_key_value(
+                config=KeyValueConfig(bucket=bucket, history=history, ttl=ttl)
+            )
+
+    async def stop(self) -> None:
+        """Delete our key (immediate offline to peers) and disconnect."""
+        for t in self._tasks:
+            t.cancel()
+        for t in self._tasks:
+            with contextlib.suppress(asyncio.CancelledError, Exception):
+                await t
+        self._tasks = []
+        if self._kv is not None:
+            with contextlib.suppress(Exception):
+                await self._kv.delete(self._local_hostname)
+        if self._presence_kv is not None:
+            with contextlib.suppress(Exception):
+                await self._presence_kv.delete(self._local_hostname)
+        if self._nc is not None:
+            # Bound the drain so a wedged connection can't hang systemd shutdown.
+            with contextlib.suppress(Exception):
+                await asyncio.wait_for(self._nc.drain(), timeout=5.0)
+            with contextlib.suppress(Exception):
+                await self._nc.close()
+            self._nc = None
+        logger.info("NATS mesh client stopped")
+
+    async def publish_registry(self, registry: NodeRegistry) -> None:
+        """PUT (or refresh) our local registry into the KV bucket."""
+        self._local_registry = registry
+        if self._kv is None:
+            return
+        await self._kv.put(
+            self._local_hostname, registry_to_json(registry).encode()
+        )
+
+    async def _seed_existing(self) -> None:
+        """Load every peer key already present in the bucket."""
+        if self._kv is None:
+            return
+        try:
+            keys = await self._kv.keys()
+        except Exception:
+            keys = []  # empty bucket raises NoKeysError in nats-py
+        for key in keys:
+            if key == self._local_hostname:
+                continue
+            with contextlib.suppress(Exception):
+                entry = await self._kv.get(key)
+                if entry.value:
+                    self._apply_put(key, entry.value.decode())
+
+    async def _watch_loop(self) -> None:
+        assert self._kv is not None
+        watcher = await self._kv.watchall()
+        async for entry in watcher:
+            if entry is None:  # "caught up with current values" sentinel
+                continue
+            key = entry.key
+            if key == self._local_hostname:
+                continue
+            try:
+                if entry.operation in ("DEL", "PURGE"):
+                    await self._apply_delete(key)
+                elif entry.value:
+                    changed = self._apply_put(key, entry.value.decode())
+                    if changed:
+                        await broadcast(
+                            "mesh", {"event": "node_updated", "hostname": key}
+                        )
+                        asyncio.create_task(refresh_remote_routes())
+            except Exception:
+                logger.exception("Error handling mesh entry for %s", key)
+
+    def _apply_put(self, hostname: str, payload: str) -> bool:
+        """Update mesh state from a peer PUT. Returns True if content changed."""
+        registry = json_to_registry(payload)
+        mesh_state.update_node(hostname, registry)  # always refresh last-seen
+        self._online.add(hostname)
+        changed = self._last_json.get(hostname) != payload
+        self._last_json[hostname] = payload
+        return changed
+
+    async def _apply_delete(self, hostname: str) -> None:
+        mesh_state.set_offline(hostname)
+        self._online.discard(hostname)
+        self._last_json.pop(hostname, None)
+        await broadcast("mesh", {"event": "node_offline", "hostname": hostname})
+        asyncio.create_task(refresh_remote_routes())
+
+    async def _heartbeat_loop(self) -> None:
+        while True:
+            await asyncio.sleep(HEARTBEAT_SEC)
+            with contextlib.suppress(Exception):
+                await self.publish_registry(self._local_registry)
+            if self._presence_kv is not None:
+                with contextlib.suppress(Exception):
+                    await self._presence_kv.put(self._local_hostname, b"online")
+
+    # --- Shared config (authority writes, followers reconcile) ---
+
+    async def get_shared_config(self, key: str) -> str | None:
+        """Read a shared-config value from the mesh (None if unset)."""
+        if self._config_kv is None:
+            return None
+        try:
+            entry = await self._config_kv.get(key)
+            return entry.value.decode() if entry.value else None
+        except Exception:
+            return None
+
+    async def put_shared_config(self, key: str, value: str) -> None:
+        """Write a shared-config value. Only the authority may write."""
+        if self.role != "authority":
+            raise PermissionError("only the authority node may write shared config")
+        if self._config_kv is None:
+            raise RuntimeError("config bucket not available")
+        await self._config_kv.put(key, value.encode())
+
+    async def _config_watch_loop(self) -> None:
+        """Watch shared config; announce changes so followers can reconcile.
+
+        The reconcile action (trigger `castle apply` on a follower) hangs off this
+        SSE event; it's inert on the authority and on a single node.
+        """
+        if self._config_kv is None:
+            return
+        watcher = await self._config_kv.watchall()
+        async for entry in watcher:
+            if entry is None:
+                continue
+            op = "delete" if entry.operation in ("DEL", "PURGE") else "put"
+            await broadcast(
+                "mesh", {"event": "config_changed", "key": entry.key, "op": op}
+            )
+
+    async def _prune_loop(self) -> None:
+        """Mark crashed peers (no refresh within the stale TTL) offline."""
+        while True:
+            await asyncio.sleep(PRUNE_SEC)
+            all_nodes = mesh_state.all_nodes(include_stale=True)
+            for host in list(self._online):
+                node = all_nodes.get(host)
+                if node is None or node.is_stale:
+                    await self._apply_delete(host)

castle-api/src/castle_api/nodes.py

@@ -66,15 +66,14 @@ def _deployed_to_summaries(registry: object, hostname: str) -> list[DeploymentSu
 @router.get("/mesh/status", response_model=MeshStatus)
 def get_mesh_status(request: Request) -> MeshStatus:
     """Get the current state of the mesh coordination layer."""
-    mqtt_client = getattr(request.app.state, "mqtt_client", None)
+    nats_client = getattr(request.app.state, "nats_client", None)
 
     peers = list(mesh_state.all_nodes(include_stale=True).keys())
 
     return MeshStatus(
-        enabled=settings.mqtt_enabled,
+        enabled=settings.nats_enabled,
-        mqtt_connected=mqtt_client.connected if mqtt_client else False,
+        connected=nats_client.connected if nats_client else False,
-        mqtt_broker_host=mqtt_client.broker_host if mqtt_client else None,
+        nats_url=str(nats_client.servers) if nats_client else None,
-        mqtt_broker_port=mqtt_client.broker_port if mqtt_client else None,
         mdns_enabled=settings.mdns_enabled,
         peer_count=len(peers),
         peers=peers,

castle-api/src/castle_api/secrets.py

@@ -1,4 +1,4 @@
-"""Secrets management — read and write ~/.castle/secrets/."""
+"""Secrets management — routes through the active backend (file or OpenBao)."""
 
 from __future__ import annotations
 
@@ -6,10 +6,15 @@ from fastapi import APIRouter, HTTPException, status
 from pydantic import BaseModel
 
 from castle_core.config import SECRETS_DIR
+from castle_core.secret_backends import build_backend
 
 router = APIRouter(prefix="/secrets", tags=["secrets"])
 
 
+def _backend():
+    return build_backend(SECRETS_DIR)
+
+
 class SecretValue(BaseModel):
     value: str
 
@@ -17,31 +22,27 @@ class SecretValue(BaseModel):
 @router.get("")
 def list_secrets() -> list[str]:
     """List all secret names (not values)."""
-    if not SECRETS_DIR.exists():
+    return _backend().list_names()
-        return []
-    return sorted(f.name for f in SECRETS_DIR.iterdir() if f.is_file())
 
 
 @router.get("/{name}")
 def get_secret(name: str) -> dict:
     """Get a secret value."""
     _validate_name(name)
-    path = SECRETS_DIR / name
+    value = _backend().read(name)
-    if not path.exists():
+    if value is None:
         raise HTTPException(
             status_code=status.HTTP_404_NOT_FOUND,
             detail=f"Secret '{name}' not found",
         )
-    return {"name": name, "value": path.read_text().strip()}
+    return {"name": name, "value": value}
 
 
 @router.put("/{name}")
 def set_secret(name: str, body: SecretValue) -> dict:
     """Set a secret value."""
     _validate_name(name)
-    SECRETS_DIR.mkdir(parents=True, exist_ok=True)
+    _backend().write(name, body.value)
-    path = SECRETS_DIR / name
-    path.write_text(body.value.strip() + "\n")
     return {"name": name, "ok": True}
 
 
@@ -49,9 +50,7 @@ def set_secret(name: str, body: SecretValue) -> dict:
 def delete_secret(name: str) -> dict:
     """Delete a secret."""
     _validate_name(name)
-    path = SECRETS_DIR / name
+    _backend().delete(name)
-    if path.exists():
-        path.unlink()
     return {"name": name, "ok": True}
 
 

castle-api/tests/test_mesh_wire.py

@@ -1,10 +1,10 @@
-"""Tests for MQTT client serialization logic."""
+"""Tests for the transport-agnostic mesh wire format (registry (de)serialization)."""
 
 import json
 
 from castle_core.registry import Deployment, NodeConfig, NodeRegistry
 
-from castle_api.mqtt_client import _json_to_registry, _registry_to_json
+from castle_api.mesh_wire import json_to_registry, registry_to_json
 
 
 def _make_registry() -> NodeRegistry:
@@ -45,15 +45,15 @@ class TestRegistrySerialization:
 
     def test_round_trip(self) -> None:
         original = _make_registry()
-        json_str = _registry_to_json(original)
+        json_str = registry_to_json(original)
-        restored = _json_to_registry(json_str)
+        restored = json_to_registry(json_str)
 
         assert restored.node.hostname == "tower"
         assert restored.node.gateway_port == 9000
 
     def test_deployed_components_preserved(self) -> None:
         original = _make_registry()
-        restored = _json_to_registry(_registry_to_json(original))
+        restored = json_to_registry(registry_to_json(original))
 
         svc = restored.get("service", "my-svc")
         assert svc is not None
@@ -68,7 +68,7 @@ class TestRegistrySerialization:
 
     def test_job_fields_preserved(self) -> None:
         original = _make_registry()
-        restored = _json_to_registry(_registry_to_json(original))
+        restored = json_to_registry(registry_to_json(original))
 
         job = restored.get("job", "my-job")
         assert job is not None
@@ -87,7 +87,7 @@ class TestRegistrySerialization:
                 ),
             },
         )
-        restored = _json_to_registry(_registry_to_json(reg))
+        restored = json_to_registry(registry_to_json(reg))
         bare = restored.get("service", "bare")
         assert bare.port is None
         assert bare.health_path is None
@@ -95,10 +95,22 @@ class TestRegistrySerialization:
         assert bare.schedule is None
         assert bare.managed is False
 
+    def test_node_role_preserved(self) -> None:
+        reg = NodeRegistry(
+            node=NodeConfig(hostname="civil", role="authority"), deployed={}
+        )
+        restored = json_to_registry(registry_to_json(reg))
+        assert restored.node.role == "authority"
+
+    def test_node_role_defaults_follower(self) -> None:
+        reg = NodeRegistry(node=NodeConfig(hostname="n"), deployed={})
+        restored = json_to_registry(registry_to_json(reg))
+        assert restored.node.role == "follower"
+
     def test_no_secrets_in_payload(self) -> None:
-        """env vars, run_cmd, and castle_root must not appear in MQTT payload."""
+        """env vars, run_cmd, and castle_root must never appear on the wire."""
         original = _make_registry()
-        json_str = _registry_to_json(original)
+        json_str = registry_to_json(original)
         data = json.loads(json_str)
 
         # No castle_root in node
@@ -106,5 +118,5 @@ class TestRegistrySerialization:
 
         # No env or run_cmd in any component
         for name, comp in data["deployed"].items():
-            assert "env" not in comp, f"{name} has env in MQTT payload"
+            assert "env" not in comp, f"{name} has env in payload"
-            assert "run_cmd" not in comp, f"{name} has run_cmd in MQTT payload"
+            assert "run_cmd" not in comp, f"{name} has run_cmd in payload"

castle-api/tests/test_nats_client.py

@@ -0,0 +1,30 @@
+"""CastleNATSClient role gating — hermetic (no live NATS server needed).
+
+The write-gate is checked before touching the KV bucket, so a follower is denied
+without any connection.
+"""
+
+from __future__ import annotations
+
+import asyncio
+
+import pytest
+from castle_core.registry import NodeConfig, NodeRegistry
+
+from castle_api.nats_client import CastleNATSClient
+
+
+def _client(role: str) -> CastleNATSClient:
+    reg = NodeRegistry(node=NodeConfig(hostname="n", role=role), deployed={})
+    return CastleNATSClient("n", reg, servers="nats://localhost:4222")
+
+
+def test_role_property() -> None:
+    assert _client("authority").role == "authority"
+    assert _client("follower").role == "follower"
+
+
+def test_follower_cannot_write_shared_config() -> None:
+    client = _client("follower")
+    with pytest.raises(PermissionError):
+        asyncio.run(client.put_shared_config("fleet/key", "value"))

cli/src/castle_cli/manifest.py

@@ -4,7 +4,6 @@ from castle_core.manifest import *  # noqa: F401, F403
 from castle_core.manifest import (  # noqa: F401 — explicit re-exports for type checkers
     BuildSpec,
     CaddyDeployment,
-    Capability,
     CommandsSpec,
     DefaultsSpec,
     DeploymentBase,

core/src/castle_core/config.py

@@ -184,6 +184,9 @@ class CastleConfig:
     # built-in defaults so tests/callers that don't care stay valid.
     data_dir: Path = field(default_factory=lambda: _DEFAULT_DATA_DIR)
     repos_dir: Path = field(default_factory=lambda: _DEFAULT_REPOS_DIR)
+    # Fleet role: "authority" (may write shared config/secrets to the mesh) or
+    # "follower" (reconciles from it). Static — pinned here, no election.
+    role: str = "follower"
     # Construction convenience only (not stored): a flat name→spec dict is routed
     # into the per-kind stores by kind_for. Lets callers/tests hand us a flat map
     # without pre-splitting it; there is still no flat `deployments` attribute.
@@ -316,10 +319,15 @@ def resolve_env_vars(
 
 
 def _read_secret(name: str) -> str:
-    """Read a secret from ~/.castle/secrets/<name>. Returns placeholder if not found."""
+    """Resolve a secret via the active backend (file by default; OpenBao opt-in).
-    secret_path = SECRETS_DIR / name
+
-    if secret_path.exists():
+    Returns a ``<MISSING_SECRET:...>`` placeholder if unresolved (never raises).
-        return secret_path.read_text().strip()
+    """
+    from castle_core.secret_backends import build_backend
+
+    value = build_backend(SECRETS_DIR).read(name)
+    if value is not None:
+        return value
     return f"<MISSING_SECRET:{name}>"
 
 
@@ -465,6 +473,7 @@ def load_config(root: Path | None = None) -> CastleConfig:
         agents=agents,
         data_dir=data_dir,
         repos_dir=repos_dir,
+        role=data.get("role", "follower"),
         **stores,
     )
     return config

core/src/castle_core/deploy.py

@@ -189,6 +189,7 @@ def _node_config(config: CastleConfig) -> NodeConfig:
         public_domain=config.gateway.public_domain,
         tunnel_id=config.gateway.tunnel_id,
         cert_hook=config.gateway.cert_hook,
+        role=config.role,
     )
 
 

core/src/castle_core/generators/caddyfile.py

@@ -121,8 +121,10 @@ def compute_routes(
     """Build the ordered list of gateway routes. Every route is a host route whose
     address is the service/frontend **name** (published at ``<name>.<domain>``);
     ``proxy`` routes reverse-proxy a local port, ``static`` routes file-serve a
-    frontend's dist. Path routes no longer exist. ``remote_registries`` is accepted
+    frontend's dist. Path routes no longer exist. When ``remote_registries`` is
-    for signature compatibility but cross-node routing is out of scope here."""
+    given (online peers, keyed by hostname), ``remote`` routes are added for
+    services this node **consumes** (a local ``requires`` ref satisfied by a peer)
+    — so a consumed cross-node service is reachable at ``<ref>.<domain>``."""
     if config is None:
         try:
             from castle_core.config import load_config
@@ -141,9 +143,50 @@ def compute_routes(
     for name, kind, target in _local_routes(config, registry):
         routes.append(GatewayRoute(name, kind, target, name, node))
 
+    if remote_registries:
+        routes.extend(_remote_routes(config, registry, remote_registries))
+
     return routes
 
 
+def _remote_routes(
+    config: CastleConfig | None,
+    registry: NodeRegistry,
+    remote_registries: dict[str, NodeRegistry],
+) -> list[GatewayRoute]:
+    """Routes to services this node consumes from online peers.
+
+    A route is emitted for each local ``requires`` ref that (a) isn't satisfied
+    locally and (b) is provided by an exposed service on some peer. The route is
+    only present while the peer is (presence expiry removes the peer from
+    ``remote_registries``, which *is* the circuit-breaker: gone → no route)."""
+    # Refs this node consumes.
+    consumed: set[str] = set()
+    local_names: set[str] = set()
+    if config is not None:
+        for _kind, name, dep in config.all_deployments():
+            local_names.add(name)
+            for req in getattr(dep, "requires", []) or []:
+                ref = getattr(req, "ref", None)
+                if ref and getattr(req, "kind", "deployment") == "deployment":
+                    consumed.add(ref)
+    # Drop refs already satisfied locally.
+    consumed -= local_names
+
+    out: list[GatewayRoute] = []
+    for host, remote in sorted(remote_registries.items()):
+        addr = remote.node.address or host
+        for _kind, name, dep in remote.all():
+            if name not in consumed:
+                continue
+            if dep.subdomain and dep.port:
+                out.append(
+                    GatewayRoute(name, "remote", f"{addr}:{dep.port}", name, host)
+                )
+                consumed.discard(name)  # first online provider wins
+    return out
+
+
 def _host_matcher_block(label: str, host: str, target: str) -> list[str]:
     """A `@host_X host <host> / handle @host_X { reverse_proxy <target> }` block.
 
@@ -159,6 +202,27 @@ def _host_matcher_block(label: str, host: str, target: str) -> list[str]:
     ]
 
 
+def _host_remote_block(label: str, host: str, target: str) -> list[str]:
+    """A remote (cross-node) host route with a fail-fast breaker: a short dial
+    timeout + passive health, so an unreachable peer 502s in ~2s instead of
+    hanging. (Presence removal drops the route entirely — this guards the
+    there-but-wedged case.)"""
+    matcher = f"@host_{label.replace('-', '_').replace('.', '_')}"
+    return [
+        f"    {matcher} host {host}",
+        f"    handle {matcher} {{",
+        f"        reverse_proxy {target} {{",
+        "            lb_try_duration 1s",
+        "            fail_duration 30s",
+        "            transport http {",
+        "                dial_timeout 2s",
+        "            }",
+        "        }",
+        "    }",
+        "",
+    ]
+
+
 def _host_static_block(label: str, host: str, serve_dir: str) -> list[str]:
     """A host matcher that file-serves a frontend's dist (with SPA fallback)."""
     matcher = f"@host_{label.replace('-', '_').replace('.', '_')}"
@@ -237,6 +301,8 @@ def generate_caddyfile_from_registry(
                 host = f"{r.address}.{domain}"
                 if r.kind == "static":
                     lines += _host_static_block(r.name or r.address, host, r.target)
+                elif r.kind == "remote":
+                    lines += _host_remote_block(r.name or r.address, host, r.target)
                 else:
                     lines += _host_matcher_block(r.name or r.address, host, r.target)
             lines.append("}")

core/src/castle_core/generators/systemd.py

@@ -174,6 +174,15 @@ SuccessExitStatus=143
             reload_argv[0] = resolved_reload
         unit += f"ExecReload={' '.join(reload_argv)}\n"
 
+    # Post-start hooks (e.g. OpenBao auto-unseal). `-` prefix → failure is ignored,
+    # so a hiccup in the hook never fails the unit.
+    for cmd in (sd.exec_start_post if sd else []):
+        argv = cmd.split()
+        resolved = shutil.which(argv[0])
+        if resolved:
+            argv[0] = resolved
+        unit += f"ExecStartPost=-{' '.join(argv)}\n"
+
     if sd and sd.no_new_privileges:
         unit += "NoNewPrivileges=true\n"
 

core/src/castle_core/manifest.py

@@ -160,6 +160,9 @@ class SystemdSpec(BaseModel):
     no_new_privileges: bool = True
     readiness: ReadinessHttpGet | None = None
     exec_reload: str | None = None
+    # Commands run after the main process starts (systemd ``ExecStartPost=``), one
+    # line each — e.g. an OpenBao auto-unseal step. Failures don't fail the unit.
+    exec_start_post: list[str] = Field(default_factory=list)
 
 
 class ManageSpec(BaseModel):
@@ -277,12 +280,6 @@ class CommandsSpec(BaseModel):
 # ---------------------
 
 
-class Capability(BaseModel):
-    type: str
-    name: str | None = None
-    meta: dict[str, str] = Field(default_factory=dict)
-
-
 class Requirement(BaseModel):
     """A precondition — another **deployment** that must exist for this one to be
     *functional* (``ref`` = the target deployment's name). ``bind`` names the env
@@ -390,9 +387,6 @@ class ProgramSpec(BaseModel):
     version: str | None = None
     build: BuildSpec | None = None
 
-    provides: list[Capability] = Field(default_factory=list)
-    consumes: list[Capability] = Field(default_factory=list)
-
     tags: list[str] = Field(default_factory=list)
 
     @property

core/src/castle_core/registry.py

@@ -32,6 +32,15 @@ class NodeConfig:
     tunnel_id: str | None = None
     # Emit the cert_obtained → `castle tls reconcile` hook (needs events-exec plugin).
     cert_hook: bool = False
+    # Routable host peers use to reach this node's services (LAN IP/hostname).
+    # Defaults to the hostname; set explicitly when the hostname isn't resolvable
+    # cross-node. Used to build `remote` gateway routes to this node.
+    address: str | None = None
+    # Fleet role: "authority" may write shared config/secrets to the mesh;
+    # "follower" reconciles from it. Static (no election) — the authority is
+    # pinned in castle.yaml. When the authority is down, shared state is
+    # read-only and every node keeps serving its own deployments from cache.
+    role: str = "follower"
 
     def __post_init__(self) -> None:
         if not self.hostname:
@@ -152,6 +161,8 @@ def load_registry(path: Path | None = None) -> NodeRegistry:
         public_domain=node_data.get("public_domain"),
         tunnel_id=node_data.get("tunnel_id"),
         cert_hook=node_data.get("cert_hook", False),
+        role=node_data.get("role", "follower"),
+        address=node_data.get("address"),
     )
 
     deployed: dict[str, Deployment] = {}
@@ -241,6 +252,10 @@ def save_registry(registry: NodeRegistry, path: Path | None = None) -> None:
         data["node"]["tunnel_id"] = registry.node.tunnel_id
     if registry.node.cert_hook:
         data["node"]["cert_hook"] = registry.node.cert_hook
+    if registry.node.role and registry.node.role != "follower":
+        data["node"]["role"] = registry.node.role
+    if registry.node.address:
+        data["node"]["address"] = registry.node.address
 
     for key, comp in registry.deployed.items():
         entry: dict = {

core/src/castle_core/relations.py

@@ -77,8 +77,6 @@ class Node:
     reach: str | None = None  # off|internal|public (systemd/caddy), else None
     endpoints: list[Endpoint] = field(default_factory=list)  # sockets it exposes
     base_url: str | None = None  # the target URL, for kind=="reference" (external)
-    provides: list[str] = field(default_factory=list)  # capability types (from program)
-    consumes: list[str] = field(default_factory=list)  # capability types (from program)
 
 
 @dataclass
@@ -242,7 +240,6 @@ def build_model(
         )
         prog_name = _program_of(name, dep)
         repo_key = repo_of.get(prog_name)
-        prog = config.programs.get(prog_name) if prog_name else None
         nodes.append(
             Node(
                 name=name,
@@ -257,8 +254,6 @@ def build_model(
                 reach=getattr(getattr(dep, "reach", None), "value", None),
                 endpoints=_endpoints_of(dep),
                 base_url=getattr(dep, "base_url", None),
-                provides=[c.type for c in prog.provides] if prog else [],
-                consumes=[c.type for c in prog.consumes] if prog else [],
             )
         )
     return Model(repos=list(repos.values()), nodes=nodes, edges=edges)

core/src/castle_core/secret_backends.py

@@ -0,0 +1,137 @@
+"""Pluggable secret backends for ``${secret:NAME}`` resolution.
+
+Default is the **file** backend (``~/.castle/secrets/<name>``) — identical to the
+historical behavior, so nothing changes unless a backend is explicitly selected
+via ``CASTLE_SECRET_BACKEND``.
+
+The **openbao** backend reads from an OpenBao/Vault KV-v2 mount and falls back to
+the file backend, which is also how it bootstraps: the OpenBao *token* itself is a
+file secret (it can't live in the vault it unlocks).
+
+Selection (env, so it works in both the CLI and the systemd-run API):
+  CASTLE_SECRET_BACKEND        file | openbao        (default: file)
+  CASTLE_OPENBAO_ADDR          http://localhost:8200
+  CASTLE_OPENBAO_MOUNT         castle                (kv-v2 mount path)
+  CASTLE_OPENBAO_TOKEN_SECRET  OPENBAO_TOKEN         (file secret holding the token)
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import urllib.request
+from pathlib import Path
+from typing import Protocol
+
+
+class SecretBackend(Protocol):
+    def read(self, name: str) -> str | None: ...
+    def write(self, name: str, value: str) -> None: ...
+    def delete(self, name: str) -> None: ...
+    def list_names(self) -> list[str]: ...
+
+
+class FileSecretBackend:
+    """Reads/writes ``<secrets_dir>/<name>`` (the historical behavior)."""
+
+    def __init__(self, secrets_dir: Path) -> None:
+        self._dir = secrets_dir
+
+    def read(self, name: str) -> str | None:
+        path = self._dir / name
+        if path.exists():
+            return path.read_text().strip()
+        return None
+
+    def write(self, name: str, value: str) -> None:
+        self._dir.mkdir(parents=True, exist_ok=True)
+        (self._dir / name).write_text(value.strip() + "\n")
+
+    def delete(self, name: str) -> None:
+        path = self._dir / name
+        if path.exists():
+            path.unlink()
+
+    def list_names(self) -> list[str]:
+        if not self._dir.exists():
+            return []
+        return sorted(f.name for f in self._dir.iterdir() if f.is_file())
+
+
+class OpenBaoBackend:
+    """Reads from an OpenBao/Vault KV-v2 mount; falls back to ``fallback``.
+
+    A missing key, an auth failure, or an unreachable server all fall through to
+    the fallback — so a partly-migrated vault and the bootstrap token both resolve.
+    """
+
+    def __init__(
+        self, addr: str, token: str, mount: str, fallback: SecretBackend
+    ) -> None:
+        self._addr = addr.rstrip("/")
+        self._token = token
+        self._mount = mount
+        self._fallback = fallback
+
+    def read(self, name: str) -> str | None:
+        value = self._read_bao(name)
+        if value is not None:
+            return value
+        return self._fallback.read(name)
+
+    def _read_bao(self, name: str) -> str | None:
+        if not self._token:
+            return None
+        url = f"{self._addr}/v1/{self._mount}/data/{name}"
+        try:
+            data = self._request("GET", url)
+            return data["data"]["data"].get("value")
+        except Exception:
+            return None
+
+    def write(self, name: str, value: str) -> None:
+        url = f"{self._addr}/v1/{self._mount}/data/{name}"
+        self._request("POST", url, {"data": {"value": value.strip()}})
+
+    def delete(self, name: str) -> None:
+        # Remove all versions (metadata delete), matching file-backend semantics.
+        url = f"{self._addr}/v1/{self._mount}/metadata/{name}"
+        self._request("DELETE", url)
+
+    def list_names(self) -> list[str]:
+        url = f"{self._addr}/v1/{self._mount}/metadata?list=true"
+        try:
+            data = self._request("GET", url)
+            return sorted(data.get("data", {}).get("keys", []))
+        except Exception:
+            return []
+
+    def _request(self, method: str, url: str, body: dict | None = None) -> dict:
+        payload = json.dumps(body).encode() if body is not None else None
+        req = urllib.request.Request(  # noqa: S310
+            url,
+            data=payload,
+            method=method,
+            headers={
+                "X-Vault-Token": self._token,
+                "Content-Type": "application/json",
+            },
+        )
+        with urllib.request.urlopen(req, timeout=5) as resp:  # noqa: S310
+            raw = resp.read()
+        return json.loads(raw) if raw else {}
+
+
+def build_backend(secrets_dir: Path) -> SecretBackend:
+    """Construct the active secret backend from the environment."""
+    file_backend = FileSecretBackend(secrets_dir)
+    kind = os.environ.get("CASTLE_SECRET_BACKEND", "file").lower()
+    if kind == "openbao":
+        addr = os.environ.get("CASTLE_OPENBAO_ADDR", "http://localhost:8200")
+        mount = os.environ.get("CASTLE_OPENBAO_MOUNT", "castle")
+        token_secret = os.environ.get("CASTLE_OPENBAO_TOKEN_SECRET", "OPENBAO_TOKEN")
+        token = file_backend.read(token_secret) or os.environ.get(
+            "CASTLE_OPENBAO_TOKEN", ""
+        )
+        return OpenBaoBackend(addr, token, mount, fallback=file_backend)
+    return file_backend

core/tests/test_caddyfile_remote.py

@@ -0,0 +1,87 @@
+"""Cross-node (remote) gateway routes + presence breaker."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import yaml
+from castle_core.config import load_config
+from castle_core.generators.caddyfile import compute_routes
+from castle_core.registry import Deployment, NodeConfig, NodeRegistry
+
+
+def _config_requiring_widget(root: Path) -> None:
+    (root / "castle.yaml").write_text(yaml.safe_dump({"gateway": {"port": 18000}}))
+    svc_dir = root / "services"
+    svc_dir.mkdir()
+    (svc_dir / "consumer.yaml").write_text(
+        yaml.safe_dump(
+            {
+                "description": "consumes a remote widget",
+                "manager": "systemd",
+                "run": {"launcher": "command", "argv": ["consumer"]},
+                "requires": [{"kind": "deployment", "ref": "widget"}],
+                "manage": {"systemd": {}},
+            }
+        )
+    )
+
+
+def _peer_with_widget(address: str | None) -> NodeRegistry:
+    widget = Deployment(
+        manager="systemd",
+        launcher="python",
+        run_cmd=[],
+        name="widget",
+        kind="service",
+        port=9099,
+        subdomain="widget",
+        managed=True,
+    )
+    return NodeRegistry(
+        node=NodeConfig(hostname="tower", address=address),
+        deployed={NodeRegistry.key("service", "widget"): widget},
+    )
+
+
+def _local() -> NodeRegistry:
+    return NodeRegistry(node=NodeConfig(hostname="civil"), deployed={})
+
+
+def test_remote_route_emitted_for_consumed_peer_service(tmp_path: Path) -> None:
+    _config_requiring_widget(tmp_path)
+    config = load_config(tmp_path)
+    routes = compute_routes(
+        _local(), config, {"tower": _peer_with_widget("10.0.0.5")}
+    )
+    remote = [r for r in routes if r.kind == "remote"]
+    assert len(remote) == 1
+    assert remote[0].address == "widget"
+    assert remote[0].target == "10.0.0.5:9099"
+    assert remote[0].node == "tower"
+
+
+def test_address_falls_back_to_hostname(tmp_path: Path) -> None:
+    _config_requiring_widget(tmp_path)
+    config = load_config(tmp_path)
+    routes = compute_routes(_local(), config, {"tower": _peer_with_widget(None)})
+    remote = [r for r in routes if r.kind == "remote"]
+    assert remote[0].target == "tower:9099"
+
+
+def test_breaker_no_route_when_peer_absent(tmp_path: Path) -> None:
+    """Presence expiry removes the peer from remote_registries -> no route."""
+    _config_requiring_widget(tmp_path)
+    config = load_config(tmp_path)
+    routes = compute_routes(_local(), config, {})  # no online peers
+    assert not [r for r in routes if r.kind == "remote"]
+
+
+def test_no_remote_route_for_unconsumed_service(tmp_path: Path) -> None:
+    """A peer service nobody requires is not routed."""
+    (tmp_path / "castle.yaml").write_text(yaml.safe_dump({"gateway": {"port": 18000}}))
+    config = load_config(tmp_path)  # no requires anywhere
+    routes = compute_routes(
+        _local(), config, {"tower": _peer_with_widget("10.0.0.5")}
+    )
+    assert not [r for r in routes if r.kind == "remote"]

core/tests/test_fleet_role.py

@@ -0,0 +1,48 @@
+"""Fleet role (authority/follower) — config loading + registry round-trip."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import yaml
+from castle_core.config import load_config
+from castle_core.registry import (
+    NodeConfig,
+    NodeRegistry,
+    load_registry,
+    save_registry,
+)
+
+
+def _write_min_config(root: Path, role: str | None) -> None:
+    data: dict = {"gateway": {"port": 18000}}
+    if role is not None:
+        data["role"] = role
+    (root / "castle.yaml").write_text(yaml.safe_dump(data))
+
+
+def test_role_defaults_to_follower(tmp_path: Path) -> None:
+    _write_min_config(tmp_path, None)
+    assert load_config(tmp_path).role == "follower"
+
+
+def test_role_loaded_from_yaml(tmp_path: Path) -> None:
+    _write_min_config(tmp_path, "authority")
+    assert load_config(tmp_path).role == "authority"
+
+
+def test_registry_role_round_trip(tmp_path: Path) -> None:
+    reg = NodeRegistry(
+        node=NodeConfig(hostname="civil", role="authority"), deployed={}
+    )
+    path = tmp_path / "registry.yaml"
+    save_registry(reg, path)
+    assert load_registry(path).node.role == "authority"
+
+
+def test_registry_role_defaults_follower(tmp_path: Path) -> None:
+    """A node with no explicit role round-trips as follower."""
+    reg = NodeRegistry(node=NodeConfig(hostname="n"), deployed={})
+    path = tmp_path / "registry.yaml"
+    save_registry(reg, path)
+    assert load_registry(path).node.role == "follower"

core/tests/test_secret_backends.py

@@ -0,0 +1,67 @@
+"""Tests for the pluggable secret backends (file default, OpenBao opt-in)."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+from castle_core.secret_backends import (
+    FileSecretBackend,
+    OpenBaoBackend,
+    build_backend,
+)
+
+
+def test_file_backend_read_hit(tmp_path: Path) -> None:
+    (tmp_path / "MY_SECRET").write_text("value\n")
+    assert FileSecretBackend(tmp_path).read("MY_SECRET") == "value"
+
+
+def test_file_backend_read_miss(tmp_path: Path) -> None:
+    assert FileSecretBackend(tmp_path).read("ABSENT") is None
+
+
+def test_file_backend_write_read_list_delete(tmp_path: Path) -> None:
+    b = FileSecretBackend(tmp_path)
+    assert b.list_names() == []
+    b.write("A", "one")
+    b.write("B", "two")
+    assert b.read("A") == "one"
+    assert b.list_names() == ["A", "B"]
+    b.delete("A")
+    assert b.read("A") is None
+    assert b.list_names() == ["B"]
+    b.delete("ABSENT")  # no error
+
+
+def test_build_backend_defaults_to_file(tmp_path: Path, monkeypatch) -> None:
+    monkeypatch.delenv("CASTLE_SECRET_BACKEND", raising=False)
+    assert isinstance(build_backend(tmp_path), FileSecretBackend)
+
+
+def test_build_backend_openbao_selected(tmp_path: Path, monkeypatch) -> None:
+    monkeypatch.setenv("CASTLE_SECRET_BACKEND", "openbao")
+    assert isinstance(build_backend(tmp_path), OpenBaoBackend)
+
+
+def test_openbao_falls_back_to_file_when_unreachable(tmp_path: Path) -> None:
+    """An unreachable vault (or empty token) resolves via the file fallback."""
+    (tmp_path / "ONLY_IN_FILE").write_text("from-file")
+    backend = OpenBaoBackend(
+        addr="http://127.0.0.1:1",  # nothing listening
+        token="dummy",
+        mount="castle",
+        fallback=FileSecretBackend(tmp_path),
+    )
+    assert backend.read("ONLY_IN_FILE") == "from-file"
+    assert backend.read("NOT_ANYWHERE") is None
+
+
+def test_openbao_empty_token_uses_fallback(tmp_path: Path) -> None:
+    (tmp_path / "K").write_text("v")
+    backend = OpenBaoBackend(
+        addr="http://127.0.0.1:8200",
+        token="",  # no token → never hits the network
+        mount="castle",
+        fallback=FileSecretBackend(tmp_path),
+    )
+    assert backend.read("K") == "v"

docs/fleet-mesh-plan.md

@@ -0,0 +1,302 @@
+# Fleet Mesh Plan — OpenBao + NATS
+
+**Status:** in progress on branch `feat/fleet-mesh-nats-openbao`.
+Phases 0–2 + Phase 4 (secret-read backend) complete + single-node verified (live).
+Phase 3 (cross-node routing/breaker) and Phase 4 hardening pending the 2nd node.
+
+## Context
+
+Castle's mesh today is a deliberately minimal, read-only gossip: each node
+publishes a secret-stripped `NodeRegistry` to Mosquitto MQTT (retained) + an
+online/offline LWT, and peers aggregate it into `MeshStateManager`. It is
+LAN-only, unauthenticated, carries no secrets, and does nothing with what it
+discovers beyond display.
+
+The goal is a **purpose-driven heterogeneous fleet** (media server, security
+appliance, TV, comms handler, AI gateway) where nodes come and go as *expected*:
+shared LAN-wide config + selected secrets, discoverability + presence, and
+binding **by purpose** (a consumer needs `media-index`, not a hostname) with
+**circuit-breakers** so churn degrades gracefully instead of hanging.
+
+Two adopted, genuinely-open components (see licensing analysis in chat history):
+
+- **NATS** (CNCF, Apache-2.0) — replaces Mosquitto as the mesh substrate and does
+  three jobs in one: pub/sub, **JetStream KV** (shared config + registry), and
+  **TTL keys as presence** (a provider's key vanishes on death → the breaker
+  trip signal, no separate health plumbing).
+- **OpenBao** (Linux Foundation / OpenSSF, MPL-2.0, Vault fork) — the secret
+  authority behind castle's existing `${secret:...}` mechanism.
+
+Design stance: **static single-writer `role` authority, no consensus** —
+availability comes from followers running cached local state, not failover.
+NATS single-node matches that topology. Consensus stays explicitly out of scope.
+
+## Architecture — planes → components → code seams
+
+| Plane | Component | Primary code seam (verified) |
+|-------|-----------|------------------------------|
+| Transport + KV + presence | NATS (`castle-nats`) | `castle-api/.../mqtt_client.py` (whole file), `main.py` lifespan 62-77, `config.py` Settings |
+| Shared config + registry + presence | NATS JetStream KV | `mesh.py` `MeshStateManager`; new KV buckets |
+| Secret authority | OpenBao (`castle-openbao`) | `core/.../config.py:307` `_read_secret` (single chokepoint), `castle-api/.../secrets.py` CRUD |
+| Binding-by-purpose + breaker | Caddy gateway | `generators/caddyfile.py` `compute_routes` (`remote` kind + ignored `remote_registries` are pre-stubbed), `deploy.py:509-540` `_target_url`/`_requires_env` (local-only today) |
+| Authority role | `NodeConfig.role` | `castle_core/registry.py` / `manifest.py` |
+
+## Phased plan
+
+Each phase is independently useful; the risky secret-transport work is last.
+
+### Phase 0 — Stand up the two services (no behavior change)
+
+- **`castle-nats`**: a `SystemdDeployment` + `RunContainer` (`nats:2` with
+  JetStream `-js`), data volume `/data/castle/castle-nats`, `reach: internal`
+  (`nats.<domain>`). Follow the container YAML shape in `docs/tcp-exposure.md`
+  (postgres example). Replace the Mosquitto provisioning in `install.sh`
+  (`setup_mqtt`, `seed_mosquitto_config`, image/port consts) with NATS — or
+  better, promote it to a bootstrap deployment YAML (today the broker is only a
+  shell-provisioned container, not a declared deployment).
+- **`castle-openbao`**: `RunContainer` (`openbao/openbao`), data volume, a file
+  storage backend, `reach: internal`, **auto-unseal** (local transit/key) so the
+  node boots unattended (Decision 2).
+
+### Phase 1 — NATS as the mesh transport (swap Mosquitto, behavior-preserving)
+
+- New `nats_client.py` (async-native `nats-py`) replacing `mqtt_client.py`.
+  Registry → a **JetStream KV bucket `castle-registry`** keyed by hostname
+  (KV last-value replaces MQTT retained). Presence → a **`castle-presence`**
+  bucket with a per-node **TTL key** the node renews (replaces LWT + the 300s
+  `STALE_TTL` poll; note `prune_stale` is currently uncalled).
+- Wire in `main.py` lifespan; rename `mqtt_*` → `nats_*` in `config.py` (keep
+  `CASTLE_API_` prefix). `MeshStateManager` logic is reused; the KV **watch**
+  callback drives `update_node`/`set_offline` — and because `nats-py` is
+  asyncio-native, the `run_coroutine_threadsafe` cross-thread hop for
+  `broadcast("mesh", …)` **goes away**.
+- **Preserve the secret-stripping invariant** from `_registry_to_json` (env /
+  run_cmd / castle_root never on the wire).
+- Rename MQTT-named fields: `models.py` `MeshStatus` (217-226) + frontend
+  `types/index.ts` (326-330), `MeshPanel.tsx`. Drop `paho-mqtt`; add `nats-py`.
+  Keep zeroconf peer-advert for LAN discovery **and** support explicit NATS seed
+  URLs for cross-network (Decision 1); drop the dangling unused `_mqtt._tcp`
+  browse. Rewrite `test_mqtt.py` → `test_nats.py`.
+- Fixes a latent gap: today the registry is published **on-connect only** (no
+  periodic/on-change republish). KV writes on `castle apply` fix this naturally.
+
+### Phase 2 — Shared config + presence via JetStream KV
+
+- Buckets: `castle-config` (shared LAN config, **authority-written only**),
+  `castle-registry` (per-node), `castle-presence` (TTL liveness).
+- Add static **`role: authority | follower`** to `NodeConfig` (castle.yaml).
+  Pin **`civil` = authority** (Decision 3); only the authority may write
+  `castle-config`. Authority-down ⇒ shared state read-only, nodes serve cached.
+- Followers **watch** `castle-config` and reconcile via `castle apply`. Presence
+  key renewal is the churn signal for Phase 3.
+
+### Phase 3 — Cross-node `requires` resolution + gateway binding + breaker (keystone)
+
+- Extend `_target_url`/`_requires_env` (`deploy.py:509-540`) to **fall through to
+  the mesh registry** when a `ref` isn't satisfied locally (today it's
+  local-only). The ref is already host-independent — this removes the
+  "must be local" restriction, not adds a concept.
+- Produce the pre-stubbed **`remote` `GatewayRoute`**; thread `remote_registries`
+  into `compute_routes` (currently accepted-but-ignored) and into the Caddyfile
+  write at `deploy.py:157` (currently passed nothing). Binding = a **stable local
+  subdomain route** the gateway re-targets to the remote node's URL — the program
+  sees one unchanging URL.
+- **Circuit-breaker:** gate the remote route on the `castle-presence` key
+  (regenerate + `_reload_gateway` when a provider appears/vanishes), backed by
+  Caddy passive health (`fail_duration` / `lb_try_duration`) on the
+  `reverse_proxy` line in `_host_matcher_block`. Consumer degraded-mode =
+  fallback target or clean 503. (Also carry `health_url` into the registry —
+  it's currently dropped on `manager: none` refs.)
+
+### Phase 4 — OpenBao as the secret backend (last; needs the security foundation)
+
+- Introduce a **`SecretBackend` seam** at the single read chokepoint
+  `_read_secret` (`config.py:307`) + the `castle-api/.../secrets.py` writer:
+  `FileSecretBackend` (default, unchanged) and `OpenBaoBackend`. `${secret:NAME}`
+  syntax is untouched; backend selected in castle.yaml or by env.
+- Followers auth to OpenBao (token / AppRole); **need-to-know scoping** via
+  per-deployment policies. Keep the file backend as fallback; migrate secrets in.
+
+### Cross-cutting prerequisite (gates Phase 4 + any cross-network)
+
+- **Transport hardening**: NATS TLS + nkey/user auth (replace Mosquitto's
+  `allow_anonymous`); OpenBao TLS listener. No secret and no cross-network link
+  moves until this is in. Cross-network overlay (e.g. Nebula) is orthogonal and
+  layers under Caddy later.
+
+## Verification (per phase, end-to-end)
+
+- **P1:** bring up two nodes; confirm registry + presence propagate over NATS and
+  the mesh view (`/mesh/status`, System Map) is identical to the MQTT behavior.
+- **P2:** write a key to `castle-config` on the authority; observe a follower
+  watch fire and `castle apply` reconcile.
+- **P3:** define a service on node A that `requires` a ref provided on node B;
+  confirm the gateway routes cross-node, then **kill node B** and confirm the
+  consumer fails fast (curl returns the degraded 503, not a hang) and **recovers**
+  when B returns.
+- **P4:** store a secret in OpenBao; confirm a service renders it via
+  `EnvironmentFile=` / `--env-file`; then disable the file fallback and confirm
+  it still resolves.
+
+## Progress
+
+### Phase 0 — DONE + verified (2026-07-07)
+
+Both services stood up **alongside** the live MQTT mesh (no cutover yet):
+
+- `castle-nats` (`nats:2`, JetStream) — deployment
+  `~/.castle/deployments/services/castle-nats.yaml`, config
+  `/data/castle/castle-nats/config/nats-server.conf`. Verified: container active,
+  `/healthz` ok, JetStream enabled (`/jsz` shows store dir + limits), and a real
+  **KV round-trip** — created `castle-registry` (put/get) and a TTL
+  `castle-presence` bucket (the presence primitive). Test buckets cleaned up.
+- `castle-openbao` (`openbao/openbao:latest`, v2.5.5) — deployment
+  `~/.castle/deployments/services/castle-openbao.yaml`, config
+  `/data/castle/castle-openbao/config/openbao.hcl`. Verified: initialized (1
+  share / threshold 1), unsealed, **KV-v2 secret put/get/delete** round-trip.
+  Unseal key + root token stored as castle secrets `OPENBAO_UNSEAL_KEY` /
+  `OPENBAO_ROOT_TOKEN`.
+
+**Consciously deferred (tracked):**
+1. **Auto-unseal on boot.** OpenBao is unsealed now but a container restart
+   re-seals it. The manifest `SystemdSpec` has no `exec_start_post` hook, so
+   proper boot-unseal wiring lands with Phase 4 (either a small manifest
+   addition or a companion oneshot). Manual recovery: unseal with
+   `OPENBAO_UNSEAL_KEY`.
+2. **`install.sh` bootstrap parity.** Fresh-install provisioning still seeds
+   Mosquitto; adding NATS/OpenBao there (and retiring Mosquitto) is bootstrap-only
+   (no runtime impact) and folds in with the Phase 1 cutover.
+
+### Phase 1 — DONE + single-node verified; live cutover done (2026-07-07)
+
+MQTT/Mosquitto transport replaced by NATS JetStream KV.
+
+- New `castle_api/mesh_wire.py` — transport-agnostic registry (de)serialization
+  (moved out of `mqtt_client.py`; preserves the secret-stripping invariant).
+- New `castle_api/nats_client.py` — `CastleNATSClient`: connects, PUTs its
+  registry to the `castle-registry` KV bucket, seeds from existing keys, watches
+  for peer PUT/DELETE, heartbeats a re-PUT (crash liveness via the existing
+  stale-TTL), and DELETEs its key on graceful stop (immediate peer-offline).
+  Async-native → the paho cross-thread `run_coroutine_threadsafe` hop is gone.
+- `main.py` lifespan, `config.py` (`nats_enabled`/`nats_url`), `models.py`
+  `MeshStatus` (`connected`/`nats_url`), `nodes.py` `/mesh/status` all rewired.
+- Frontend: `types/index.ts` `MeshStatus` + `MeshPanel.tsx` updated to the new
+  fields. `mqtt_client.py` + `test_mqtt.py` deleted; `test_mesh_wire.py` added.
+  `mdns.py` dead `_mqtt._tcp` broker-browse removed. `paho-mqtt` → `nats-py`.
+- **Live cutover:** `castle-api.yaml` env → `CASTLE_API_NATS_*`, `requires:
+  castle-nats`; applied. castle-api healthy on NATS, `/mesh/status` connected,
+  registry (9.7 KiB, 48 deployments) published to KV, **no secrets on the wire**.
+- Verification run: ruff clean; 89 api + 196 core tests pass; frontend `tsc`
+  clean; runtime KV round-trip confirmed against the live `castle-nats`.
+
+**Pending second node:** two-node mesh-view parity (peer sees peer, offline on
+departure) — needs the second LAN node running the NATS-enabled castle-api
+pointed at civil's NATS (seed URL) or a clustered `castle-nats`. Revert path if
+needed: restore `CASTLE_API_MQTT_*` env — but note `mqtt_client.py` is deleted,
+so revert = `git checkout main -- castle-api` then re-apply. The old Mosquitto
+`mqtt` service is left running (dormant) as a safety net; retire it once the
+second node is proven on NATS.
+
+### Phase 2 — DONE + single-node verified + tested (2026-07-07)
+
+- **`role` field** on `NodeConfig` + `CastleConfig`, wired end-to-end:
+  castle.yaml top-level `role:` → config → `_node_config` → registry.yaml →
+  mesh wire. **`civil` pinned `role: authority`**; default `follower`.
+- **Presence** (`castle-presence`) — a TTL KV bucket each node renews on the
+  heartbeat; expiry = the node is gone. Delete-on-stop for immediate departure.
+- **Shared config** (`castle-config`) — `get_shared_config` / `put_shared_config`
+  (authority-gated: followers raise `PermissionError`), plus a watch loop that
+  broadcasts a `config_changed` SSE (the follower `castle apply` reconcile hook
+  hangs off this — inert on one node).
+- Hardened `stop()` to bound the NATS drain (can't hang systemd shutdown).
+- Verified live: all three buckets present, `civil=online` presence,
+  `"role": "authority"` on the wire, authority write + follower-deny exercised
+  against the running `castle-nats`.
+- **Tests added:** `core/tests/test_fleet_role.py` (role config load + registry
+  round-trip), `castle-api/tests/test_nats_client.py` (role gating, hermetic),
+  role round-trip in `test_mesh_wire.py`. Suites: **200 core + 93 api** pass.
+
+**Pending second node:** the follower-side reconcile (watch `castle-config` →
+`castle apply`) is wired as an SSE hook but only meaningful with a peer.
+
+### Phase 4 — secret-read backend DONE + verified + tested (2026-07-07)
+
+- New `core/castle_core/secret_backends.py`: `SecretBackend` protocol,
+  `FileSecretBackend` (the historical behavior), `OpenBaoBackend` (KV-v2 read with
+  file fallback), and `build_backend()` selecting via `CASTLE_SECRET_BACKEND`
+  (default **file** — production is byte-for-byte unchanged until opted in).
+- `_read_secret` (the single chokepoint, `config.py`) now delegates to the active
+  backend. `${secret:NAME}` syntax untouched.
+- OpenBao token bootstraps from the file backend (it can't live in the vault it
+  unlocks); a missing key / auth failure / unreachable server all fall through to
+  file, so a partly-migrated vault keeps working.
+- Verified live against the running `castle-openbao`: a secret stored in the vault
+  resolves through `${secret:...}` in openbao mode; file-only secrets resolve via
+  fallback; missing → placeholder; **default file mode unchanged**.
+- **Tests:** `core/tests/test_secret_backends.py` (file hit/miss, backend
+  selection, unreachable-vault + empty-token fallback). Suites: **206 core + 93 api**.
+
+**Hardening:**
+1. **Write path — DONE + verified live.** `SecretBackend` gained
+   `write`/`delete`/`list_names`; `castle-api/secrets.py` routes all CRUD through
+   the active backend, so in openbao mode the dashboard writes to the vault.
+   Verified: write→vault→read→list→delete round-trip against live OpenBao.
+2. **Auto-unseal on boot — DONE + verified.** Added `exec_start_post` to the
+   systemd spec (general, `-`-prefixed so a hook failure never fails the unit);
+   `castle-openbao` runs `/data/castle/castle-openbao/unseal.sh` (polls up, unseals
+   with `OPENBAO_UNSEAL_KEY`). Verified: manual seal → restart → auto-unsealed.
+3. **TLS hardening — REMAINING (deliberately deferred).** NATS + OpenBao are
+   plaintext on the trusted LAN. This is the gate before cross-*network* or moving
+   real secrets — neither of which is here yet (the vault holds no production
+   secrets). Doing NATS TLS/auth touches the *live* civil↔primer mesh, so it's a
+   deliberate, staged change (regenerate certs, update both nodes' clients), not a
+   tail-end one. Next dedicated step.
+
+### Phase 3 — cross-node routing + breaker: logic DONE + verified against a real peer (2026-07-07)
+
+**Real second node:** `primer` (192.168.8.129) migrated onto the NATS mesh
+(branch checked out, castle-api pointed at `nats://civil:4222` via a systemd
+drop-in). civil ↔ primer mesh confirmed over the LAN — **Phase 1 two-node parity
+done for real**, not simulated. (This also *restored* the civil↔primer mesh my
+Phase 1 cutover had split — primer was still on MQTT→civil.)
+
+- `NodeConfig.address` — routable host peers proxy to (wired through registry +
+  wire; falls back to hostname, which civil resolves for primer).
+- `compute_routes` now emits **`remote`** routes for services this node
+  **consumes** (a local `requires` ref satisfied by an online peer), targeting
+  `<peer-address>:<port>`. `_host_remote_block` renders a **fail-fast breaker**
+  (2s dial timeout + passive health) for the there-but-wedged case; **presence
+  expiry removes the peer from the route set entirely** (gone → no route) — the
+  primary breaker.
+- `castle_api/mesh_gateway.py` — on peer join/leave/change (+ startup) the API
+  re-renders the Caddyfile (same generator `apply` uses + remote routes for
+  online peers) and reloads the gateway **iff content changed**.
+- **Verified:** 4 hermetic tests (route emitted / address fallback / breaker:
+  peer-absent → no route / unconsumed → no route); **resolution against primer's
+  real registry** pulled live from the mesh → `castle-api → primer:9020`; and the
+  live integration proven a **no-op** on civil (Caddyfile hash unchanged, gateway
+  healthy) since civil consumes nothing cross-node yet. Suites: **210 core + 93 api**.
+
+**Remaining:** the full live curl+kill E2E (civil routing to a peer service, then
+failing fast on kill) needs a peer-unique service civil consumes — every
+underlying piece is verified, but the end-to-end demo needs that provisioning.
+primer is now a permanent mesh member on the branch (revert: remove the drop-in +
+`git checkout main` on primer).
+
+## Decisions (resolved)
+
+1. **Discovery — both.** Keep mDNS for zero-config LAN peer discovery *and*
+   support explicit NATS seed URLs so the mesh can span networks. mDNS is the
+   convenience path on-LAN; seed URLs are the cross-network path. Client connects
+   via seeds when present, falls back to mDNS-discovered peers on-LAN.
+2. **OpenBao unseal — auto for now.** Auto-unseal (local transit/key) so a home
+   node boots unattended; designed so we can migrate to manual/stricter unseal
+   later (a supported rekey/unseal-migration, not zero-effort but planned for).
+   Keep the backend seam clean so the switch is config, not code.
+3. **Authority = `civil`.** `civil` (the acme node, `civil.payne.io`) is pinned
+   as `role: authority`; all others are followers. **Confirmed:** when `civil` is
+   down, shared config/secrets go **read-only** fleet-wide and every node keeps
+   serving its own deployments from cached local state.
+4. **NATS — single server now, cluster-ready.** Run one `castle-nats` today
+   (matches the single-writer authority). Clustering is deferred but must be a
+   pure config addition (cluster/routes block + a few more nodes) — no re-architecture.

uv.lock

@@ -48,7 +48,7 @@ dependencies = [
     { name = "castle-core" },
     { name = "fastapi" },
     { name = "httpx" },
-    { name = "paho-mqtt" },
+    { name = "nats-py" },
     { name = "pydantic-settings" },
     { name = "uvicorn", extra = ["standard"] },
     { name = "zeroconf" },
@@ -67,7 +67,7 @@ requires-dist = [
     { name = "castle-core", editable = "core" },
     { name = "fastapi", specifier = ">=0.115.0" },
     { name = "httpx", specifier = ">=0.27.0" },
-    { name = "paho-mqtt", specifier = ">=2.0.0" },
+    { name = "nats-py", specifier = ">=2.9.0" },
     { name = "pydantic-settings", specifier = ">=2.0.0" },
     { name = "uvicorn", extras = ["standard"], specifier = ">=0.34.0" },
     { name = "zeroconf", specifier = ">=0.131.0" },
@@ -349,6 +349,15 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/70/bc/6f1c2f612465f5fa89b95bead1f44dcb607670fd42891d8fdcd5d039f4f4/markupsafe-3.0.3-cp314-cp314t-win_arm64.whl", hash = "sha256:32001d6a8fc98c8cb5c947787c5d08b0a50663d139f1305bac5885d98d9b40fa", size = 14146, upload-time = "2025-09-27T18:37:28.327Z" },
 ]
 
+[[package]]
+name = "nats-py"
+version = "2.15.0"
+source = { registry = "https://pypi.org/simple" }
+sdist = { url = "https://files.pythonhosted.org/packages/02/f0/fc5e93f2b0dd14a202590ad9d30eda1955ea872039b5204357348d0f4b1e/nats_py-2.15.0.tar.gz", hash = "sha256:6622c547d9a7d2313d9c147d46c386188f4ec2c7b5c9f9a0438a4d1b55f54a93", size = 75995, upload-time = "2026-06-05T07:34:03.904Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/db/a8/b55606c7c621fb813c8ec78baf201d2c78bf6051091ec0c7ada572999e95/nats_py-2.15.0-py3-none-any.whl", hash = "sha256:9f8d36aa52a9926a88b8f1d70cf1fdce0ad387941479b500ee9ab3e51073cefd", size = 90334, upload-time = "2026-06-05T07:34:02.81Z" },
+]
+
 [[package]]
 name = "nodeenv"
 version = "1.10.0"
@@ -367,15 +376,6 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/b7/b9/c538f279a4e237a006a2c98387d081e9eb060d203d8ed34467cc0f0b9b53/packaging-26.0-py3-none-any.whl", hash = "sha256:b36f1fef9334a5588b4166f8bcd26a14e521f2b55e6b9de3aaa80d3ff7a37529", size = 74366, upload-time = "2026-01-21T20:50:37.788Z" },
 ]
 
-[[package]]
-name = "paho-mqtt"
-version = "2.1.0"
-source = { registry = "https://pypi.org/simple" }
-sdist = { url = "https://files.pythonhosted.org/packages/39/15/0a6214e76d4d32e7f663b109cf71fb22561c2be0f701d67f93950cd40542/paho_mqtt-2.1.0.tar.gz", hash = "sha256:12d6e7511d4137555a3f6ea167ae846af2c7357b10bc6fa4f7c3968fc1723834", size = 148848, upload-time = "2024-04-29T19:52:55.591Z" }
-wheels = [
-    { url = "https://files.pythonhosted.org/packages/c4/cb/00451c3cf31790287768bb12c6bec834f5d292eaf3022afc88e14b8afc94/paho_mqtt-2.1.0-py3-none-any.whl", hash = "sha256:6db9ba9b34ed5bc6b6e3812718c7e06e2fd7444540df2455d2c51bd58808feee", size = 67219, upload-time = "2024-04-29T19:52:48.345Z" },
-]
-
 [[package]]
 name = "pluggy"
 version = "1.6.0"