feat(mesh): fleet role + shared config/presence KV (Phase 2)

- add static 'role' (authority|follower) to NodeConfig/CastleConfig, wired from
  castle.yaml through the registry to the mesh wire; civil pinned authority
- castle-presence: TTL KV bucket each node renews (churn signal); delete-on-stop
- castle-config: authority-gated get/put + change-watch SSE (follower reconcile
  hook hangs off it)
- bound the NATS drain on stop() so shutdown can't hang
- tests: role config/registry round-trip, wire round-trip, role write-gating

Single-node verified; follower reconcile pending the second node.

castle-api/src/castle_api/mesh_wire.py

@@ -26,6 +26,8 @@ def registry_to_json(registry: NodeRegistry) -> str:
             # 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,
         },
         "deployed": {},
     }
@@ -73,6 +75,7 @@ def json_to_registry(payload: str) -> NodeRegistry:
         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"),
     )
     deployed: dict[str, Deployment] = {}
     for key, comp_data in data.get("deployed", {}).items():

castle-api/src/castle_api/nats_client.py

@@ -34,8 +34,11 @@ 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:
@@ -52,6 +55,8 @@ class CastleNATSClient:
         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()
@@ -64,6 +69,11 @@ class CastleNATSClient:
     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(
@@ -72,22 +82,39 @@ class CastleNATSClient:
             max_reconnect_attempts=-1,  # reconnect forever — nodes come and go
         )
         js = self._nc.jetstream()
-        try:
+        self._kv = await self._ensure_bucket(js, REGISTRY_BUCKET, history=1)
-            self._kv = await js.key_value(REGISTRY_BUCKET)
+        # Presence: a short-TTL key each node renews; its expiry = the node is gone.
-        except Exception:
+        self._presence_kv = await self._ensure_bucket(
-            self._kv = await js.create_key_value(
+            js, PRESENCE_BUCKET, history=1, ttl=PRESENCE_TTL
-                config=KeyValueConfig(bucket=REGISTRY_BUCKET, history=1)
+        )
-            )
+        # 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()
 
         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)", self._servers)
+        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."""
@@ -100,9 +127,15 @@ class CastleNATSClient:
         if self._kv is not None:
             with contextlib.suppress(Exception):
                 await self._kv.delete(self._local_hostname)
-        if self._nc is not None:
+        if self._presence_kv is not None:
             with contextlib.suppress(Exception):
-                await self._nc.drain()
+                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")
 
@@ -172,6 +205,46 @@ class CastleNATSClient:
             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."""

castle-api/tests/test_mesh_wire.py

@@ -95,6 +95,18 @@ 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 never appear on the wire."""
         original = _make_registry()

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"))

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.
@@ -465,6 +468,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/registry.py

@@ -32,6 +32,11 @@ class NodeConfig:
     tunnel_id: str | None = None
     # Emit the cert_obtained → `castle tls reconcile` hook (needs events-exec plugin).
     cert_hook: bool = False
+    # 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 +157,7 @@ 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"),
     )
 
     deployed: dict[str, Deployment] = {}
@@ -241,6 +247,8 @@ 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
 
     for key, comp in registry.deployed.items():
         entry: dict = {

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"

docs/fleet-mesh-plan.md

@@ -1,7 +1,7 @@
 # Fleet Mesh Plan — OpenBao + NATS
 
 **Status:** in progress on branch `feat/fleet-mesh-nats-openbao`.
-Phases 0–1 complete + single-node verified (live cutover done). Phases 2–4 pending.
+Phases 0–2 complete + single-node verified (live). Phases 3–4 pending (need 2nd node).
 
 ## Context
 
@@ -196,6 +196,28 @@ 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.
+
 ## Decisions (resolved)
 
 1. **Discovery — both.** Keep mDNS for zero-config LAN peer discovery *and*