Refactor configuration handling to use directory-per-resource layout

- Introduced `_write_castle_config` function to scatter nested config into separate YAML files for programs, services, and jobs.
- Updated `castle_root` fixture to utilize the new config writing method.
- Added tests for configuration aggregation and scattering in `test_health.py`.
- Removed legacy unit handling from `manifest.py` and refactored related code in `config.py`.
- Updated documentation to reflect new configuration directory structure.
- Removed obsolete unit tests related to unit expansion.

CLAUDE.md

@@ -6,12 +6,11 @@ with code in this repository.
 ## Overview
 
 Castle is a personal software platform — a monorepo of independent projects
-(services, tools, libraries) managed by the `castle` CLI. The registry
+(services, tools, libraries) managed by the `castle` CLI. The registry config is split into three directories under your config root:
-(`castle.yaml`) has three top-level sections:
 
-- **`programs:`** — Software catalog (source, behavior, stack, system_dependencies, build)
+- **`programs/`** — Software catalog (source, behavior, stack, system_dependencies, build)
-- **`services:`** — Long-running daemons (run, expose, proxy, systemd)
+- **`services/`** — Long-running daemons (run, expose, proxy, systemd)
-- **`jobs:`** — Scheduled tasks (run, cron schedule, systemd timer)
+- **`jobs/`** — Scheduled tasks (run, cron schedule, systemd timer)
 
 Each program has a **stack** (development toolchain: python-fastapi,
 python-cli, react-vite) and a **behavior** (runtime role: daemon, tool,

README.md

@@ -19,14 +19,11 @@ cd cli && uv tool install --editable . && cd ..
 # Run the installer (sets up Docker, Caddy, MQTT, Postgres, Neo4j, directory tree)
 ./install.sh
 
-# Initialize castle.yaml (the registry that tracks everything)
+# Initialize the global castle.yaml (the registry that tracks everything)
+mkdir -p ~/.castle/programs ~/.castle/services ~/.castle/jobs
 cat > ~/.castle/castle.yaml << 'EOF'
 gateway:
   port: 9000
-
-programs: {}
-services: {}
-jobs: {}
 EOF
 
 # See what's here
@@ -82,51 +79,57 @@ Tools are programs with `behavior: tool` — list them with
 
 ## Registry
 
-`castle.yaml` lives at `~/.castle/castle.yaml` and is the single source of truth. It has four top-level sections:
+The registry lives under `~/.castle/` and is the single source of truth, split
+into a global `castle.yaml` plus one file per resource under `programs/`,
+`services/`, and `jobs/`:
 
-- **`programs:`** — Software catalog (source, stack, behavior, build config)
+- **`castle.yaml`** — Global settings (`gateway`, `repo`)
-- **`services:`** — Long-running daemons (run, expose, proxy, systemd)
+- **`programs/<name>.yaml`** — Software catalog (source, stack, behavior, build config)
-- **`jobs:`** — Scheduled tasks (run, cron schedule, systemd timer)
+- **`services/<name>.yaml`** — Long-running daemons (run, expose, proxy, systemd)
-- **`units:`** — Compact shorthand that expands into programs + services/jobs
+- **`jobs/<name>.yaml`** — Scheduled tasks (run, cron schedule, systemd timer)
 
-Services and jobs can reference a program via `component:` for description fallthrough.
+Services and jobs can reference a program via `program:` for description fallthrough.
 
 ```yaml
+# ~/.castle/castle.yaml
 gateway:
   port: 9000
 repo: /path/to/castle
+```
 
-programs:
+```yaml
-  central-context:
+# ~/.castle/programs/central-context.yaml
-    description: Content storage API
+description: Content storage API
-    behavior: daemon
+behavior: daemon
-    source: code/central-context
+source: code/central-context
-    stack: python-fastapi
+stack: python-fastapi
+```
 
-services:
+```yaml
-  central-context:
+# ~/.castle/services/central-context.yaml
-    component: central-context
+program: central-context
-    run:
+run:
-      runner: python
+  runner: python
-      program: central-context
+  program: central-context
-    expose:
+expose:
-      http:
+  http:
-        internal: { port: 9001 }
+    internal: { port: 9001 }
-        health_path: /health
+    health_path: /health
-    proxy:
+proxy:
-      caddy: { path_prefix: /central-context }
+  caddy: { path_prefix: /central-context }
-    manage:
+manage:
-      systemd: {}
+  systemd: {}
+```
 
-jobs:
+```yaml
-  backup-collect:
+# ~/.castle/jobs/backup-collect.yaml
-    component: backup-collect
+program: backup-collect
-    run:
+run:
-      runner: command
+  runner: command
-      argv: [backup-collect]
+  argv: [backup-collect]
-    schedule: "0 2 * * *"
+schedule: "0 2 * * *"
-    manage:
+manage:
-      systemd: {}
+  systemd: {}
 ```
 
 Convention-based env vars (`<PREFIX>_DATA_DIR`, `<PREFIX>_PORT`) are generated

castle-api/src/castle_api/config_editor.py

@@ -9,7 +9,14 @@ import yaml
 from fastapi import APIRouter, HTTPException, status
 from pydantic import BaseModel
 
-from castle_core.config import save_config
+from castle_core.config import (
+    CastleConfig,
+    GatewayConfig,
+    _program_to_yaml_dict,
+    _spec_to_yaml_dict,
+    load_config,
+    save_config,
+)
 from castle_core.manifest import ProgramSpec, JobSpec, ServiceSpec
 
 from castle_api.config import get_castle_root, get_config, get_registry
@@ -61,13 +68,31 @@ def _require_repo() -> None:
         )
 
 
+def _aggregate_yaml(config: CastleConfig) -> str:
+    """Build a unified virtual castle.yaml from the directory-per-resource config."""
+    data: dict = {"gateway": {"port": config.gateway.port}}
+    if config.repo:
+        data["repo"] = str(config.repo)
+    if config.programs:
+        data["programs"] = {
+            n: _program_to_yaml_dict(s, config) for n, s in config.programs.items()
+        }
+    if config.services:
+        data["services"] = {
+            n: _spec_to_yaml_dict(s) for n, s in config.services.items()
+        }
+    if config.jobs:
+        data["jobs"] = {n: _spec_to_yaml_dict(s) for n, s in config.jobs.items()}
+    return yaml.dump(data, default_flow_style=False, sort_keys=False)
+
+
 @router.get("", response_model=ConfigResponse)
 def get_config_yaml() -> ConfigResponse:
