feat(core): single-source data_dir/repos_dir via castle.yaml

The `castle` CLI and the `castle-api` service are two independent in-process
drivers of `castle_core`. Each resolved DATA_DIR/REPOS_DIR at import time from
its own process env (default /data/castle), persisted nowhere — so they silently
diverged, and `apply`/dashboard-apply crashed on a non-existent /data.

Make the loaded CastleConfig the single source of truth:
- Resolve data_dir/repos_dir only in load_config (env > castle.yaml > default),
  anchored to the config root; drop the DATA_DIR/REPOS_DIR module globals and the
  import-time file read entirely — no global twin that can disagree with the file.
- Thread config.data_dir/repos_dir through ensure_dirs, _env_context, tls_dir_for
  (now unified — deploy no longer inlines the tls path), and create/add/clone.
- ensure_dirs raises an actionable CastleDirError instead of a bare PermissionError;
  the api surfaces it as 422.
- doctor: "data dir writable" check + WARN when CASTLE_DATA_DIR/REPOS_DIR env
  overrides the file (the one remaining cross-process divergence vector).
- install.sh persists data_dir/repos_dir into castle.yaml (idempotent, non-default).
- Docs: registry.md globals + AGENTS.md roots.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

core/src/castle_core/config.py

@@ -34,30 +34,32 @@ def _resolve_castle_home() -> Path:
     return Path.home() / ".castle"
 
 
-def _resolve_data_dir() -> Path:
-    """Resolve the program data directory (service/program data I/O).
-
-    Decoupled from CASTLE_HOME so bulk data can live on a dedicated volume.
-    Defaults to /data/castle. Override with the CASTLE_DATA_DIR environment
-    variable (supports ~ and relative paths, which are expanded and made absolute).
-    """
-    override = os.environ.get("CASTLE_DATA_DIR")
-    if override:
-        return Path(override).expanduser().resolve()
-    return Path("/data/castle")
+_DEFAULT_DATA_DIR = Path("/data/castle")
+_DEFAULT_REPOS_DIR = Path("/data/repos")
 
 
-def _resolve_repos_dir() -> Path:
-    """Resolve where program source repos live by default.
+class CastleDirError(RuntimeError):
+    """A required castle directory can't be created (e.g. data_dir outside a writable
+    parent). Carries an actionable message; surfaced to the CLI and the api instead of
+    a bare PermissionError traceback."""
 
-    `castle create` scaffolds and `castle add` adopts repos under here. Programs
-    may also live anywhere (source: is an absolute path); this is just the default
-    home for new ones. Override with CASTLE_REPOS_DIR. Defaults to /data/repos.
-    """
-    override = os.environ.get("CASTLE_REPOS_DIR")
-    if override:
-        return Path(override).expanduser().resolve()
-    return Path("/data/repos")
+
+def _resolve_root_path(
+    env_var: str, yaml_value: object, anchor: Path, default: Path
+) -> Path:
+    """Resolve a configurable root with precedence: env var > castle.yaml > default.
+
+    `~` is expanded; a relative path is anchored to `anchor` (the dir containing
+    castle.yaml) — never cwd, so the CLI (shell cwd) and the api service (unit cwd)
+    resolve identically. The built-in default is returned as-is (so it compares equal
+    for the "persist only when non-default" check in save_config)."""
+    raw = os.environ.get(env_var) or yaml_value
+    if not raw:
+        return default
+    p = Path(str(raw)).expanduser()
+    if not p.is_absolute():
+        p = anchor / p
+    return p.resolve()
 
 
 CASTLE_HOME = _resolve_castle_home()
@@ -65,9 +67,13 @@ CODE_DIR = CASTLE_HOME / "code"
 ARTIFACTS_DIR = CASTLE_HOME / "artifacts"
 SPECS_DIR = ARTIFACTS_DIR / "specs"
 CONTENT_DIR = ARTIFACTS_DIR / "content"
-DATA_DIR = _resolve_data_dir()
 SECRETS_DIR = CASTLE_HOME / "secrets"
