From 38c33224e7effbd5ee9977218ed401caadf61c63 Mon Sep 17 00:00:00 2001 From: Paul Payne Date: Tue, 7 Jul 2026 04:54:50 -0700 Subject: [PATCH] refactor: remove dead provides/consumes capability model The program-level Capability model (provides/consumes) was a vestige of an earlier design superseded by the deployment-level requires/Requirement graph. It was read only into a diagnostic node and surfaced as an always-empty 'Capabilities' panel; nothing acted on it. Removed from manifest, relations, the CLI re-export, and the System Map. --- app/src/pages/SystemMap.tsx | 24 -- castle-api/src/castle_api/mqtt_client.py | 283 ----------------------- castle-api/tests/test_mqtt.py | 110 --------- cli/src/castle_cli/manifest.py | 1 - core/src/castle_core/manifest.py | 9 - core/src/castle_core/relations.py | 5 - 6 files changed, 432 deletions(-) delete mode 100644 castle-api/src/castle_api/mqtt_client.py delete mode 100644 castle-api/tests/test_mqtt.py diff --git a/app/src/pages/SystemMap.tsx b/app/src/pages/SystemMap.tsx index af38d50..e056356 100644 --- a/app/src/pages/SystemMap.tsx +++ b/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({ )} - {(info.provides.length > 0 || info.capsConsumes.length > 0) && ( - - {info.provides.map((t) => ( - - provides {t} - - ))} - {info.capsConsumes.map((t) => ( - - needs {t} - - ))} - - )} {info.consumes.length === 0 && } {info.consumes.map((c) => ( diff --git a/castle-api/src/castle_api/mqtt_client.py b/castle-api/src/castle_api/mqtt_client.py deleted file mode 100644 index 7a107e5..0000000 --- a/castle-api/src/castle_api/mqtt_client.py +++ /dev/null @@ -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 (.) - # 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") diff --git a/castle-api/tests/test_mqtt.py b/castle-api/tests/test_mqtt.py deleted file mode 100644 index 541c8c7..0000000 --- a/castle-api/tests/test_mqtt.py +++ /dev/null @@ -1,110 +0,0 @@ -"""Tests for MQTT client serialization logic.""" - -import json - -from castle_core.registry import Deployment, NodeConfig, NodeRegistry - -from castle_api.mqtt_client import _json_to_registry, _registry_to_json - - -def _make_registry() -> NodeRegistry: - return NodeRegistry( - node=NodeConfig( - hostname="tower", castle_root="/data/repos/castle", gateway_port=9000 - ), - deployed={ - "my-svc": Deployment( - manager="systemd", - launcher="python", - run_cmd=["uv", "run", "my-svc"], - env={"PORT": "9001", "SECRET_KEY": "super-secret"}, - description="My service", - name="my-svc", - kind="service", - stack="python-fastapi", - port=9001, - health_path="/health", - subdomain="my-svc", - managed=True, - ), - "my-job": Deployment( - manager="systemd", - launcher="command", - run_cmd=["my-job"], - name="my-job", - kind="job", - stack="python-cli", - schedule="0 2 * * *", - ), - }, - ) - - -class TestRegistrySerialization: - """Round-trip serialization of NodeRegistry to/from JSON.""" - - def test_round_trip(self) -> None: - original = _make_registry() - json_str = _registry_to_json(original) - 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)) - - svc = restored.get("service", "my-svc") - assert svc is not None - assert svc.manager == "systemd" - assert svc.launcher == "python" - assert svc.port == 9001 - assert svc.health_path == "/health" - assert svc.subdomain == "my-svc" - assert svc.managed is True - assert svc.kind == "service" - assert svc.stack == "python-fastapi" - - def test_job_fields_preserved(self) -> None: - original = _make_registry() - restored = _json_to_registry(_registry_to_json(original)) - - job = restored.get("job", "my-job") - assert job is not None - assert job.launcher == "command" - assert job.schedule == "0 2 * * *" - assert job.kind == "job" - assert job.stack == "python-cli" - - def test_optional_fields_omitted(self) -> None: - """Fields like port, health_path are None when not set.""" - reg = NodeRegistry( - node=NodeConfig(hostname="minimal"), - deployed={ - "bare": Deployment( - manager="systemd", launcher="command", run_cmd=["bare"], name="bare" - ), - }, - ) - restored = _json_to_registry(_registry_to_json(reg)) - bare = restored.get("service", "bare") - assert bare.port is None - assert bare.health_path is None - assert bare.subdomain is None - assert bare.schedule is None - assert bare.managed is False - - def test_no_secrets_in_payload(self) -> None: - """env vars, run_cmd, and castle_root must not appear in MQTT payload.""" - original = _make_registry() - json_str = _registry_to_json(original) - data = json.loads(json_str) - - # No castle_root in node - assert "castle_root" not in data["node"] - - # 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 "run_cmd" not in comp, f"{name} has run_cmd in MQTT payload" diff --git a/cli/src/castle_cli/manifest.py b/cli/src/castle_cli/manifest.py index 3b285e0..9b73fc1 100644 --- a/cli/src/castle_cli/manifest.py +++ b/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, diff --git a/core/src/castle_core/manifest.py b/core/src/castle_core/manifest.py index 82a4ee5..06adc39 100644 --- a/core/src/castle_core/manifest.py +++ b/core/src/castle_core/manifest.py @@ -277,12 +277,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 +384,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 diff --git a/core/src/castle_core/relations.py b/core/src/castle_core/relations.py index 3dfb929..ac2559b 100644 --- a/core/src/castle_core/relations.py +++ b/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)