-    """Get the raw castle.yaml content."""
+    """Get a unified virtual castle.yaml aggregated from all resource files."""
     _require_repo()
     root = get_castle_root()
-    config_path = root / "castle.yaml"
+    config = load_config(root)
-    return ConfigResponse(yaml_content=config_path.read_text())
+    return ConfigResponse(yaml_content=_aggregate_yaml(config))
 
 
 @router.put("", response_model=ConfigSaveResponse)
@@ -92,37 +117,58 @@ def save_yaml(request: ConfigSaveRequest) -> ConfigSaveResponse:
             detail="YAML must be a mapping",
         )
 
+    # repo: drives repo-relative source resolution (fall back to existing config)
+    repo_path = None
+    if data.get("repo"):
+        from pathlib import Path
+
+        repo_path = Path(data["repo"]).expanduser()
+    else:
+        try:
+            repo_path = load_config(root).repo
+        except Exception:
+            repo_path = None
+
+    def _resolve_source(spec: ProgramSpec) -> None:
+        from pathlib import Path
+
+        if not spec.source:
+            return
+        if spec.source.startswith("repo:") and repo_path:
+            spec.source = str(repo_path / spec.source[5:])
+        elif not Path(spec.source).is_absolute():
+            spec.source = str(root / spec.source)
+
     # Validate programs
-    prog_count = 0
+    programs: dict[str, ProgramSpec] = {}
-    programs_data = data.get("programs") or data.get("components") or {}
+    programs_data = data.get("programs") or {}
     for name, comp_data in programs_data.items():
         try:
             comp_data_copy = dict(comp_data) if comp_data else {}
             comp_data_copy["id"] = name
-            ProgramSpec.model_validate(comp_data_copy)
+            spec = ProgramSpec.model_validate(comp_data_copy)
-            prog_count += 1
+            _resolve_source(spec)
+            programs[name] = spec
         except Exception as e:
             errors.append(f"programs.{name}: {e}")
 
     # Validate services
-    svc_count = 0
+    services: dict[str, ServiceSpec] = {}
     for name, svc_data in data.get("services", {}).items():
         try:
             svc_data_copy = dict(svc_data) if svc_data else {}
             svc_data_copy["id"] = name
-            ServiceSpec.model_validate(svc_data_copy)
+            services[name] = ServiceSpec.model_validate(svc_data_copy)
-            svc_count += 1
         except Exception as e:
             errors.append(f"services.{name}: {e}")
 
     # Validate jobs
-    job_count = 0
+    jobs: dict[str, JobSpec] = {}
     for name, job_data in data.get("jobs", {}).items():
         try:
             job_data_copy = dict(job_data) if job_data else {}
             job_data_copy["id"] = name
-            JobSpec.model_validate(job_data_copy)
+            jobs[name] = JobSpec.model_validate(job_data_copy)
-            job_count += 1
         except Exception as e:
             errors.append(f"jobs.{name}: {e}")
 
@@ -132,15 +178,27 @@ def save_yaml(request: ConfigSaveRequest) -> ConfigSaveResponse:
             detail={"errors": errors},
         )
 
-    # Backup and save
+    prog_count = len(programs)
-    config_path = root / "castle.yaml"
+    svc_count = len(services)
-    backup_path = config_path.with_suffix(".yaml.bak")
+    job_count = len(jobs)
-    shutil.copy2(config_path, backup_path)
+
-    config_path.write_text(request.yaml_content)
+    gateway_data = data.get("gateway", {})
+    config = CastleConfig(
+        root=root,
+        repo=repo_path,
+        gateway=GatewayConfig(port=gateway_data.get("port", 9000)),
+        programs=programs,
+        services=services,
+        jobs=jobs,
+    )
+    save_config(config)
 
     return ConfigSaveResponse(
-        ok=True, program_count=prog_count, service_count=svc_count,
+        ok=True,
-        job_count=job_count, errors=[],
+        program_count=prog_count,
+        service_count=svc_count,
+        job_count=job_count,
+        errors=[],
     )
 
 
@@ -160,9 +218,7 @@ def save_program(name: str, request: ProgramConfigRequest) -> dict:
         )
 
     config = get_config()
-    config.programs[name] = ProgramSpec.model_validate(
+    config.programs[name] = ProgramSpec.model_validate({**request.config, "id": name})
-        {**request.config, "id": name}
-    )
     save_config(config)
     return {"ok": True, "program": name}
 
@@ -211,9 +267,7 @@ def save_service(name: str, request: ServiceConfigRequest) -> dict:
         )
 
     config = get_config()
-    config.services[name] = ServiceSpec.model_validate(
+    config.services[name] = ServiceSpec.model_validate({**request.config, "id": name})
-        {**request.config, "id": name}
-    )
     save_config(config)
     return {"ok": True, "service": name}
 
@@ -248,9 +302,7 @@ def save_job(name: str, request: JobConfigRequest) -> dict:
         )
 
     config = get_config()
-    config.jobs[name] = JobSpec.model_validate(
+    config.jobs[name] = JobSpec.model_validate({**request.config, "id": name})
-        {**request.config, "id": name}
-    )
     save_config(config)
     return {"ok": True, "job": name}
 

castle-api/tests/conftest.py

@@ -17,10 +17,25 @@ from castle_core.registry import (
 )
 
 
+def _write_castle_config(root: Path, config: dict) -> None:
+    """Scatter a nested castle config dict into the directory-per-resource layout."""
+    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 {}
+        if not entries:
+            continue
+        section_dir = root / section
+        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)
+            )
+
+
 @pytest.fixture
 def castle_root(tmp_path: Path) -> Generator[Path, None, None]:
-    """Create a temporary castle root with castle.yaml."""
+    """Create a temporary castle root with directory-per-resource config."""
-    castle_yaml = tmp_path / "castle.yaml"
     config = {
         "gateway": {"port": 9000},
         "programs": {
@@ -77,7 +92,7 @@ def castle_root(tmp_path: Path) -> Generator[Path, None, None]:
             },
         },
     }
