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)