-REPOS_DIR = _resolve_repos_dir()
+# data_dir and repos_dir are deliberately NOT module constants. Unlike the CASTLE_HOME
+# family above (env-or-default — the dir that *holds* castle.yaml can't be configured
+# inside it), these are per-instance settings read from castle.yaml. A module global
+# would be a second copy of that value, resolved once at import against one process's
+# environment — exactly what let the CLI and the api service drift. They live only on
+# the loaded CastleConfig; read config.data_dir / config.repos_dir (see load_config).
 
 # User tool directories — the single source of truth for "where our CLIs live".
 # Used both at build time (dev-verb subprocess PATH) and at run time (generated
@@ -173,6 +179,11 @@ class CastleConfig:
     # Launchable agent CLIs for the dashboard terminal UX (assistant-agnostic).
     # Optional; empty means the API falls back to a built-in default set.
     agents: dict[str, AgentSpec] = field(default_factory=dict)
+    # Configurable roots — the single source of truth (no module-constant twin).
+    # load_config sets them (env > castle.yaml > default); a bare constructor gets the
+    # built-in defaults so tests/callers that don't care stay valid.
+    data_dir: Path = field(default_factory=lambda: _DEFAULT_DATA_DIR)
+    repos_dir: Path = field(default_factory=lambda: _DEFAULT_REPOS_DIR)
     # Construction convenience only (not stored): a flat name→spec dict is routed
     # into the per-kind stores by kind_for. Lets callers/tests hand us a flat map
     # without pre-splitting it; there is still no flat `deployments` attribute.
@@ -416,6 +427,16 @@ def load_config(root: Path | None = None) -> CastleConfig:
     if data.get("repo"):
         repo_path = Path(data["repo"]).expanduser()
 
+    # Configurable roots: env > this file's data_dir/repos_dir > default, anchored to
+    # `root` (the dir holding this castle.yaml) so a per-call load_config is correct
+    # regardless of the import-time constants (which resolved against CASTLE_HOME).
+    data_dir = _resolve_root_path(
+        "CASTLE_DATA_DIR", data.get("data_dir"), root, _DEFAULT_DATA_DIR
+    )
+    repos_dir = _resolve_root_path(
+        "CASTLE_REPOS_DIR", data.get("repos_dir"), root, _DEFAULT_REPOS_DIR
+    )
+
     programs: dict[str, ProgramSpec] = {}
     for name, comp_data in _load_resource_dir(root / "programs").items():
         prog = _parse_program(name, comp_data)
@@ -442,6 +463,8 @@ def load_config(root: Path | None = None) -> CastleConfig:
         gateway=gateway,
         programs=programs,
         agents=agents,
+        data_dir=data_dir,
+        repos_dir=repos_dir,
         **stores,
     )
     return config
@@ -622,6 +645,13 @@ def save_config(config: CastleConfig) -> None:
     data: dict = {"gateway": gateway_data}
     if config.repo:
         data["repo"] = str(config.repo)
