refactor: Separate runtime from build. Enhance configuration and registry management, migrate to registry-based component handling

castle-api/src/castle_api/config.py

@@ -4,11 +4,13 @@ from pathlib import Path
 
 from pydantic_settings import BaseSettings
 
+from castle_core.config import CastleConfig, load_config
+from castle_core.registry import NodeRegistry, load_registry
+
 
 class Settings(BaseSettings):
     """Service settings loaded from environment variables."""
 
-    castle_root: Path = Path("/data/repos/castle")
     host: str = "0.0.0.0"
     port: int = 9020
 
@@ -19,3 +21,32 @@ class Settings(BaseSettings):
 
 
 settings = Settings()
+
+
+def get_registry() -> NodeRegistry:
+    """Load the node registry. Raises if not found."""
+    return load_registry()
+
+
+def get_castle_root() -> Path | None:
+    """Get the castle repo root from the registry, if available."""
+    try:
+        registry = load_registry()
+        if registry.node.castle_root:
+            return Path(registry.node.castle_root)
+    except (FileNotFoundError, ValueError):
+        pass
+    return None
+
+
+def get_config() -> CastleConfig:
+    """Load castle.yaml via the registry's castle_root.
+
+    Raises FileNotFoundError if repo not available.
+    """
+    root = get_castle_root()
+    if root is None:
+        raise FileNotFoundError(
+            "Castle repo not available. Set castle_root in registry."
+        )
+    return load_config(root)

castle-api/src/castle_api/config_editor.py

@@ -9,10 +9,10 @@ import yaml
 from fastapi import APIRouter, HTTPException, status
 from pydantic import BaseModel
 
-from castle_core.config import load_config, save_config
+from castle_core.config import save_config
 from castle_core.manifest import ComponentManifest
 
-from castle_api.config import settings
+from castle_api.config import get_castle_root, get_config, get_registry
 from castle_api.stream import broadcast
 
 router = APIRouter(prefix="/config", tags=["config"])
@@ -42,16 +42,29 @@ class ComponentConfigRequest(BaseModel):
     config: dict
 
 
+def _require_repo() -> None:
+    """Raise 503 if repo is not available."""
+    if get_castle_root() is None:
+        raise HTTPException(
+            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
+            detail="Castle repo not available on this node.",
+        )
+
+
 @router.get("", response_model=ConfigResponse)
-def get_config() -> ConfigResponse:
+def get_config_yaml() -> ConfigResponse:
     """Get the raw castle.yaml content."""
-    config_path = settings.castle_root / "castle.yaml"
+    _require_repo()
+    root = get_castle_root()
+    config_path = root / "castle.yaml"
     return ConfigResponse(yaml_content=config_path.read_text())
 
 
 @router.put("", response_model=ConfigSaveResponse)
 def save_yaml(request: ConfigSaveRequest) -> ConfigSaveResponse:
     """Validate and save castle.yaml. Does NOT apply changes."""
+    _require_repo()
+    root = get_castle_root()
     errors: list[str] = []
 
     # Parse YAML
@@ -87,7 +100,7 @@ def save_yaml(request: ConfigSaveRequest) -> ConfigSaveResponse:
         )
 
     # Backup and save
-    config_path = settings.castle_root / "castle.yaml"
+    config_path = root / "castle.yaml"
     backup_path = config_path.with_suffix(".yaml.bak")
     shutil.copy2(config_path, backup_path)
     config_path.write_text(request.yaml_content)
@@ -98,6 +111,8 @@ def save_yaml(request: ConfigSaveRequest) -> ConfigSaveResponse:
 @router.put("/components/{name}")
 def save_component(name: str, request: ComponentConfigRequest) -> dict:
     """Update a single component's config in castle.yaml."""
+    _require_repo()
+
     # Validate
     try:
         comp_data = dict(request.config)
@@ -109,7 +124,7 @@ def save_component(name: str, request: ComponentConfigRequest) -> dict:
             detail=f"Invalid component config: {e}",
         )
 
-    config = load_config(settings.castle_root)
+    config = get_config()
     config.components[name] = ComponentManifest.model_validate(
         {**request.config, "id": name}
     )
@@ -120,7 +135,7 @@ def save_component(name: str, request: ComponentConfigRequest) -> dict:
 @router.delete("/components/{name}")
 def delete_component(name: str) -> dict:
     """Remove a component from castle.yaml."""
