api+ui: kind-scoped save/delete endpoints and links

API: /config/{services,jobs,tools,static}/{name} now pin the twin they
target — _save_deployment/_delete_deployment take an explicit kind so a
patch to a 'backup' service can't bleed into a 'backup' job/tool. Add
/tools and /static endpoints; keep /deployments/{name} kind-agnostic.
New test_kind_twins proves per-kind save/delete isolation on disk.

UI: ConfigPanel and CreateDeploymentForm write to the kind-scoped
resource; GatewayPanel/NodeDetail/DeploymentsSection link via a shared
detailPath(name, kind) helper instead of the ambiguous /deployment/:name.

Includes incidental ruff-format reflow of untouched api files.

app/src/components/GatewayPanel.tsx

@@ -3,7 +3,7 @@ import { Link } from "react-router-dom"
 import { Globe, RefreshCw, FileText, ExternalLink, Cable } from "lucide-react"
 import type { GatewayInfo, HealthStatus } from "@/types"
 import { useApply, useCaddyfile } from "@/services/api/hooks"
-import { subdomainUrl } from "@/lib/labels"
+import { subdomainUrl, detailPath } from "@/lib/labels"
 import { HealthBadge } from "./HealthBadge"
 import { GatewaySettings } from "./GatewaySettings"
 