+    # Persist the configurable roots only when non-default, keeping castle.yaml minimal.
+    # These MUST round-trip: save_config rewrites the file from scratch, so a root that
+    # isn't re-emitted here would be silently dropped on the next apply.
+    if config.data_dir != _DEFAULT_DATA_DIR:
+        data["data_dir"] = str(config.data_dir)
+    if config.repos_dir != _DEFAULT_REPOS_DIR:
+        data["repos_dir"] = str(config.repos_dir)
     if config.agents:
         data["agents"] = {
             n: s.model_dump(exclude_none=True, exclude_defaults=True)
@@ -649,13 +679,25 @@ def save_config(config: CastleConfig) -> None:
             path.unlink()
 
 
-def ensure_dirs() -> None:
-    """Ensure castle directories exist."""
+def ensure_dirs(config: CastleConfig) -> None:
+    """Ensure castle directories exist. Takes the config so the data dir comes from the
+    one source of truth (config.data_dir), not a process-resolved global."""
     CASTLE_HOME.mkdir(parents=True, exist_ok=True)
     CODE_DIR.mkdir(parents=True, exist_ok=True)
     SPECS_DIR.mkdir(parents=True, exist_ok=True)
     CONTENT_DIR.mkdir(parents=True, exist_ok=True)
-    DATA_DIR.mkdir(parents=True, exist_ok=True)
+    # The data dir can live outside $HOME (a dedicated volume), so its parent may be
+    # unwritable or absent — fail loud with a fix, not a bare PermissionError.
+    data_dir = config.data_dir
+    try:
+        data_dir.mkdir(parents=True, exist_ok=True)
+    except OSError as e:
+        raise CastleDirError(
+            f"Cannot create data dir {data_dir}: {e.strerror or e}. "
+            f"Set data_dir: in {CASTLE_HOME / 'castle.yaml'} (or export CASTLE_DATA_DIR) "
+            f"to a writable path, or create it: "
+            f"sudo mkdir -p {data_dir} && sudo chown $(id -un) {data_dir}"
+        ) from e
     SECRETS_DIR.mkdir(parents=True, exist_ok=True)
     os.chmod(SECRETS_DIR, 0o700)
     # Generated per-deployment secret env files (EnvironmentFile= / --env-file)

core/src/castle_core/deploy.py

@@ -15,7 +15,6 @@ from dataclasses import dataclass, field
 from pathlib import Path
 
 from castle_core.config import (
-    DATA_DIR,
     SECRETS_DIR,
     SPECS_DIR,
     CastleConfig,
@@ -110,7 +109,7 @@ def deploy(target_name: str | None = None, root: Path | None = None) -> DeployRe
     config = load_config(root)
     result = DeployResult()
 
-    ensure_dirs()
+    ensure_dirs(config)
 
     # Build node config
     node = _node_config(config)
@@ -195,7 +194,9 @@ def _node_config(config: CastleConfig) -> NodeConfig:
 
 def _unit_file_for(name: str, kind: str) -> Path:
     """On-disk systemd unit path for a deployment (timer if it's a job)."""
-    return SYSTEMD_USER_DIR / (timer_name(name) if kind == "job" else unit_name(name, kind))
+    return SYSTEMD_USER_DIR / (
+        timer_name(name) if kind == "job" else unit_name(name, kind)
+    )
 
 
 def _unit_bytes(name: str, kind: str) -> str | None:
@@ -563,14 +564,15 @@ def _env_context(
     name: str,
     config_key: str,
     port: int | None,
+    data_dir: Path,
     public_url: str | None = None,
     supabase_app_schemas: str | None = None,
 ) -> dict[str, str]:
     """Placeholder values for defaults.env: ${name}/${data_dir}/${port}/${public_url}/
-    ${supabase_app_schemas}."""
+    ${supabase_app_schemas}. `data_dir` is the instance root (config.data_dir)."""
     ctx = {
         "name": name,
-        "data_dir": str(DATA_DIR / config_key),
+        "data_dir": str(data_dir / config_key),
         "uid": str(os.getuid()),
         "gid": str(os.getgid()),
     }
@@ -710,14 +712,21 @@ def _build_deployed(
         raw_env.setdefault(var, url)
     public_url = _public_url(config, name, expose, port)
     ctx = _env_context(
-        name, config_key, port, public_url, _supabase_app_schemas(config)
+        name,
+        config_key,
+        port,
+        config.data_dir,
+        public_url,
+        _supabase_app_schemas(config),
     )
     # ${tls_*}: paths to castle-materialized cert files for a TLS-material TCP
     # service. The deployment maps them into its own config (mount ${tls_dir} for a
     # container, or reference ${tls_cert}/${tls_key} directly for a native service).
     tls = dep.expose.tcp.tls if (dep.expose and dep.expose.tcp) else None
     if tls and tls.material != TlsMaterial.OFF:
-        tls_dir = DATA_DIR / config_key / "tls"
+        from castle_core.tls import tls_dir_for
+
+        tls_dir = tls_dir_for(config.data_dir, config_key)
         ctx.update(
             {
                 "tls_dir": str(tls_dir),
@@ -1077,5 +1086,7 @@ def _generate_systemd_units(config: CastleConfig, registry: NodeRegistry) -> Non
     SYSTEMD_USER_DIR.mkdir(parents=True, exist_ok=True)
 
     for _key, deployed in registry.deployed.items():
-        for fname, content in _render_unit_files(config, deployed.name, deployed).items():
+        for fname, content in _render_unit_files(
+            config, deployed.name, deployed
+        ).items():
             (SYSTEMD_USER_DIR / fname).write_text(content)

core/src/castle_core/tls.py

@@ -21,7 +21,7 @@ import subprocess
 import time
 from pathlib import Path
 
-from castle_core.config import DATA_DIR, CastleConfig
+from castle_core.config import CastleConfig
 from castle_core.manifest import SystemdDeployment, TlsMaterial
 
 _KEY_MODE_FILES = {"key.pem", "combined.pem"}  # secret → 0600; certs → 0644
@@ -55,9 +55,10 @@ def wildcard_cert(domain: str) -> tuple[Path, Path] | None:
     return None
 
 
-def tls_dir_for(config_key: str) -> Path:
-    """Where a deployment's materialized cert files live (``${tls_dir}``)."""
-    return DATA_DIR / config_key / "tls"
+def tls_dir_for(data_dir: Path, config_key: str) -> Path:
+    """Where a deployment's materialized cert files live (``${tls_dir}``). `data_dir`
+    is the instance root (config.data_dir) — the single source of truth."""
+    return data_dir / config_key / "tls"
 
 
 def _tls_of(dep: object) -> object | None:
@@ -121,7 +122,7 @@ def materialize_tls(config: CastleConfig, name: str, dep: object) -> bool:
     crt, key = crt_path.read_bytes(), key_path.read_bytes()
 
     config_key = dep.program or name  # type: ignore[attr-defined]
-    tls_dir = tls_dir_for(config_key)
+    tls_dir = tls_dir_for(config.data_dir, config_key)
     wanted = _wanted_files(tls_dir, tls.material, crt, key)  # type: ignore[attr-defined]
 
     if all(p.exists() and p.read_bytes() == c for p, c in wanted.items()):

core/tests/test_config.py

@@ -329,3 +329,141 @@ class TestConfigRoundTrip:
         assert g.public_domain == "pub.io"
         assert g.tunnel_id == "uuid-123"
         assert g.cert_hook is True
+
+
+class TestConfigurableRoots:
+    """data_dir / repos_dir: env > castle.yaml > default. The single source of truth
+    that keeps the CLI and the api service from resolving different data dirs."""
+
+    @pytest.fixture(autouse=True)
+    def _no_root_env(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        # The test host may export CASTLE_DATA_DIR (that's the bug we're fixing);
+        # clear it so yaml/default precedence is exercised deterministically.
+        monkeypatch.delenv("CASTLE_DATA_DIR", raising=False)
+        monkeypatch.delenv("CASTLE_REPOS_DIR", raising=False)
+
+    def test_resolve_precedence_env_over_yaml(
+        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        from castle_core.config import _DEFAULT_DATA_DIR, _resolve_root_path
+
+        monkeypatch.setenv("X_ROOT_ENV", "/from/env")
+        p = _resolve_root_path("X_ROOT_ENV", "/from/yaml", tmp_path, _DEFAULT_DATA_DIR)
+        assert p == Path("/from/env")
+
+    def test_resolve_yaml_over_default(self, tmp_path: Path) -> None:
+        from castle_core.config import _DEFAULT_DATA_DIR, _resolve_root_path
+
+        p = _resolve_root_path(
+            "UNSET_ROOT_ENV", "/from/yaml", tmp_path, _DEFAULT_DATA_DIR
+        )
+        assert p == Path("/from/yaml")
+
+    def test_resolve_default_when_neither(self, tmp_path: Path) -> None:
+        from castle_core.config import _DEFAULT_DATA_DIR, _resolve_root_path
+
+        p = _resolve_root_path("UNSET_ROOT_ENV", None, tmp_path, _DEFAULT_DATA_DIR)
+        assert p == _DEFAULT_DATA_DIR  # returned as-is so save_config can compare equal
+
+    def test_resolve_expanduser(
+        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        from castle_core.config import _DEFAULT_DATA_DIR, _resolve_root_path
+
+        monkeypatch.setenv("X_ROOT_ENV", "~/box")
+        p = _resolve_root_path("X_ROOT_ENV", None, tmp_path, _DEFAULT_DATA_DIR)
+        assert p == (Path.home() / "box").resolve()
+
+    def test_resolve_relative_anchored_to_anchor_not_cwd(self, tmp_path: Path) -> None:
+        """A relative root is anchored to the dir holding castle.yaml — never cwd, or
+        the CLI (shell cwd) and api (unit cwd) would diverge again."""
+        from castle_core.config import _DEFAULT_DATA_DIR, _resolve_root_path
+
+        p = _resolve_root_path(
+            "UNSET_ROOT_ENV", "sub/data", tmp_path, _DEFAULT_DATA_DIR
+        )
+        assert p == (tmp_path / "sub" / "data").resolve()
+
+    def test_load_config_reads_data_dir_from_yaml(self, tmp_path: Path) -> None:
+        (tmp_path / "castle.yaml").write_text(
+            yaml.dump({"gateway": {"port": 9000}, "data_dir": "/srv/box/data"})
+        )
+        config = load_config(tmp_path)
+        assert config.data_dir == Path("/srv/box/data")
+
+    def test_load_config_data_dir_defaults(self, tmp_path: Path) -> None:
+        from castle_core.config import _DEFAULT_DATA_DIR
+
+        (tmp_path / "castle.yaml").write_text(yaml.dump({"gateway": {"port": 9000}}))
+        config = load_config(tmp_path)
+        assert config.data_dir == _DEFAULT_DATA_DIR
+
+    def test_save_round_trips_nondefault_roots(self, tmp_path: Path) -> None:
+        from castle_core.config import GatewayConfig
+
+        cfg = CastleConfig(
+            root=tmp_path,
+            gateway=GatewayConfig(port=9000),
+            repo=None,
+            programs={},
+            data_dir=Path("/srv/box/data"),
+            repos_dir=Path("/srv/box/repos"),
+        )
+        save_config(cfg)
+        text = (tmp_path / "castle.yaml").read_text()
+        assert "data_dir: /srv/box/data" in text
+        assert "repos_dir: /srv/box/repos" in text
+        reloaded = load_config(tmp_path)
+        assert reloaded.data_dir == Path("/srv/box/data")
+        assert reloaded.repos_dir == Path("/srv/box/repos")
+
+    def test_save_omits_default_roots(self, tmp_path: Path) -> None:
+        from castle_core.config import (
+            _DEFAULT_DATA_DIR,
+            _DEFAULT_REPOS_DIR,
+            GatewayConfig,
+        )
+
+        cfg = CastleConfig(
+            root=tmp_path,
+            gateway=GatewayConfig(port=9000),
+            repo=None,
+            programs={},
+            data_dir=_DEFAULT_DATA_DIR,
+            repos_dir=_DEFAULT_REPOS_DIR,
+        )
+        save_config(cfg)
+        text = (tmp_path / "castle.yaml").read_text()
+        assert "data_dir" not in text
+        assert "repos_dir" not in text
+
+    def test_ensure_dirs_raises_actionable_error(
+        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        """An uncreatable data dir yields a CastleDirError with a fix, not a bare OSError."""
+        import castle_core.config as C
+        from castle_core.config import GatewayConfig
+
+        # Redirect the in-$HOME dirs to tmp so the test never touches the real ~/.castle.
+        for name in (
+            "CASTLE_HOME",
+            "CODE_DIR",
+            "SPECS_DIR",
+            "CONTENT_DIR",
+            "SECRETS_DIR",
+        ):
+            monkeypatch.setattr(C, name, tmp_path / name.lower())
+        # data_dir whose parent is a FILE → mkdir raises NotADirectoryError (OSError),
+        # deterministically, even if the suite runs as root.
+        blocker = tmp_path / "afile"
+        blocker.write_text("x")
+        cfg = CastleConfig(
+            root=tmp_path,
+            gateway=GatewayConfig(port=9000),
+            repo=None,
+            programs={},
+            data_dir=blocker / "sub",
+        )
+        with pytest.raises(C.CastleDirError) as ei:
+            C.ensure_dirs(cfg)
+        assert "data_dir" in str(ei.value)

core/tests/test_tls.py

@@ -20,21 +20,27 @@ def _write_wildcard(xdg: Path, domain: str, tag: str, acme_dir: str) -> None:
 
 @pytest.fixture
 def tls_env(tmp_path: Path, monkeypatch: pytest.MonkeyPatch):
-    """Isolate Caddy's cert store (XDG_DATA_HOME, read live) and DATA_DIR (patched
-    on the tls module) to a temp dir — no importlib.reload, so no global leak."""
+    """Isolate Caddy's cert store (XDG_DATA_HOME, read live) to a temp dir. The data
+    dir is carried on the config (config.data_dir) — the single source of truth — and
+    returned so tests can locate materialized certs via tls_dir_for(data_dir, ...)."""
     domain = "civil.payne.io"
     xdg = tmp_path / "xdg"
     _write_wildcard(xdg, domain, "PROD", "acme-v02.api.letsencrypt.org-directory")
-    _write_wildcard(xdg, domain, "STAGING", "acme-staging-v02.api.letsencrypt.org-directory")
+    _write_wildcard(
+        xdg, domain, "STAGING", "acme-staging-v02.api.letsencrypt.org-directory"
+    )
     monkeypatch.setenv("XDG_DATA_HOME", str(xdg))
-    monkeypatch.setattr(T, "DATA_DIR", tmp_path / "data")  # tls_dir_for reads this
-    return T, C, domain
+    return T, C, domain, tmp_path / "data"
 
 
-def _cfg(C, domain, dep):
+def _cfg(C, domain, dep, data_dir):
     return C.CastleConfig(
-        root=None, gateway=C.GatewayConfig(port=9000, domain=domain), repo=None,
-        programs={}, deployments={"postgres": dep},
+        root=None,
+        gateway=C.GatewayConfig(port=9000, domain=domain),
+        repo=None,
+        programs={},
+        data_dir=data_dir,
+        deployments={"postgres": dep},
     )
 
 
@@ -51,31 +57,33 @@ def _pg(material: str):
 
 
 def test_prefers_prod_over_staging(tls_env) -> None:
-    T, _, domain = tls_env
+    T, _, domain, _dd = tls_env
     crt, _ = T.wildcard_cert(domain)
     assert crt.read_text().strip() == "CERT-PROD"
 
 
 def test_pair_material_and_idempotency(tls_env) -> None:
-    T, C, domain = tls_env
+    T, C, domain, dd = tls_env
     pg = _pg("pair")
-    cfg = _cfg(C, domain, pg)
-    assert T.materialize_tls(cfg, "postgres", pg) is True   # first write
+    cfg = _cfg(C, domain, pg, dd)
+    assert T.materialize_tls(cfg, "postgres", pg) is True  # first write
     assert T.materialize_tls(cfg, "postgres", pg) is False  # idempotent
-    td = T.tls_dir_for("postgres")
+    td = T.tls_dir_for(dd, "postgres")
     assert sorted(p.name for p in td.iterdir()) == ["cert.pem", "chain.pem", "key.pem"]
     assert (td / "cert.pem").read_text().strip() == "CERT-PROD"
-    assert oct((td / "key.pem").stat().st_mode)[-3:] == "600"   # secret
+    assert oct((td / "key.pem").stat().st_mode)[-3:] == "600"  # secret
     assert oct((td / "cert.pem").stat().st_mode)[-3:] == "644"  # public
 
 
 def test_material_switch_cleans_stale(tls_env) -> None:
-    T, C, domain = tls_env
+    T, C, domain, dd = tls_env
     pair = _pg("pair")
-    T.materialize_tls(_cfg(C, domain, pair), "postgres", pair)
+    T.materialize_tls(_cfg(C, domain, pair, dd), "postgres", pair)
     combined = _pg("combined")
-    assert T.materialize_tls(_cfg(C, domain, combined), "postgres", combined) is True
-    td = T.tls_dir_for("postgres")
+    assert (
+        T.materialize_tls(_cfg(C, domain, combined, dd), "postgres", combined) is True
+    )
+    td = T.tls_dir_for(dd, "postgres")
     assert sorted(p.name for p in td.iterdir()) == ["chain.pem", "combined.pem"]
     assert (td / "combined.pem").read_text() == "KEY-PROD\nCERT-PROD\n"  # key + cert
     assert oct((td / "combined.pem").stat().st_mode)[-3:] == "600"
@@ -85,28 +93,36 @@ def test_pair_chain_is_issuer_not_leaf(tls_env, tmp_path) -> None:
     """`chain.pem` (${tls_ca}) is the issuer chain — the intermediates only, leaf
     stripped — so it's a real CA bundle distinct from the leaf-bearing cert.pem
     (regression: they used to be byte-identical)."""
-    T, C, domain = tls_env
+    T, C, domain, dd = tls_env
     leaf = b"-----BEGIN CERTIFICATE-----\nLEAF\n-----END CERTIFICATE-----\n"
     inter = b"-----BEGIN CERTIFICATE-----\nINTERMEDIATE\n-----END CERTIFICATE-----\n"
     crt_dir = (
-        Path(tmp_path) / "xdg" / "caddy" / "certificates"
-        / "acme-v02.api.letsencrypt.org-directory" / f"wildcard_.{domain}"
+        Path(tmp_path)
+        / "xdg"
+        / "caddy"
+        / "certificates"
+        / "acme-v02.api.letsencrypt.org-directory"
+        / f"wildcard_.{domain}"
     )
     (crt_dir / f"wildcard_.{domain}.crt").write_bytes(leaf + inter)
     pg = _pg("pair")
-    assert T.materialize_tls(_cfg(C, domain, pg), "postgres", pg) is True
-    td = T.tls_dir_for("postgres")
+    assert T.materialize_tls(_cfg(C, domain, pg, dd), "postgres", pg) is True
+    td = T.tls_dir_for(dd, "postgres")
     assert (td / "cert.pem").read_bytes() == leaf + inter  # server presents leaf+chain
-    assert (td / "chain.pem").read_bytes() == inter        # CA bundle = intermediates
+    assert (td / "chain.pem").read_bytes() == inter  # CA bundle = intermediates
     assert (td / "cert.pem").read_bytes() != (td / "chain.pem").read_bytes()
 
 
 def test_material_off_is_noop(tls_env) -> None:
-    T, C, domain = tls_env
+    T, C, domain, dd = tls_env
     off = SystemdDeployment.model_validate(
-        {"manager": "systemd", "program": "postgres",
-         "run": {"launcher": "container", "image": "postgres:17"},
-         "reach": "internal", "expose": {"tcp": {"port": 5432}}}
+        {
+            "manager": "systemd",
+            "program": "postgres",
+            "run": {"launcher": "container", "image": "postgres:17"},
+            "reach": "internal",
+            "expose": {"tcp": {"port": 5432}},
+        }
     )
-    assert T.materialize_tls(_cfg(C, domain, off), "postgres", off) is False
-    assert not T.tls_dir_for("postgres").exists()
+    assert T.materialize_tls(_cfg(C, domain, off, dd), "postgres", off) is False
+    assert not T.tls_dir_for(dd, "postgres").exists()