-    config = load_config(settings.castle_root)
+    config = get_config()
     if name not in config.components:
         raise HTTPException(
             status_code=status.HTTP_404_NOT_FOUND,
@@ -133,13 +148,15 @@ def delete_component(name: str) -> dict:
 
 @router.post("/apply", response_model=ApplyResponse)
 async def apply_config() -> ApplyResponse:
-    """Apply config: regenerate systemd units for managed services + reload gateway."""
-    config = load_config(settings.castle_root)
+    """Apply config: restart managed services + regenerate and reload gateway."""
+    registry = get_registry()
     actions: list[str] = []
     errors: list[str] = []
 
-    # Regenerate and restart managed services
-    for name in config.managed:
+    # Restart managed services
+    for name, deployed in registry.deployed.items():
+        if not deployed.managed:
+            continue
         unit = f"castle-{name}.service"
         ok, output = await _systemctl("restart", unit)
         if ok:
@@ -149,17 +166,17 @@ async def apply_config() -> ApplyResponse:
 
     # Reload gateway
     from castle_core.config import GENERATED_DIR, ensure_dirs
-    from castle_core.generators import generate_caddyfile
+    from castle_core.generators.caddyfile import generate_caddyfile_from_registry
 
     ensure_dirs()
     caddyfile_path = GENERATED_DIR / "Caddyfile"
-    caddyfile_path.write_text(generate_caddyfile(config))
+    caddyfile_path.write_text(generate_caddyfile_from_registry(registry))
     actions.append("Generated Caddyfile")
 
     if shutil.which("caddy"):
-        ok, output = await _run("caddy", "reload",
-                                "--config", str(caddyfile_path),
-                                "--adapter", "caddyfile")
+        ok, output = await _run(
+            "caddy", "reload", "--config", str(caddyfile_path), "--adapter", "caddyfile"
+        )
         if ok:
             actions.append("Reloaded gateway")
         else:
@@ -171,7 +188,10 @@ async def apply_config() -> ApplyResponse:
 
 async def _systemctl(action: str, unit: str) -> tuple[bool, str]:
     proc = await asyncio.create_subprocess_exec(
-        "systemctl", "--user", action, unit,
+        "systemctl",
+        "--user",
+        action,
+        unit,
         stdout=asyncio.subprocess.PIPE,
         stderr=asyncio.subprocess.PIPE,
     )

castle-api/src/castle_api/health.py

@@ -7,23 +7,18 @@ import time
 
 import httpx
 
-from castle_core.config import CastleConfig
+from castle_core.registry import NodeRegistry
 
 from castle_api.models import HealthStatus
 
 
-async def check_all_health(config: CastleConfig) -> list[HealthStatus]:
-    """Check health of all components with expose.http and a health_path."""
+async def check_all_health(registry: NodeRegistry) -> list[HealthStatus]:
+    """Check health of all deployed components with a port and health_path."""
     targets: list[tuple[str, str]] = []
-    for name, manifest in config.components.items():
-        if not (manifest.expose and manifest.expose.http):
+    for name, deployed in registry.deployed.items():
+        if not deployed.port or not deployed.health_path:
             continue
-        http = manifest.expose.http
-        if not http.health_path:
-            continue
-        host = http.internal.host or "127.0.0.1"
-        port = http.internal.port
-        url = f"http://{host}:{port}{http.health_path}"
+        url = f"http://127.0.0.1:{deployed.port}{deployed.health_path}"
         targets.append((name, url))
 
     if not targets:

castle-api/src/castle_api/logs.py

@@ -42,7 +42,13 @@ async def get_logs(
 
     # Static tail
     proc = await asyncio.create_subprocess_exec(
-        "journalctl", "--user", "-u", unit, "-n", str(n), "--no-pager",
+        "journalctl",
+        "--user",
+        "-u",
+        unit,
+        "-n",
+        str(n),
+        "--no-pager",
         stdout=asyncio.subprocess.PIPE,
         stderr=asyncio.subprocess.PIPE,
     )
@@ -54,7 +60,14 @@ async def get_logs(
 async def _follow_logs(unit: str, n: int) -> AsyncGenerator[str, None]:
     """Stream journalctl -f output as SSE events."""
     proc = await asyncio.create_subprocess_exec(
-        "journalctl", "--user", "-u", unit, "-n", str(n), "-f", "--no-pager",
+        "journalctl",
+        "--user",
+        "-u",
+        unit,
+        "-n",
+        str(n),
+        "-f",
+        "--no-pager",
         stdout=asyncio.subprocess.PIPE,
         stderr=asyncio.subprocess.PIPE,
     )

castle-api/src/castle_api/main.py

@@ -17,7 +17,12 @@ from castle_api.logs import router as logs_router
 from castle_api.routes import router as dashboard_router
 from castle_api.secrets import router as secrets_router
 from castle_api.services import router as services_router
-from castle_api.stream import close_all_subscribers, health_poll_loop, subscribe, unsubscribe
+from castle_api.stream import (
+    close_all_subscribers,
+    health_poll_loop,
+    subscribe,
+    unsubscribe,
+)
 from castle_api.tools import router as tools_router
 
 # Set by _watch_shutdown when uvicorn begins its shutdown sequence.

castle-api/src/castle_api/routes.py

@@ -7,9 +7,9 @@ from pathlib import Path
 
 from fastapi import APIRouter, HTTPException, status
 
-from castle_core.config import load_config
+from castle_core.generators.caddyfile import generate_caddyfile_from_registry
 
-from castle_api.config import settings
+from castle_api.config import get_castle_root, get_registry
 from castle_api.health import check_all_health
 from castle_api.models import (
     ComponentDetail,
@@ -22,8 +22,43 @@ from castle_api.models import (
 router = APIRouter(tags=["dashboard"])
 
 
+def _summary_from_deployed(name: str, deployed: object) -> ComponentSummary:
+    """Build a ComponentSummary from a DeployedComponent."""
+    managed = deployed.managed
+
+    systemd_info: SystemdInfo | None = None
+    if managed:
+        unit_name = f"castle-{name}.service"
+        unit_path = str(Path("~/.config/systemd/user") / unit_name)
+        has_timer = deployed.schedule is not None
+        systemd_info = SystemdInfo(
+            unit_name=unit_name,
+            unit_path=unit_path,
+            timer=has_timer,
+        )
+
+    # Check if tool is installed on PATH
+    installed: bool | None = None
+    if "tool" in deployed.roles:
+        installed = shutil.which(name) is not None
+
+    return ComponentSummary(
+        id=name,
+        description=deployed.description,
+        roles=deployed.roles,
+        runner=deployed.runner,
+        port=deployed.port,
+        health_path=deployed.health_path,
+        proxy_path=deployed.proxy_path,
+        managed=managed,
+        systemd=systemd_info,
+        schedule=deployed.schedule,
+        installed=installed,
+    )
+
+
 def _summary_from_manifest(name: str, manifest: object, root: Path) -> ComponentSummary:
-    """Build a ComponentSummary from a manifest."""
+    """Build a ComponentSummary from a manifest (for non-deployed components)."""
     port = None
     health_path = None
     proxy_path = None
@@ -37,26 +72,25 @@ def _summary_from_manifest(name: str, manifest: object, root: Path) -> Component
         manifest.manage and manifest.manage.systemd and manifest.manage.systemd.enable
     )
 
-    # Systemd info for managed components
     systemd_info: SystemdInfo | None = None
     if managed:
         unit_name = f"castle-{name}.service"
         unit_path = str(Path("~/.config/systemd/user") / unit_name)
-        has_timer = any(getattr(t, "type", None) == "schedule" for t in manifest.triggers)
+        has_timer = any(
+            getattr(t, "type", None) == "schedule" for t in manifest.triggers
+        )
         systemd_info = SystemdInfo(
             unit_name=unit_name,
             unit_path=unit_path,
             timer=has_timer,
         )
 
-    # Extract cron schedule from first schedule trigger, if any
     schedule = None
     for t in manifest.triggers:
         if t.type == "schedule":
             schedule = t.cron
             break
 
-    # Infer runner — from run block or from tool source
     runner = manifest.run.runner if manifest.run else None
     if runner is None and manifest.tool and manifest.tool.source:
         source_dir = root / manifest.tool.source
@@ -65,7 +99,6 @@ def _summary_from_manifest(name: str, manifest: object, root: Path) -> Component
         elif source_dir.is_file():
             runner = "command"
 
-    # Check if tool is actually installed on PATH
     installed: bool | None = None
     if manifest.install and manifest.install.path:
         alias = manifest.install.path.alias or name
@@ -91,53 +124,95 @@ def _summary_from_manifest(name: str, manifest: object, root: Path) -> Component
 
 @router.get("/components", response_model=list[ComponentSummary])
 def list_components() -> list[ComponentSummary]:
-    """List all registered components."""
-    config = load_config(settings.castle_root)
-    return [
-        _summary_from_manifest(name, m, config.root)
-        for name, m in config.components.items()
-    ]
+    """List all components — deployed from registry, non-deployed from castle.yaml."""
+    registry = get_registry()
+    summaries: list[ComponentSummary] = []
+    seen: set[str] = set()
+
+    # Deployed components from registry
+    for name, deployed in registry.deployed.items():
+        summaries.append(_summary_from_deployed(name, deployed))
+        seen.add(name)
+
+    # Non-deployed components from castle.yaml (if repo available)
+    root = get_castle_root()
+    if root:
+        try:
+            from castle_core.config import load_config
+
+            config = load_config(root)
+            for name, manifest in config.components.items():
+                if name not in seen:
+                    summaries.append(_summary_from_manifest(name, manifest, root))
+        except FileNotFoundError:
+            pass
+
+    return summaries
 
 
 @router.get("/components/{name}", response_model=ComponentDetail)
 def get_component(name: str) -> ComponentDetail:
     """Get detailed info for a single component."""
-    config = load_config(settings.castle_root)
-    if name not in config.components:
+    registry = get_registry()
+
+    if name in registry.deployed:
+        deployed = registry.deployed[name]
+        summary = _summary_from_deployed(name, deployed)
+        raw = {
+            "runner": deployed.runner,
+            "run_cmd": deployed.run_cmd,
+            "env": deployed.env,
+            "port": deployed.port,
+            "health_path": deployed.health_path,
+            "proxy_path": deployed.proxy_path,
+            "managed": deployed.managed,
+            "roles": deployed.roles,
+        }
+        return ComponentDetail(**summary.model_dump(), manifest=raw)
+
+    # Fall back to castle.yaml
+    root = get_castle_root()
+    if root:
+        from castle_core.config import load_config
+
+        config = load_config(root)
+        if name in config.components:
+            manifest = config.components[name]
+            summary = _summary_from_manifest(name, manifest, root)
+            raw = manifest.model_dump(mode="json", exclude_none=True)
+            return ComponentDetail(**summary.model_dump(), manifest=raw)
+
     raise HTTPException(
         status_code=status.HTTP_404_NOT_FOUND,
         detail=f"Component '{name}' not found",
     )
-    manifest = config.components[name]
-    summary = _summary_from_manifest(name, manifest, config.root)
-    raw = manifest.model_dump(mode="json", exclude_none=True)
-    return ComponentDetail(**summary.model_dump(), manifest=raw)
 
 
 @router.get("/status", response_model=StatusResponse)
 async def get_status() -> StatusResponse:
-    """Get live health status for all exposed services."""
-    config = load_config(settings.castle_root)
-    statuses = await check_all_health(config)
+    """Get live health status for all deployed services."""
+    registry = get_registry()
+    statuses = await check_all_health(registry)
     return StatusResponse(statuses=statuses)
 
 
 @router.get("/gateway", response_model=GatewayInfo)
 def get_gateway() -> GatewayInfo:
     """Get gateway configuration summary."""
-    config = load_config(settings.castle_root)
+    registry = get_registry()
+    deployed_count = len(registry.deployed)
+    service_count = sum(1 for d in registry.deployed.values() if d.port is not None)
+    managed_count = sum(1 for d in registry.deployed.values() if d.managed)
     return GatewayInfo(
-        port=config.gateway.port,
-        component_count=len(config.components),
-        service_count=len(config.services),
-        managed_count=len(config.managed),
+        port=registry.node.gateway_port,
+        component_count=deployed_count,
+        service_count=service_count,
+        managed_count=managed_count,
     )
 
 
 @router.get("/gateway/caddyfile")
 def get_caddyfile() -> dict[str, str]:
     """Return the generated Caddyfile content."""
-    from castle_core.generators import generate_caddyfile
-
-    config = load_config(settings.castle_root)
-    return {"content": generate_caddyfile(config)}
+    registry = get_registry()
+    return {"content": generate_caddyfile_from_registry(registry)}

castle-api/src/castle_api/services.py

@@ -8,9 +8,12 @@ import time
 from fastapi import APIRouter, HTTPException, status
 from starlette.responses import JSONResponse
 
-from castle_core.config import load_config
+from castle_core.generators.systemd import (
+    generate_timer,
+    generate_unit_from_deployed,
+)
 
-from castle_api.config import settings
+from castle_api.config import get_castle_root, get_registry
 from castle_api.health import check_all_health
 from castle_api.models import HealthStatus
 from castle_api.stream import broadcast
@@ -24,7 +27,10 @@ SELF_NAME = "castle-api"
 async def _systemctl(action: str, unit: str) -> tuple[bool, str]:
     """Run a systemctl --user command. Returns (success, output)."""
     proc = await asyncio.create_subprocess_exec(
-        "systemctl", "--user", action, unit,
+        "systemctl",
+        "--user",
+        action,
+        unit,
         stdout=asyncio.subprocess.PIPE,
         stderr=asyncio.subprocess.PIPE,
     )
@@ -36,7 +42,10 @@ async def _systemctl(action: str, unit: str) -> tuple[bool, str]:
 async def _get_unit_status(unit: str) -> str:
     """Get the active status of a systemd unit."""
     proc = await asyncio.create_subprocess_exec(
-        "systemctl", "--user", "is-active", unit,
+        "systemctl",
+        "--user",
+        "is-active",
+        unit,
         stdout=asyncio.subprocess.PIPE,
         stderr=asyncio.subprocess.PIPE,
     )
@@ -45,9 +54,9 @@ async def _get_unit_status(unit: str) -> str:
 
 
 def _validate_managed(name: str) -> None:
-    """Raise 404 if the component isn't systemd-managed."""
-    config = load_config(settings.castle_root)
-    if name not in config.managed:
+    """Raise 404 if the component isn't managed in the registry."""
+    registry = get_registry()
+    if name not in registry.deployed or not registry.deployed[name].managed:
         raise HTTPException(
             status_code=status.HTTP_404_NOT_FOUND,
             detail=f"'{name}' is not a managed service",
@@ -58,25 +67,29 @@ async def _broadcast_health_with_override(
     override_name: str, override_status: str
 ) -> None:
     """Run health checks but override one component's status from systemd."""
-    config = load_config(settings.castle_root)
-    statuses = await check_all_health(config)
+    registry = get_registry()
+    statuses = await check_all_health(registry)
 
-    # Replace the overridden component's status with the systemd truth
     result = []
     for s in statuses:
         if s.id == override_name:
-            result.append(HealthStatus(
+            result.append(
+                HealthStatus(
                     id=override_name,
                     status="down" if override_status != "active" else "up",
                     latency_ms=None,
-            ))
+                )
+            )
         else:
             result.append(s)
 
-    await broadcast("health", {
+    await broadcast(
+        "health",
+        {
             "statuses": [s.model_dump() for s in result],
             "timestamp": time.time(),
-    })
+        },
+    )
 
 
 async def _deferred_systemctl(action: str, unit: str, delay: float = 0.5) -> None:
@@ -104,7 +117,6 @@ async def _do_action(name: str, action: str) -> JSONResponse:
     if not ok:
         raise HTTPException(status_code=500, detail=output or f"Failed to {action}")
 
-    # Broadcast immediately with systemd status as the source of truth
     await _broadcast_health_with_override(name, unit_status)
 
     return JSONResponse(
@@ -115,13 +127,25 @@ async def _do_action(name: str, action: str) -> JSONResponse:
 @router.get("/{name}/unit")
 def get_unit(name: str) -> dict[str, str | None]:
     """Return the generated systemd unit file(s) for a managed component."""
-    from castle_core.generators import generate_timer, generate_unit
-
     _validate_managed(name)
-    config = load_config(settings.castle_root)
-    manifest = config.managed[name]
-    unit = generate_unit(config, name, manifest)
-    timer = generate_timer(name, manifest)
+    registry = get_registry()
+    deployed = registry.deployed[name]
+
+    # Get systemd spec from manifest if repo available
+    systemd_spec = None
+    root = get_castle_root()
+    manifest = None
+    if root:
+        from castle_core.config import load_config
+
+        config = load_config(root)
+        if name in config.components:
+            manifest = config.components[name]
+            if manifest.manage and manifest.manage.systemd:
+                systemd_spec = manifest.manage.systemd
+
+    unit = generate_unit_from_deployed(name, deployed, systemd_spec)
+    timer = generate_timer(name, manifest) if manifest else None
     return {"service": unit, "timer": timer}
 
 

castle-api/src/castle_api/stream.py

@@ -7,9 +7,7 @@ import json
 import logging
 import time
 
-from castle_core.config import load_config
-
-from castle_api.config import settings
+from castle_api.config import get_registry
 from castle_api.health import check_all_health
 
 logger = logging.getLogger(__name__)
@@ -60,12 +58,15 @@ async def health_poll_loop(interval: float = 10.0) -> None:
     """Background task that polls health and broadcasts updates."""
     while True:
         try:
-            config = load_config(settings.castle_root)
-            statuses = await check_all_health(config)
-            await broadcast("health", {
+            registry = get_registry()
+            statuses = await check_all_health(registry)
+            await broadcast(
+                "health",
+                {
                     "statuses": [s.model_dump() for s in statuses],
                     "timestamp": time.time(),
-            })
+                },
+            )
         except Exception:
             logger.exception("Health poll failed")
         await asyncio.sleep(interval)

castle-api/src/castle_api/tools.py

@@ -7,20 +7,23 @@ from pathlib import Path
 
 from fastapi import APIRouter, HTTPException, status
 
-from castle_core.config import load_config
 from castle_core.manifest import ComponentManifest
 
-from castle_api.config import settings
+from castle_api.config import get_config
 from castle_api.models import ToolDetail, ToolSummary
 
 router = APIRouter(tags=["tools"])
 
 
-def _tool_summary(name: str, manifest: ComponentManifest, root: Path | None = None) -> ToolSummary:
+def _tool_summary(
+    name: str, manifest: ComponentManifest, root: Path | None = None
+) -> ToolSummary:
     """Build a ToolSummary from a manifest that has a tool spec."""
     t = manifest.tool
     assert t is not None
-    installed = bool(manifest.install and manifest.install.path and manifest.install.path.enable)
+    installed = bool(
+        manifest.install and manifest.install.path and manifest.install.path.enable
+    )
 
     # Infer runner from run block or source directory
     runner = manifest.run.runner if manifest.run else None
@@ -44,12 +47,15 @@ def _tool_summary(name: str, manifest: ComponentManifest, root: Path | None = No
 
 @router.get("/tools", response_model=list[ToolSummary])
 def list_tools() -> list[ToolSummary]:
-    """List all registered tools."""
-    config = load_config(settings.castle_root)
+    """List all registered tools (requires repo access)."""
+    config = get_config()
     tools = {k: v for k, v in config.components.items() if v.tool}
 
     return sorted(
-        [_tool_summary(name, manifest, config.root) for name, manifest in tools.items()],
+        [
+            _tool_summary(name, manifest, config.root)
+            for name, manifest in tools.items()
+        ],
         key=lambda t: t.id,
     )
 
@@ -57,7 +63,7 @@ def list_tools() -> list[ToolSummary]:
 @router.get("/tools/{name}", response_model=ToolDetail)
 def get_tool(name: str) -> ToolDetail:
     """Get detailed info for a single tool."""
-    config = load_config(settings.castle_root)
+    config = get_config()
 
     if name not in config.components:
         raise HTTPException(
@@ -79,20 +85,31 @@ def get_tool(name: str) -> ToolDetail:
 @router.post("/tools/{name}/install")
 async def install_tool(name: str) -> dict:
     """Install a tool to PATH via uv tool install."""
-    config = load_config(settings.castle_root)
+    config = get_config()
     if name not in config.components:
-        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=f"'{name}' not found")
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND, detail=f"'{name}' not found"
+        )
 
     manifest = config.components[name]
     if not manifest.tool or not manifest.tool.source:
-        raise HTTPException(status_code=400, detail=f"'{name}' has no tool source to install")
+        raise HTTPException(
+            status_code=400, detail=f"'{name}' has no tool source to install"
+        )
 
     source_dir = config.root / manifest.tool.source
     if not (source_dir / "pyproject.toml").exists():
-        raise HTTPException(status_code=400, detail=f"No pyproject.toml in {manifest.tool.source}")
+        raise HTTPException(
+            status_code=400, detail=f"No pyproject.toml in {manifest.tool.source}"
+        )
 
     proc = await asyncio.create_subprocess_exec(
-        "uv", "tool", "install", "--editable", str(source_dir), "--force",
+        "uv",
+        "tool",
+        "install",
+        "--editable",
+        str(source_dir),
+        "--force",
         stdout=asyncio.subprocess.PIPE,
         stderr=asyncio.subprocess.PIPE,
     )
@@ -108,9 +125,11 @@ async def install_tool(name: str) -> dict:
 @router.post("/tools/{name}/uninstall")
 async def uninstall_tool(name: str) -> dict:
     """Uninstall a tool from PATH via uv tool uninstall."""
-    config = load_config(settings.castle_root)
+    config = get_config()
     if name not in config.components:
-        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=f"'{name}' not found")
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND, detail=f"'{name}' not found"
+        )
 
     manifest = config.components[name]
     if not manifest.tool or not manifest.tool.source:
@@ -118,17 +137,20 @@ async def uninstall_tool(name: str) -> dict:
 
     # uv tool uninstall uses the package name from pyproject.toml
     source_dir = config.root / manifest.tool.source
-    # Try to read the package name; fall back to the source dir name
     pkg_name = source_dir.name
     pyproject = source_dir / "pyproject.toml"
     if pyproject.exists():
         import tomllib
+
         with open(pyproject, "rb") as f:
             data = tomllib.load(f)
         pkg_name = data.get("project", {}).get("name", pkg_name)
 
     proc = await asyncio.create_subprocess_exec(
-        "uv", "tool", "uninstall", pkg_name,
+        "uv",
+        "tool",
+        "uninstall",
+        pkg_name,
         stdout=asyncio.subprocess.PIPE,
         stderr=asyncio.subprocess.PIPE,
     )

castle-api/tests/conftest.py

@@ -7,8 +7,14 @@ import pytest
 import yaml
 from fastapi.testclient import TestClient
 
-from castle_api.config import settings
+import castle_api.config as api_config
 from castle_api.main import app
+from castle_core.registry import (
+    DeployedComponent,
+    NodeConfig,
+    NodeRegistry,
+    save_registry,
+)
 
 
 @pytest.fixture
@@ -20,10 +26,10 @@ def castle_root(tmp_path: Path) -> Generator[Path, None, None]:
         "components": {
             "test-svc": {
                 "description": "Test service",
+                "source": "test-svc",
                 "run": {
                     "runner": "python_uv_tool",
                     "tool": "test-svc",
-                    "cwd": "test-svc",
                 },
                 "expose": {
                     "http": {
@@ -52,15 +58,67 @@ def castle_root(tmp_path: Path) -> Generator[Path, None, None]:
         },
     }
     castle_yaml.write_text(yaml.dump(config, default_flow_style=False))
-
-    original = settings.castle_root
-    settings.castle_root = tmp_path
     yield tmp_path
-    settings.castle_root = original
 
 
 @pytest.fixture
-def client(castle_root: Path) -> Generator[TestClient, None, None]:
-    """Create a test client pointing to temporary castle root."""
+def registry_path(tmp_path: Path, castle_root: Path) -> Generator[Path, None, None]:
+    """Create a temporary registry.yaml and patch the module to use it."""
+    reg_path = tmp_path / "registry.yaml"
+    registry = NodeRegistry(
+        node=NodeConfig(
+            hostname="test-node",
+            castle_root=str(castle_root),
+            gateway_port=9000,
+        ),
+        deployed={
+            "test-svc": DeployedComponent(
+                runner="python_uv_tool",
+                run_cmd=["uv", "run", "test-svc"],
+                env={
+                    "TEST_SVC_PORT": "19000",
+                    "TEST_SVC_DATA_DIR": "/data/castle/test-svc",
+                },
+                description="Test service",
+                roles=["service"],
+                port=19000,
+                health_path="/health",
+                proxy_path="/test-svc",
+                managed=True,
+            ),
+        },
+    )
+    save_registry(registry, reg_path)
+
+    # Patch the registry path and helper functions
+    import castle_core.registry as reg_mod
+
+    original_path = reg_mod.REGISTRY_PATH
+    reg_mod.REGISTRY_PATH = reg_path
+
+    original_get_registry = api_config.get_registry
+    original_get_castle_root = api_config.get_castle_root
+
+    def _get_registry() -> NodeRegistry:
+        from castle_core.registry import load_registry
+
+        return load_registry(reg_path)
+
+    def _get_castle_root() -> Path | None:
+        return castle_root
+
+    api_config.get_registry = _get_registry
+    api_config.get_castle_root = _get_castle_root
+
+    yield reg_path
+
+    reg_mod.REGISTRY_PATH = original_path
+    api_config.get_registry = original_get_registry
+    api_config.get_castle_root = original_get_castle_root
+
+
+@pytest.fixture
+def client(registry_path: Path) -> Generator[TestClient, None, None]:
+    """Create a test client with temporary registry."""
     with TestClient(app) as client:
         yield client

castle-api/tests/test_health.py

@@ -55,7 +55,7 @@ class TestComponentDetail:
         data = response.json()
         assert data["id"] == "test-svc"
         assert "manifest" in data
-        assert data["manifest"]["run"]["runner"] == "python_uv_tool"
+        assert data["manifest"]["runner"] == "python_uv_tool"
 
     def test_not_found(self, client: TestClient) -> None:
         """Returns 404 for unknown component."""
@@ -67,11 +67,12 @@ class TestGateway:
     """Gateway info endpoint tests."""
 
     def test_gateway_info(self, client: TestClient) -> None:
-        """Returns gateway configuration."""
+        """Returns gateway configuration from registry."""
         response = client.get("/gateway")
         assert response.status_code == 200
         data = response.json()
         assert data["port"] == 9000
-        assert data["component_count"] == 3
+        # Registry has 1 deployed component (test-svc)
+        assert data["component_count"] == 1
         assert data["service_count"] == 1
         assert data["managed_count"] == 1

castle-api/tests/test_tools.py

@@ -1,7 +1,5 @@
 """Tests for tools endpoints."""
 
-from pathlib import Path
-
 from fastapi.testclient import TestClient
 
 

castle.yaml

@@ -5,7 +5,6 @@ components:
     description: Caddy reverse proxy gateway
     run:
       runner: command
-      working_dir: .
       argv:
       - caddy
       - run
@@ -25,12 +24,9 @@ components:
   central-context:
     description: Content storage API useful to get context into my data dir from anywhere
       on the LAN
+    source: components/central-context
     run:
       runner: python_uv_tool
-      working_dir: components/central-context
-      env:
-        CENTRAL_CONTEXT_DATA_DIR: /data/castle/central-context
-        CENTRAL_CONTEXT_PORT: '9001'
       tool: central-context
     manage:
       systemd: {}
@@ -46,14 +42,15 @@ components:
     description: Desktop notification forwarder. This taps into your notifications
       system on the desktop and will forward all notifications the the central context
       server.
+    source: components/notification-bridge
     run:
       runner: python_uv_tool
-      working_dir: components/notification-bridge
+      tool: notification-bridge
+    defaults:
       env:
         CENTRAL_CONTEXT_URL: http://localhost:9001
         BUCKET_NAME: notifications
         PORT: '9002'
-      tool: notification-bridge
     manage:
       systemd: {}
     expose:
@@ -66,11 +63,9 @@ components:
         path_prefix: /notifications
   castle-api:
     description: Castle API
+    source: castle-api
     run:
       runner: python_uv_tool
-      working_dir: castle-api
-      env:
-        CASTLE_API_CASTLE_ROOT: /data/repos/castle
       tool: castle-api
     manage:
       systemd: {}
@@ -84,15 +79,16 @@ components:
         path_prefix: /api
   protonmail:
     description: ProtonMail email sync via Bridge
+    source: components/protonmail
     run:
       runner: command
-      working_dir: components/protonmail
-      env:
-        PROTONMAIL_USERNAME: paul@payne.io
-        PROTONMAIL_API_KEY: ${secret:PROTONMAIL_API_KEY}
       argv:
       - protonmail
       - sync
+    defaults:
+      env:
+        PROTONMAIL_USERNAME: paul@payne.io
+        PROTONMAIL_API_KEY: ${secret:PROTONMAIL_API_KEY}
     triggers:
     - type: schedule
       cron: '*/5 * * * *'
@@ -106,13 +102,14 @@ components:
       source: components/protonmail/
   backup-collect:
     description: Collect files from various sources into backup directory
+    source: components/backup-collect
     run:
       runner: command
-      working_dir: .
-      env:
-        DBACKUP: /data/backup
       argv:
       - backup-collect
+    defaults:
+      env:
+        DBACKUP: /data/backup
     triggers:
     - type: schedule
       cron: 0 2 * * *
@@ -127,12 +124,12 @@ components:
     description: Nightly restic backup of /data to /storage
     run:
       runner: command
-      working_dir: .
+      argv:
+      - backup-data
+    defaults:
       env:
         RESTIC_REPOSITORY: /storage/restic-data
         RESTIC_PASSWORD_FILE: /home/payne/.config/restic-password
-      argv:
-      - backup-data
     triggers:
     - type: schedule
       cron: 30 3 * * *
@@ -142,8 +139,8 @@ components:
     tool: {}
   castle-app:
     description: Castle web app
+    source: app
     build:
-      working_dir: app
       commands:
       - - pnpm
         - build
@@ -151,6 +148,7 @@ components:
       - dist/
   devbox-connect:
     description: SSH tunnel manager with auto-reconnect
+    source: components/devbox-connect
     install:
       path:
         alias: devbox-connect
@@ -158,6 +156,7 @@ components:
       source: components/devbox-connect/
   mbox2eml:
     description: MBOX to EML email converter
+    source: components/mbox2eml
     install:
       path:
         alias: mbox2eml
@@ -165,6 +164,7 @@ components:
       source: components/mbox2eml/
   android-backup:
     description: Backup Android device using ADB
+    source: components/android-backup
     install:
       path:
         alias: android-backup
@@ -174,6 +174,7 @@ components:
       - adb
   browser:
     description: Browse the web using natural language via browser-use
+    source: components/browser
     install:
       path:
         alias: browser
@@ -181,6 +182,7 @@ components:
       source: components/browser/
   docx-extractor:
     description: Extract content and metadata from Word .docx files
+    source: components/docx-extractor
     install:
       path:
         alias: docx-extractor
@@ -190,6 +192,7 @@ components:
       - pandoc
   docx2md:
     description: Convert Word .docx files to Markdown
+    source: components/docx2md
     install:
       path:
         alias: docx2md
@@ -199,6 +202,7 @@ components:
       - pandoc
   gpt:
     description: OpenAI text generation utility
+    source: components/gpt
     install:
       path:
         alias: gpt
@@ -206,6 +210,7 @@ components:
       source: components/gpt/
   html2text:
     description: Convert HTML content to plain text
+    source: components/html2text
     install:
       path:
         alias: html2text
@@ -213,6 +218,7 @@ components:
       source: components/html2text/
   md2pdf:
     description: Convert Markdown files to PDF
+    source: components/md2pdf
     install:
       path:
         alias: md2pdf
@@ -223,6 +229,7 @@ components:
       - texlive-latex-base
   mdscraper:
     description: Combine text files into a single markdown document
+    source: components/mdscraper
     install:
       path:
         alias: mdscraper
@@ -230,6 +237,7 @@ components:
       source: components/mdscraper/
   pdf-extractor:
     description: Extract content and metadata from PDF files
+    source: components/pdf-extractor
     install:
       path:
         alias: pdf-extractor
@@ -237,6 +245,7 @@ components:
       source: components/pdf-extractor/
   pdf2md:
     description: Convert PDF files to Markdown
+    source: components/pdf2md
     install:
       path:
         alias: pdf2md
@@ -247,6 +256,7 @@ components:
       - poppler-utils
   schedule:
     description: Manage systemd user timers and scheduled tasks
+    source: components/schedule
     install:
       path:
         alias: schedule
@@ -254,6 +264,7 @@ components:
       source: components/schedule/
   search:
     description: Manage self-contained searchable collections of files
+    source: components/search
     install:
       path:
         alias: search
@@ -261,6 +272,7 @@ components:
       source: components/search/
   text-extractor:
     description: Extract content and metadata from text files
+    source: components/text-extractor
     install:
       path:
         alias: text-extractor

cli/src/castle_cli/commands/create.py

@@ -73,17 +73,14 @@ def run_create(args: argparse.Namespace) -> int:
     )
 
     # Build manifest entry
-    env_prefix = package_name.upper()
-
     if proj_type == "service":
         manifest = ComponentManifest(
             id=name,
             description=args.description or f"A castle {proj_type}",
+            source=f"components/{name}",
             run=RunPythonUvTool(
                 runner="python_uv_tool",
                 tool=name,
-                cwd=f"components/{name}",
-                env={f"{env_prefix}_DATA_DIR": f"/data/castle/{name}"},
             ),
             expose=ExposeSpec(
                 http=HttpExposeSpec(
@@ -98,6 +95,7 @@ def run_create(args: argparse.Namespace) -> int:
         manifest = ComponentManifest(
             id=name,
             description=args.description or f"A castle {proj_type}",
+            source=f"components/{name}",
             tool=ToolSpec(source=f"components/{name}/"),
             install=InstallSpec(path=PathInstallSpec(alias=name)),
         )
@@ -106,6 +104,7 @@ def run_create(args: argparse.Namespace) -> int:
         manifest = ComponentManifest(
             id=name,
             description=args.description or f"A castle {proj_type}",
+            source=f"components/{name}",
         )
 
     config.components[name] = manifest
@@ -120,8 +119,7 @@ def run_create(args: argparse.Namespace) -> int:
     print("  uv sync")
     if proj_type == "service":
         print(f"  uv run {name}  # starts on port {port}")
+        print(f"  castle deploy {name}  # deploy to ~/.castle/")
     print(f"  castle test {name}")
 
     return 0
-
-

cli/src/castle_cli/commands/deploy.py

@@ -0,0 +1,285 @@
+"""castle deploy — bridge spec (castle.yaml) to runtime (~/.castle/)."""
+
+from __future__ import annotations
+
+import argparse
+import shutil
+import subprocess
+from pathlib import Path
+
+from castle_core.config import (
+    GENERATED_DIR,
+    STATIC_DIR,
+    CastleConfig,
+    ensure_dirs,
+    load_config,
+    resolve_env_vars,
+)
+from castle_core.generators.caddyfile import generate_caddyfile_from_registry
+from castle_core.generators.systemd import (
+    generate_timer,
+    generate_unit_from_deployed,
+    get_schedule_trigger,
+    timer_name,
+    unit_name,
+)
+from castle_core.manifest import ComponentManifest
+from castle_core.registry import (
+    REGISTRY_PATH,
+    DeployedComponent,
+    NodeConfig,
+    NodeRegistry,
+    load_registry,
+    save_registry,
+)
+
+DATA_ROOT = Path("/data/castle")
+SYSTEMD_USER_DIR = Path.home() / ".config" / "systemd" / "user"
+
+
+def run_deploy(args: argparse.Namespace) -> int:
+    """Deploy components from castle.yaml to ~/.castle/."""
+    config = load_config()
+    component_name = getattr(args, "component", None)
+
+    if component_name:
+        if component_name not in config.components:
+            print(f"Error: component '{component_name}' not found in castle.yaml")
+            return 1
+        names = [component_name]
+    else:
+        names = list(config.components.keys())
+
+    ensure_dirs()
+
+    # Build node config
+    node = NodeConfig(castle_root=str(config.root), gateway_port=config.gateway.port)
+
+    # Load existing registry to preserve components not being redeployed,
+    # or start fresh if deploying all
+    if component_name and REGISTRY_PATH.exists():
+        try:
+            existing = load_registry()
+            registry = NodeRegistry(node=node, deployed=dict(existing.deployed))
+        except (FileNotFoundError, ValueError):
+            registry = NodeRegistry(node=node)
+    else:
+        registry = NodeRegistry(node=node)
+
+    deployed_count = 0
+    for name in names:
+        manifest = config.components[name]
+
+        # Only deploy components with a run spec
+        if not manifest.run:
+            continue
+
+        deployed = _build_deployed(config, name, manifest)
+        registry.deployed[name] = deployed
+        deployed_count += 1
+        _print_deployed(name, deployed)
+
+    # Handle castle-app build artifacts
+    _copy_app_static(config)
+
+    # Save registry
+    save_registry(registry)
+    print(f"\nRegistry written: {REGISTRY_PATH}")
+
+    # Generate systemd units from registry
+    _generate_systemd_units(config, registry)
+
+    # Generate Caddyfile from registry
+    caddyfile_path = GENERATED_DIR / "Caddyfile"
+    caddyfile_content = generate_caddyfile_from_registry(registry)
+    caddyfile_path.write_text(caddyfile_content)
+    print(f"Caddyfile written: {caddyfile_path}")
+
+    # Reload systemd daemon
+    subprocess.run(["systemctl", "--user", "daemon-reload"], check=False)
+
+    print(f"\nDeployed {deployed_count} component(s).")
+    print("Run 'castle services start' to start all services.")
+    return 0
+
+
+def _env_prefix(name: str) -> str:
+    """Derive env var prefix from component name: central-context → CENTRAL_CONTEXT."""
+    return name.replace("-", "_").upper()
+
+
+def _build_deployed(
+    config: CastleConfig, name: str, manifest: ComponentManifest
+) -> DeployedComponent:
+    """Build a DeployedComponent from a manifest spec."""
+    run = manifest.run
+    assert run is not None
+
+    # 1. Convention-based env vars
+    prefix = _env_prefix(name)
+    env: dict[str, str] = {}
+
+    # Data dir convention (for all managed components)
+    if manifest.manage and manifest.manage.systemd:
+        env[f"{prefix}_DATA_DIR"] = str(DATA_ROOT / name)
+
+    # Port convention (if exposed)
+    if manifest.expose and manifest.expose.http:
+        env[f"{prefix}_PORT"] = str(manifest.expose.http.internal.port)
+
+    # 2. Merge defaults.env (overrides conventions)
+    if manifest.defaults and manifest.defaults.env:
+        env.update(manifest.defaults.env)
+
+    # 3. Resolve secrets
+    env = resolve_env_vars(env, manifest)
+
+    # 4. Build run_cmd
+    run_cmd = _build_run_cmd(run, env)
+
+    # 5. Extract metadata
+    port = None
+    health_path = None
+    if manifest.expose and manifest.expose.http:
+        port = manifest.expose.http.internal.port
+        health_path = manifest.expose.http.health_path
+
+    proxy_path = None
+    if manifest.proxy and manifest.proxy.caddy and manifest.proxy.caddy.enable:
+        proxy_path = manifest.proxy.caddy.path_prefix or f"/{name}"
+
+    schedule = None
+    sched_trigger = get_schedule_trigger(manifest)
+    if sched_trigger:
+        schedule = sched_trigger.cron
+
+    managed = bool(manifest.manage and manifest.manage.systemd and manifest.manage.systemd.enable)
+
+    roles = [r.value for r in manifest.roles]
+
+    return DeployedComponent(
+        runner=run.runner,
+        run_cmd=run_cmd,
+        env=env,
+        description=manifest.description,
+        roles=roles,
+        port=port,
+        health_path=health_path,
+        proxy_path=proxy_path,
+        schedule=schedule,
+        managed=managed,
+    )
+
+
+def _build_run_cmd(run: object, env: dict[str, str]) -> list[str]:
+    """Build a run command list from a RunSpec."""
+    match run.runner:
+        case "python_uv_tool":
+            uv = shutil.which("uv") or "uv"
+            cmd = [uv, "run", run.tool]
+            if run.args:
+                cmd.extend(run.args)
+            return cmd
+        case "python_module":
+            python = run.python or shutil.which("python3") or "python3"
+            cmd = [python, "-m", run.module]
+            if run.args:
+                cmd.extend(run.args)
+            return cmd
+        case "command":
+            cmd = list(run.argv)
+            resolved = shutil.which(cmd[0])
+            if resolved:
+                cmd[0] = resolved
+            return cmd
+        case "container":
+            runtime = shutil.which("podman") or shutil.which("docker") or "podman"
+            image_name = run.image.split("/")[-1].split(":")[0]
+            cmd = [runtime, "run", "--rm", f"--name=castle-{image_name}"]
+            for container_port, host_port in run.ports.items():
+                cmd.extend(["-p", f"{host_port}:{container_port}"])
+            for vol in run.volumes:
+                cmd.extend(["-v", vol])
+            # Container env comes from both run.env (container-specific) and deployed env
+            for key, val in run.env.items():
+                cmd.extend(["-e", f"{key}={val}"])
+            for key, val in env.items():
+                cmd.extend(["-e", f"{key}={val}"])
+            if run.workdir:
+                cmd.extend(["-w", run.workdir])
+            cmd.append(run.image)
+            if run.command:
+                cmd.extend(run.command)
+            if run.args:
+                cmd.extend(run.args)
+            return cmd
+        case "node":
+            cmd = [run.package_manager, "run", run.script]
+            if run.args:
+                cmd.extend(run.args)
+            return cmd
+        case _:
+            raise ValueError(f"Unsupported runner: {run.runner}")
+
+
+def _print_deployed(name: str, deployed: DeployedComponent) -> None:
+    """Print deployment summary for a component."""
+    parts = [f"  {name}"]
+    if deployed.port:
+        parts.append(f"port={deployed.port}")
+    if deployed.schedule:
+        parts.append(f"schedule={deployed.schedule}")
+    if deployed.proxy_path:
+        parts.append(f"proxy={deployed.proxy_path}")
+    print(" ".join(parts))
+
+
+def _copy_app_static(config: CastleConfig) -> None:
+    """Copy castle-app build output to ~/.castle/static/castle-app/."""
+    if "castle-app" not in config.components:
+        return
+
+    manifest = config.components["castle-app"]
+    if not (manifest.build and manifest.build.outputs):
+        return
+
+    # Find the source dist directory
+    source_dir = manifest.source_dir or "app"
+    for output in manifest.build.outputs:
+        src = config.root / source_dir / output
+        if src.exists():
+            dest = STATIC_DIR / "castle-app"
+            if dest.exists():
+                shutil.rmtree(dest)
+            shutil.copytree(src, dest)
+            print(f"  Static: {src} → {dest}")
+
+
+def _generate_systemd_units(config: CastleConfig, registry: NodeRegistry) -> None:
+    """Generate systemd units from the registry."""
+    SYSTEMD_USER_DIR.mkdir(parents=True, exist_ok=True)
+
+    for name, deployed in registry.deployed.items():
+        if not deployed.managed:
+            continue
+
+        # Get systemd spec from manifest (for restart policy, exec_reload, etc.)
+        systemd_spec = None
+        if name in config.components:
+            manifest = config.components[name]
+            if manifest.manage and manifest.manage.systemd:
+                systemd_spec = manifest.manage.systemd
+
+        # Generate and write service unit
+        svc_name = unit_name(name)
+        svc_content = generate_unit_from_deployed(name, deployed, systemd_spec)
+        (SYSTEMD_USER_DIR / svc_name).write_text(svc_content)
+
+        # Generate timer if scheduled
+        if name in config.components:
+            timer_content = generate_timer(name, config.components[name])
+            if timer_content:
+                tmr_name = timer_name(name)
+                (SYSTEMD_USER_DIR / tmr_name).write_text(timer_content)
+
+    print(f"Systemd units written: {SYSTEMD_USER_DIR}")

cli/src/castle_cli/commands/dev.py

@@ -23,9 +23,7 @@ def _has_pyproject(project_dir: Path) -> bool:
     return (project_dir / "pyproject.toml").exists()
 
 
-def _run_in_project(
-    project_dir: Path, cmd: list[str], label: str
-) -> bool:
+def _run_in_project(project_dir: Path, cmd: list[str], label: str) -> bool:
     """Run a command in a project directory. Returns True on success."""
     if not _has_pyproject(project_dir):
         return True  # Skip projects without pyproject.toml

cli/src/castle_cli/commands/gateway.py

@@ -5,27 +5,28 @@ from __future__ import annotations
 import argparse
 import subprocess
 
-from castle_cli.config import GENERATED_DIR, CastleConfig, ensure_dirs, load_config
-from castle_core.generators.caddyfile import find_app_dist, generate_caddyfile
+from castle_core.config import GENERATED_DIR
+from castle_core.generators.caddyfile import generate_caddyfile_from_registry
+from castle_core.registry import REGISTRY_PATH, load_registry
 
+from castle_cli.config import CastleConfig, ensure_dirs, load_config
 
 GATEWAY_COMPONENT = "castle-gateway"
 GATEWAY_UNIT = "castle-castle-gateway.service"
 
 
-def _write_generated_files(config: CastleConfig) -> None:
-    """Write generated Caddyfile."""
+def _write_generated_files() -> None:
+    """Write generated Caddyfile from registry."""
     ensure_dirs()
 
-    caddyfile_path = GENERATED_DIR / "Caddyfile"
-    caddyfile_path.write_text(generate_caddyfile(config))
-    print(f"  Generated {caddyfile_path}")
+    if not REGISTRY_PATH.exists():
+        print("Error: no registry found. Run 'castle deploy' first.")
+        return
 
-    app_dist = find_app_dist(config)
-    if app_dist:
-        print(f"  App: {app_dist}")
-    else:
-        print("  App: dist/ not found, using fallback")
+    registry = load_registry()
+    caddyfile_path = GENERATED_DIR / "Caddyfile"
+    caddyfile_path.write_text(generate_caddyfile_from_registry(registry))
+    print(f"  Generated {caddyfile_path}")
 
 
 def run_gateway(args: argparse.Namespace) -> int:
@@ -38,24 +39,29 @@ def run_gateway(args: argparse.Namespace) -> int:
 
     if args.gateway_command == "start":
         if getattr(args, "dry_run", False):
-            return _gateway_dry_run(config)
+            return _gateway_dry_run()
         return _gateway_start(config)
     elif args.gateway_command == "stop":
         return _gateway_stop()
     elif args.gateway_command == "reload":
         if getattr(args, "dry_run", False):
-            return _gateway_dry_run(config)
-        return _gateway_reload(config)
+            return _gateway_dry_run()
+        return _gateway_reload()
     elif args.gateway_command == "status":
         return _gateway_status()
 
     return 1
 
 
-def _gateway_dry_run(config: CastleConfig) -> int:
+def _gateway_dry_run() -> int:
     """Print generated Caddyfile without applying."""
+    if not REGISTRY_PATH.exists():
+        print("Error: no registry found. Run 'castle deploy' first.")
+        return 1
+
+    registry = load_registry()
     print("# Caddyfile")
-    print(generate_caddyfile(config))
+    print(generate_caddyfile_from_registry(registry))
     return 0
 
 
@@ -68,7 +74,7 @@ def _gateway_start(config: CastleConfig) -> int:
         return 1
 
     print("Generating gateway configuration...")
-    _write_generated_files(config)
+    _write_generated_files()
 
     print(f"\nStarting gateway on port {config.gateway.port}...")
     return _service_enable(config, GATEWAY_COMPONENT)
@@ -81,14 +87,15 @@ def _gateway_stop() -> int:
     return _service_disable(GATEWAY_COMPONENT)
 
 
-def _gateway_reload(config: CastleConfig) -> int:
+def _gateway_reload() -> int:
     """Regenerate config and reload Caddy."""
     print("Regenerating gateway configuration...")
-    _write_generated_files(config)
+    _write_generated_files()
 
     result = subprocess.run(
         ["systemctl", "--user", "reload", GATEWAY_UNIT],
-        capture_output=True, text=True,
+        capture_output=True,
+        text=True,
     )
 
     if result.returncode == 0:
@@ -98,7 +105,8 @@ def _gateway_reload(config: CastleConfig) -> int:
         print("Reload signal sent. Verifying...")
         result = subprocess.run(
             ["systemctl", "--user", "is-active", GATEWAY_UNIT],
-            capture_output=True, text=True,
+            capture_output=True,
+            text=True,
         )
         if result.stdout.strip() == "active":
             print("Gateway running.")
@@ -112,7 +120,8 @@ def _gateway_status() -> int:
     """Show gateway status via systemd."""
     result = subprocess.run(
         ["systemctl", "--user", "is-active", GATEWAY_UNIT],
-        capture_output=True, text=True,
+        capture_output=True,
+        text=True,
     )
     status = result.stdout.strip()
 

cli/src/castle_cli/commands/info.py

@@ -42,20 +42,24 @@ def run_info(args: argparse.Namespace) -> int:
     if manifest.description:
         print(f"  {BOLD}description{RESET}: {manifest.description}")
 
+    # Source
+    if manifest.source:
+        print(f"  {BOLD}source{RESET}:      {manifest.source}")
+
     # Run spec
     if manifest.run:
         print(f"  {BOLD}runner{RESET}:      {manifest.run.runner}")
-        if manifest.run.working_dir:
-            print(f"  {BOLD}working_dir{RESET}: {manifest.run.working_dir}")
         if hasattr(manifest.run, "tool"):
             print(f"  {BOLD}tool{RESET}:        {manifest.run.tool}")
         elif hasattr(manifest.run, "argv"):
             print(f"  {BOLD}argv{RESET}:        {manifest.run.argv}")
         elif hasattr(manifest.run, "image"):
             print(f"  {BOLD}image{RESET}:       {manifest.run.image}")
-        if manifest.run.env:
-            print(f"  {BOLD}env{RESET}:")
-            for key, val in manifest.run.env.items():
+
+    # Defaults env
+    if manifest.defaults and manifest.defaults.env:
+        print(f"  {BOLD}defaults.env{RESET}:")
+        for key, val in manifest.defaults.env.items():
             print(f"    {key}: {val}")
 
     # Expose
@@ -85,7 +89,7 @@ def run_info(args: argparse.Namespace) -> int:
     if manifest.tool:
         t = manifest.tool
         if t.source:
-            print(f"  {BOLD}source{RESET}:      {t.source}")
+            print(f"  {BOLD}tool.source{RESET}: {t.source}")
         if t.system_dependencies:
             print(f"  {BOLD}requires{RESET}:    {', '.join(t.system_dependencies)}")
 
@@ -104,8 +108,8 @@ def run_info(args: argparse.Namespace) -> int:
             print(f"    - {cap.type}" + (f" ({cap.name})" if cap.name else ""))
 
     # Show CLAUDE.md if it exists
-    cwd = manifest.run.working_dir if manifest.run else None
-    claude_md = _find_claude_md(config.root, cwd or name)
+    source_dir = manifest.source_dir or name
+    claude_md = _find_claude_md(config.root, source_dir)
     if claude_md:
         print(f"\n{BOLD}{CYAN}CLAUDE.md{RESET}")
         print(f"{CYAN}{'─' * 40}{RESET}")
@@ -115,9 +119,9 @@ def run_info(args: argparse.Namespace) -> int:
     return 0
 
 
-def _find_claude_md(root: Path, working_dir: str) -> str | None:
+def _find_claude_md(root: Path, source_dir: str) -> str | None:
     """Read CLAUDE.md from project directory if it exists."""
-    claude_path = root / working_dir / "CLAUDE.md"
+    claude_path = root / source_dir / "CLAUDE.md"
     if claude_path.exists():
         return claude_path.read_text()
     return None

cli/src/castle_cli/commands/list_cmd.py

@@ -32,9 +32,7 @@ def run_list(args: argparse.Namespace) -> int:
 
     filter_role = getattr(args, "role", None)
     if filter_role:
-        components = {
-            k: v for k, v in components.items() if filter_role in v.roles
-        }
+        components = {k: v for k, v in components.items() if filter_role in v.roles}
 
     if getattr(args, "json", False):
         output = []

cli/src/castle_cli/commands/run_cmd.py

@@ -4,93 +4,36 @@ from __future__ import annotations
 
 import argparse
 import os
-import shutil
 import subprocess
-import sys
 
-from castle_cli.config import load_config, resolve_env_vars
+from castle_core.registry import REGISTRY_PATH, load_registry
 
 
 def run_run(args: argparse.Namespace) -> int:
-    """Run a component in the foreground (dev mode)."""
-    config = load_config()
+    """Run a component in the foreground using the registry."""
+    if not REGISTRY_PATH.exists():
+        print("Error: no registry found. Run 'castle deploy' first.")
+        return 1
+
+    registry = load_registry()
     name = args.name
 
-    if name not in config.components:
-        print(f"Error: component '{name}' not found in castle.yaml")
+    if name not in registry.deployed:
+        print(f"Error: component '{name}' not found in registry.")
+        print("Run 'castle deploy' to update the registry.")
         return 1
 
-    manifest = config.components[name]
-    run = manifest.run
+    deployed = registry.deployed[name]
 
-    if run is None:
-        print(f"Error: component '{name}' has no run spec")
-        return 1
-
-    # Build command
+    # Build command with any extra args
     extra_args = getattr(args, "extra", []) or []
-    cmd = _build_command(run, extra_args)
-    if cmd is None:
-        print(f"Error: unsupported runner '{run.runner}' for foreground execution")
-        return 1
-
-    # Working directory
-    cwd = config.root / (run.working_dir or name)
-    if not cwd.exists():
-        print(f"Error: working directory '{cwd}' does not exist")
-        return 1
+    cmd = list(deployed.run_cmd) + extra_args
 
     # Merge environment
     env = dict(os.environ)
-    resolved = resolve_env_vars(run.env, manifest)
-    env.update(resolved)
+    env.update(deployed.env)
 
-    # Run in foreground
-    result = subprocess.run(cmd, cwd=cwd, env=env)
+    # Run in foreground (no cwd — registry-based, no repo dependency)
+    print(f"Running {name}: {' '.join(cmd)}")
+    result = subprocess.run(cmd, env=env)
     return result.returncode
-
-
-def _build_command(run: object, extra_args: list[str]) -> list[str] | None:
-    """Build command list from RunSpec."""
-    match run.runner:
-        case "python_uv_tool":
-            uv = shutil.which("uv") or "uv"
-            cmd = [uv, "run", run.tool]
-            cmd.extend(run.args)
-            cmd.extend(extra_args)
-            return cmd
-        case "python_module":
-            python = run.python or sys.executable
-            cmd = [python, "-m", run.module]
-            cmd.extend(run.args)
-            cmd.extend(extra_args)
-            return cmd
-        case "command":
-            cmd = list(run.argv)
-            cmd.extend(extra_args)
-            return cmd
-        case "container":
-            runtime = shutil.which("podman") or shutil.which("docker") or "podman"
-            cmd = [runtime, "run", "--rm", "-it"]
-            for cp, hp in run.ports.items():
-                cmd.extend(["-p", f"{hp}:{cp}"])
-            for vol in run.volumes:
-                cmd.extend(["-v", vol])
-            for key, val in run.env.items():
-                cmd.extend(["-e", f"{key}={val}"])
-            if run.workdir:
-                cmd.extend(["-w", run.workdir])
-            cmd.append(run.image)
-            if run.command:
-                cmd.extend(run.command)
-            cmd.extend(run.args)
-            cmd.extend(extra_args)
-            return cmd
-        case "node":
-            pm = run.package_manager
-            cmd = [pm, "run", run.script]
-            cmd.extend(run.args)
-            cmd.extend(extra_args)
-            return cmd
-        case _:
-            return None

cli/src/castle_cli/commands/service.py

@@ -5,19 +5,21 @@ from __future__ import annotations
 import argparse
 import subprocess
 
+from castle_core.generators.systemd import (
+    SYSTEMD_USER_DIR,
+    generate_timer,
+    generate_unit_from_deployed,
+    get_schedule_trigger,
+    timer_name,
+    unit_name,
+)
+from castle_core.registry import REGISTRY_PATH, load_registry
+
 from castle_cli.config import (
     CastleConfig,
     ensure_dirs,
     load_config,
 )
-from castle_core.generators.systemd import (
-    SYSTEMD_USER_DIR,
-    generate_timer,
-    generate_unit,
-    get_schedule_trigger,
-    timer_name,
-    unit_name,
-)
 
 
 def _install_unit(uname: str, content: str) -> None:
@@ -74,25 +76,38 @@ def run_services(args: argparse.Namespace) -> int:
 
 def _service_enable(config: CastleConfig, name: str) -> int:
     """Enable and start a single service (or timer for scheduled jobs)."""
-    managed = config.managed
-    if name not in managed:
-        print(f"Error: '{name}' is not a managed service")
+    # Require registry
+    if not REGISTRY_PATH.exists():
+        print("Error: no registry found. Run 'castle deploy' first.")
         return 1
 
-    manifest = managed[name]
-    if not manifest.run:
-        print(f"Error: '{name}' has no run spec defined")
+    registry = load_registry()
+    if name not in registry.deployed:
+        print(f"Error: '{name}' not found in registry. Run 'castle deploy' first.")
+        return 1
+
+    deployed = registry.deployed[name]
+    if not deployed.managed:
+        print(f"Error: '{name}' is not a managed service")
         return 1
 
     ensure_dirs()
 
-    # Generate and install the service unit
+    # Get systemd spec from manifest for restart policy etc.
+    systemd_spec = None
+    if name in config.components:
+        manifest = config.components[name]
+        if manifest.manage and manifest.manage.systemd:
+            systemd_spec = manifest.manage.systemd
+
+    # Generate and install the service unit from registry
     svc_unit = unit_name(name)
-    svc_content = generate_unit(config, name, manifest)
+    svc_content = generate_unit_from_deployed(name, deployed, systemd_spec)
     _install_unit(svc_unit, svc_content)
 
-    # Check for timer
-    timer_content = generate_timer(name, manifest)
+    # Check for timer (still uses manifest for schedule config)
+    manifest = config.components.get(name)
+    timer_content = generate_timer(name, manifest) if manifest else None
     if timer_content:
         tmr_unit = timer_name(name)
         _install_unit(tmr_unit, timer_content)
@@ -103,7 +118,8 @@ def _service_enable(config: CastleConfig, name: str) -> int:
 
         result = subprocess.run(
             ["systemctl", "--user", "is-active", tmr_unit],
-            capture_output=True, text=True,
+            capture_output=True,
+            text=True,
         )
         status = result.stdout.strip()
         if status in ("active", "waiting"):
@@ -117,12 +133,13 @@ def _service_enable(config: CastleConfig, name: str) -> int:
 
         result = subprocess.run(
             ["systemctl", "--user", "is-active", svc_unit],
-            capture_output=True, text=True,
+            capture_output=True,
+            text=True,
         )
         status = result.stdout.strip()
         port_str = ""
-        if manifest.expose and manifest.expose.http:
-            port_str = f" (port {manifest.expose.http.internal.port})"
+        if deployed.port:
+            port_str = f" (port {deployed.port})"
         if status == "active":
             print(f"  {name}: running{port_str}")
         else:
@@ -166,7 +183,8 @@ def _service_status(config: CastleConfig) -> int:
             tmr_unit = timer_name(name)
             result = subprocess.run(
                 ["systemctl", "--user", "is-active", tmr_unit],
-                capture_output=True, text=True,
+                capture_output=True,
+                text=True,
             )
             status = result.stdout.strip()
             if status in ("active", "waiting"):
@@ -181,7 +199,8 @@ def _service_status(config: CastleConfig) -> int:
             svc_unit = unit_name(name)
             result = subprocess.run(
                 ["systemctl", "--user", "is-active", svc_unit],
-                capture_output=True, text=True,
+                capture_output=True,
+                text=True,
             )
             status = result.stdout.strip()
             if status == "active":
@@ -203,41 +222,55 @@ def _service_status(config: CastleConfig) -> int:
 
 def _service_dry_run(config: CastleConfig, name: str) -> int:
     """Print the generated systemd unit(s) without installing."""
-    managed = config.managed
-    if name not in managed:
-        print(f"Error: '{name}' is not a managed service")
-        return 1
-
-    manifest = managed[name]
-    if not manifest.run:
-        print(f"Error: '{name}' has no run spec defined")
-        return 1
+    # Try registry first, fall back to showing what deploy would generate
+    if REGISTRY_PATH.exists():
+        registry = load_registry()
+        if name in registry.deployed:
+            deployed = registry.deployed[name]
+            systemd_spec = None
+            if name in config.components:
+                manifest = config.components[name]
+                if manifest.manage and manifest.manage.systemd:
+                    systemd_spec = manifest.manage.systemd
 
             svc_unit = unit_name(name)
-    svc_content = generate_unit(config, name, manifest)
+            svc_content = generate_unit_from_deployed(name, deployed, systemd_spec)
             print(f"# {svc_unit}")
             print(svc_content)
 
+            manifest = config.components.get(name)
+            if manifest:
                 timer_content = generate_timer(name, manifest)
                 if timer_content:
                     print(f"# {timer_name(name)}")
                     print(timer_content)
-
             return 0
 
+    print(f"Error: '{name}' not found in registry. Run 'castle deploy' first.")
+    return 1
+
 
 def _services_start(config: CastleConfig) -> int:
     """Start all managed services and gateway."""
+    # Require registry
+    if not REGISTRY_PATH.exists():
+        print("Error: no registry found. Run 'castle deploy' first.")
+        return 1
+
     ensure_dirs()
 
-    # Generate Caddyfile before starting gateway
-    from castle_cli.commands.gateway import _write_generated_files
+    # Generate Caddyfile from registry
+    from castle_core.config import GENERATED_DIR
+    from castle_core.generators.caddyfile import generate_caddyfile_from_registry
 
-    print("Generating gateway configuration...")
-    _write_generated_files(config)
+    registry = load_registry()
+    caddyfile_path = GENERATED_DIR / "Caddyfile"
+    caddyfile_path.write_text(generate_caddyfile_from_registry(registry))
+    print(f"Generated {caddyfile_path}")
 
-    for name, manifest in config.managed.items():
-        if not manifest.run:
+    for name in config.managed:
+        if name not in registry.deployed:
+            print(f"  {name}: skipped (not in registry, run 'castle deploy')")
             continue
         _service_enable(config, name)
 

cli/src/castle_cli/commands/sync.py

@@ -64,9 +64,9 @@ def run_sync(args: argparse.Namespace) -> int:
                 continue
             print(f"\nInstalling {name}...")
             result = subprocess.run(
-                [uv_path, "tool", "install", "--editable", str(source_dir),
-                 "--force"],
-                capture_output=True, text=True,
+                [uv_path, "tool", "install", "--editable", str(source_dir), "--force"],
+                capture_output=True,
+                text=True,
             )
             if result.returncode != 0:
                 if "already installed" in result.stderr.lower():

cli/src/castle_cli/config.py

@@ -5,6 +5,7 @@ from castle_core.config import (  # noqa: F401 — explicit re-exports for type
     CASTLE_HOME,
     GENERATED_DIR,
     SECRETS_DIR,
+    STATIC_DIR,
     CastleConfig,
     GatewayConfig,
     ensure_dirs,
@@ -13,3 +14,11 @@ from castle_core.config import (  # noqa: F401 — explicit re-exports for type
     resolve_env_vars,
     save_config,
 )
+from castle_core.registry import (  # noqa: F401
+    REGISTRY_PATH,
+    DeployedComponent,
+    NodeConfig,
+    NodeRegistry,
+    load_registry,
+    save_registry,
+)

cli/src/castle_cli/main.py

@@ -14,9 +14,7 @@ def build_parser() -> argparse.ArgumentParser:
         prog="castle",
         description="Castle platform CLI - manage projects, services, and infrastructure",
     )
-    parser.add_argument(
-        "--version", action="version", version=f"castle {__version__}"
-    )
+    parser.add_argument("--version", action="version", version=f"castle {__version__}")
 
     subparsers = parser.add_subparsers(dest="command", help="Available commands")
 
@@ -27,9 +25,7 @@ def build_parser() -> argparse.ArgumentParser:
         choices=["service", "tool", "worker", "job", "frontend", "remote", "containerized"],
         help="Filter by role",
     )
-    list_parser.add_argument(
-        "--json", action="store_true", help="Output as JSON"
-    )
+    list_parser.add_argument("--json", action="store_true", help="Output as JSON")
 
     # castle create
     create_parser = subparsers.add_parser("create", help="Create a new project")
@@ -40,18 +36,12 @@ def build_parser() -> argparse.ArgumentParser:
         required=True,
         help="Project type",
     )
-    create_parser.add_argument(
-        "--description", default="", help="Project description"
-    )
-    create_parser.add_argument(
-        "--port", type=int, help="Port number (services only)"
-    )
+    create_parser.add_argument("--description", default="", help="Project description")
+    create_parser.add_argument("--port", type=int, help="Port number (services only)")
     # castle info
     info_parser = subparsers.add_parser("info", help="Show component details")
     info_parser.add_argument("project", help="Component name")
-    info_parser.add_argument(
-        "--json", action="store_true", help="Output as JSON"
-    )
+    info_parser.add_argument("--json", action="store_true", help="Output as JSON")
 
     # castle test
     test_parser = subparsers.add_parser("test", help="Run tests")
@@ -86,16 +76,12 @@ def build_parser() -> argparse.ArgumentParser:
     enable_parser.add_argument(
         "--dry-run", action="store_true", help="Print generated unit without installing"
     )
-    disable_parser = service_sub.add_parser(
-        "disable", help="Stop and disable a service"
-    )
+    disable_parser = service_sub.add_parser("disable", help="Stop and disable a service")
     disable_parser.add_argument("name", help="Service name")
     service_sub.add_parser("status", help="Show status of all services")
 
     # castle services (plural - manage all services)
-    services_parser = subparsers.add_parser(
-        "services", help="Manage all services together"
-    )
+    services_parser = subparsers.add_parser("services", help="Manage all services together")
     services_sub = services_parser.add_subparsers(dest="services_command")
     services_sub.add_parser("start", help="Start all services and gateway")
     services_sub.add_parser("stop", help="Stop all services and gateway")
@@ -103,9 +89,7 @@ def build_parser() -> argparse.ArgumentParser:
     # castle logs
     logs_parser = subparsers.add_parser("logs", help="View component logs")
     logs_parser.add_argument("name", help="Component name")
-    logs_parser.add_argument(
-        "-f", "--follow", action="store_true", help="Follow log output"
-    )
+    logs_parser.add_argument("-f", "--follow", action="store_true", help="Follow log output")
     logs_parser.add_argument(
         "-n", "--lines", type=int, default=50, help="Number of lines to show (default: 50)"
     )
@@ -117,6 +101,12 @@ def build_parser() -> argparse.ArgumentParser:
         "extra", nargs=argparse.REMAINDER, help="Extra arguments passed to the component"
     )
 
+    # castle deploy
+    deploy_parser = subparsers.add_parser(
+        "deploy", help="Deploy components to ~/.castle/ (spec → runtime)"
+    )
+    deploy_parser.add_argument("component", nargs="?", help="Component to deploy (default: all)")
+
     # castle tool
     tool_parser = subparsers.add_parser("tool", help="Manage tools")
     tool_sub = tool_parser.add_subparsers(dest="tool_command")
@@ -187,6 +177,11 @@ def main() -> int:
 
         return run_logs(args)
 
+    elif args.command == "deploy":
+        from castle_cli.commands.deploy import run_deploy
+
+        return run_deploy(args)
+
     elif args.command == "run":
         from castle_cli.commands.run_cmd import run_run
 

cli/src/castle_cli/manifest.py

@@ -6,6 +6,7 @@ from castle_core.manifest import (  # noqa: F401 — explicit re-exports for typ
     CaddySpec,
     Capability,
     ComponentManifest,
+    DefaultsSpec,
     EnvMap,
     ExposeSpec,
     HttpExposeSpec,

cli/tests/conftest.py

@@ -18,10 +18,12 @@ def castle_root(tmp_path: Path) -> Generator[Path, None, None]:
         "components": {
             "test-svc": {
                 "description": "Test service",
+                "source": "test-svc",
                 "run": {
                     "runner": "python_uv_tool",
                     "tool": "test-svc",
-                    "working_dir": "test-svc",
+                },
+                "defaults": {
                     "env": {"TEST_SVC_DATA_DIR": str(tmp_path / "data" / "test-svc")},
                 },
                 "expose": {

cli/tests/test_create.py

@@ -15,9 +15,10 @@ class TestCreateCommand:
 
     def test_create_service(self, castle_root: Path) -> None:
         """Create a new service project."""
-        with patch("castle_cli.commands.create.load_config") as mock_load, patch(
-            "castle_cli.commands.create.save_config"
-        ) as mock_save:
+        with (
+            patch("castle_cli.commands.create.load_config") as mock_load,
+            patch("castle_cli.commands.create.save_config") as mock_save,
+        ):
             config = load_config(castle_root)
             mock_load.return_value = config
 
@@ -50,17 +51,16 @@ class TestCreateCommand:
 
     def test_create_tool(self, castle_root: Path) -> None:
         """Create a new tool project."""
-        with patch("castle_cli.commands.create.load_config") as mock_load, patch(
-            "castle_cli.commands.create.save_config"
+        with (
+            patch("castle_cli.commands.create.load_config") as mock_load,
+            patch("castle_cli.commands.create.save_config"),
         ):
             config = load_config(castle_root)
             mock_load.return_value = config
 
             from castle_cli.commands.create import run_create
 
-            args = Namespace(
-                name="my-tool", type="tool", description="My tool", port=None
-            )
+            args = Namespace(name="my-tool", type="tool", description="My tool", port=None)
             result = run_create(args)
 
         assert result == 0
@@ -74,17 +74,16 @@ class TestCreateCommand:
 
     def test_create_library(self, castle_root: Path) -> None:
         """Create a new library project."""
-        with patch("castle_cli.commands.create.load_config") as mock_load, patch(
-            "castle_cli.commands.create.save_config"
+        with (
+            patch("castle_cli.commands.create.load_config") as mock_load,
+            patch("castle_cli.commands.create.save_config"),
         ):
             config = load_config(castle_root)
             mock_load.return_value = config
 
             from castle_cli.commands.create import run_create
 
-            args = Namespace(
-                name="my-lib", type="library", description="My library", port=None
-            )
+            args = Namespace(name="my-lib", type="library", description="My library", port=None)
             result = run_create(args)
 
         assert result == 0
@@ -114,8 +113,9 @@ class TestCreateCommand:
 
     def test_create_auto_port(self, castle_root: Path) -> None:
         """Service creation auto-assigns next available port."""
-        with patch("castle_cli.commands.create.load_config") as mock_load, patch(
-            "castle_cli.commands.create.save_config"
+        with (
+            patch("castle_cli.commands.create.load_config") as mock_load,
+            patch("castle_cli.commands.create.save_config"),
         ):
             config = load_config(castle_root)
             mock_load.return_value = config

core/src/castle_core/config.py

@@ -31,6 +31,7 @@ def find_castle_root() -> Path:
 CASTLE_HOME = Path.home() / ".castle"
 GENERATED_DIR = CASTLE_HOME / "generated"
 SECRETS_DIR = CASTLE_HOME / "secrets"
+STATIC_DIR = CASTLE_HOME / "static"
 
 
 @dataclass
@@ -51,23 +52,17 @@ class CastleConfig:
     @property
     def services(self) -> dict[str, ComponentManifest]:
         """Return components with the SERVICE role."""
-        return {
-            k: v for k, v in self.components.items() if Role.SERVICE in v.roles
-        }
+        return {k: v for k, v in self.components.items() if Role.SERVICE in v.roles}
 
     @property
     def tools(self) -> dict[str, ComponentManifest]:
         """Return components with the TOOL role."""
-        return {
-            k: v for k, v in self.components.items() if Role.TOOL in v.roles
-        }
+        return {k: v for k, v in self.components.items() if Role.TOOL in v.roles}
 
     @property
     def workers(self) -> dict[str, ComponentManifest]:
         """Return components with the WORKER role."""
-        return {
-            k: v for k, v in self.components.items() if Role.WORKER in v.roles
-        }
+        return {k: v for k, v in self.components.items() if Role.WORKER in v.roles}
 
     @property
     def managed(self) -> dict[str, ComponentManifest]:
@@ -158,7 +153,16 @@ def _clean_for_yaml(data: object, preserve_keys: set[str] | None = None) -> obje
 
 # Keys whose presence is structurally significant even with all-default values.
 # We serialize these as empty dicts `{}` so they survive a roundtrip.
-_STRUCTURAL_KEYS = {"manage", "systemd", "install", "path", "tool", "expose", "proxy", "caddy"}
+_STRUCTURAL_KEYS = {
+    "manage",
+    "systemd",
+    "install",
+    "path",
+    "tool",
+    "expose",
+    "proxy",
+    "caddy",
+}
 
 
 def _manifest_to_yaml_dict(manifest: ComponentManifest) -> dict:
@@ -214,4 +218,5 @@ def ensure_dirs() -> None:
     CASTLE_HOME.mkdir(parents=True, exist_ok=True)
     GENERATED_DIR.mkdir(parents=True, exist_ok=True)
     SECRETS_DIR.mkdir(parents=True, exist_ok=True)
+    STATIC_DIR.mkdir(parents=True, exist_ok=True)
     os.chmod(SECRETS_DIR, 0o700)

core/src/castle_core/generators/__init__.py

@@ -1,12 +1,17 @@
 """Castle infrastructure generators."""
 
-from castle_core.generators.caddyfile import find_app_dist, generate_caddyfile
+from castle_core.generators.caddyfile import (
+    find_app_dist,
+    generate_caddyfile,
+    generate_caddyfile_from_registry,
+)
 from castle_core.generators.systemd import (
     build_podman_command,
     cron_to_interval_sec,
     cron_to_oncalendar,
     generate_timer,
     generate_unit,
+    generate_unit_from_deployed,
     get_schedule_trigger,
     manifest_to_exec_start,
     timer_name,
@@ -19,8 +24,10 @@ __all__ = [
     "cron_to_oncalendar",
     "find_app_dist",
     "generate_caddyfile",
+    "generate_caddyfile_from_registry",
     "generate_timer",
     "generate_unit",
+    "generate_unit_from_deployed",
     "get_schedule_trigger",
     "manifest_to_exec_start",
     "timer_name",

core/src/castle_core/generators/caddyfile.py

@@ -2,11 +2,12 @@
 
 from __future__ import annotations
 
-from castle_core.config import GENERATED_DIR, CastleConfig
+from castle_core.config import GENERATED_DIR, STATIC_DIR, CastleConfig
+from castle_core.registry import NodeRegistry
 
 
 def find_app_dist(config: CastleConfig) -> str | None:
-    """Find the app dist/ directory if it exists."""
+    """Find the app dist/ directory if it exists (legacy, checks repo)."""
     dist = config.root / "app" / "dist"
     if dist.exists() and (dist / "index.html").exists():
         return str(dist)
@@ -14,12 +15,17 @@ def find_app_dist(config: CastleConfig) -> str | None:
 
 
 def generate_caddyfile(config: CastleConfig) -> str:
-    """Generate Caddyfile content from castle config."""
+    """Generate Caddyfile content from castle config (legacy, uses manifest).
+
+    Prefer generate_caddyfile_from_registry() for registry-based generation.
+    """
     lines = [f":{config.gateway.port} {{"]
 
     # Reverse proxy for each component with proxy.caddy and expose.http
     for name, manifest in config.components.items():
-        if not (manifest.proxy and manifest.proxy.caddy and manifest.proxy.caddy.enable):
+        if not (
+            manifest.proxy and manifest.proxy.caddy and manifest.proxy.caddy.enable
+        ):
             continue
         if not (manifest.expose and manifest.expose.http):
             continue
@@ -53,3 +59,39 @@ def generate_caddyfile(config: CastleConfig) -> str:
 
     lines.append("}")
     return "\n".join(lines)
+
+
+def generate_caddyfile_from_registry(registry: NodeRegistry) -> str:
+    """Generate Caddyfile from the node registry.
+
+    Static files served from ~/.castle/static/castle-app/.
+    No repo-relative paths.
+    """
+    lines = [f":{registry.node.gateway_port} {{"]
+
+    for name, deployed in registry.deployed.items():
+        if not deployed.proxy_path or not deployed.port:
+            continue
+
+        lines.append(f"    handle_path {deployed.proxy_path}/* {{")
+        lines.append(f"        reverse_proxy localhost:{deployed.port}")
+        lines.append("    }")
+        lines.append("")
+
+    # SPA from static dir
+    static_app = STATIC_DIR / "castle-app"
+    if (static_app / "index.html").exists():
+        lines.append("    handle {")
+        lines.append(f"        root * {static_app}")
+        lines.append("        try_files {path} /index.html")
+        lines.append("        file_server")
+        lines.append("    }")
+    else:
+        fallback = GENERATED_DIR / "app"
+        lines.append("    handle / {")
+        lines.append(f"        root * {fallback}")
+        lines.append("        file_server")
+        lines.append("    }")
+
+    lines.append("}")
+    return "\n".join(lines)

core/src/castle_core/generators/systemd.py

@@ -6,7 +6,8 @@ import shutil
 from pathlib import Path
 
 from castle_core.config import CastleConfig, resolve_env_vars
-from castle_core.manifest import ComponentManifest, RestartPolicy
+from castle_core.manifest import ComponentManifest, RestartPolicy, SystemdSpec
+from castle_core.registry import DeployedComponent
 
 SYSTEMD_USER_DIR = Path.home() / ".config" / "systemd" / "user"
 UNIT_PREFIX = "castle-"
@@ -42,7 +43,13 @@ def cron_to_oncalendar(cron: str) -> str:
     minute, hour, dom, month, dow = parts
 
     # */N minutes → run every N minutes
-    if minute.startswith("*/") and hour == "*" and dom == "*" and month == "*" and dow == "*":
+    if (
+        minute.startswith("*/")
+        and hour == "*"
+        and dom == "*"
+        and month == "*"
+        and dow == "*"
+    ):
         return ""  # Use OnUnitActiveSec instead
 
     # Specific time daily: "0 2 * * *" → "*-*-* 02:00:00"
@@ -60,7 +67,13 @@ def cron_to_interval_sec(cron: str) -> int | None:
     if len(parts) != 5:
         return None
     minute, hour, dom, month, dow = parts
-    if minute.startswith("*/") and hour == "*" and dom == "*" and month == "*" and dow == "*":
+    if (
+        minute.startswith("*/")
+        and hour == "*"
+        and dom == "*"
+        and month == "*"
+        and dow == "*"
+    ):
         try:
             return int(minute[2:]) * 60
         except ValueError:
@@ -132,15 +145,19 @@ def build_podman_command(manifest: ComponentManifest) -> str:
 
 
 def generate_unit(config: CastleConfig, name: str, manifest: ComponentManifest) -> str:
-    """Generate a systemd user unit file for a component."""
+    """Generate a systemd user unit file for a component (legacy, uses manifest).
+
+    Prefer generate_unit_from_deployed() for registry-based generation.
+    """
     run = manifest.run
     if run is None:
         raise ValueError(f"Component '{name}' has no run spec")
 
-    working_dir = config.root / (run.working_dir or name)
     exec_start = manifest_to_exec_start(manifest, config.root)
 
-    resolved_env = resolve_env_vars(run.env, manifest)
+    # Env vars now come from manifest.defaults.env instead of run.env
+    raw_env = manifest.defaults.env if manifest.defaults else {}
+    resolved_env = resolve_env_vars(raw_env, manifest)
     env_lines = ""
     for key, value in resolved_env.items():
         env_lines += f"Environment={key}={value}\n"
@@ -166,7 +183,6 @@ After={after}
 
 [Service]
 Type=oneshot
-WorkingDirectory={working_dir}
 ExecStart={exec_start}
 {env_lines}"""
     else:
@@ -178,7 +194,68 @@ After={after}
 
 [Service]
 Type=simple
-WorkingDirectory={working_dir}
+ExecStart={exec_start}
+{env_lines}Restart={restart}
+RestartSec={restart_sec}
+SuccessExitStatus=143
+"""
+
+    if sd and sd.exec_reload:
+        reload_argv = sd.exec_reload.split()
+        resolved_reload = shutil.which(reload_argv[0])
+        if resolved_reload:
+            reload_argv[0] = resolved_reload
+        unit += f"ExecReload={' '.join(reload_argv)}\n"
+
+    if sd and sd.no_new_privileges:
+        unit += "NoNewPrivileges=true\n"
+
+    unit += f"""
+[Install]
+WantedBy={wanted_by}
+"""
+    return unit
+
+
+def generate_unit_from_deployed(
+    name: str,
+    deployed: DeployedComponent,
+    systemd_spec: SystemdSpec | None = None,
+) -> str:
+    """Generate a systemd unit from a deployed component (registry-based).
+
+    No repo-relative paths — uses only resolved run_cmd and env from the registry.
+    """
+    exec_start = " ".join(deployed.run_cmd)
+
+    env_lines = ""
+    for key, value in deployed.env.items():
+        env_lines += f"Environment={key}={value}\n"
+    env_lines += f'Environment="PATH={Path.home() / ".local/bin"}:/usr/local/bin:/usr/bin:/bin"\n'
+
+    sd = systemd_spec
+    description = deployed.description or name
+    after = " ".join(sd.after) if sd and sd.after else "network.target"
+    wanted_by = " ".join(sd.wanted_by) if sd else "default.target"
+
+    if deployed.schedule:
+        unit = f"""[Unit]
+Description=Castle: {description}
+After={after}
+
+[Service]
+Type=oneshot
+ExecStart={exec_start}
+{env_lines}"""
+    else:
+        restart = (sd.restart if sd else RestartPolicy.ON_FAILURE).value
+        restart_sec = sd.restart_sec if sd else 5
+        unit = f"""[Unit]
+Description=Castle: {description}
+After={after}
+
+[Service]
+Type=simple
 ExecStart={exec_start}
 {env_lines}Restart={restart}
 RestartSec={restart_sec}

core/src/castle_core/manifest.py

@@ -39,8 +39,6 @@ class Role(str, Enum):
 
 class RunBase(BaseModel):
     runner: str
-    working_dir: str | None = None
-    env: EnvMap = Field(default_factory=dict)
 
 
 class RunCommand(RunBase):
@@ -68,6 +66,7 @@ class RunContainer(RunBase):
     args: list[str] = Field(default_factory=list)
     ports: dict[int, int] = Field(default_factory=dict)
     volumes: list[str] = Field(default_factory=list)
+    env: EnvMap = Field(default_factory=dict)
     workdir: str | None = None
 
 
@@ -85,7 +84,9 @@ class RunRemote(RunBase):
 
 
 RunSpec = Annotated[
-    Union[RunCommand, RunPythonModule, RunPythonUvTool, RunContainer, RunNode, RunRemote],
+    Union[
+        RunCommand, RunPythonModule, RunPythonUvTool, RunContainer, RunNode, RunRemote
+    ],
     Field(discriminator="runner"),
 ]
 
@@ -221,7 +222,6 @@ class ProxySpec(BaseModel):
 
 
 class BuildSpec(BaseModel):
-    working_dir: str | None = None
     commands: list[list[str]] = Field(default_factory=list)
     outputs: list[str] = Field(default_factory=list)
 
@@ -237,6 +237,15 @@ class Capability(BaseModel):
     meta: dict[str, str] = Field(default_factory=dict)
 
 
+# ---------------------
+# Defaults
+# ---------------------
+
+
+class DefaultsSpec(BaseModel):
+    env: EnvMap = Field(default_factory=dict)
+
+
 # ---------------------
 # Component manifest
 # ---------------------
@@ -247,6 +256,8 @@ class ComponentManifest(BaseModel):
     name: str | None = None
     description: str | None = None
 
+    source: str | None = None
+
     run: RunSpec | None = None
 
     triggers: list[TriggerSpec] = Field(default_factory=list)
@@ -258,6 +269,8 @@ class ComponentManifest(BaseModel):
     proxy: ProxySpec | None = None
     build: BuildSpec | None = None
 
+    defaults: DefaultsSpec | None = None
+
     provides: list[Capability] = Field(default_factory=list)
     consumes: list[Capability] = Field(default_factory=list)
 
@@ -306,13 +319,11 @@ class ComponentManifest(BaseModel):
     def source_dir(self) -> str | None:
         """Best-effort relative directory for this component's source.
 
-        Resolution order: run.working_dir → build.working_dir → tool.source (strip trailing /).
+        Resolution order: source → tool.source (strip trailing /).
         Returns None if no directory can be determined.
         """
-        if self.run and self.run.working_dir:
-            return self.run.working_dir
-        if self.build and self.build.working_dir:
-            return self.build.working_dir
+        if self.source:
+            return self.source.rstrip("/")
         if self.tool and self.tool.source:
             return self.tool.source.rstrip("/")
         return None

core/src/castle_core/registry.py

@@ -0,0 +1,138 @@
+"""Node registry — per-machine deployment state."""
+
+from __future__ import annotations
+
+import socket
+from dataclasses import dataclass, field
+from pathlib import Path
+
+import yaml
+
+from castle_core.config import CASTLE_HOME
+
+REGISTRY_PATH = CASTLE_HOME / "registry.yaml"
+STATIC_DIR = CASTLE_HOME / "static"
+
+
+@dataclass
+class NodeConfig:
+    """Per-node identity and settings."""
+
+    hostname: str = ""
+    castle_root: str | None = None  # repo path, for dev commands
+    gateway_port: int = 9000
+
+    def __post_init__(self) -> None:
+        if not self.hostname:
+            self.hostname = socket.gethostname()
+
+
+@dataclass
+class DeployedComponent:
+    """A component deployed on this node with resolved runtime config."""
+
+    runner: str
+    run_cmd: list[str]
+    env: dict[str, str] = field(default_factory=dict)
+    description: str | None = None
+    roles: list[str] = field(default_factory=list)
+    port: int | None = None
+    health_path: str | None = None
+    proxy_path: str | None = None
+    schedule: str | None = None
+    managed: bool = False
+
+
+@dataclass
+class NodeRegistry:
+    """What's deployed on this node."""
+
+    node: NodeConfig
+    deployed: dict[str, DeployedComponent] = field(default_factory=dict)
+
+
+def load_registry(path: Path | None = None) -> NodeRegistry:
+    """Load the node registry from ~/.castle/registry.yaml."""
+    if path is None:
+        path = REGISTRY_PATH
+
+    if not path.exists():
+        raise FileNotFoundError(
+            f"Registry not found: {path}\n"
+            "Run 'castle deploy' to generate it from castle.yaml."
+        )
+
+    with open(path) as f:
+        data = yaml.safe_load(f)
+
+    if not data:
+        raise ValueError(f"Empty registry: {path}")
+
+    node_data = data.get("node", {})
+    node = NodeConfig(
+        hostname=node_data.get("hostname", ""),
+        castle_root=node_data.get("castle_root"),
+        gateway_port=node_data.get("gateway_port", 9000),
+    )
+
+    deployed: dict[str, DeployedComponent] = {}
+    for name, comp_data in data.get("deployed", {}).items():
+        deployed[name] = DeployedComponent(
+            runner=comp_data.get("runner", "command"),
+            run_cmd=comp_data.get("run_cmd", []),
+            env=comp_data.get("env", {}),
+            description=comp_data.get("description"),
+            roles=comp_data.get("roles", []),
+            port=comp_data.get("port"),
+            health_path=comp_data.get("health_path"),
+            proxy_path=comp_data.get("proxy_path"),
+            schedule=comp_data.get("schedule"),
+            managed=comp_data.get("managed", False),
+        )
+
+    return NodeRegistry(node=node, deployed=deployed)
+
+
+def save_registry(registry: NodeRegistry, path: Path | None = None) -> None:
+    """Write the node registry to ~/.castle/registry.yaml."""
+    if path is None:
+        path = REGISTRY_PATH
+
+    path.parent.mkdir(parents=True, exist_ok=True)
+
+    data: dict = {
+        "node": {
+            "hostname": registry.node.hostname,
+            "gateway_port": registry.node.gateway_port,
+        },
+        "deployed": {},
+    }
+
+    if registry.node.castle_root:
+        data["node"]["castle_root"] = registry.node.castle_root
+
+    for name, comp in registry.deployed.items():
+        entry: dict = {
+            "runner": comp.runner,
+            "run_cmd": comp.run_cmd,
+        }
+        if comp.env:
+            entry["env"] = comp.env
+        if comp.description:
+            entry["description"] = comp.description
+        if comp.roles:
+            entry["roles"] = comp.roles
+        if comp.port is not None:
+            entry["port"] = comp.port
+        if comp.health_path:
+            entry["health_path"] = comp.health_path
+        if comp.proxy_path:
+            entry["proxy_path"] = comp.proxy_path
+        if comp.schedule:
+            entry["schedule"] = comp.schedule
+        if comp.managed:
+            entry["managed"] = comp.managed
+        data["deployed"][name] = entry
+
+    with open(path, "w") as f:
+        yaml.dump(data, f, default_flow_style=False, sort_keys=False)

core/tests/conftest.py

@@ -18,10 +18,12 @@ def castle_root(tmp_path: Path) -> Generator[Path, None, None]:
         "components": {
             "test-svc": {
                 "description": "Test service",
+                "source": "test-svc",
                 "run": {
                     "runner": "python_uv_tool",
                     "tool": "test-svc",
-                    "working_dir": "test-svc",
+                },
+                "defaults": {
                     "env": {"TEST_SVC_DATA_DIR": str(tmp_path / "data" / "test-svc")},
                 },
                 "expose": {

core/tests/test_config.py

@@ -62,7 +62,7 @@ class TestLoadConfig:
         svc = config.components["test-svc"]
         assert svc.run.runner == "python_uv_tool"
         assert svc.run.tool == "test-svc"
-        assert svc.run.working_dir == "test-svc"
+        assert svc.source == "test-svc"
 
     def test_tool_no_run(self, castle_root: Path) -> None:
         """Tool without run block has no run spec."""

core/tests/test_manifest.py

@@ -33,9 +33,7 @@ class TestRoleDerivation:
         m = ComponentManifest(
             id="svc",
             run=RunPythonUvTool(runner="python_uv_tool", tool="svc"),
-            expose=ExposeSpec(
-                http=HttpExposeSpec(internal=HttpInternal(port=8000))
-            ),
+            expose=ExposeSpec(http=HttpExposeSpec(internal=HttpInternal(port=8000))),
         )
         assert Role.SERVICE in m.roles
 
@@ -117,9 +115,7 @@ class TestRoleDerivation:
         m = ComponentManifest(
             id="multi",
             run=RunPythonUvTool(runner="python_uv_tool", tool="multi"),
-            expose=ExposeSpec(
-                http=HttpExposeSpec(internal=HttpInternal(port=8000))
-            ),
+            expose=ExposeSpec(http=HttpExposeSpec(internal=HttpInternal(port=8000))),
             install=InstallSpec(path=PathInstallSpec(alias="multi")),
         )
         assert Role.SERVICE in m.roles
@@ -130,9 +126,7 @@ class TestRoleDerivation:
         m = ComponentManifest(
             id="svc",
             run=RunPythonUvTool(runner="python_uv_tool", tool="svc"),
-            expose=ExposeSpec(
-                http=HttpExposeSpec(internal=HttpInternal(port=8000))
-            ),
+            expose=ExposeSpec(http=HttpExposeSpec(internal=HttpInternal(port=8000))),
             manage=ManageSpec(systemd=SystemdSpec()),
         )
         assert Role.SERVICE in m.roles
@@ -144,7 +138,9 @@ class TestConsistencyValidation:
 
     def test_remote_with_systemd_raises(self) -> None:
         """Remote runner + systemd management is invalid."""
-        with pytest.raises(ValueError, match="manage.systemd cannot be enabled for runner=remote"):
+        with pytest.raises(
+            ValueError, match="manage.systemd cannot be enabled for runner=remote"
+        ):
             ComponentManifest(
                 id="bad",
                 run=RunRemote(runner="remote", base_url="http://example.com"),

core/tests/test_systemd.py

@@ -5,7 +5,12 @@ from __future__ import annotations
 from pathlib import Path
 
 from castle_core.config import load_config
-from castle_core.generators.systemd import generate_unit, unit_name
+from castle_core.generators.systemd import (
+    generate_unit,
+    generate_unit_from_deployed,
+    unit_name,
+)
+from castle_core.registry import DeployedComponent
 
 
 class TestUnitName:
@@ -27,15 +32,15 @@ class TestUnitGeneration:
         unit = generate_unit(config, "test-svc", manifest)
         assert "Description=Castle: Test service" in unit
 
-    def test_contains_working_dir(self, castle_root: Path) -> None:
-        """Unit file has correct working directory."""
+    def test_no_working_directory(self, castle_root: Path) -> None:
+        """Unit file has no WorkingDirectory (source/runtime separation)."""
         config = load_config(castle_root)
         manifest = config.components["test-svc"]
         unit = generate_unit(config, "test-svc", manifest)
-        assert f"WorkingDirectory={castle_root / 'test-svc'}" in unit
+        assert "WorkingDirectory" not in unit
 
     def test_contains_environment(self, castle_root: Path) -> None:
-        """Unit file has environment variables."""
+        """Unit file has environment variables from defaults.env."""
         config = load_config(castle_root)
         manifest = config.components["test-svc"]
         unit = generate_unit(config, "test-svc", manifest)
@@ -55,3 +60,47 @@ class TestUnitGeneration:
         manifest = config.components["test-svc"]
         unit = generate_unit(config, "test-svc", manifest)
         assert "run test-svc" in unit
+
+
+class TestUnitFromDeployed:
+    """Tests for registry-based systemd unit generation."""
+
+    def test_basic_service(self) -> None:
+        """Generate a unit from a deployed component."""
+        deployed = DeployedComponent(
+            runner="python_uv_tool",
+            run_cmd=["/home/user/.local/bin/uv", "run", "my-svc"],
+            env={"MY_SVC_PORT": "9001", "MY_SVC_DATA_DIR": "/data/castle/my-svc"},
+            description="My service",
+        )
+        unit = generate_unit_from_deployed("my-svc", deployed)
+        assert "Description=Castle: My service" in unit
+        assert "ExecStart=/home/user/.local/bin/uv run my-svc" in unit
+        assert "Environment=MY_SVC_PORT=9001" in unit
+        assert "Environment=MY_SVC_DATA_DIR=/data/castle/my-svc" in unit
+        assert "WorkingDirectory" not in unit
+        assert "Restart=on-failure" in unit
+
+    def test_scheduled_job(self) -> None:
+        """Scheduled component generates oneshot unit."""
+        deployed = DeployedComponent(
+            runner="command",
+            run_cmd=["/home/user/.local/bin/my-job"],
+            env={},
+            description="Nightly job",
+            schedule="0 2 * * *",
+        )
+        unit = generate_unit_from_deployed("my-job", deployed)
+        assert "Type=oneshot" in unit
+        assert "Restart=" not in unit
+
+    def test_no_repo_paths(self) -> None:
+        """Generated units must not reference repo paths."""
+        deployed = DeployedComponent(
+            runner="python_uv_tool",
+            run_cmd=["/home/user/.local/bin/uv", "run", "my-svc"],
+            env={"DATA_DIR": "/data/castle/my-svc"},
+            description="Test",
+        )
+        unit = generate_unit_from_deployed("my-svc", deployed)
+        assert "/data/repos/" not in unit