-    castle_yaml.write_text(yaml.dump(config, default_flow_style=False))
+    _write_castle_config(tmp_path, config)
     yield tmp_path
 
 
@@ -143,7 +158,14 @@ def registry_path(tmp_path: Path, castle_root: Path) -> Generator[Path, None, No
         "config_editor.get_castle_root": config_editor_mod.get_castle_root,
     }
 
-    for mod in [api_config, routes_mod, services_mod, nodes_mod, stream_mod, config_editor_mod]:
+    for mod in [
+        api_config,
+        routes_mod,
+        services_mod,
+        nodes_mod,
+        stream_mod,
+        config_editor_mod,
+    ]:
         if hasattr(mod, "get_registry"):
             mod.get_registry = _get_registry
         if hasattr(mod, "get_castle_root"):

castle-api/tests/test_health.py

@@ -268,3 +268,36 @@ class TestGateway:
         assert data["routes"]
         for r in data["routes"]:
             assert r["kind"] in ("static", "proxy", "remote")
+
+
+class TestConfigEditor:
+    """Virtual castle.yaml aggregation/scatter endpoints."""
+
+    def test_get_aggregates_resources(self, client: TestClient) -> None:
+        """GET /config returns a unified YAML aggregating all resource files."""
+        import yaml
+
+        response = client.get("/config")
+        assert response.status_code == 200
+        data = yaml.safe_load(response.json()["yaml_content"])
+        assert "test-tool" in data["programs"]
+        assert "test-svc" in data["services"]
+        assert "test-job" in data["jobs"]
+
+    def test_put_scatters_and_prunes(self, client: TestClient, castle_root) -> None:
+        """PUT /config writes resource files and prunes removed ones."""
+        import yaml
+
+        current = yaml.safe_load(client.get("/config").json()["yaml_content"])
+        current["services"].pop("test-svc")
+        current["programs"]["new-tool"] = {
+            "description": "Brand new",
+            "behavior": "tool",
+        }
+        resp = client.put("/config", json={"yaml_content": yaml.dump(current)})
+        assert resp.status_code == 200, resp.text
+        assert not (castle_root / "services" / "test-svc.yaml").exists()
+        assert (castle_root / "programs" / "new-tool.yaml").exists()
+        after = yaml.safe_load(client.get("/config").json()["yaml_content"])
+        assert "new-tool" in after["programs"]
+        assert "test-svc" not in (after.get("services") or {})

cli/src/castle_cli/manifest.py

@@ -28,6 +28,4 @@ from castle_core.manifest import (  # noqa: F401 — explicit re-exports for typ
     ServiceSpec,
     SystemdSpec,
     TLSMode,
-    UnitKind,
-    UnitSpec,
 )

cli/tests/conftest.py

@@ -9,10 +9,23 @@ import pytest
 import yaml
 
 
+def _write_castle_config(root: Path, config: dict) -> None:
+    """Scatter a nested castle config dict into the directory-per-resource layout."""
+    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 {}
+        if not entries:
+            continue
+        section_dir = root / section
+        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))
+
+
 @pytest.fixture
 def castle_root(tmp_path: Path) -> Generator[Path, None, None]:
-    """Create a temporary castle root with castle.yaml."""
+    """Create a temporary castle root with directory-per-resource config."""
-    castle_yaml = tmp_path / "castle.yaml"
     config = {
         "gateway": {"port": 18000},
         "programs": {
@@ -61,7 +74,7 @@ def castle_root(tmp_path: Path) -> Generator[Path, None, None]:
             },
         },
     }
-    castle_yaml.write_text(yaml.dump(config, default_flow_style=False))
+    _write_castle_config(tmp_path, config)
 
     # Create project directories
     svc_dir = tmp_path / "test-svc"

core/src/castle_core/config.py

@@ -4,27 +4,15 @@ from __future__ import annotations
 
 import os
 import re
-from dataclasses import dataclass, field
+from dataclasses import dataclass
 from pathlib import Path
 
 import yaml
 
 from castle_core.manifest import (
-    CaddySpec,
-    DefaultsSpec,
-    ExposeSpec,
-    HttpExposeSpec,
-    HttpInternal,
     JobSpec,
-    ManageSpec,
     ProgramSpec,
-    ProxySpec,
-    RunCommand,
-    RunPython,
     ServiceSpec,
-    SystemdSpec,
-    UnitKind,
-    UnitSpec,
 )
 
 
@@ -132,8 +120,6 @@ class CastleConfig:
     programs: dict[str, ProgramSpec]
     services: dict[str, ServiceSpec]
     jobs: dict[str, JobSpec]
-    _unit_names: set[str] = field(default_factory=set)
-    _units_raw: dict[str, dict] = field(default_factory=dict)
 
     @property
     def tools(self) -> dict[str, ProgramSpec]:
@@ -230,103 +216,26 @@ def _parse_job(name: str, data: dict) -> JobSpec:
     return JobSpec.model_validate(data_copy)
 
 
