feat: cross-node consumption edges over the mesh

Carry each deployment's `requires` (deployment refs, no secrets) from config →
registry → the mesh MQTT payload, and expose it on /mesh/deployments. The System
Map resolves a remote deployment's ref against the provider set across nodes
(same machine → local provider → other machine) and draws a dashed cross-node
edge. This is the multi-node payoff: e.g. primer's castle-api → civil's mqtt.

app/src/pages/SystemMap.tsx

@@ -324,6 +324,8 @@ function RemoteNode({ data }: NodeProps) {
         </div>
         {d.sub && <div className="truncate text-[9px] text-[var(--muted)]">{d.sub}</div>}
       </div>
+      <Handle id="rs" type="source" position={Position.Bottom} isConnectable={false} style={{ ...HANDLE, opacity: 0 }} />
+      <Handle id="lt" type="target" position={Position.Left} isConnectable={false} style={{ ...HANDLE, opacity: 0 }} />
     </div>
   )
 }
@@ -655,6 +657,47 @@ export function SystemMapPage() {
       })
     }
 
+    // Cross-node consumption — a remote deployment's `requires` resolved against
+    // the provider set across nodes: same machine first, then a local provider
+    // (a cross-machine edge), then another machine. This is the multi-node payoff.
+    const meshDeps = meshResp?.deployments ?? []
+    const meshByNode = new Map<string, Set<string>>()
+    for (const md of meshDeps) {
+      const s = meshByNode.get(md.node) ?? new Set<string>()
+      s.add(md.name)
+      meshByNode.set(md.node, s)
+    }
+    const remoteId = (node: string, name: string) => `__remote_${node}_${name}__`
+    for (const md of meshDeps) {
+      for (const ref of md.requires ?? []) {
+        let target: string | null = null
+        if (meshByNode.get(md.node)?.has(ref) && present.has(remoteId(md.node, ref)))
+          target = remoteId(md.node, ref) // same machine
+        else if (present.has(ref))
+          target = ref // a local provider — cross-machine edge
+        else
+          for (const [mn, names] of meshByNode)
+            if (mn !== md.node && names.has(ref) && present.has(remoteId(mn, ref))) {
+              target = remoteId(mn, ref)
+              break
+            }
+        if (!target) continue
+        edges.push({
+          id: `xnode:${md.node}:${md.name}->${ref}`,
+          source: remoteId(md.node, md.name),
+          target,
+          sourceHandle: "rs",
+          targetHandle: "lt",
+          animated: true,
+          deletable: false,
+          label: `${md.node} → ${ref}`,
+          style: { stroke: "#e879f9", strokeWidth: 1.5, strokeDasharray: "6 3" },
+          labelStyle: { fill: "#e879f9", fontSize: 9 },
+          labelBgStyle: { fill: "#0d1117" },
+        })
+      }
+    }
+
     return { nodes, edges, kindOf, reachOf }
   }, [graph, jobs, programs, gateway, suggestionsResp, meshResp])
 
@@ -1176,6 +1219,7 @@ function Legend() {
     { c: PROTO_COLOR.system, label: "consumes — other" },
     { c: "#bc8cff", label: "same program", dashed: true },
     { c: "#f59e0b", label: "suggested — click to accept", dashed: true },
+    { c: "#e879f9", label: "cross-node", dashed: true },
     { c: "#58a6ff", label: "internal — LAN" },
     { c: "#2ea043", label: "public — internet" },
   ]

app/src/types/index.ts

@@ -172,6 +172,7 @@ export interface MeshDeployment {
   base_url: string | null
   subdomain: string | null
   endpoints: GraphEndpoint[]
+  requires: string[] // deployment refs it consumes (for cross-node edges)
 }
 
 // POST /programs/{name}/sync — a fast-forward pull (no build/apply/restart).

castle-api/src/castle_api/mqtt_client.py

@@ -67,6 +67,9 @@ def _registry_to_json(registry: NodeRegistry) -> str:
             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)
@@ -101,6 +104,7 @@ def _json_to_registry(payload: str) -> NodeRegistry:
             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/nodes.py

@@ -126,6 +126,9 @@ def mesh_deployments() -> dict:
                     "base_url": getattr(d, "base_url", None),
                     "subdomain": d.subdomain,
                     "endpoints": _endpoints_of_registry(d),
+                    "requires": [
+                        r.get("ref") for r in (getattr(d, "requires", None) or []) if r.get("ref")
+                    ],
                 }
             )
     return {"deployments": out}

core/src/castle_core/deploy.py

@@ -615,6 +615,15 @@ def _resolve_description(config: CastleConfig, spec: DeploymentBase) -> str | No
     return None
 
 
+def _registry_requires(dep: DeploymentSpec) -> list[dict]:
+    """A deployment's `requires` as plain dicts for the registry — carried into the
+    mesh payload so peers can draw cross-node consumption."""
+    return [
+        {"kind": r.kind, "ref": r.ref, "bind": r.bind}
+        for r in (getattr(dep, "requires", None) or [])
+    ]
+
+
 def _build_deployed(
     config: CastleConfig, name: str, dep: DeploymentSpec, messages: list[str]
 ) -> Deployment:
@@ -642,6 +651,7 @@ def _build_deployed(
             static_root=static_root,
             managed=False,
             enabled=dep.enabled,
+            requires=_registry_requires(dep),
         )
     if isinstance(dep, PathDeployment):
         return Deployment(
@@ -652,6 +662,7 @@ def _build_deployed(
             stack=stack,
             managed=False,
             enabled=dep.enabled,
+            requires=_registry_requires(dep),
         )
     if isinstance(dep, RemoteDeployment):
         return Deployment(
@@ -663,6 +674,7 @@ def _build_deployed(
             base_url=dep.base_url,
             managed=False,
             enabled=dep.enabled,
+            requires=_registry_requires(dep),
         )
 
     # systemd: a supervised process (a service, or a job when scheduled).
@@ -766,6 +778,7 @@ def _build_deployed(
         schedule=getattr(dep, "schedule", None),
         managed=managed,
         enabled=dep.enabled,
+        requires=_registry_requires(dep),
     )
 
 

core/src/castle_core/registry.py

@@ -86,6 +86,10 @@ class Deployment:
     # Declared desired state (from the deployment's `enabled:`). `castle apply`
     # activates enabled deployments and deactivates disabled ones. Default True.
     enabled: bool = True
+    # Deployment `requires` (list of {kind, ref, bind}) — carried so the mesh can
+    # draw cross-node consumption. A remote consumer's ref resolves against the
+    # provider set across nodes.
+    requires: list[dict] = field(default_factory=list)
 
 
 @dataclass