@@ -118,7 +118,10 @@ export function GatewayPanel({ gateway, statuses }: GatewayPanelProps) {
                   </td>
                   <td className="px-4 py-2 font-mono text-xs text-[var(--muted)]">
                     {route.name ? (
-                      <Link to={`/deployment/${route.name}`} className="hover:text-[var(--primary)]">
+                      <Link
+                        to={detailPath(route.name, route.kind)}
+                        className="hover:text-[var(--primary)]"
+                      >
                         {route.kind === "static" ? shortDir(route.target) : route.target}
                       </Link>
                     ) : (

app/src/components/detail/ConfigPanel.tsx

@@ -31,9 +31,10 @@ export function ConfigPanel({ deployment, configSection, onRefetch }: ConfigPane
   const [pendingApply, setPendingApply] = useState(false)
   const [applying, setApplying] = useState(false)
   const isDeployment = configSection !== "programs"
-  // Programs are their own catalog; every deployment kind (service/job/tool/
+  // Save/delete against the kind-scoped resource (services/jobs/tools/static) so
-  // static) lives in the single deployments/ collection.
+  // a save can't hit a same-named twin of another kind. `configSection` already
-  const writeSection = isDeployment ? "deployments" : "programs"
+  // carries the kind; programs write to their own catalog.
+  const writeSection = configSection
 
   const handleSave = async (compName: string, config: Record<string, unknown>) => {
     setMessage(null)

app/src/components/detail/CreateDeploymentForm.tsx

@@ -112,7 +112,10 @@ export function CreateDeploymentForm({
     setError("")
     try {
       setBusy("Saving…")
-      await apiClient.put(`/config/deployments/${name}`, { config: buildConfig() })
+      // Post to the kind-scoped resource so creating a twin (a `backup` job when
+      // a `backup` service exists) targets the right collection, not a guess.
+      const section = kind === "static" ? "static" : `${kind}s`
+      await apiClient.put(`/config/${section}/${name}`, { config: buildConfig() })
       // Converge: render + activate the new deployment in one step.
       setBusy("Applying…")
       await apiClient.post(`/apply`, { name })

app/src/components/detail/DeploymentsSection.tsx

@@ -4,6 +4,7 @@ import { Plus, ChevronRight } from "lucide-react"
 import type { ProgramDetail } from "@/types"
 import { useServices, useJobs } from "@/services/api/hooks"
 import { KindBadge } from "@/components/KindBadge"
+import { detailPath } from "@/lib/labels"
 import { CreateDeploymentForm, type CreatePrefill } from "./CreateDeploymentForm"
 
 /** How a program is deployed. A program → 0-N deployments; each row links to its
@@ -27,9 +28,6 @@ export function DeploymentsSection({ program }: { program: ProgramDetail }) {
     launcher: program.stack?.startsWith("python") || !program.stack ? "python" : "command",
   }
 
-  const detailPath = (name: string, kind: string) =>
-    kind === "tool" ? `/tools/${name}` : kind === "job" ? `/jobs/${name}` : `/services/${name}`
-
   return (
     <div className="bg-[var(--card)] border border-[var(--border)] rounded-lg p-5 mb-6">
       <div className="flex items-center justify-between mb-1">

app/src/components/detail/SystemdPanel.tsx

@@ -27,6 +27,14 @@ export function SystemdPanel({ name, systemd }: SystemdPanelProps) {
       <div className="grid grid-cols-2 gap-x-6 gap-y-2 text-sm mt-3">
         <span className="text-[var(--muted)]">Unit</span>
         <span className="font-mono break-all">{systemd.unit_name}</span>
+        {systemd.timer && (
+          <>
+            <span className="text-[var(--muted)]">Timer unit</span>
+            <span className="font-mono break-all">
+              {systemd.unit_name.replace(".service", ".timer")}
+            </span>
+          </>
+        )}
         <span className="text-[var(--muted)]">Path</span>
         <span className="font-mono break-all">{systemd.unit_path}</span>
         {systemd.timer && (

app/src/lib/labels.ts

@@ -48,6 +48,17 @@ export function stackLabel(stack: string): string {
   return STACK_LABELS[stack] ?? stack
 }
 
+/**
+ * The kind-scoped detail route for a deployment. A name can be shared across
+ * kinds (a `backup` service + job + tool), so links must carry the kind to
+ * reach the right twin. Statics share the service detail page.
+ */
+export function detailPath(name: string, kind: string): string {
+  if (kind === "tool") return `/tools/${name}`
+  if (kind === "job") return `/jobs/${name}`
+  return `/services/${name}`
+}
+
 /**
  * Full URL for a service exposed at <subdomain>.<gateway.domain>. The domain is
  * derived from the dashboard's own host (it is served at castle.<domain>), so

app/src/pages/NodeDetail.tsx

@@ -3,6 +3,7 @@ import { ArrowLeft, Server } from "lucide-react"
 import { useNode } from "@/services/api/hooks"
 import { KindBadge } from "@/components/KindBadge"
 import { StackBadge } from "@/components/StackBadge"
+import { detailPath } from "@/lib/labels"
 import { cn } from "@/lib/utils"
 
 export function NodeDetailPage() {
@@ -77,7 +78,7 @@ export function NodeDetailPage() {
                 >
                   <td className="px-3 py-2.5">
                     <Link
-                      to={`/deployment/${comp.id}`}
+                      to={detailPath(comp.id, comp.kind ?? "service")}
                       className="font-medium hover:text-[var(--primary)] transition-colors"
                     >
                       {comp.id}

app/src/pages/ScheduledDetail.tsx

@@ -50,24 +50,16 @@ export function ScheduledDetailPage() {
               <Clock size={14} className="shrink-0 text-[var(--muted)]" />
               {deployment.schedule}
             </span>
-            {deployment.systemd && (
-              <>
-                <span className="text-[var(--muted)]">Timer unit</span>
-                <span className="font-mono break-all">
-                  {deployment.systemd.unit_name.replace(".service", ".timer")}
-                </span>
-              </>
-            )}
           </div>
         </div>
       )}
 
+      <ConfigPanel deployment={deployment} configSection="jobs" onRefetch={refetch} />
+
       {deployment.systemd && (
         <SystemdPanel name={deployment.id} systemd={deployment.systemd} />
       )}
 
-      <ConfigPanel deployment={deployment} configSection="jobs" onRefetch={refetch} />
-
       {deployment.managed && (
         <div className="mb-6">
           <h2 className="text-lg font-semibold mb-3">Logs</h2>

app/src/pages/ServiceDetail.tsx

@@ -146,10 +146,6 @@ export function ServiceDetailPage() {
         </div>
       </div>
 
-      {deployment.systemd && (
-        <SystemdPanel name={deployment.id} systemd={deployment.systemd} />
-      )}
-
       {isGateway && caddyfile?.content && (
         <div className="bg-[var(--card)] border border-[var(--border)] rounded-lg p-5 mb-6">
           <h2 className="text-sm font-semibold text-[var(--muted)] uppercase tracking-wider mb-1">
@@ -170,6 +166,10 @@ export function ServiceDetailPage() {
         onRefetch={refetch}
       />
 
+      {deployment.systemd && (
+        <SystemdPanel name={deployment.id} systemd={deployment.systemd} />
+      )}
+
       {deployment.managed && (
         <div className="mb-6">
           <h2 className="text-lg font-semibold mb-3">Logs</h2>

castle-api/src/castle_api/config_editor.py

@@ -279,10 +279,15 @@ async def delete_program(name: str, cascade: bool = False) -> dict:
         except Exception:
             pass
 
-    return {"ok": True, "program": name, "action": "deleted", "removed_deployments": removed}
+    return {
+        "ok": True,
+        "program": name,
+        "action": "deleted",
+        "removed_deployments": removed,
+    }
 
 
-def _save_deployment(name: str, config_dict: dict) -> dict:
+def _save_deployment(name: str, config_dict: dict, kind: str | None = None) -> dict:
     """Create/update a deployment (any manager) with PATCH semantics.
 
     The incoming config is shallow-merged over the existing spec, so a save can
@@ -290,18 +295,26 @@ def _save_deployment(name: str, config_dict: dict) -> dict:
     a present key replaces wholesale, an **omitted** key is preserved, and an
     explicit ``null`` clears the key (back to its default). On CREATE there's no
     base, so the incoming config stands alone.
+
+    ``kind`` pins the twin this save targets — a kind-scoped endpoint
+    (``/services|/jobs|/tools|/static``) passes it so a partial patch to a
+    ``backup`` service can never bleed into a ``backup`` job/tool sharing the
+    name. The kind-agnostic ``/deployments/{name}`` leaves it None and infers.
     """
     _require_repo()
     config = get_config()
     incoming = dict(config_dict)
 
-    # Resolve the (name, kind) this save targets. A partial patch (e.g. just
+    # Resolve the (name, kind) this save targets. An explicit kind is
+    # authoritative (kind-scoped endpoint). Otherwise: a partial patch (e.g. just
     # {reach: off}) has no manager, so we can't derive kind from it — prefer the
-    # existing same-named deployment when there's exactly one; otherwise derive the
+    # existing same-named deployment when there's exactly one; else derive the
     # kind from the incoming spec (a create, or disambiguating a shared name).
     named = config.deployments_named(name)
     existing = None
-    if len(named) == 1:
+    if kind is not None:
+        existing = config.deployment(kind, name)
+    elif len(named) == 1:
         existing = named[0][1]
     else:
         try:
@@ -332,17 +345,25 @@ def _save_deployment(name: str, config_dict: dict) -> dict:
             status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
             detail=f"Invalid deployment config: {e}",
         )
-    config.store_for(kind_for(dep))[name] = dep
+    target_kind = kind_for(dep)
+    # A field edit that changes the derived kind (e.g. adds a schedule) moves the
+    # spec to the new store; drop the stale entry under the requested kind.
+    if kind is not None and target_kind != kind:
+        config.store_for(kind).pop(name, None)
+    config.store_for(target_kind)[name] = dep
     save_config(config)
     return {"ok": True, "deployment": name}
 
 
-def _delete_deployment(name: str) -> dict:
+def _delete_deployment(name: str, kind: str | None = None) -> dict:
+    """Remove a deployment. A kind-scoped delete drops only that twin; the
+    kind-agnostic path removes every kind sharing the name."""
     config = get_config()
     removed = False
-    for kind in KINDS:
+    kinds = (kind,) if kind is not None else KINDS
-        if name in config.store_for(kind):
+    for k in kinds:
-            del config.store_for(kind)[name]
+        if name in config.store_for(k):
+            del config.store_for(k)[name]
             removed = True
     if not removed:
         raise HTTPException(
@@ -391,26 +412,50 @@ def set_deployment_enabled(name: str, request: EnabledRequest) -> dict:
     return {"ok": True, "deployment": name, "enabled": request.enabled}
 
 
+# Kind-scoped endpoints — pin the twin so a save/delete can't hit a same-named
+# deployment of another kind (a `backup` service vs job vs tool).
 @router.put("/services/{name}")
 def save_service(name: str, request: ServiceConfigRequest) -> dict:
-    """Alias of PUT /deployments/{name} (kept for the existing dashboard)."""
+    """Create/update the *service* named `name`."""
-    return _save_deployment(name, request.config)
+    return _save_deployment(name, request.config, kind="service")
 
 
 @router.delete("/services/{name}")
 def delete_service(name: str) -> dict:
-    return _delete_deployment(name)
+    return _delete_deployment(name, kind="service")
 
 
 @router.put("/jobs/{name}")
 def save_job(name: str, request: JobConfigRequest) -> dict:
-    """Alias of PUT /deployments/{name} (kept for the existing dashboard)."""
+    """Create/update the *job* named `name`."""
-    return _save_deployment(name, request.config)
+    return _save_deployment(name, request.config, kind="job")
 
 
 @router.delete("/jobs/{name}")
 def delete_job(name: str) -> dict:
-    return _delete_deployment(name)
+    return _delete_deployment(name, kind="job")
+
+
+@router.put("/tools/{name}")
+def save_tool(name: str, request: ServiceConfigRequest) -> dict:
+    """Create/update the *tool* named `name`."""
+    return _save_deployment(name, request.config, kind="tool")
+
+
+@router.delete("/tools/{name}")
+def delete_tool(name: str) -> dict:
+    return _delete_deployment(name, kind="tool")
+
+
+@router.put("/static/{name}")
+def save_static(name: str, request: ServiceConfigRequest) -> dict:
+    """Create/update the *static* frontend named `name`."""
+    return _save_deployment(name, request.config, kind="static")
+
+
+@router.delete("/static/{name}")
+def delete_static(name: str) -> dict:
+    return _delete_deployment(name, kind="static")
 
 
 @router.post("/apply", response_model=ApplyResponse)

castle-api/src/castle_api/health.py

@@ -60,7 +60,10 @@ async def _check_systemd(name: str) -> HealthStatus:
     """Check a managed service's health via its systemd unit state."""
     unit = f"castle-{name}.service"
     proc = await asyncio.create_subprocess_exec(
-        "systemctl", "--user", "is-active", unit,
+        "systemctl",
+        "--user",
+        "is-active",
+        unit,
         stdout=asyncio.subprocess.PIPE,
         stderr=asyncio.subprocess.PIPE,
     )

castle-api/src/castle_api/logs.py

@@ -31,7 +31,11 @@ async def get_logs(
         config = load_config(root)
         # A name may span kinds — the managed (systemd) one owns the journal.
         dep_kind = next(
-            ((k, s) for k, s in config.deployments_named(name) if getattr(s, "manage", None)),
+            (
+                (k, s)
+                for k, s in config.deployments_named(name)
+                if getattr(s, "manage", None)
+            ),
             None,
         )
         if dep_kind is None:

castle-api/src/castle_api/mdns.py

@@ -35,7 +35,9 @@ class CastleMDNS:
         self._service_info: ServiceInfo | None = None
 
         # Discovered state
-        self.peers: dict[str, dict] = {}  # hostname -> {gateway_port, api_port, addresses}
+        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(
@@ -67,14 +69,18 @@ class CastleMDNS:
     def _handle_castle_peer(self, info: ServiceInfo) -> None:
         """Process a discovered castle peer."""
         props = {
-            k.decode() if isinstance(k, bytes) else k: v.decode() if isinstance(v, bytes) else v
+            k.decode() if isinstance(k, bytes) else k: v.decode()
+            if isinstance(v, bytes)
+            else v
             for k, v in info.properties.items()
         }
         peer_hostname = props.get("hostname", "")
         if not peer_hostname or peer_hostname == self._hostname:
             return
 
-        addresses = [socket.inet_ntoa(addr) for addr in info.addresses if len(addr) == 4]
+        addresses = [
+            socket.inet_ntoa(addr) for addr in info.addresses if len(addr) == 4
+        ]
 
         self.peers[peer_hostname] = {
             "gateway_port": int(props.get("gateway_port", 9000)),
@@ -85,13 +91,17 @@ class CastleMDNS:
 
     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]
+        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)
+            logger.info(
+                "mDNS: discovered MQTT broker at %s:%d", addresses[0], info.port
+            )
 
     def start(self) -> None:
         """Start advertising and browsing."""
@@ -109,14 +119,24 @@ class CastleMDNS:
             },
         )
         self._zeroconf.register_service(self._service_info)
-        logger.info("mDNS: advertising %s on port %d", self._hostname, self._gateway_port)
+        logger.info(
+            "mDNS: advertising %s on port %d", self._hostname, self._gateway_port
+        )
 
         # Browse for peers and MQTT broker
         self._browsers.append(
-            ServiceBrowser(self._zeroconf, CASTLE_SERVICE_TYPE, handlers=[self._on_service_state_change])
+            ServiceBrowser(
+                self._zeroconf,
+                CASTLE_SERVICE_TYPE,
+                handlers=[self._on_service_state_change],
+            )
         )
         self._browsers.append(
-            ServiceBrowser(self._zeroconf, MQTT_SERVICE_TYPE, handlers=[self._on_service_state_change])
+            ServiceBrowser(
+                self._zeroconf,
+                MQTT_SERVICE_TYPE,
+                handlers=[self._on_service_state_change],
+            )
         )
 
     def stop(self) -> None:

castle-api/src/castle_api/mesh.py

@@ -40,7 +40,9 @@ class MeshStateManager:
     def update_node(self, hostname: str, registry: NodeRegistry) -> None:
         """Add or update a remote node's registry."""
         self._nodes[hostname] = RemoteNode(registry=registry)
-        logger.info("Mesh: updated node %s (%d deployed)", hostname, len(registry.deployed))
+        logger.info(
+            "Mesh: updated node %s (%d deployed)", hostname, len(registry.deployed)
+        )
 
     def set_offline(self, hostname: str) -> None:
         """Mark a node as offline (LWT received)."""

castle-api/src/castle_api/mqtt_client.py

@@ -171,7 +171,9 @@ class CastleMQTTClient:
             return
 
         self._connected = True
-        logger.info("Connected to MQTT broker at %s:%d", self._broker_host, self._broker_port)
+        logger.info(
+            "Connected to MQTT broker at %s:%d", self._broker_host, self._broker_port
+        )
 
         # Publish our status as online (retained)
         client.publish(
@@ -222,7 +224,9 @@ class CastleMQTTClient:
                 if payload == "offline":
                     mesh_state.set_offline(hostname)
                     asyncio.run_coroutine_threadsafe(
-                        broadcast("mesh", {"event": "node_offline", "hostname": hostname}),
+                        broadcast(
+                            "mesh", {"event": "node_offline", "hostname": hostname}
+                        ),
                         self._loop,
                     )
 

castle-api/src/castle_api/routes.py

@@ -154,7 +154,9 @@ def _summary_from_service(
     )
 
 
-def _summary_from_job(name: str, job: SystemdDeployment, config: object) -> DeploymentSummary:
+def _summary_from_job(
+    name: str, job: SystemdDeployment, config: object
+) -> DeploymentSummary:
     """Build a DeploymentSummary from a systemd deployment (job, non-deployed)."""
     managed = bool(job.manage and job.manage.systemd and job.manage.systemd.enable)
 
@@ -280,7 +282,9 @@ def _service_from_deployed(name: str, deployed: object) -> ServiceSummary:
     )
 
 
-def _service_from_spec(name: str, svc: SystemdDeployment, config: object) -> ServiceSummary:
+def _service_from_spec(
+    name: str, svc: SystemdDeployment, config: object
+) -> ServiceSummary:
     """Build a ServiceSummary from a systemd deployment."""
     port = None
     health_path = None
@@ -911,7 +915,8 @@ def get_gateway() -> GatewayInfo:
     # is not a service, so filtering to services dropped its public_url (calculator).
     public_domain = registry.node.public_domain
     public_names = {
-        name for _k, name, dep in (config.all_deployments() if config else [])
+        name
+        for _k, name, dep in (config.all_deployments() if config else [])
         if getattr(dep, "public", False)
     }
 
@@ -937,7 +942,8 @@ def get_gateway() -> GatewayInfo:
     tunnel_connected = (
         subprocess.run(
             ["systemctl", "--user", "is-active", "castle-castle-tunnel.service"],
-            capture_output=True, text=True,
+            capture_output=True,
+            text=True,
         ).stdout.strip()
         == "active"
     )
@@ -970,7 +976,7 @@ def save_gateway_config(request: GatewayConfigRequest) -> dict[str, str]:
     from castle_core.config import load_config, save_config
 
     config = load_config(root)
-    norm = lambda v: (v or None)  # noqa: E731 — empty string clears
+    norm = lambda v: v or None  # noqa: E731 — empty string clears
     config.gateway.tls = norm(request.tls)
     config.gateway.domain = norm(request.domain)
     config.gateway.public_domain = norm(request.public_domain)

castle-api/src/castle_api/services.py

@@ -107,7 +107,9 @@ async def _deferred_systemctl(action: str, unit: str, delay: float = 0.5) -> Non
 async def _do_action(name: str, action: str) -> JSONResponse:
     """Execute a systemctl action and broadcast updated health."""
     deployed = _managed(name)
-    unit = unit_name(name, deployed.kind) if deployed else f"{UNIT_PREFIX}{name}.service"
+    unit = (
+        unit_name(name, deployed.kind) if deployed else f"{UNIT_PREFIX}{name}.service"
+    )
 
     # Self-restart: defer the systemctl call so the response can be sent first
     if name == SELF_NAME and action in ("restart", "stop"):

castle-api/tests/test_deployment_roundtrip.py

@@ -18,9 +18,9 @@ class TestDeploymentEditSafety:
         the shape the edit form consumes (launcher nested under `run`, plus
         reach/expose) — not the flat runtime view (`run_cmd`, top-level launcher)."""
         m = client.get("/deployments/test-svc").json()["manifest"]
-        assert m["run"]["launcher"] == "python"          # spec shape (nested)
+        assert m["run"]["launcher"] == "python"  # spec shape (nested)
-        assert "run_cmd" not in m                         # runtime-only key absent
+        assert "run_cmd" not in m  # runtime-only key absent
-        assert m.get("reach") == "internal"               # normalized from proxy:true
+        assert m.get("reach") == "internal"  # normalized from proxy:true
         assert m["expose"]["http"]["internal"]["port"] == 19000
 
     def test_save_roundtrip_preserves_spec_fields(self, client: TestClient) -> None:
@@ -39,11 +39,13 @@ class TestDeploymentEditSafety:
         This is what makes the astro-class regression structurally impossible —
         even a client that sends a lossy payload can't nuke fields it omitted."""
         before = client.get("/deployments/test-svc").json()["manifest"]
-        resp = client.put("/config/deployments/test-svc", json={"config": {"reach": "off"}})
+        resp = client.put(
+            "/config/deployments/test-svc", json={"config": {"reach": "off"}}
+        )
         assert resp.status_code == 200, resp.text
         after = client.get("/deployments/test-svc").json()["manifest"]
-        assert after["reach"] == "off"                        # the change applied
+        assert after["reach"] == "off"  # the change applied
-        assert after["program"] == before["program"]         # untouched → preserved
+        assert after["program"] == before["program"]  # untouched → preserved
         assert after["run"] == before["run"]
         assert after["expose"] == before["expose"]
 
@@ -58,9 +60,9 @@ class TestDeploymentEditSafety:
         )
         assert resp.status_code == 200, resp.text
         after = client.get("/deployments/test-svc").json()["manifest"]
-        assert after.get("expose") is None                    # cleared
+        assert after.get("expose") is None  # cleared
         assert after["reach"] == "off"
-        assert after["program"] == "test-svc-comp"            # rest preserved
+        assert after["program"] == "test-svc-comp"  # rest preserved
 
 
 class TestProgramEditSafety:
@@ -73,6 +75,6 @@ class TestProgramEditSafety:
         )
         assert resp.status_code == 200, resp.text
         m = client.get("/programs/wired-in").json()["manifest"]
-        assert m["description"] == "renamed"                  # change applied
+        assert m["description"] == "renamed"  # change applied
-        assert m["source"].endswith("wired-in")               # source preserved
+        assert m["source"].endswith("wired-in")  # source preserved
-        assert m["commands"]["lint"] == [["make", "lint"]]    # commands preserved
+        assert m["commands"]["lint"] == [["make", "lint"]]  # commands preserved

castle-api/tests/test_exposure_derivation.py

@@ -37,12 +37,21 @@ def public_client(
         )
     )
     (root / "programs").mkdir()
-    (root / "programs" / "calc.yaml").write_text(yaml.dump({"source": str(root / "calc")}))
+    (root / "programs" / "calc.yaml").write_text(
-    (root / "calc" / "public").mkdir(parents=True)  # static build dir must exist to route
+        yaml.dump({"source": str(root / "calc")})
+    )
+    (root / "calc" / "public").mkdir(
+        parents=True
+    )  # static build dir must exist to route
 
     deps = {
         # public STATIC (the calculator case)
-        "calc": {"program": "calc", "manager": "caddy", "root": "public", "reach": "public"},
+        "calc": {
+            "program": "calc",
+            "manager": "caddy",
+            "root": "public",
+            "reach": "public",
+        },
         # public systemd service
         "web": {
             "manager": "systemd",
@@ -80,20 +89,42 @@ def public_client(
         ),
         deployed={
             "calc": Deployment(
-                manager="caddy", run_cmd=[], kind="static", subdomain="calc",
+                manager="caddy",
-                public=True, static_root=str(root / "calc" / "public"),
+                run_cmd=[],
+                kind="static",
+                subdomain="calc",
+                public=True,
+                static_root=str(root / "calc" / "public"),
             ),
             "web": Deployment(
-                manager="systemd", launcher="python", run_cmd=["x"], kind="service",
+                manager="systemd",
-                port=9001, subdomain="web", public=True, managed=True,
+                launcher="python",
+                run_cmd=["x"],
+                kind="service",
+                port=9001,
+                subdomain="web",
+                public=True,
+                managed=True,
             ),
             "intern": Deployment(
-                manager="systemd", launcher="python", run_cmd=["x"], kind="service",
+                manager="systemd",
-                port=9002, subdomain="intern", public=False, managed=True,
+                launcher="python",
+                run_cmd=["x"],
+                kind="service",
+                port=9002,
+                subdomain="intern",
+                public=False,
+                managed=True,
             ),
             "pg": Deployment(
-                manager="systemd", launcher="container", run_cmd=["x"], kind="service",
+                manager="systemd",
-                port=None, subdomain=None, tcp_port=5432, managed=True,
+                launcher="container",
+                run_cmd=["x"],
+                kind="service",
+                port=None,
+                subdomain=None,
+                tcp_port=5432,
+                managed=True,
             ),
         },
     )
@@ -125,7 +156,9 @@ class TestGatewayPublicUrl:
     def test_public_service_has_public_url(self, public_client: TestClient) -> None:
         assert _routes(public_client)["web"]["public_url"] == "https://web.pub.test"
 
-    def test_internal_service_has_no_public_url(self, public_client: TestClient) -> None:
+    def test_internal_service_has_no_public_url(
+        self, public_client: TestClient
+    ) -> None:
         assert _routes(public_client)["intern"]["public_url"] is None
 
 
@@ -152,10 +185,10 @@ class TestDetailEndpointInvariant:
     @pytest.mark.parametrize(
         "endpoint,expected_reach",
         [
-            ("/deployments/calc", "public"),   # static, unified endpoint
+            ("/deployments/calc", "public"),  # static, unified endpoint
-            ("/services/calc", "public"),       # static, /services endpoint (broke here)
+            ("/services/calc", "public"),  # static, /services endpoint (broke here)
-            ("/deployments/web", "public"),     # service, unified endpoint
+            ("/deployments/web", "public"),  # service, unified endpoint
-            ("/services/web", "public"),         # service, /services endpoint
+            ("/services/web", "public"),  # service, /services endpoint
             ("/deployments/intern", "internal"),
             ("/services/intern", "internal"),
         ],
@@ -164,8 +197,8 @@ class TestDetailEndpointInvariant:
         self, public_client: TestClient, endpoint: str, expected_reach: str
     ) -> None:
         m = public_client.get(endpoint).json()["manifest"]
-        assert m.get("reach") == expected_reach       # spec field present
+        assert m.get("reach") == expected_reach  # spec field present
-        assert "run_cmd" not in m                      # not the runtime view
+        assert "run_cmd" not in m  # not the runtime view
 
     @pytest.mark.parametrize("name", ["calc", "web", "intern"])
     def test_deployments_and_services_endpoints_agree(

castle-api/tests/test_graph_repos.py

@@ -7,15 +7,23 @@ from pathlib import Path
 from fastapi.testclient import TestClient
 
 _ENV = {
-    "GIT_AUTHOR_NAME": "t", "GIT_AUTHOR_EMAIL": "t@t",
+    "GIT_AUTHOR_NAME": "t",
-    "GIT_COMMITTER_NAME": "t", "GIT_COMMITTER_EMAIL": "t@t",
+    "GIT_AUTHOR_EMAIL": "t@t",
-    "GIT_CONFIG_GLOBAL": "/dev/null", "GIT_CONFIG_SYSTEM": "/dev/null",
+    "GIT_COMMITTER_NAME": "t",
+    "GIT_COMMITTER_EMAIL": "t@t",
+    "GIT_CONFIG_GLOBAL": "/dev/null",
+    "GIT_CONFIG_SYSTEM": "/dev/null",
 }
 
 
 def _git(cwd: Path, *args: str) -> None:
-    subprocess.run(["git", "-C", str(cwd), *args], check=True,
+    subprocess.run(
-                   capture_output=True, text=True, env={**os.environ, **_ENV})
+        ["git", "-C", str(cwd), *args],
+        check=True,
+        capture_output=True,
+        text=True,
+        env={**os.environ, **_ENV},
+    )
 
 
 def _commit(cwd: Path, fname: str) -> None:

castle-api/tests/test_kind_twins.py

@@ -0,0 +1,100 @@
+"""A tool, a service, and a job may share a name — kind-scoped endpoints must
+address (and mutate) exactly one twin.
+
+These guard the collision case the per-kind identity refactor enables: a
+`backup` service + job + tool coexisting. The risk is a save/delete against one
+kind bleeding into a same-named twin of another kind. We assert against the
+on-disk config (load_config) so we're testing the persisted invariant, not a
+response echo.
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+from fastapi.testclient import TestClient
+
+from castle_core.config import load_config
+
+# Minimal, valid specs for each kind — all named "backup", all referencing the
+# same program. Only the service is HTTP-exposed (none claims a subdomain here),
+# so the trio passes subdomain-uniqueness validation.
+_SVC = {
+    "program": "backup",
+    "run": {"runner": "python", "program": "backup"},
+    "manage": {"systemd": {}},
+}
+_JOB = {
+    "program": "backup",
+    "run": {"runner": "command", "argv": ["backup"]},
+    "schedule": "0 3 * * *",
+}
+_TOOL = {"program": "backup", "run": {"runner": "path"}}
+
+
+def _put(client: TestClient, section: str, name: str, cfg: dict) -> None:
+    r = client.put(f"/config/{section}/{name}", json={"config": cfg})
+    assert r.status_code == 200, r.text
+
+
+class TestKindScopedTwins:
+    def test_same_name_across_kinds_coexist(
+        self, client: TestClient, castle_root: Path
+    ) -> None:
+        """Creating a service, job, and tool all named `backup` yields three
+        distinct deployments — one per kind, none overwriting another."""
+        _put(client, "services", "backup", _SVC)
+        _put(client, "jobs", "backup", _JOB)
+        _put(client, "tools", "backup", _TOOL)
+
+        cfg = load_config(castle_root)
+        assert "backup" in cfg.services
+        assert "backup" in cfg.jobs
+        assert "backup" in cfg.tools
+
+    def test_detail_endpoints_resolve_the_right_twin(
+        self, client: TestClient, castle_root: Path
+    ) -> None:
+        """`/services/backup` and `/jobs/backup` each return their own kind, not
+        whichever twin happens to sort first."""
+        _put(client, "services", "backup", _SVC)
+        _put(client, "jobs", "backup", _JOB)
+
+        svc = client.get("/services/backup").json()
+        job = client.get("/jobs/backup").json()
+        assert svc["kind"] == "service"
+        # Each endpoint returns its own twin: the job carries the schedule, the
+        # service does not — so neither resolved to the other.
+        assert job["manifest"].get("schedule") == "0 3 * * *"
+        assert "schedule" not in svc["manifest"]
+
+    def test_kind_scoped_save_does_not_touch_the_twin(
+        self, client: TestClient, castle_root: Path
+    ) -> None:
+        """A partial patch to the *service* backup must leave the *job* backup
+        (and its schedule) untouched — the wrong-twin bleed this refactor closes."""
+        _put(client, "services", "backup", _SVC)
+        _put(client, "jobs", "backup", _JOB)
+
+        _put(client, "services", "backup", {"reach": "off"})
+
+        cfg = load_config(castle_root)
+        assert cfg.jobs["backup"].schedule == "0 3 * * *"  # job untouched
+        assert "backup" in cfg.services  # service still there
+
+    def test_kind_scoped_delete_removes_only_that_twin(
+        self, client: TestClient, castle_root: Path
+    ) -> None:
+        """Deleting `/config/tools/backup` drops only the tool; the service and
+        job twins survive."""
+        _put(client, "services", "backup", _SVC)
+        _put(client, "jobs", "backup", _JOB)
+        _put(client, "tools", "backup", _TOOL)
+
+        r = client.delete("/config/tools/backup")
+        assert r.status_code == 200, r.text
+
+        cfg = load_config(castle_root)
+        assert "backup" not in cfg.tools  # tool gone
+        assert "backup" in cfg.services  # twins survive
+        assert "backup" in cfg.jobs

castle-api/tests/test_mqtt.py

@@ -9,7 +9,9 @@ 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),
+        node=NodeConfig(
+            hostname="tower", castle_root="/data/repos/castle", gateway_port=9000
+        ),
         deployed={
             "my-svc": Deployment(
                 manager="systemd",
@@ -80,7 +82,9 @@ class TestRegistrySerialization:
         reg = NodeRegistry(
             node=NodeConfig(hostname="minimal"),
             deployed={
-                "bare": Deployment(manager="systemd", launcher="command", run_cmd=["bare"], name="bare"),
+                "bare": Deployment(
+                    manager="systemd", launcher="command", run_cmd=["bare"], name="bare"
+                ),
             },
         )
         restored = _json_to_registry(_registry_to_json(reg))

castle-api/tests/test_nodes.py

@@ -31,7 +31,9 @@ class TestNodesList:
         assert local["deployed_count"] == 2  # test-svc + test-tool
         assert local["service_count"] == 1  # only test-svc is a service
 
-    def test_includes_remote_nodes(self, client: TestClient, registry_path: Path) -> None:
+    def test_includes_remote_nodes(
+        self, client: TestClient, registry_path: Path
+    ) -> None:
         """Remote nodes from mesh state are included."""
         import castle_api.mesh as mesh_mod
 
@@ -42,7 +44,8 @@ class TestNodesList:
                 node=NodeConfig(hostname="devbox", gateway_port=9000),
                 deployed={
                     "remote-svc": Deployment(
-                        manager="systemd", launcher="python",
+                        manager="systemd",
+                        launcher="python",
                         run_cmd=["svc"],
                         port=9050,
                         kind="service",

castle-api/tests/test_programs_commands.py

@@ -4,7 +4,9 @@ from fastapi.testclient import TestClient
 
 
 class TestProgramCommands:
-    def test_wired_in_program_surfaces_commands_and_repo(self, client: TestClient) -> None:
+    def test_wired_in_program_surfaces_commands_and_repo(
+        self, client: TestClient
+    ) -> None:
         """A stack-less adopted program exposes its declared commands + repo."""
         resp = client.get("/programs")
         assert resp.status_code == 200