-def _parse_unit(name: str, data: dict) -> UnitSpec:
+def _load_resource_dir(directory: Path) -> dict[str, dict]:
-    """Parse a units: entry into a UnitSpec."""
+    """Load every *.yaml file in a resource directory.
-    data_copy = dict(data)
-    data_copy["id"] = name
-    return UnitSpec.model_validate(data_copy)
 
-
+    The filename stem becomes the resource id. Returns a mapping of
-# Stack convention defaults used during unit expansion.
+    id → parsed YAML dict (empty mappings normalized to {}).
-_STACK_DEFAULTS: dict[str, dict[str, str]] = {
+    """
-    "python-fastapi": {"runner": "python", "health_path": "/health"},
+    result: dict[str, dict] = {}
-    "python-cli": {"runner": "command"},
+    if not directory.is_dir():
-    "react-vite": {},
+        return result
-}
+    for path in sorted(directory.glob("*.yaml")):
-
+        with open(path) as f:
-# Kind → ProgramSpec.behavior mapping.
+            data = yaml.safe_load(f) or {}
-_KIND_BEHAVIOR: dict[str, str] = {
+        if not isinstance(data, dict):
-    "tool": "tool",
+            raise ValueError(f"{path} must contain a YAML mapping")
-    "service": "daemon",
+        result[path.stem] = data
-    "site": "frontend",
+    return result
-    "job": "tool",
-}
-
-
-def _expand_units(
-    units: dict[str, UnitSpec],
-    programs: dict[str, ProgramSpec],
-    services: dict[str, ServiceSpec],
-    jobs: dict[str, JobSpec],
-) -> None:
-    """Expand units: entries into programs/services/jobs dicts (in-place)."""
-    for name, unit in units.items():
-        if name in programs or name in services or name in jobs:
-            raise ValueError(
-                f"Unit '{name}' conflicts with existing entry in programs/services/jobs"
-            )
-
-        defaults = _STACK_DEFAULTS.get(unit.stack or "", {})
-
-        # Always create a ProgramSpec
-        programs[name] = ProgramSpec(
-            id=name,
-            description=unit.description,
-            behavior=_KIND_BEHAVIOR[unit.kind.value],
-            source=unit.source,
-            stack=unit.stack,
-            system_dependencies=unit.system_dependencies,
-            install_extras=unit.install_extras,
-            version=unit.version,
-            build=unit.build,
-            tags=unit.tags,
-        )
-
-        if unit.kind == UnitKind.SERVICE:
-            assert unit.port is not None  # guaranteed by validator
-            runner = defaults.get("runner", "python")
-            run_spec: RunCommand | RunPython
-            if runner == "python":
-                run_spec = RunPython(runner="python", program=name)
-            else:
-                run_spec = RunCommand(runner="command", argv=[name])
-
-            health = unit.health_path or defaults.get("health_path")
-
-            services[name] = ServiceSpec(
-                id=name,
-                program=name,
-                run=run_spec,
-                expose=ExposeSpec(
-                    http=HttpExposeSpec(
-                        internal=HttpInternal(port=unit.port),
-                        health_path=health,
-                    )
-                ),
-                proxy=ProxySpec(caddy=CaddySpec(path_prefix=unit.path_prefix))
-                if unit.path_prefix
-                else None,
-                manage=ManageSpec(systemd=SystemdSpec()),
-                defaults=DefaultsSpec(env=dict(unit.env)) if unit.env else None,
-            )
-
-        elif unit.kind == UnitKind.JOB:
-            assert unit.schedule is not None  # guaranteed by validator
-            assert unit.argv is not None  # guaranteed by validator
-            jobs[name] = JobSpec(
-                id=name,
-                program=name,
-                description=unit.description,
-                run=RunCommand(runner="command", argv=list(unit.argv)),
-                schedule=unit.schedule,
-                timezone=unit.timezone,
-                manage=ManageSpec(systemd=SystemdSpec()),
-                defaults=DefaultsSpec(env=dict(unit.env)) if unit.env else None,
-            )
 
 
 def load_config(root: Path | None = None) -> CastleConfig:
-    """Load castle.yaml and return parsed configuration."""
+    """Load castle config: global castle.yaml + programs/, services/, jobs/ dirs."""
     if root is None:
         root = find_castle_root()
 
@@ -335,7 +244,7 @@ def load_config(root: Path | None = None) -> CastleConfig:
         raise FileNotFoundError(f"Castle config not found: {config_path}")
 
     with open(config_path) as f:
-        data = yaml.safe_load(f)
+        data = yaml.safe_load(f) or {}
 
     gateway_data = data.get("gateway", {})
     gateway = GatewayConfig(port=gateway_data.get("port", 9000))
@@ -346,9 +255,7 @@ def load_config(root: Path | None = None) -> CastleConfig:
         repo_path = Path(data["repo"]).expanduser()
 
     programs: dict[str, ProgramSpec] = {}
-    # Support both "programs:" and legacy "components:" key
+    for name, comp_data in _load_resource_dir(root / "programs").items():
-    programs_data = data.get("programs") or data.get("components") or {}
-    for name, comp_data in programs_data.items():
         prog = _parse_program(name, comp_data)
         # Resolve source paths to absolute
         if prog.source:
@@ -360,32 +267,13 @@ def load_config(root: Path | None = None) -> CastleConfig:
         programs[name] = prog
 
     services: dict[str, ServiceSpec] = {}
-    for name, svc_data in data.get("services", {}).items():
+    for name, svc_data in _load_resource_dir(root / "services").items():
         services[name] = _parse_service(name, svc_data)
 
     jobs: dict[str, JobSpec] = {}
-    for name, job_data in data.get("jobs", {}).items():
+    for name, job_data in _load_resource_dir(root / "jobs").items():
         jobs[name] = _parse_job(name, job_data)
 
-    # Expand units: section into programs/services/jobs
-    units_data = data.get("units") or {}
-    units: dict[str, UnitSpec] = {}
-    for name, unit_data in units_data.items():
-        units[name] = _parse_unit(name, unit_data)
-
-    unit_names: set[str] = set()
-    if units:
-        _expand_units(units, programs, services, jobs)
-        unit_names = set(units.keys())
-        # Resolve source paths for unit-generated programs
-        for name in unit_names:
-            prog = programs[name]
-            if prog.source:
-                if prog.source.startswith("repo:") and repo_path:
-                    prog.source = str(repo_path / prog.source[5:])
-                elif not Path(prog.source).is_absolute():
-                    prog.source = str(root / prog.source)
-
     return CastleConfig(
         root=root,
         repo=repo_path,
@@ -393,8 +281,6 @@ def load_config(root: Path | None = None) -> CastleConfig:
         programs=programs,
         services=services,
         jobs=jobs,
-        _unit_names=unit_names,
-        _units_raw=dict(units_data),
     )
 
 
@@ -468,65 +354,61 @@ def _spec_to_yaml_dict(spec: ProgramSpec | ServiceSpec | JobSpec) -> dict:
     return _clean_for_yaml(result)
 
 
