core: drop pre-migration read-compat now both nodes are current-format

Both machines (civil, primer) are on the current on-disk layout, so the
read-compat for older formats is dead weight. Remove it:

- config loader: only deployments/<kind>/<name>.yaml (drop flat deployments/*.yaml
  and the services/+jobs/ split); drop the run.runner→manager normalizer.
- registry loader: require composite <kind>/<name> keys + manager/kind (registry.yaml
  is regenerated every apply, so it's always current).
- manifest: drop the proxy/public→reach legacy input and the component→program alias
  (the derived proxy/public read accessors stay). reach field defaults already match.
- config_editor: parse incoming specs directly (frontend sends the current shape).
- save_config: drop the flat-file migration cleanup.

Bootstrap seeds move to bootstrap/deployments/<kind>/ and install.sh seeds that
per-kind tree directly. The legacy→current translation for terse test fixtures
moves into each test conftest (out of production). All suites green (362).

bootstrap/deployments/services/castle-api.yaml

@@ -9,7 +9,7 @@ expose:
     internal:
       port: 9020
     health_path: /health
-proxy: true
+reach: internal
 manage:
   systemd: {}
 defaults:

castle-api/src/castle_api/config_editor.py

@@ -13,7 +13,6 @@ from castle_core.config import (
     KINDS,
     CastleConfig,
     _DEPLOYMENT_ADAPTER,
-    _normalize_deployment_dict,
     _program_to_yaml_dict,
     _spec_to_yaml_dict,
     load_config,
@@ -160,16 +159,11 @@ def save_yaml(request: ConfigSaveRequest) -> ConfigSaveResponse:
         except Exception as e:
             errors.append(f"programs.{name}: {e}")
 
-    # Validate deployments (accepting a legacy services:/jobs: split too, which
+    # Validate deployments (a flat name→spec map in the manager-discriminated shape).
-    # the normalizer folds into the single manager-discriminated collection).
     deployments = {}
-    raw_deps: dict = dict(data.get("deployments") or {})
+    for name, dep_data in (data.get("deployments") or {}).items():
-    for legacy in ("services", "jobs"):
-        raw_deps.update(data.get(legacy) or {})
-    for name, dep_data in raw_deps.items():
         try:
-            dep_copy = _normalize_deployment_dict(dict(dep_data) if dep_data else {})
+            dep_copy = dict(dep_data) if dep_data else {}
-            dep_copy = dict(dep_copy)
             dep_copy["id"] = name
             deployments[name] = _DEPLOYMENT_ADAPTER.validate_python(dep_copy)
         except Exception as e:
@@ -331,9 +325,7 @@ def _save_deployment(name: str, config_dict: dict, kind: str | None = None) -> d
         existing = named[0][1]
     else:
         try:
-            probe = _DEPLOYMENT_ADAPTER.validate_python(
+            probe = _DEPLOYMENT_ADAPTER.validate_python({**incoming, "id": name})
-                _normalize_deployment_dict({**incoming, "id": name})
-            )
             existing = config.deployment(kind_for(probe), name)
         except Exception:
             existing = None
@@ -352,7 +344,7 @@ def _save_deployment(name: str, config_dict: dict, kind: str | None = None) -> d
                 merged["description"] = config.programs[prog].description
 
     try:
-        dep = _DEPLOYMENT_ADAPTER.validate_python(_normalize_deployment_dict(merged))
+        dep = _DEPLOYMENT_ADAPTER.validate_python(merged)
     except Exception as e:
         raise HTTPException(
             status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,

castle-api/tests/conftest.py

@@ -69,20 +69,68 @@ def nats_url() -> Generator[str, None, None]:
         subprocess.run(["docker", "rm", "-f", name], capture_output=True)
 
 
+def _modernize_deployment(spec: dict) -> dict:
+    """Translate a test's terse legacy deployment dict to the current
+    manager-discriminated shape (production dropped this read-compat post-migration).
+    ``proxy``/``public`` → ``reach``; ``run.runner`` → ``manager`` (+ ``run.launcher``)."""
+    d = dict(spec)
+    proxy = bool(d.pop("proxy", False))
+    public = bool(d.pop("public", False))
+    if "reach" not in d:
+        if public:
+            d["reach"] = "public"
+        elif proxy:
+            d["reach"] = "internal"
+    if "manager" not in d:
+        run = dict(d.pop("run", None) or {})
+        runner = run.get("runner")
+        if runner == "static":
+            d["manager"] = "caddy"
+            if run.get("root"):
+                d["root"] = run["root"]
+        elif runner == "path":
+            d["manager"] = "path"
+        elif runner == "remote":
+            d["manager"] = "none"
+            for k in ("base_url", "health_url"):
+                if run.get(k):
+                    d[k] = run[k]
+        else:
+            launch = {k: v for k, v in run.items() if k != "runner"}
+            launch["launcher"] = runner
+            d["manager"] = "systemd"
+            d["run"] = launch
+    return d
+
+
+def _store_for(spec: dict) -> str:
+    if spec.get("schedule"):
+        return "jobs"
+    return {"systemd": "services", "caddy": "statics", "path": "tools", "none": "references"}[
+        spec["manager"]
+    ]
+
+
 def _write_castle_config(root: Path, config: dict) -> None:
-    """Scatter a nested castle config dict into the directory-per-resource layout."""
+    """Scatter a nested castle config dict into the on-disk layout: castle.yaml globals,
+    programs/<name>.yaml, and deployments/<kind>/<name>.yaml (fields modernized)."""
     globals_data = {k: v for k, v in config.items() if k in ("gateway", "repo")}
     (root / "castle.yaml").write_text(yaml.dump(globals_data, default_flow_style=False))
-    for section in ("programs", "services", "jobs"):
+    programs = config.get("programs") or {}
-        entries = config.get(section) or {}
+    if programs:
-        if not entries:
+        (root / "programs").mkdir(parents=True, exist_ok=True)
-            continue
+        for name, spec in programs.items():
-        section_dir = root / section
+            (root / "programs" / f"{name}.yaml").write_text(
-        section_dir.mkdir(parents=True, exist_ok=True)
-        for name, spec in entries.items():
-            (section_dir / f"{name}.yaml").write_text(
                 yaml.dump(spec, default_flow_style=False)
             )
+    for section in ("services", "jobs", "deployments"):
+        for name, spec in (config.get(section) or {}).items():
+            modern = _modernize_deployment(spec)
+            store_dir = root / "deployments" / _store_for(modern)
+            store_dir.mkdir(parents=True, exist_ok=True)
+            (store_dir / f"{name}.yaml").write_text(
+                yaml.dump(modern, default_flow_style=False)
+            )
 
 
 @pytest.fixture
@@ -161,7 +209,7 @@ def registry_path(tmp_path: Path, castle_root: Path) -> Generator[Path, None, No
             gateway_port=9000,
         ),
         deployed={
-            "test-svc": Deployment(
+            NodeRegistry.key("service", "test-svc"): Deployment(
                 manager="systemd",
                 launcher="python",
                 run_cmd=["uv", "run", "test-svc"],
@@ -170,6 +218,7 @@ def registry_path(tmp_path: Path, castle_root: Path) -> Generator[Path, None, No
                     "TEST_SVC_DATA_DIR": "/home/user/.castle/data/test-svc",
                 },
                 description="Test service",
+                name="test-svc",
                 kind="service",
                 port=19000,
                 health_path="/health",
@@ -177,10 +226,11 @@ def registry_path(tmp_path: Path, castle_root: Path) -> Generator[Path, None, No
                 managed=True,
             ),
             # A deployed tool (path) — must NOT leak into the /services list.
-            "test-tool": Deployment(
+            NodeRegistry.key("tool", "test-tool"): Deployment(
                 manager="path",
                 run_cmd=[],
                 description="Test tool",
+                name="test-tool",
                 kind="tool",
             ),
         },

castle-api/tests/test_exposure_derivation.py

@@ -74,9 +74,12 @@ def public_client(
             "reach": "internal",
         },
     }
-    (root / "deployments").mkdir()
+    _store = {"caddy": "statics", "path": "tools", "none": "references"}
     for name, spec in deps.items():
-        (root / "deployments" / f"{name}.yaml").write_text(yaml.dump(spec))
+        store = "jobs" if spec.get("schedule") else _store.get(spec["manager"], "services")
+        store_dir = root / "deployments" / store
+        store_dir.mkdir(parents=True, exist_ok=True)
+        (store_dir / f"{name}.yaml").write_text(yaml.dump(spec))
 
     reg = NodeRegistry(
         node=NodeConfig(
@@ -88,38 +91,42 @@ def public_client(
             public_domain="pub.test",
         ),
         deployed={
-            "calc": Deployment(
+            NodeRegistry.key("static", "calc"): Deployment(
                 manager="caddy",
                 run_cmd=[],
+                name="calc",
                 kind="static",
                 subdomain="calc",
                 public=True,
                 static_root=str(root / "calc" / "public"),
             ),
-            "web": Deployment(
+            NodeRegistry.key("service", "web"): Deployment(
                 manager="systemd",
                 launcher="python",
                 run_cmd=["x"],
+                name="web",
                 kind="service",
                 port=9001,
                 subdomain="web",
                 public=True,
                 managed=True,
             ),
-            "intern": Deployment(
+            NodeRegistry.key("service", "intern"): Deployment(
                 manager="systemd",
                 launcher="python",
                 run_cmd=["x"],
+                name="intern",
                 kind="service",
                 port=9002,
                 subdomain="intern",
                 public=False,
                 managed=True,
             ),
-            "pg": Deployment(
+            NodeRegistry.key("service", "pg"): Deployment(
                 manager="systemd",
                 launcher="container",
                 run_cmd=["x"],
+                name="pg",
                 kind="service",
                 port=None,
                 subdomain=None,

castle-api/tests/test_kind_twins.py

@@ -21,15 +21,17 @@ from castle_core.config import load_config
 # so the trio passes subdomain-uniqueness validation.
 _SVC = {
     "program": "backup",
-    "run": {"runner": "python", "program": "backup"},
+    "manager": "systemd",
+    "run": {"launcher": "python", "program": "backup"},
     "manage": {"systemd": {}},
 }
 _JOB = {
     "program": "backup",
-    "run": {"runner": "command", "argv": ["backup"]},
+    "manager": "systemd",
+    "run": {"launcher": "command", "argv": ["backup"]},
     "schedule": "0 3 * * *",
 }
-_TOOL = {"program": "backup", "run": {"runner": "path"}}
+_TOOL = {"program": "backup", "manager": "path"}
 
 
 def _put(client: TestClient, section: str, name: str, cfg: dict) -> None:

cli/tests/conftest.py

@@ -9,18 +9,64 @@ import pytest
 import yaml
 
 
+def _modernize_deployment(spec: dict) -> dict:
+    """Translate a test's terse legacy deployment dict to the current
+    manager-discriminated shape (production dropped this read-compat post-migration).
+    ``proxy``/``public`` → ``reach``; ``run.runner`` → ``manager`` (+ ``run.launcher``)."""
+    d = dict(spec)
+    proxy = bool(d.pop("proxy", False))
+    public = bool(d.pop("public", False))
+    if "reach" not in d:
+        if public:
+            d["reach"] = "public"
+        elif proxy:
+            d["reach"] = "internal"
+    if "manager" not in d:
+        run = dict(d.pop("run", None) or {})
+        runner = run.get("runner")
+        if runner == "static":
+            d["manager"] = "caddy"
+            if run.get("root"):
+                d["root"] = run["root"]
+        elif runner == "path":
+            d["manager"] = "path"
+        elif runner == "remote":
+            d["manager"] = "none"
+            for k in ("base_url", "health_url"):
+                if run.get(k):
+                    d[k] = run[k]
+        else:
+            launch = {k: v for k, v in run.items() if k != "runner"}
+            launch["launcher"] = runner
+            d["manager"] = "systemd"
+            d["run"] = launch
+    return d
+
+
+def _store_for(spec: dict) -> str:
+    if spec.get("schedule"):
+        return "jobs"
+    return {"systemd": "services", "caddy": "statics", "path": "tools", "none": "references"}[
+        spec["manager"]
+    ]
+
+
 def _write_castle_config(root: Path, config: dict) -> None:
-    """Scatter a nested castle config dict into the directory-per-resource layout."""
+    """Scatter a nested castle config dict into the on-disk layout: castle.yaml globals,
+    programs/<name>.yaml, and deployments/<kind>/<name>.yaml (fields modernized)."""
     globals_data = {k: v for k, v in config.items() if k in ("gateway", "repo")}
     (root / "castle.yaml").write_text(yaml.dump(globals_data, default_flow_style=False))
-    for section in ("programs", "services", "jobs"):
+    programs = config.get("programs") or {}
-        entries = config.get(section) or {}
+    if programs:
-        if not entries:
+        (root / "programs").mkdir(parents=True, exist_ok=True)
-            continue
+        for name, spec in programs.items():
-        section_dir = root / section
+            (root / "programs" / f"{name}.yaml").write_text(yaml.dump(spec, default_flow_style=False))
-        section_dir.mkdir(parents=True, exist_ok=True)
+    for section in ("services", "jobs", "deployments"):
-        for name, spec in entries.items():
+        for name, spec in (config.get(section) or {}).items():
-            (section_dir / f"{name}.yaml").write_text(yaml.dump(spec, default_flow_style=False))
+            modern = _modernize_deployment(spec)
+            store_dir = root / "deployments" / _store_for(modern)
+            store_dir.mkdir(parents=True, exist_ok=True)
+            (store_dir / f"{name}.yaml").write_text(yaml.dump(modern, default_flow_style=False))
 
 
 @pytest.fixture

core/src/castle_core/config.py

@@ -352,42 +352,9 @@ def _parse_program(name: str, data: dict) -> ProgramSpec:
     return ProgramSpec.model_validate(data_copy)
 
 
-def _normalize_deployment_dict(data: dict) -> dict:
-    """Map a legacy service/job entry to the manager-discriminated shape.
-
-    Legacy entries carry `run.runner` (including static/path/remote); new entries
-    carry `manager` and (for systemd) `run.launcher`. New-shape entries pass through.
-    """
-    if "manager" in data:
-        return data
-    d = dict(data)
-    run = dict(d.pop("run", None) or {})
-    runner = run.get("runner")
-    if runner == "static":
-        d["manager"] = "caddy"
-        if run.get("root"):
-            d["root"] = run["root"]
-    elif runner == "path":
-        d["manager"] = "path"
-    elif runner == "remote":
-        d["manager"] = "none"
-        if run.get("base_url"):
-            d["base_url"] = run["base_url"]
-        if run.get("health_url"):
-            d["health_url"] = run["health_url"]
-    else:
-        # A process launcher (python/command/container/compose/node) → systemd.
-        d["manager"] = "systemd"
-        launch = {k: v for k, v in run.items() if k != "runner"}
-        launch["launcher"] = runner
-        d["run"] = launch
-    return d
-
-
 def _parse_deployment(name: str, data: dict) -> DeploymentSpec:
-    """Parse a deployment entry (new or legacy shape) into a DeploymentSpec."""
+    """Parse a deployment entry (manager-discriminated shape) into a DeploymentSpec."""
-    data_copy = _normalize_deployment_dict(data)
+    data_copy = dict(data)
-    data_copy = dict(data_copy)
     data_copy["id"] = name
     return _DEPLOYMENT_ADAPTER.validate_python(data_copy)
 
@@ -515,31 +482,16 @@ def _validate_subdomains(stores: dict[str, dict[str, DeploymentSpec]]) -> None:
 def _load_deployments(root: Path) -> dict[str, dict[str, DeploymentSpec]]:
     """Load the per-kind deployment stores for a config root.
 
-    New layout: ``deployments/<store>/<name>.yaml`` (store = services|jobs|tools|
+    Layout: ``deployments/<store>/<name>.yaml`` (store = services|jobs|tools|statics|
-    statics|references). Read-compat for the pre-migration layouts: flat
+    references). Every file is routed to its store by ``kind_for(spec)`` — the dir is
-    ``deployments/*.yaml`` files, and the older ``services/``+``jobs/`` split. Every
+    a namespace, the spec's manager/schedule is the source of truth.
-    file is routed to its store by ``kind_for(spec)`` — the dir is a namespace, the
-    spec's manager/schedule is the source of truth (they must agree post-migration).
     """
     stores: dict[str, dict[str, DeploymentSpec]] = {s: {} for s in _KIND_STORE.values()}
-
-    def route(name: str, data: dict) -> None:
-        spec = _parse_deployment(name, data)
-        stores[_KIND_STORE[kind_for(spec)]][name] = spec
-
     dep_dir = root / "deployments"
-    # New per-kind subdirs.
     for store in _KIND_STORE.values():
         for name, data in _load_resource_dir(dep_dir / store).items():
-            route(name, data)
+            spec = _parse_deployment(name, data)
-    # Legacy flat deployments/*.yaml (top-level only — subdirs handled above).
+            stores[_KIND_STORE[kind_for(spec)]][name] = spec
-    for name, data in _load_resource_dir(dep_dir).items():
-        route(name, data)
-    # Oldest layout: services/ + jobs/ dirs, only if there's no deployments/ dir.
-    if not dep_dir.is_dir():
-        for legacy in ("services", "jobs"):
-            for name, data in _load_resource_dir(root / legacy).items():
-                route(name, data)
     return stores
 
 
@@ -750,10 +702,6 @@ def save_config(config: CastleConfig) -> None:
             dep_dir / store,
             {n: _spec_to_yaml_dict(d) for n, d in config.store_for(kind).items()},
         )
-    # Migration cleanup: drop any pre-migration flat deployments/*.yaml files.
-    if dep_dir.is_dir():
-        for path in dep_dir.glob("*.yaml"):
-            path.unlink()
 
 
 def ensure_dirs(config: CastleConfig) -> None:

core/src/castle_core/manifest.py

@@ -5,7 +5,7 @@ from __future__ import annotations
 from enum import Enum
 from typing import Annotated, Literal, Union
 
-from pydantic import AliasChoices, BaseModel, ConfigDict, Field, model_validator
+from pydantic import BaseModel, ConfigDict, Field, model_validator
 
 EnvMap = dict[str, str]
 
@@ -25,8 +25,7 @@ class Reach(str, Enum):
     ``public``   → *also* projected to the internet (HTTP via the tunnel origin;
                    TCP via ``cloudflared access tcp``). Implies ``internal``.
 
-    Replaces the old ``proxy``/``public`` booleans; ``proxy``/``public`` survive as
+    ``proxy``/``public`` survive as derived read-only accessors on the deployment.
-    derived read-only accessors and as accepted *legacy input* (normalized below).
     """
 
     OFF = "off"
@@ -34,29 +33,6 @@ class Reach(str, Enum):
     PUBLIC = "public"
 
 
-def _reach_from_legacy(data: object, default: Reach) -> object:
-    """Map legacy ``proxy``/``public`` booleans on a raw deployment dict to ``reach``.
-
-    Runs as a ``mode="before"`` validator. When ``reach`` is given explicitly it
-    wins (legacy keys are dropped); otherwise ``reach`` is derived from the old
-    booleans: ``public`` → PUBLIC, ``proxy`` → INTERNAL, else ``default``. Non-dict
-    input (e.g. model re-validation) passes through untouched.
-    """
-    if not isinstance(data, dict):
-        return data
-    d = dict(data)
-    proxy = bool(d.pop("proxy", False))
-    public = bool(d.pop("public", False))
-    if "reach" not in d:
-        if public:
-            d["reach"] = Reach.PUBLIC
-        elif proxy:
-            d["reach"] = Reach.INTERNAL
-        else:
-            d["reach"] = default
-    return d
-
-
 # ---------------------
 # Launch specs — how systemd starts a process (discriminated union on `launcher`)
 # ---------------------
@@ -416,10 +392,8 @@ class DeploymentBase(BaseModel):
     model_config = ConfigDict(populate_by_name=True)
 
     id: str = ""
-    # The program this deployment materializes. (`component` = legacy alias.)
+    # The program this deployment materializes.
-    program: str | None = Field(
+    program: str | None = None
-        default=None, validation_alias=AliasChoices("program", "component")
-    )
     description: str | None = None
     defaults: DefaultsSpec | None = None
     # Deployment-to-deployment preconditions: other deployments this one needs
@@ -447,11 +421,6 @@ class SystemdDeployment(DeploymentBase):
     reach: Reach = Reach.OFF
     manage: ManageSpec | None = None
 
-    @model_validator(mode="before")
-    @classmethod
-    def _normalize_reach(cls, data: object) -> object:
-        return _reach_from_legacy(data, default=Reach.OFF)
-
     @model_validator(mode="after")
     def _validate_reach(self) -> SystemdDeployment:
         # An exposed reach needs a port to expose. Without an `expose` block the
@@ -519,11 +488,6 @@ class CaddyDeployment(DeploymentBase):
     # `internal` or `public` (never `off`). `public` = also project via the tunnel.
     reach: Reach = Reach.INTERNAL
 
-    @model_validator(mode="before")
-    @classmethod
-    def _normalize_reach(cls, data: object) -> object:
-        return _reach_from_legacy(data, default=Reach.INTERNAL)
-
     @model_validator(mode="after")
     def _validate_reach(self) -> CaddyDeployment:
         if self.reach == Reach.OFF:

core/src/castle_core/registry.py

@@ -167,34 +167,12 @@ def load_registry(path: Path | None = None) -> NodeRegistry:
 
     deployed: dict[str, Deployment] = {}
     for key, comp_data in data.get("deployed", {}).items():
-        # Key is the composite "<kind>/<name>" (new) or a bare name (legacy).
+        # Key is the composite "<kind>/<name>"; manager/kind are always present
-        key_kind, name = key.split("/", 1) if "/" in key else (None, key)
+        # (save_registry writes them). registry.yaml is regenerated every apply.
-        # New shape carries manager/launcher/kind; legacy carries runner/behavior.
+        kind, name = key.split("/", 1)
-        manager = comp_data.get("manager")
-        launcher = comp_data.get("launcher")
-        if manager is None:
-            runner = comp_data.get("runner", "command")
-            manager = {"static": "caddy", "path": "path", "remote": "none"}.get(
-                runner, "systemd"
-            )
-            if manager == "systemd":
-                launcher = runner
-        kind = comp_data.get("kind") or key_kind
-        if kind is None:
-            behavior = comp_data.get("behavior")
-            if comp_data.get("schedule"):
-                kind = "job"
-            elif manager == "caddy" or behavior == "frontend":
-                kind = "static"
-            elif manager == "path" or behavior == "tool":
-                kind = "tool"
-            elif manager == "none":
-                kind = "reference"
-            else:
-                kind = "service"
         deployed[NodeRegistry.key(kind, name)] = Deployment(
-            manager=manager,
+            manager=comp_data["manager"],
-            launcher=launcher,
+            launcher=comp_data.get("launcher"),
             run_cmd=comp_data.get("run_cmd", []),
             stop_cmd=comp_data.get("stop_cmd", []),
             env=comp_data.get("env", {}),

core/tests/conftest.py

@@ -14,26 +14,84 @@ import pytest
 import yaml
 
 
-def write_castle_config(root: Path, config: dict) -> None:
+def _modernize_deployment(spec: dict) -> dict:
-    """Scatter a nested castle config dict into the directory-per-resource layout.
+    """Translate a test's legacy deployment dict to the current manager-discriminated
+    shape. Production dropped this read-compat once every machine migrated; the tests
+    keep authoring the terse legacy shape, so the translation lives here instead.
 
-    `config` uses the legacy nested shape (gateway/repo at top level, plus
+    - ``proxy``/``public`` booleans → ``reach`` (internal/public).
-    programs/services/jobs mappings); this writes castle.yaml with globals and
+    - ``run.runner`` → ``manager`` (+ ``run.launcher`` for the systemd process kinds).
-    one file per resource under programs/, services/, jobs/.
+    """
+    d = dict(spec)
+    proxy = bool(d.pop("proxy", False))
+    public = bool(d.pop("public", False))
+    if "reach" not in d:
+        if public:
+            d["reach"] = "public"
+        elif proxy:
+            d["reach"] = "internal"
+    if "manager" not in d:
+        run = dict(d.pop("run", None) or {})
+        runner = run.get("runner")
+        if runner == "static":
+            d["manager"] = "caddy"
+            if run.get("root"):
+                d["root"] = run["root"]
+        elif runner == "path":
+            d["manager"] = "path"
+        elif runner == "remote":
+            d["manager"] = "none"
+            for k in ("base_url", "health_url"):
+                if run.get(k):
+                    d[k] = run[k]
+        else:
+            launch = {k: v for k, v in run.items() if k != "runner"}
+            launch["launcher"] = runner
+            d["manager"] = "systemd"
+            d["run"] = launch
+    return d
+
+
+def _store_for(spec: dict) -> str:
+    """The deployments/<store>/ subdir for a (modernized) deployment spec."""
+    if spec.get("schedule"):
+        return "jobs"
+    return {
+        "systemd": "services",
+        "caddy": "statics",
+        "path": "tools",
+        "none": "references",
+    }[spec["manager"]]
+
+
+def write_castle_config(root: Path, config: dict) -> None:
+    """Scatter a nested castle config dict into the on-disk layout.
+
+    `config` uses the terse nested shape (gateway/repo at top level, plus
+    programs/services/jobs mappings); this writes castle.yaml with globals, one file
+    per program under programs/, and each deployment under deployments/<kind>/ after
+    modernizing its legacy fields (see `_modernize_deployment`).
     """
     globals_data = {k: v for k, v in config.items() if k in ("gateway", "repo")}
     (root / "castle.yaml").write_text(yaml.dump(globals_data, default_flow_style=False))
-    for section in ("programs", "services", "jobs"):
+
-        entries = config.get(section) or {}
+    programs = config.get("programs") or {}
-        if not entries:
+    if programs:
-            continue
+        (root / "programs").mkdir(parents=True, exist_ok=True)
-        section_dir = root / section
+        for name, spec in programs.items():
-        section_dir.mkdir(parents=True, exist_ok=True)
+            (root / "programs" / f"{name}.yaml").write_text(
-        for name, spec in entries.items():
-            (section_dir / f"{name}.yaml").write_text(
                 yaml.dump(spec, default_flow_style=False)
             )
 
+    for section in ("services", "jobs", "deployments"):
+        for name, spec in (config.get(section) or {}).items():
+            modern = _modernize_deployment(spec)
+            store_dir = root / "deployments" / _store_for(modern)
+            store_dir.mkdir(parents=True, exist_ok=True)
+            (store_dir / f"{name}.yaml").write_text(
+                yaml.dump(modern, default_flow_style=False)
+            )
+
 
 @pytest.fixture
 def castle_root(tmp_path: Path) -> Generator[Path, None, None]:

core/tests/test_apply.py

@@ -37,7 +37,7 @@ class TestApplyPlan:
     def test_disabled_active_deployment_deactivates(self, castle_root: Path) -> None:
         """A deployment with enabled:false that's currently up → 'deactivate'."""
         # Turn the tool off in config.
-        tool = castle_root / "services" / "test-tool.yaml"
+        tool = castle_root / "deployments" / "tools" / "test-tool.yaml"
         tool.write_text(tool.read_text() + "enabled: false\n")
 
         result = _plan(castle_root, active={"test-tool": True})
@@ -47,7 +47,7 @@ class TestApplyPlan:
 
     def test_disabled_inactive_is_unchanged(self, castle_root: Path) -> None:
         """enabled:false and already down → nothing to do."""
-        tool = castle_root / "services" / "test-tool.yaml"
+        tool = castle_root / "deployments" / "tools" / "test-tool.yaml"
         tool.write_text(tool.read_text() + "enabled: false\n")
 
         result = _plan(castle_root, active={})

core/tests/test_caddyfile.py

@@ -16,6 +16,7 @@ from castle_core.manifest import (
     HttpExposeSpec,
     HttpInternal,
     ProgramSpec,
+    Reach,
     RunPython,
     SystemdDeployment,
 )
@@ -157,7 +158,7 @@ def _service(port: int, *, expose: bool) -> SystemdDeployment:
         manager="systemd",
         run=RunPython(launcher="python", program="svc"),
         expose=ExposeSpec(http=HttpExposeSpec(internal=HttpInternal(port=port))),
-        proxy=expose,
+        reach=Reach.INTERNAL if expose else Reach.OFF,
     )
 
 

core/tests/test_caddyfile_remote.py

@@ -12,8 +12,8 @@ from castle_core.registry import Deployment, NodeConfig, NodeRegistry
 
 def _config_requiring_widget(root: Path) -> None:
     (root / "castle.yaml").write_text(yaml.safe_dump({"gateway": {"port": 18000}}))
-    svc_dir = root / "services"
+    svc_dir = root / "deployments" / "services"
-    svc_dir.mkdir()
+    svc_dir.mkdir(parents=True)
     (svc_dir / "consumer.yaml").write_text(
         yaml.safe_dump(
             {

core/tests/test_manifest.py

@@ -111,9 +111,9 @@ class TestSystemdDeployment:
         )
         assert s.manage.systemd.enable is True
 
-    def test_reach_ladder_and_legacy_mapping(self) -> None:
+    def test_reach_ladder_and_accessors(self) -> None:
-        """`reach` is canonical; legacy proxy/public map to it, and the derived
+        """`reach` is canonical; the derived proxy/public accessors reflect it
-        proxy/public accessors reflect it (public implies internal)."""
+        (public implies internal)."""
         # An exposed reach needs an expose block (see test_reach_requires_expose),
         # so give the base one; reach off doesn't, tested separately below.
         base = dict(
@@ -122,21 +122,15 @@ class TestSystemdDeployment:
             run=RunPython(launcher="python", program="svc"),
             expose={"http": {"internal": {"port": 9001}}},
         )
-        # legacy input still parses
+        s_internal = SystemdDeployment(**base, reach=Reach.INTERNAL)
-        s_proxy = SystemdDeployment.model_validate({**base, "proxy": True})
+        assert s_internal.proxy is True and s_internal.public is False
-        assert s_proxy.reach == Reach.INTERNAL
+        s_pub = SystemdDeployment(**base, reach=Reach.PUBLIC)
-        assert s_proxy.proxy is True and s_proxy.public is False
-        s_pub = SystemdDeployment.model_validate({**base, "proxy": True, "public": True})
-        assert s_pub.reach == Reach.PUBLIC
         assert s_pub.proxy is True and s_pub.public is True
-        # legacy public alone now simply means public (which implies internal)
+        # reach off needs no expose block
-        assert SystemdDeployment.model_validate({**base, "public": True}).reach == Reach.PUBLIC
-        # new canonical field (reach off needs no expose block)
         no_expose = dict(
             id="svc", manager="systemd", run=RunPython(launcher="python", program="svc")
         )
         assert SystemdDeployment(**no_expose, reach=Reach.OFF).reach == Reach.OFF
-        assert SystemdDeployment(**base, reach=Reach.PUBLIC).public is True
 
     def test_reach_requires_expose(self) -> None:
         """An exposed reach with no expose block is rejected (it would otherwise
@@ -146,9 +140,6 @@ class TestSystemdDeployment:
         for reach in ("internal", "public"):
             with pytest.raises(ValueError, match="requires an `expose` block"):
                 SystemdDeployment.model_validate({**base, "reach": reach})
-        # legacy public: true with no expose maps to reach public → same rejection
-        with pytest.raises(ValueError, match="requires an `expose` block"):
-            SystemdDeployment.model_validate({**base, "public": True})
 
     def test_tcp_exposure_is_not_http_exposed(self) -> None:
         """A raw-TCP service is reachable by name+port but never HTTP-routed."""

install.sh

@@ -351,13 +351,17 @@ seed_control_plane() {
         printf 'repo: %s\n' "$CASTLE_ROOT" >> "${CASTLE_HOME}/castle.yaml"
     fi
 
-    local seeded=0 f dst
+    local seeded=0 f dst rel
     for f in "${CASTLE_ROOT}"/bootstrap/programs/*.yaml; do
         dst="${CASTLE_HOME}/programs/$(basename "$f")"
         [ -f "$dst" ] || { cp "$f" "$dst"; seeded=1; }
     done
-    for f in "${CASTLE_ROOT}"/bootstrap/deployments/*.yaml; do
+    # Deployments are seeded into the per-kind layout (deployments/<kind>/<name>.yaml),
-        dst="${CASTLE_HOME}/deployments/$(basename "$f")"
+    # mirroring bootstrap/deployments/<kind>/ — the shape the loader reads.
+    for f in "${CASTLE_ROOT}"/bootstrap/deployments/*/*.yaml; do
+        rel="${f#"${CASTLE_ROOT}"/bootstrap/deployments/}"
+        dst="${CASTLE_HOME}/deployments/${rel}"
+        mkdir -p "$(dirname "$dst")"
         [ -f "$dst" ] || { sed "s#__SPECS_DIR__#${specs}#g" "$f" > "$dst"; seeded=1; }
     done