feat: Implement secret management for environment variables and enhance systemd unit generation

core/src/castle_core/config.py

@@ -150,10 +150,16 @@ class CastleConfig:
         }
 
 
-def resolve_env_vars(
+def resolve_env_split(
     env: dict[str, str], context: dict[str, str] | None = None
-) -> dict[str, str]:
-    """Resolve placeholders in env values.
+) -> tuple[dict[str, str], dict[str, str]]:
+    """Resolve placeholders, splitting secret-bearing vars from plain ones.
+
+    Returns ``(plain, secret)``. A var is *secret-bearing* if its raw value
+    contained a ``${secret:...}`` reference — including composite values like
+    ``neo4j/${secret:NEO4J_PASSWORD}``. Both dicts hold fully-resolved values;
+    partitioning lets callers keep secrets out of unit files and process argv
+    (routing them through a mode-0600 env file) while inlining the rest.
 
     - ``${secret:NAME}`` reads `~/.castle/secrets/NAME`.
     - ``${port}`` / ``${data_dir}`` / ``${name}`` (and anything else in
@@ -162,7 +168,8 @@ def resolve_env_vars(
       hardcoding or castle silently injecting a guessed var name.
     """
     context = context or {}
-    resolved = {}
+    plain: dict[str, str] = {}
+    secret: dict[str, str] = {}
     for key, value in env.items():
 
         def replace_var(match: re.Match[str]) -> str:
@@ -173,8 +180,25 @@ def resolve_env_vars(
                 return context[ref]
             return match.group(0)
 
-        resolved[key] = re.sub(r"\$\{([^}]+)\}", replace_var, value)
-    return resolved
+        resolved = re.sub(r"\$\{([^}]+)\}", replace_var, value)
+        if re.search(r"\$\{secret:[^}]+\}", value):
+            secret[key] = resolved
+        else:
+            plain[key] = resolved
+    return plain, secret
+
+
+def resolve_env_vars(
+    env: dict[str, str], context: dict[str, str] | None = None
+) -> dict[str, str]:
+    """Resolve placeholders in env values (secrets included), preserving order.
+
+    Convenience wrapper over :func:`resolve_env_split` for callers that want a
+    single flat dict. Prefer ``resolve_env_split`` when secrets must be kept out
+    of generated artifacts.
+    """
+    plain, secret = resolve_env_split(env, context)
+    return {k: secret[k] if k in secret else plain[k] for k in env}
 
 
 def _read_secret(name: str) -> str:
@@ -513,3 +537,8 @@ def ensure_dirs() -> None:
     DATA_DIR.mkdir(parents=True, exist_ok=True)
     SECRETS_DIR.mkdir(parents=True, exist_ok=True)
     os.chmod(SECRETS_DIR, 0o700)
+    # Generated per-deployment secret env files (EnvironmentFile= / --env-file)
+    # live here, kept out of unit files and process argv.
+    secret_env_dir = SECRETS_DIR / "env"
+    secret_env_dir.mkdir(parents=True, exist_ok=True)
+    os.chmod(secret_env_dir, 0o700)

core/src/castle_core/deploy.py

@@ -8,6 +8,7 @@ copies frontend build outputs.
 
 from __future__ import annotations
 
+import os
 import shutil
 import subprocess
 from dataclasses import dataclass, field
@@ -19,13 +20,16 @@ from castle_core.config import (
     CastleConfig,
     ensure_dirs,
     load_config,
-    resolve_env_vars,
+    resolve_env_split,
 )
 from castle_core.generators.caddyfile import generate_caddyfile_from_registry
 from castle_core.generators.systemd import (
+    SECRET_ENV_DIR,
     generate_timer,
     generate_unit_from_deployed,
+    secret_env_path,
     timer_name,
+    unit_env_file,
     unit_name,
 )
 from castle_core.manifest import JobSpec, ServiceSpec
@@ -143,9 +147,13 @@ def _reload_gateway(messages: list[str]) -> None:
         text=True,
     )
     if active.stdout.strip() != "active":
-        messages.append("Gateway not running — skipped reload (start it with 'castle gateway start').")
+        messages.append(
+            "Gateway not running — skipped reload (start it with 'castle gateway start')."
+        )
         return
-    result = subprocess.run(["systemctl", "--user", "reload", gw_unit], capture_output=True, text=True)
+    result = subprocess.run(
+        ["systemctl", "--user", "reload", gw_unit], capture_output=True, text=True
+    )
     if result.returncode == 0:
         messages.append("Gateway reloaded.")
     else:
@@ -165,7 +173,32 @@ def _env_context(name: str, config_key: str, port: int | None) -> dict[str, str]
     return ctx
 
 
-def _resolve_description(config: CastleConfig, spec: ServiceSpec | JobSpec) -> str | None:
+def _write_secret_env_file(name: str, secret_env: dict[str, str]) -> Path | None:
+    """Write a deployment's resolved secrets to a mode-0600 env file.
+
+    Keeps secrets out of the generated unit file and the process argv: systemd
+    loads it via ``EnvironmentFile=`` and docker via ``--env-file``. Returns the
+    path, or ``None`` (after removing any stale file) when there are no secrets,
+    so a service that drops its last secret doesn't leave a dangling file.
+    """
+    path = secret_env_path(name)
+    if not secret_env:
+        path.unlink(missing_ok=True)
+        return None
+    SECRET_ENV_DIR.mkdir(parents=True, exist_ok=True)
+    os.chmod(SECRET_ENV_DIR, 0o700)
+    # O_TRUNC keeps a pre-existing file's mode, so chmod explicitly afterwards.
+    fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
+    with os.fdopen(fd, "w") as f:
+        for key, value in secret_env.items():
+            f.write(f"{key}={value}\n")
+    os.chmod(path, 0o600)
+    return path
+
+
+def _resolve_description(
+    config: CastleConfig, spec: ServiceSpec | JobSpec
+) -> str | None:
     """Get description, falling through to program if referenced."""
     if spec.description:
         return spec.description
@@ -196,18 +229,26 @@ def _build_deployed_service(
 
     # Env is exactly what's declared in defaults.env — no hidden convention
     # injection. ${port}/${data_dir}/${name} let the program's own env var names
-    # map to castle's computed values without hardcoding them.
-    env = dict(svc.defaults.env) if (svc.defaults and svc.defaults.env) else {}
-    env = resolve_env_vars(env, _env_context(name, config_key, port))
+    # map to castle's computed values without hardcoding them. Secret-bearing vars
+    # are split out so they never land in the unit file or process argv — they're
+    # written to a mode-0600 env file referenced via EnvironmentFile=/--env-file.
+    raw_env = dict(svc.defaults.env) if (svc.defaults and svc.defaults.env) else {}
+    env, secret_env = resolve_env_split(raw_env, _env_context(name, config_key, port))
+    secret_env_file = _write_secret_env_file(name, secret_env)
 
     # `command`-runner services resolve a tool on PATH → ensure it's installed.
     # `python`-runner services run in place via `uv run` (below) — no tool venv.
     if run.runner == "command":
         _ensure_python_tool(config, svc.program, messages)
 
-    # Build run_cmd
+    # Build run_cmd (container runners get --env-file for the secrets).
     run_cmd = _build_run_cmd(
-        name, run, env, messages, _program_source_dir(config, svc.program)
+        name,
+        run,
+        env,
+        messages,
+        _program_source_dir(config, svc.program),
+        secret_env_file=secret_env_file,
     )
 
     # Proxy: a path prefix (handle_path on the gateway) and/or a hostname (a
@@ -235,6 +276,7 @@ def _build_deployed_service(
         runner=run.runner,
         run_cmd=run_cmd,
         env=env,
+        secret_env_keys=sorted(secret_env),
         description=_resolve_description(config, svc),
         behavior="daemon",
         stack=stack,
@@ -255,12 +297,18 @@ def _build_deployed_job(
     # ${data_dir} is keyed by the program the job runs, not the job name — see
     # _build_deployed_service. Falls back to the job name.
     config_key = job.program or name
-    env = dict(job.defaults.env) if (job.defaults and job.defaults.env) else {}
-    env = resolve_env_vars(env, _env_context(name, config_key, None))
+    raw_env = dict(job.defaults.env) if (job.defaults and job.defaults.env) else {}
+    env, secret_env = resolve_env_split(raw_env, _env_context(name, config_key, None))
+    secret_env_file = _write_secret_env_file(name, secret_env)
     if run.runner == "command":
         _ensure_python_tool(config, job.program, messages)
     run_cmd = _build_run_cmd(
-        name, run, env, messages, _program_source_dir(config, job.program)
+        name,
+        run,
+        env,
+        messages,
+        _program_source_dir(config, job.program),
+        secret_env_file=secret_env_file,
     )
 
     stack = None
@@ -271,6 +319,7 @@ def _build_deployed_job(
         runner=run.runner,
         run_cmd=run_cmd,
         env=env,
+        secret_env_keys=sorted(secret_env),
         description=_resolve_description(config, job),
         behavior="tool",
         stack=stack,
@@ -340,7 +389,9 @@ def _ensure_python_tool(
         text=True,
     )
     if result.returncode != 0:
-        messages.append(f"Error: {program} install failed:\n{result.stdout}{result.stderr}")
+        messages.append(
+            f"Error: {program} install failed:\n{result.stdout}{result.stderr}"
+        )
     else:
         messages.append(f"Installed {program}")
 
@@ -351,8 +402,16 @@ def _build_run_cmd(
     env: dict[str, str],
     messages: list[str],
     source_dir: Path | None = None,
+    secret_env_file: Path | None = None,
 ) -> list[str]:
-    """Build a run command list from a RunSpec."""
+    """Build a run command list from a RunSpec.
+
+    ``env`` holds plain (non-secret) vars only; ``secret_env_file`` is the
+    mode-0600 file holding the deployment's secrets. For container runners the
+    secrets are passed via ``--env-file`` (keeping them out of the argv);
+    systemd-launched runners get them via ``EnvironmentFile=`` on the unit, so
+    ``secret_env_file`` is unused here for those.
+    """
     match run.runner:  # type: ignore[union-attr]
         case "python":
             # Run the program in place from its own project venv via `uv run`,
@@ -392,8 +451,11 @@ def _build_run_cmd(
                 cmd.extend(["-v", vol])
             for key, val in run.env.items():  # type: ignore[union-attr]
                 cmd.extend(["-e", f"{key}={val}"])
+            # env is plain-only; secrets go via --env-file so they never hit argv.
             for key, val in env.items():
                 cmd.extend(["-e", f"{key}={val}"])
+            if secret_env_file is not None:
+                cmd.extend(["--env-file", str(secret_env_file)])
             if run.workdir:  # type: ignore[union-attr]
                 cmd.extend(["-w", run.workdir])  # type: ignore[union-attr]
             cmd.append(run.image)  # type: ignore[union-attr]
@@ -446,6 +508,9 @@ def _teardown_unit(unit_file: str, messages: list[str]) -> None:
     subprocess.run(["systemctl", "--user", "disable", unit_file], check=False)
     path.unlink()
     messages.append(f"Pruned orphan unit: {unit_file}")
+    # Remove the matching secret env file (only services have one; not timers).
+    if unit_file.endswith(".service"):
+        (SECRET_ENV_DIR / f"{unit_file}.env").unlink(missing_ok=True)
 
 
 def _prune_orphans(registry: NodeRegistry, messages: list[str]) -> None:
@@ -484,7 +549,9 @@ def _generate_systemd_units(config: CastleConfig, registry: NodeRegistry) -> Non
                 systemd_spec = job.manage.systemd
 
         svc_name = unit_name(name)
-        svc_content = generate_unit_from_deployed(name, deployed, systemd_spec)
+        svc_content = generate_unit_from_deployed(
+            name, deployed, systemd_spec, env_file=unit_env_file(deployed, name)
+        )
         (SYSTEMD_USER_DIR / svc_name).write_text(svc_content)
 
         if deployed.schedule:
@@ -494,4 +561,4 @@ def _generate_systemd_units(config: CastleConfig, registry: NodeRegistry) -> Non
                 description=deployed.description,
             )
             tmr_name = timer_name(name)
-            (SYSTEMD_USER_DIR / tmr_name).write_text(timer_content)
+            (SYSTEMD_USER_DIR / tmr_name).write_text(timer_content)

core/src/castle_core/generators/systemd.py

@@ -5,19 +5,40 @@ from __future__ import annotations
 import shutil
 from pathlib import Path
 
-from castle_core.config import USER_TOOL_PATH_DIRS
+from castle_core.config import SECRETS_DIR, USER_TOOL_PATH_DIRS
 from castle_core.manifest import RestartPolicy, SystemdSpec
 from castle_core.registry import Deployment
 
 SYSTEMD_USER_DIR = Path.home() / ".config" / "systemd" / "user"
 UNIT_PREFIX = "castle-"
 
+# Generated mode-0600 env files holding a deployment's resolved secrets, kept out
+# of the unit file and the process argv (loaded via EnvironmentFile= / --env-file).
+SECRET_ENV_DIR = SECRETS_DIR / "env"
+
 
 def unit_name(service_name: str) -> str:
     """Get the systemd unit name for a service."""
     return f"{UNIT_PREFIX}{service_name}.service"
 
 
+def secret_env_path(service_name: str) -> Path:
+    """Path to a deployment's generated secret env file (1:1 with its unit name)."""
+    return SECRET_ENV_DIR / f"{unit_name(service_name)}.env"
+
+
+def unit_env_file(deployed: Deployment, name: str) -> Path | None:
+    """The ``EnvironmentFile=`` path for a systemd-launched runner, or None.
+
+    Container runners load secrets via docker ``--env-file`` (baked into run_cmd),
+    so systemd must not also read them — return None there. Only deployments that
+    actually have secrets get a file.
+    """
+    if deployed.runner == "container" or not deployed.secret_env_keys:
+        return None
+    return secret_env_path(name)
+
+
 def timer_name(service_name: str) -> str:
     """Get the systemd timer name for a scheduled service."""
     return f"{UNIT_PREFIX}{service_name}.timer"
@@ -77,10 +98,14 @@ def generate_unit_from_deployed(
     name: str,
     deployed: Deployment,
     systemd_spec: SystemdSpec | None = None,
+    env_file: Path | 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.
+    Secrets are never inlined as ``Environment=`` lines: ``env_file`` (when set)
+    is loaded via ``EnvironmentFile=`` so the values stay out of the unit. The
+    path is referenced fail-loud (no ``-`` prefix): a missing file blocks start.
     """
     exec_start = " ".join(deployed.run_cmd)
 
@@ -89,6 +114,8 @@ def generate_unit_from_deployed(
         env_lines += f"Environment={key}={value}\n"
     tool_path = ":".join(str(d) for d in USER_TOOL_PATH_DIRS if d.exists())
     env_lines += f'Environment="PATH={tool_path}:/usr/local/bin:/usr/bin:/bin"\n'
+    if env_file is not None:
+        env_lines += f"EnvironmentFile={env_file}\n"
 
     sd = systemd_spec
     description = deployed.description or name

core/src/castle_core/lifecycle.py

@@ -20,7 +20,9 @@ from castle_core.generators.systemd import (
     SYSTEMD_USER_DIR,
     generate_timer,
     generate_unit_from_deployed,
+    secret_env_path,
     timer_name,
+    unit_env_file,
     unit_name,
 )
 from castle_core.registry import REGISTRY_PATH, load_registry
@@ -86,13 +88,19 @@ def is_active(name: str, config: CastleConfig) -> bool:
 def enable_service(name: str, config: CastleConfig) -> ActionResult:
     """Generate+install the unit (and timer) from the registry, enable and start it."""
     if not REGISTRY_PATH.exists():
-        return ActionResult(name, "activate", "error", "No registry. Run 'castle deploy' first.")
+        return ActionResult(
+            name, "activate", "error", "No registry. Run 'castle deploy' first."
+        )
     registry = load_registry()
     if name not in registry.deployed:
-        return ActionResult(name, "activate", "error", f"'{name}' not in registry; run 'castle deploy'.")
+        return ActionResult(
+            name, "activate", "error", f"'{name}' not in registry; run 'castle deploy'."
+        )
     deployed = registry.deployed[name]
     if not deployed.managed:
-        return ActionResult(name, "activate", "error", f"'{name}' is not a managed service.")
+        return ActionResult(
+            name, "activate", "error", f"'{name}' is not a managed service."
+        )
 
     systemd_spec = None
     if name in config.services and config.services[name].manage:
@@ -102,12 +110,18 @@ def enable_service(name: str, config: CastleConfig) -> ActionResult:
 
     SYSTEMD_USER_DIR.mkdir(parents=True, exist_ok=True)
     svc_unit = unit_name(name)
-    (SYSTEMD_USER_DIR / svc_unit).write_text(generate_unit_from_deployed(name, deployed, systemd_spec))
+    (SYSTEMD_USER_DIR / svc_unit).write_text(
+        generate_unit_from_deployed(
+            name, deployed, systemd_spec, env_file=unit_env_file(deployed, name)
+        )
+    )
     primary = svc_unit
     if deployed.schedule:
         tmr_unit = timer_name(name)
         (SYSTEMD_USER_DIR / tmr_unit).write_text(
-            generate_timer(name, schedule=deployed.schedule, description=deployed.description)
+            generate_timer(
+                name, schedule=deployed.schedule, description=deployed.description
+            )
         )
         primary = tmr_unit
 
@@ -115,8 +129,9 @@ def enable_service(name: str, config: CastleConfig) -> ActionResult:
     subprocess.run(["systemctl", "--user", "enable", primary], check=False)
     subprocess.run(["systemctl", "--user", "start", primary], check=False)
     status = "active" if _systemctl_active(primary) else "inactive"
-    return ActionResult(name, "activate", "ok" if status == "active" else "error",
-                        f"{name}: {status}")
+    return ActionResult(
+        name, "activate", "ok" if status == "active" else "error", f"{name}: {status}"
+    )
 
 
 def disable_service(name: str) -> ActionResult:
@@ -127,6 +142,8 @@ def disable_service(name: str) -> ActionResult:
             subprocess.run(["systemctl", "--user", "stop", unit], check=False)
             subprocess.run(["systemctl", "--user", "disable", unit], check=False)
             path.unlink()
+    # Drop the generated secret env file alongside the unit.
+    secret_env_path(name).unlink(missing_ok=True)
     subprocess.run(["systemctl", "--user", "daemon-reload"], check=False)
     return ActionResult(name, "deactivate", "ok", f"{name}: deactivated")
 

core/src/castle_core/registry.py

@@ -34,6 +34,10 @@ class Deployment:
     runner: str
     run_cmd: list[str]
     env: dict[str, str] = field(default_factory=dict)
+    # Names (never values) of secret-bearing env vars. Their resolved values live
+    # only in the mode-0600 env file, never in env/run_cmd/registry — this is for
+    # visibility (which secrets a deployment expects).
+    secret_env_keys: list[str] = field(default_factory=list)
     description: str | None = None
     behavior: str = "daemon"
     stack: str | None = None
@@ -97,6 +101,7 @@ def load_registry(path: Path | None = None) -> NodeRegistry:
             runner=comp_data.get("runner", "command"),
             run_cmd=comp_data.get("run_cmd", []),
             env=comp_data.get("env", {}),
+            secret_env_keys=comp_data.get("secret_env_keys", []),
             description=comp_data.get("description"),
             behavior=behavior,
             stack=comp_data.get("stack"),
@@ -137,6 +142,8 @@ def save_registry(registry: NodeRegistry, path: Path | None = None) -> None:
         }
         if comp.env:
             entry["env"] = comp.env
+        if comp.secret_env_keys:
+            entry["secret_env_keys"] = comp.secret_env_keys
         if comp.description:
             entry["description"] = comp.description
         entry["behavior"] = comp.behavior

core/tests/test_config.py

@@ -8,6 +8,7 @@ import pytest
 from castle_core.config import (
     CastleConfig,
     load_config,
+    resolve_env_split,
     resolve_env_vars,
     save_config,
 )
@@ -151,3 +152,71 @@ class TestResolveEnvVars:
         env = {"API_KEY": "${secret:NONEXISTENT}"}
         resolved = resolve_env_vars(env)
         assert resolved["API_KEY"] == "<MISSING_SECRET:NONEXISTENT>"
+
+
+class TestResolveEnvSplit:
+    """Tests for the secret/plain partition used to keep secrets out of units."""
+
+    def _secrets(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch, **vals: str):
+        secrets_dir = tmp_path / "secrets"
+        secrets_dir.mkdir()
+        for name, val in vals.items():
+            (secrets_dir / name).write_text(val + "\n")
+        monkeypatch.setattr("castle_core.config.SECRETS_DIR", secrets_dir)
+
+    def test_plain_only(self) -> None:
+        plain, secret = resolve_env_split({"PORT": "9001", "URL": "http://x"})
+        assert plain == {"PORT": "9001", "URL": "http://x"}
+        assert secret == {}
+
+    def test_context_placeholders_are_plain(self) -> None:
+        plain, secret = resolve_env_split(
+            {"P": "${port}", "D": "${data_dir}"}, {"port": "9001", "data_dir": "/d"}
+        )
+        assert plain == {"P": "9001", "D": "/d"}
+        assert secret == {}
+
+    def test_pure_secret_partitioned(
+        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        self._secrets(tmp_path, monkeypatch, API_KEY="sk-123")
+        plain, secret = resolve_env_split({"API_KEY": "${secret:API_KEY}"})
+        assert plain == {}
+        assert secret == {"API_KEY": "sk-123"}
+
+    def test_composite_secret_partitioned(
+        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        """A value embedding a secret is secret-bearing; the whole value resolves."""
+        self._secrets(tmp_path, monkeypatch, NEO4J_PASSWORD="pw")
+        plain, secret = resolve_env_split(
+            {"NEO4J_AUTH": "neo4j/${secret:NEO4J_PASSWORD}"}
+        )
+        assert plain == {}
+        assert secret == {"NEO4J_AUTH": "neo4j/pw"}
+
+    def test_mixed(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+        self._secrets(tmp_path, monkeypatch, K="v")
+        plain, secret = resolve_env_split(
+            {"PORT": "${port}", "K": "${secret:K}"}, {"port": "9001"}
+        )
+        assert plain == {"PORT": "9001"}
+        assert secret == {"K": "v"}
+
+    def test_missing_secret_still_partitioned(
+        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        self._secrets(tmp_path, monkeypatch)
+        plain, secret = resolve_env_split({"K": "${secret:NOPE}"})
+        assert plain == {}
+        assert secret == {"K": "<MISSING_SECRET:NOPE>"}
+
+    def test_resolve_env_vars_matches_merged_split(
+        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        """The flat wrapper equals the merged split and preserves key order."""
+        self._secrets(tmp_path, monkeypatch, K="v")
+        env = {"PORT": "${port}", "K": "${secret:K}", "Z": "lit"}
+        flat = resolve_env_vars(env, {"port": "9001"})
+        assert flat == {"PORT": "9001", "K": "v", "Z": "lit"}
+        assert list(flat.keys()) == ["PORT", "K", "Z"]

core/tests/test_deploy_run_cmd.py

@@ -6,7 +6,7 @@ from pathlib import Path
 from unittest.mock import patch
 
 from castle_core.deploy import _build_run_cmd
-from castle_core.manifest import RunPython
+from castle_core.manifest import RunContainer, RunPython
 
 
 def test_python_runner_uses_uv_run_from_source(tmp_path: Path) -> None:
@@ -35,7 +35,9 @@ def test_python_runner_appends_args(tmp_path: Path) -> None:
 def test_python_runner_falls_back_to_path_without_source() -> None:
     """No resolvable source → PATH lookup of the script (no uv run)."""
     run = RunPython(runner="python", program="my-svc")
-    with patch("castle_core.deploy.shutil.which", return_value="/home/u/.local/bin/my-svc"):
+    with patch(
+        "castle_core.deploy.shutil.which", return_value="/home/u/.local/bin/my-svc"
+    ):
         cmd = _build_run_cmd("my-svc", run, {}, [], source_dir=None)
     assert cmd == ["/home/u/.local/bin/my-svc"]
 
@@ -48,3 +50,25 @@ def test_python_runner_warns_when_unresolvable() -> None:
         cmd = _build_run_cmd("my-svc", run, {}, messages, source_dir=None)
     assert cmd == ["my-svc"]
     assert any("my-svc" in m for m in messages)
+
+
+def test_container_secrets_use_env_file_not_argv() -> None:
+    """Secrets go through --env-file; plain vars stay as -e; no secret in argv."""
+    run = RunContainer(runner="container", image="img:latest", env={"PLAIN": "1"})
+    env_file = Path("/home/u/.castle/secrets/env/castle-svc.service.env")
+    with patch("castle_core.deploy.shutil.which", return_value="/usr/bin/docker"):
+        cmd = _build_run_cmd("svc", run, {"PORT": "9001"}, [], secret_env_file=env_file)
+    joined = " ".join(cmd)
+    assert "--env-file" in cmd
+    assert str(env_file) in cmd
+    # plain (non-secret) vars are still inlined as -e
+    assert "-e" in cmd and "PORT=9001" in joined and "PLAIN=1" in joined
+    # no resolved secret value leaks into argv (only the file path is referenced)
+    assert "SECRET=" not in joined
+
+
+def test_container_without_secrets_has_no_env_file() -> None:
+    run = RunContainer(runner="container", image="img:latest")
+    with patch("castle_core.deploy.shutil.which", return_value="/usr/bin/docker"):
+        cmd = _build_run_cmd("svc", run, {}, [], secret_env_file=None)
+    assert "--env-file" not in cmd

core/tests/test_deploy_secret_env.py

@@ -0,0 +1,87 @@
+"""Tests for the generated secret env file and its registry visibility."""
+
+from __future__ import annotations
+
+import stat
+from pathlib import Path
+
+import pytest
+
+import castle_core.deploy as deploy
+from castle_core.registry import (
+    Deployment,
+    NodeConfig,
+    NodeRegistry,
+    load_registry,
+    save_registry,
+)
+
+
+@pytest.fixture
+def secret_env_dir(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path:
+    """Redirect the secret-env file location into a temp dir."""
+    d = tmp_path / "secrets" / "env"
+    monkeypatch.setattr(deploy, "SECRET_ENV_DIR", d)
+    monkeypatch.setattr(
+        deploy, "secret_env_path", lambda name: d / f"castle-{name}.service.env"
+    )
+    return d
+
+
+class TestWriteSecretEnvFile:
+    def test_writes_mode_0600_in_0700_dir(self, secret_env_dir: Path) -> None:
+        path = deploy._write_secret_env_file("svc", {"API_KEY": "sk-123"})
+        assert path is not None and path.exists()
+        assert stat.S_IMODE(path.stat().st_mode) == 0o600
+        assert stat.S_IMODE(secret_env_dir.stat().st_mode) == 0o700
+
+    def test_content_format(self, secret_env_dir: Path) -> None:
+        path = deploy._write_secret_env_file(
+            "svc", {"A": "1", "NEO4J_AUTH": "neo4j/pw"}
+        )
+        assert path is not None
+        assert path.read_text() == "A=1\nNEO4J_AUTH=neo4j/pw\n"
+
+    def test_empty_unlinks_and_returns_none(self, secret_env_dir: Path) -> None:
+        path = deploy.secret_env_path("svc")
+        secret_env_dir.mkdir(parents=True)
+        path.write_text("STALE=1\n")
+        result = deploy._write_secret_env_file("svc", {})
+        assert result is None
+        assert not path.exists()
+
+    def test_rewrite_truncates_and_fixes_mode(self, secret_env_dir: Path) -> None:
+        deploy._write_secret_env_file("svc", {"A": "old", "B": "x"})
+        path = deploy._write_secret_env_file("svc", {"A": "new"})
+        assert path is not None
+        assert path.read_text() == "A=new\n"
+        assert stat.S_IMODE(path.stat().st_mode) == 0o600
+
+
+class TestRegistrySecretKeys:
+    def test_round_trip_persists_keys_only(self, tmp_path: Path) -> None:
+        reg_path = tmp_path / "registry.yaml"
+        registry = NodeRegistry(
+            node=NodeConfig(hostname="h"),
+            deployed={
+                "svc": Deployment(
+                    runner="container",
+                    run_cmd=["docker", "run"],
+                    env={"PORT": "9001"},
+                    secret_env_keys=["ANTHROPIC_API_KEY", "OPENAI_API_KEY"],
+                    managed=True,
+                )
+            },
+        )
+        save_registry(registry, reg_path)
+
+        text = reg_path.read_text()
+        assert "ANTHROPIC_API_KEY" in text  # names are fine
+        assert "sk-ant" not in text  # but no values ever
+
+        loaded = load_registry(reg_path)
+        assert loaded.deployed["svc"].secret_env_keys == [
+            "ANTHROPIC_API_KEY",
+            "OPENAI_API_KEY",
+        ]
+        assert loaded.deployed["svc"].env == {"PORT": "9001"}

core/tests/test_systemd.py

@@ -2,9 +2,13 @@
 
 from __future__ import annotations
 
+from pathlib import Path
+
 from castle_core.generators.systemd import (
     generate_timer,
     generate_unit_from_deployed,
+    secret_env_path,
+    unit_env_file,
     unit_name,
 )
 from castle_core.registry import Deployment
@@ -82,7 +86,10 @@ class TestUnitFromDeployed:
         deployed = Deployment(
             runner="python",
             run_cmd=["/home/user/.local/bin/uv", "run", "my-svc"],
-            env={"MY_SVC_PORT": "9001", "MY_SVC_DATA_DIR": "/home/user/.castle/data/my-svc"},
+            env={
+                "MY_SVC_PORT": "9001",
+                "MY_SVC_DATA_DIR": "/home/user/.castle/data/my-svc",
+            },
             description="My service",
         )
         unit = generate_unit_from_deployed("my-svc", deployed)
@@ -118,6 +125,70 @@ class TestUnitFromDeployed:
         assert "/data/repos/" not in unit
 
 
+class TestSecretEnvFile:
+    """Secrets are referenced via EnvironmentFile=, never inlined."""
+
+    def test_environment_file_added_for_simple_unit(self) -> None:
+        deployed = Deployment(
+            runner="python",
+            run_cmd=["/uv", "run", "my-svc"],
+            env={"PORT": "9001"},
+            secret_env_keys=["API_KEY"],
+        )
+        path = Path("/home/u/.castle/secrets/env/castle-my-svc.service.env")
+        unit = generate_unit_from_deployed("my-svc", deployed, env_file=path)
+        assert f"EnvironmentFile={path}" in unit
+        # fail-loud: no '-' prefix
+        assert f"EnvironmentFile=-{path}" not in unit
+
+    def test_environment_file_added_for_oneshot_job(self) -> None:
+        deployed = Deployment(
+            runner="command",
+            run_cmd=["/bin/job"],
+            env={},
+            schedule="0 2 * * *",
+            secret_env_keys=["TOKEN"],
+        )
+        path = Path("/home/u/.castle/secrets/env/castle-my-job.service.env")
+        unit = generate_unit_from_deployed("my-job", deployed, env_file=path)
+        assert "Type=oneshot" in unit
+        assert f"EnvironmentFile={path}" in unit
+
+    def test_no_environment_file_when_none(self) -> None:
+        deployed = Deployment(runner="python", run_cmd=["/uv", "run", "x"], env={})
+        unit = generate_unit_from_deployed("x", deployed, env_file=None)
+        assert "EnvironmentFile" not in unit
+
+    def test_secret_values_never_in_unit(self) -> None:
+        """The unit references the file path; resolved secret values never appear."""
+        deployed = Deployment(
+            runner="python",
+            run_cmd=["/uv", "run", "x"],
+            env={"PORT": "9001"},
+            secret_env_keys=["API_KEY"],
+        )
+        path = secret_env_path("x")
+        unit = generate_unit_from_deployed("x", deployed, env_file=path)
+        assert "sk-secret" not in unit  # value is in the file, not the unit
+
+
+class TestUnitEnvFile:
+    """unit_env_file decides which runners get an EnvironmentFile= path."""
+
+    def test_none_without_secrets(self) -> None:
+        d = Deployment(runner="python", run_cmd=[], env={}, secret_env_keys=[])
+        assert unit_env_file(d, "x") is None
+
+    def test_path_for_python_with_secrets(self) -> None:
+        d = Deployment(runner="python", run_cmd=[], env={}, secret_env_keys=["K"])
+        assert unit_env_file(d, "x") == secret_env_path("x")
+
+    def test_none_for_container(self) -> None:
+        """Containers load secrets via docker --env-file, not systemd."""
+        d = Deployment(runner="container", run_cmd=[], env={}, secret_env_keys=["K"])
+        assert unit_env_file(d, "x") is None
+
+
 class TestGenerateTimer:
     """Tests for timer generation from schedule strings."""