-def save_config(config: CastleConfig) -> None:
+def _program_to_yaml_dict(spec: ProgramSpec, config: CastleConfig) -> dict:
-    """Save castle configuration to castle.yaml."""
+    """Serialize a ProgramSpec, rewriting absolute source paths to relative."""
-    data: dict = {"gateway": {"port": config.gateway.port}}
+    d = _spec_to_yaml_dict(spec)
+    if d.get("source") and Path(d["source"]).is_absolute():
+        src = Path(d["source"])
+        # If source is under repo, store as repo:relative
+        if config.repo:
+            try:
+                d["source"] = "repo:" + str(src.relative_to(config.repo))
+                return d
+            except ValueError:
+                pass
+        # Otherwise store relative to config root
+        try:
+            d["source"] = str(src.relative_to(config.root))
+        except ValueError:
+            pass  # not under root — keep absolute
+    return d
 
+
+def _write_resource_dir(directory: Path, specs: dict[str, dict]) -> None:
+    """Write each spec to <directory>/<name>.yaml and prune orphaned files."""
+    directory.mkdir(parents=True, exist_ok=True)
+    for name, d in specs.items():
+        with open(directory / f"{name}.yaml", "w") as f:
+            yaml.dump(d, f, default_flow_style=False, sort_keys=False)
+    # Prune files with no corresponding in-memory entry
+    for path in directory.glob("*.yaml"):
+        if path.stem not in specs:
+            path.unlink()
+
+
+def save_config(config: CastleConfig) -> None:
+    """Save castle config: global castle.yaml + programs/, services/, jobs/ dirs."""
+    data: dict = {"gateway": {"port": config.gateway.port}}
     if config.repo:
         data["repo"] = str(config.repo)
 
-    # Write units: section (raw roundtrip preserves user's original YAML)
-    if config._units_raw:
-        data["units"] = dict(config._units_raw)
-
-    # Write programs: (excluding unit-expanded ones)
-    non_unit_programs = {
-        k: v for k, v in config.programs.items() if k not in config._unit_names
-    }
-    if non_unit_programs:
-        data["programs"] = {}
-        for name, spec in non_unit_programs.items():
-            d = _spec_to_yaml_dict(spec)
-            # Store relative source paths in YAML
-            if d.get("source") and Path(d["source"]).is_absolute():
-                src = Path(d["source"])
-                # If source is under repo, store as repo:relative
-                if config.repo:
-                    try:
-                        d["source"] = "repo:" + str(src.relative_to(config.repo))
-                        data["programs"][name] = d
-                        continue
-                    except ValueError:
-                        pass
-                # Otherwise store relative to config root
-                try:
-                    d["source"] = str(src.relative_to(config.root))
-                except ValueError:
-                    pass  # not under root — keep absolute
-            data["programs"][name] = d
-
-    # Write services: (excluding unit-expanded ones)
-    non_unit_services = {
-        k: v for k, v in config.services.items() if k not in config._unit_names
-    }
-    if non_unit_services:
-        data["services"] = {}
-        for name, spec in non_unit_services.items():
-            data["services"][name] = _spec_to_yaml_dict(spec)
-
-    # Write jobs: (excluding unit-expanded ones)
-    non_unit_jobs = {
-        k: v for k, v in config.jobs.items() if k not in config._unit_names
-    }
-    if non_unit_jobs:
-        data["jobs"] = {}
-        for name, spec in non_unit_jobs.items():
-            data["jobs"][name] = _spec_to_yaml_dict(spec)
-
     config_path = config.root / "castle.yaml"
     with open(config_path, "w") as f:
         yaml.dump(data, f, default_flow_style=False, sort_keys=False)
 
+    _write_resource_dir(
+        config.root / "programs",
+        {n: _program_to_yaml_dict(s, config) for n, s in config.programs.items()},
+    )
+    _write_resource_dir(
+        config.root / "services",
+        {n: _spec_to_yaml_dict(s) for n, s in config.services.items()},
+    )
+    _write_resource_dir(
+        config.root / "jobs",
+        {n: _spec_to_yaml_dict(s) for n, s in config.jobs.items()},
+    )
+
 
 def ensure_dirs() -> None:
     """Ensure castle directories exist."""

core/src/castle_core/manifest.py

@@ -298,58 +298,3 @@ class JobSpec(BaseModel):
 
     manage: ManageSpec | None = None
     defaults: DefaultsSpec | None = None
-
-
-# ---------------------
-# Unit spec — unified definition
-# ---------------------
-
-
-class UnitKind(str, Enum):
-    TOOL = "tool"
-    SERVICE = "service"
-    SITE = "site"
-    JOB = "job"
-
-
-class UnitSpec(BaseModel):
-    """Unified program+deployment definition.
-
-    Expands into ProgramSpec + optional ServiceSpec/JobSpec at config load time.
-    """
-
-    id: str = ""
-    kind: UnitKind
-    description: str | None = None
-
-    # Program identity
-    source: str | None = None
-    stack: str | None = None
-    system_dependencies: list[str] = Field(default_factory=list)
-    install_extras: list[str] = Field(default_factory=list)
-    version: str | None = None
-    build: BuildSpec | None = None
-    tags: list[str] = Field(default_factory=list)
-
-    # Service fields (kind=service)
-    port: int | None = Field(default=None, ge=1, le=65535)
-    path_prefix: str | None = None
-    health_path: str | None = None
-
-    # Job fields (kind=job)
-    schedule: str | None = None
-    timezone: str = "America/Los_Angeles"
-    argv: list[str] | None = None
-
-    # Simplified env (flat dict, replaces defaults.env nesting)
-    env: EnvMap = Field(default_factory=dict)
-
-    @model_validator(mode="after")
-    def _validate_kind_fields(self) -> UnitSpec:
-        if self.kind == UnitKind.SERVICE and self.port is None:
-            raise ValueError("kind=service requires 'port'")
-        if self.kind == UnitKind.JOB and self.schedule is None:
-            raise ValueError("kind=job requires 'schedule'")
-        if self.kind == UnitKind.JOB and self.argv is None:
-            raise ValueError("kind=job requires 'argv'")
-        return self

core/tests/conftest.py

@@ -9,10 +9,30 @@ import pytest
 import yaml
 
 
+def write_castle_config(root: Path, config: dict) -> None:
+    """Scatter a nested castle config dict into the directory-per-resource layout.
+
+    `config` uses the legacy nested shape (gateway/repo at top level, plus
+    programs/services/jobs mappings); this writes castle.yaml with globals and
+    one file per resource under programs/, services/, jobs/.
+    """
+    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 {}
+        if not entries:
+            continue
+        section_dir = root / section
+        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)
+            )
+
+
 @pytest.fixture
 def castle_root(tmp_path: Path) -> Generator[Path, None, None]:
-    """Create a temporary castle root with castle.yaml."""
+    """Create a temporary castle root with directory-per-resource config."""
-    castle_yaml = tmp_path / "castle.yaml"
     config = {
         "gateway": {"port": 18000},
         "programs": {
@@ -57,7 +77,7 @@ def castle_root(tmp_path: Path) -> Generator[Path, None, None]:
             },
         },
     }
-    castle_yaml.write_text(yaml.dump(config, default_flow_style=False))
+    write_castle_config(tmp_path, config)
 
     # Create project directories
     svc_dir = tmp_path / "test-svc"

core/tests/test_config.py

@@ -5,6 +5,7 @@ from __future__ import annotations
 from pathlib import Path
 
 import pytest
+import yaml
 from castle_core.config import (
     CastleConfig,
     load_config,
@@ -112,6 +113,30 @@ class TestSaveConfig:
         assert svc.manage is not None
         assert svc.manage.systemd is not None
 
+    def test_writes_directory_layout(self, castle_root: Path) -> None:
+        """Save writes one file per resource under programs/services/jobs dirs."""
+        config = load_config(castle_root)
+        save_config(config)
+        assert (castle_root / "programs" / "test-tool.yaml").exists()
+        assert (castle_root / "services" / "test-svc.yaml").exists()
+        assert (castle_root / "jobs" / "test-job.yaml").exists()
+        # Global file holds gateway only, no resource sections
+        global_data = yaml.safe_load((castle_root / "castle.yaml").read_text())
+        assert global_data["gateway"]["port"] == 18000
+        assert "programs" not in global_data
+        assert "services" not in global_data
+        assert "jobs" not in global_data
+
+    def test_delete_prunes_file(self, castle_root: Path) -> None:
+        """Removing a resource and saving deletes its on-disk file."""
+        config = load_config(castle_root)
+        del config.services["test-svc"]
+        save_config(config)
+        assert not (castle_root / "services" / "test-svc.yaml").exists()
+        config2 = load_config(castle_root)
+        assert "test-svc" not in config2.services
+        assert "test-tool" in config2.programs
+
 
 class TestResolveEnvVars:
     """Tests for environment variable resolution."""

core/tests/test_units.py

@@ -1,374 +0,0 @@
-"""Tests for units: expansion."""
-
-from __future__ import annotations
-
-from pathlib import Path
-
-import pytest
-import yaml
-
-from castle_core.config import load_config, save_config
-from castle_core.manifest import UnitKind, UnitSpec
-
-
-@pytest.fixture
-def units_root(tmp_path: Path) -> Path:
-    """Castle root with a units: section."""
-    config = {
-        "gateway": {"port": 18000},
-        "units": {
-            "my-svc": {
-                "kind": "service",
-                "stack": "python-fastapi",
-                "description": "Test service",
-                "source": "code/my-svc",
-                "port": 9050,
-                "path_prefix": "/my-svc",
-            },
-            "my-tool": {
-                "kind": "tool",
-                "stack": "python-cli",
-                "description": "Test tool",
-                "source": "code/my-tool",
-                "system_dependencies": ["pandoc"],
-            },
-            "my-site": {
-                "kind": "site",
-                "stack": "react-vite",
-                "description": "Test frontend",
-                "source": "code/my-site",
-                "build": {
-                    "commands": [["pnpm", "build"]],
-                    "outputs": ["dist/"],
-                },
-            },
-            "my-job": {
-                "kind": "job",
-                "stack": "python-cli",
-                "description": "Test job",
-                "source": "code/my-job",
-                "schedule": "0 2 * * *",
-                "argv": ["my-job", "run"],
-                "env": {"DATA_DIR": "/tmp/data"},
-            },
-        },
-    }
-    (tmp_path / "castle.yaml").write_text(
-        yaml.dump(config, default_flow_style=False)
-    )
-    return tmp_path
-
-
-class TestUnitExpansion:
-
-    def test_service_creates_program_and_service(self, units_root: Path) -> None:
-        config = load_config(units_root)
-        assert "my-svc" in config.programs
-        assert "my-svc" in config.services
-        prog = config.programs["my-svc"]
-        assert prog.behavior == "daemon"
-        assert prog.stack == "python-fastapi"
-        assert prog.description == "Test service"
-        svc = config.services["my-svc"]
-        assert svc.run.runner == "python"
-        assert svc.run.program == "my-svc"
-        assert svc.expose.http.internal.port == 9050
-        assert svc.expose.http.health_path == "/health"
-        assert svc.proxy.caddy.path_prefix == "/my-svc"
-        assert svc.manage.systemd is not None
-        assert svc.program == "my-svc"
-
-    def test_tool_creates_program_only(self, units_root: Path) -> None:
-        config = load_config(units_root)
-        assert "my-tool" in config.programs
-        assert "my-tool" not in config.services
-        assert "my-tool" not in config.jobs
-        prog = config.programs["my-tool"]
-        assert prog.behavior == "tool"
-        assert prog.system_dependencies == ["pandoc"]
-
-    def test_site_creates_program_with_build(self, units_root: Path) -> None:
-        config = load_config(units_root)
-        assert "my-site" in config.programs
-        assert "my-site" not in config.services
-        prog = config.programs["my-site"]
-        assert prog.behavior == "frontend"
-        assert prog.build is not None
-        assert prog.build.outputs == ["dist/"]
-
-    def test_job_creates_program_and_job(self, units_root: Path) -> None:
-        config = load_config(units_root)
-        assert "my-job" in config.programs
-        assert "my-job" in config.jobs
-        assert "my-job" not in config.services
-        prog = config.programs["my-job"]
-        assert prog.behavior == "tool"
-        job = config.jobs["my-job"]
-        assert job.schedule == "0 2 * * *"
-        assert job.run.runner == "command"
-        assert job.run.argv == ["my-job", "run"]
-        assert job.defaults.env["DATA_DIR"] == "/tmp/data"
-        assert job.program == "my-job"
-
-    def test_service_without_path_prefix(self, tmp_path: Path) -> None:
-        config_data = {
-            "gateway": {"port": 18000},
-            "units": {
-                "internal-svc": {
-                    "kind": "service",
-                    "stack": "python-fastapi",
-                    "port": 9060,
-                },
-            },
-        }
-        (tmp_path / "castle.yaml").write_text(
-            yaml.dump(config_data, default_flow_style=False)
-        )
-        config = load_config(tmp_path)
-        svc = config.services["internal-svc"]
-        assert svc.proxy is None
-
-    def test_service_with_env(self, tmp_path: Path) -> None:
-        config_data = {
-            "gateway": {"port": 18000},
-            "units": {
-                "svc-env": {
-                    "kind": "service",
-                    "stack": "python-fastapi",
-                    "port": 9060,
-                    "env": {"API_KEY": "test123"},
-                },
-            },
-        }
-        (tmp_path / "castle.yaml").write_text(
-            yaml.dump(config_data, default_flow_style=False)
-        )
-        config = load_config(tmp_path)
-        svc = config.services["svc-env"]
-        assert svc.defaults.env["API_KEY"] == "test123"
-
-    def test_service_custom_health_path(self, tmp_path: Path) -> None:
-        config_data = {
-            "gateway": {"port": 18000},
-            "units": {
-                "custom-health": {
-                    "kind": "service",
-                    "stack": "python-fastapi",
-                    "port": 9060,
-                    "health_path": "/status",
-                },
-            },
-        }
-        (tmp_path / "castle.yaml").write_text(
-            yaml.dump(config_data, default_flow_style=False)
-        )
-        config = load_config(tmp_path)
-        svc = config.services["custom-health"]
-        assert svc.expose.http.health_path == "/status"
-
-    def test_source_paths_resolved(self, units_root: Path) -> None:
-        config = load_config(units_root)
-        prog = config.programs["my-tool"]
-        assert prog.source == str(units_root / "code" / "my-tool")
-
-    def test_repo_source_paths_resolved(self, tmp_path: Path) -> None:
-        repo_dir = tmp_path / "repo"
-        repo_dir.mkdir()
-        config_data = {
-            "gateway": {"port": 18000},
-            "repo": str(repo_dir),
-            "units": {
-                "repo-svc": {
-                    "kind": "service",
-                    "stack": "python-fastapi",
-                    "port": 9060,
-                    "source": "repo:my-api",
-                },
-            },
-        }
-        (tmp_path / "castle.yaml").write_text(
-            yaml.dump(config_data, default_flow_style=False)
-        )
-        config = load_config(tmp_path)
-        assert config.programs["repo-svc"].source == str(repo_dir / "my-api")
-
-
-class TestUnitValidation:
-
-    def test_service_requires_port(self) -> None:
-        with pytest.raises(ValueError, match="requires 'port'"):
-            UnitSpec(id="bad", kind=UnitKind.SERVICE, stack="python-fastapi")
-
-    def test_job_requires_schedule(self) -> None:
-        with pytest.raises(ValueError, match="requires 'schedule'"):
-            UnitSpec(id="bad", kind=UnitKind.JOB, argv=["x"])
-
-    def test_job_requires_argv(self) -> None:
-        with pytest.raises(ValueError, match="requires 'argv'"):
-            UnitSpec(id="bad", kind=UnitKind.JOB, schedule="0 * * * *")
-
-    def test_tool_needs_no_port(self) -> None:
-        unit = UnitSpec(id="ok", kind=UnitKind.TOOL, stack="python-cli")
-        assert unit.port is None
-
-    def test_site_needs_no_port(self) -> None:
-        unit = UnitSpec(id="ok", kind=UnitKind.SITE, stack="react-vite")
-        assert unit.port is None
-
-
-class TestUnitConflicts:
-
-    def test_conflict_with_existing_program(self, tmp_path: Path) -> None:
-        config = {
-            "gateway": {"port": 18000},
-            "programs": {
-                "conflict": {
-                    "description": "Existing",
-                    "behavior": "tool",
-                },
-            },
-            "units": {
-                "conflict": {
-                    "kind": "tool",
-                    "stack": "python-cli",
-                },
-            },
-        }
-        (tmp_path / "castle.yaml").write_text(
-            yaml.dump(config, default_flow_style=False)
-        )
-        with pytest.raises(ValueError, match="conflicts"):
-            load_config(tmp_path)
-
-    def test_conflict_with_existing_service(self, tmp_path: Path) -> None:
-        config = {
-            "gateway": {"port": 18000},
-            "services": {
-                "conflict": {
-                    "run": {"runner": "command", "argv": ["svc"]},
-                },
-            },
-            "units": {
-                "conflict": {
-                    "kind": "service",
-                    "stack": "python-fastapi",
-                    "port": 9999,
-                },
-            },
-        }
-        (tmp_path / "castle.yaml").write_text(
-            yaml.dump(config, default_flow_style=False)
-        )
-        with pytest.raises(ValueError, match="conflicts"):
-            load_config(tmp_path)
-
-
-class TestUnitCoexistence:
-
-    def test_units_and_explicit_coexist(self, tmp_path: Path) -> None:
-        config = {
-            "gateway": {"port": 18000},
-            "programs": {
-                "manual-tool": {
-                    "description": "Manual",
-                    "behavior": "tool",
-                },
-            },
-            "services": {
-                "manual-svc": {
-                    "run": {"runner": "command", "argv": ["svc"]},
-                    "expose": {"http": {"internal": {"port": 19000}}},
-                },
-            },
-            "units": {
-                "unit-svc": {
-                    "kind": "service",
-                    "stack": "python-fastapi",
-                    "port": 9070,
-                },
-            },
-        }
-        (tmp_path / "castle.yaml").write_text(
-            yaml.dump(config, default_flow_style=False)
-        )
-        cfg = load_config(tmp_path)
-        assert "manual-tool" in cfg.programs
-        assert "manual-svc" in cfg.services
-        assert "unit-svc" in cfg.programs
-        assert "unit-svc" in cfg.services
-
-
-class TestUnitRoundTrip:
-
-    def test_save_preserves_units_section(self, units_root: Path) -> None:
-        config = load_config(units_root)
-        save_config(config)
-        config2 = load_config(units_root)
-        assert "my-svc" in config2.programs
-        assert "my-svc" in config2.services
-        assert "my-tool" in config2.programs
-        assert "my-job" in config2.jobs
-
-    def test_save_does_not_duplicate_into_explicit_sections(
-        self, units_root: Path
-    ) -> None:
-        config = load_config(units_root)
-        save_config(config)
-        raw = yaml.safe_load((units_root / "castle.yaml").read_text())
-        # Unit entries should NOT appear in programs/services/jobs sections
-        assert "my-svc" not in raw.get("programs", {})
-        assert "my-svc" not in raw.get("services", {})
-        assert "my-job" not in raw.get("jobs", {})
-        # They should still be in units:
-        assert "my-svc" in raw.get("units", {})
-        assert "my-tool" in raw.get("units", {})
-        assert "my-job" in raw.get("units", {})
-
-    def test_mixed_save_separates_correctly(self, tmp_path: Path) -> None:
-        config_data = {
-            "gateway": {"port": 18000},
-            "programs": {
-                "manual-tool": {
-                    "description": "Manual",
-                    "behavior": "tool",
-                    "source": "code/manual",
-                },
-            },
-            "units": {
-                "unit-tool": {
-                    "kind": "tool",
-                    "stack": "python-cli",
-                    "source": "code/unit-tool",
-                },
-            },
-        }
-        (tmp_path / "castle.yaml").write_text(
-            yaml.dump(config_data, default_flow_style=False)
-        )
-        config = load_config(tmp_path)
-        save_config(config)
-        raw = yaml.safe_load((tmp_path / "castle.yaml").read_text())
-        # manual-tool stays in programs, unit-tool stays in units
-        assert "manual-tool" in raw.get("programs", {})
-        assert "unit-tool" not in raw.get("programs", {})
-        assert "unit-tool" in raw.get("units", {})
-
-
-class TestNoUnitsSection:
-
-    def test_config_without_units_works(self, tmp_path: Path) -> None:
-        config_data = {
-            "gateway": {"port": 18000},
-            "programs": {
-                "some-tool": {
-                    "description": "A tool",
-                    "behavior": "tool",
-                },
-            },
-        }
-        (tmp_path / "castle.yaml").write_text(
-            yaml.dump(config_data, default_flow_style=False)
-        )
-        config = load_config(tmp_path)
-        assert "some-tool" in config.programs
-        assert not config._unit_names

docs/registry.md

@@ -31,56 +31,78 @@ may have neither (a tool you just install), a **service** (always-on), or a
 service; a `tool`-behavior program may back a job or just be installed for
 manual use.
 
-## castle.yaml
+## Configuration Directory Layout
 
-The single source of truth. Lives at `~/.castle/castle.yaml`. Three top-level
+Castle splits its configuration across a root directory (`~/.castle/` or your config root) instead of a single file:
-sections:
+
+```
+~/.castle/
+├── castle.yaml        # Global settings (gateway, repo, etc.)
+├── programs/          # Program configuration files (one file per program)
+│   └── my-tool.yaml
+├── services/          # Service configuration files (one file per service)
+│   └── my-service.yaml
+└── jobs/              # Job configuration files (one file per job)
+    └── my-job.yaml
+```
+
+### castle.yaml (Globals)
+
+The core `castle.yaml` contains configuration settings that apply globally to your Castle platform instance:
 
 ```yaml
 gateway:
   port: 9000
-
+repo: /data/repos/castle
-programs:
-  my-tool:
-    description: Does something useful
-    source: /data/repos/my-tool
-    stack: python-cli
-    behavior: tool
-    system_dependencies: [pandoc]
-
-services:
-  my-service:
-    program: my-service
-    run:
-      runner: python
-      program: my-service
-    expose:
-      http:
-        internal: { port: 9001 }
-        health_path: /health
-    proxy:
-      caddy: { path_prefix: /my-service }
-    manage:
-      systemd: {}
-
-jobs:
-  my-job:
-    program: my-tool
-    run:
-      runner: command
-      argv: [my-tool, sync]
-    schedule: "0 2 * * *"
-    manage:
-      systemd: {}
 ```
 
-### Section semantics
+### Resource Configuration Files (`programs/`, `services/`, `jobs/`)
 
-| Section | Purpose | Category |
+Each resource (program, service, or job) is configured in its own YAML file named after the resource's unique ID (e.g., `services/my-service.yaml` defines the service `my-service`).
-|---------|---------|----------|
+
-| `programs:` | Software catalog — what exists | tool, frontend, daemon |
+**programs/my-tool.yaml:**
-| `services:` | Long-running daemons — how they run | service |
+```yaml
-| `jobs:` | Scheduled tasks — when they run | job |
+description: Does something useful
+source: /data/repos/my-tool
+stack: python-cli
+behavior: tool
+system_dependencies: [pandoc]
+```
+
+**services/my-service.yaml:**
+```yaml
+program: my-service
+run:
+  runner: python
+  program: my-service
+expose:
+  http:
+    internal: { port: 9001 }
+    health_path: /health
+proxy:
+  caddy: { path_prefix: /my-service }
+manage:
+  systemd: {}
+```
+
+**jobs/my-job.yaml:**
+```yaml
+program: my-tool
+run:
+  runner: command
+  argv: [my-tool, sync]
+schedule: "0 2 * * *"
+manage:
+  systemd: {}
+```
+
+### Resource Categories
+
+| Category | Location | Purpose | Role / Types |
+|----------|----------|---------|--------------|
+| **programs** | `programs/*.yaml` | Software catalog — what software exists | tool, frontend, daemon |
+| **services** | `services/*.yaml` | Long-running daemons — how they run | service |
+| **jobs** | `jobs/*.yaml` | Scheduled tasks — when they run | job |
 
 Services and jobs can reference a program via `program:` for description
 fallthrough and source code linking. They can also exist independently