feat: raw-TCP exposure with castle-managed TLS + reach model

Expose raw-TCP services (postgres) by name at <name>.<domain>:<port> and
cut the gateway's ACME wildcard cert onto them so they present a trusted
cert. Replaces the proxy/public booleans with a single `reach` enum
(off|internal|public); legacy input still parses via derived accessors.

Core:
- expose.tcp{port,tls} + TlsSpec(material: pair|combined|off, reload)
- tls.py: materialize cert files from Caddy's wildcard, reconcile on
  renewal; `castle tls` CLI; optional cert_obtained events-exec hook
  (gateway.cert_hook, gated so a plugin-less Caddy still parses)
- apply waits (bounded) for the wildcard to issue before materializing so
  a fresh-node TLS service starts with its cert in place, then scopes
  materialization to the deployments being applied
- reach: internal|public requires an expose block (no silent no-op);
  public raw-TCP guarded until tunnel support lands
- chain.pem (${tls_ca}) is the issuer chain (leaf stripped), a real CA
  bundle distinct from cert.pem
- one shared ${...} resolver (resolve_placeholders) for env and container
  run fields; run.env now expands like volumes/args; $$ escapes a literal
- validate the generated Caddyfile before reloading the gateway so an
  invalid config never degrades routing

Docs: docs/tcp-exposure.md. Tests cover reach/expose validation,
placeholder expansion + escape, issuer-chain material, TLS materialize.

core/src/castle_core/config.py

@@ -128,6 +128,12 @@ class GatewayConfig:
     # names never appear in public DNS. `tunnel_id` is the cloudflared tunnel UUID.
     public_domain: str | None = None
     tunnel_id: str | None = None
+    # acme mode only: emit the `events { on cert_obtained exec castle tls reconcile }`
+    # hook so certs materialized onto raw-TCP services refresh on renewal. Requires
+    # the events-exec plugin in the gateway's Caddy build — set true only once that
+    # Caddy is installed (see docs/tcp-exposure.md §5). Default false keeps a
+    # plugin-less gateway parseable.
+    cert_hook: bool = False
 
 
 @dataclass
@@ -228,6 +234,30 @@ def resolve_env_split(
     return plain, secret
 
 
+def resolve_placeholders(value: str, context: dict[str, str] | None) -> str:
+    """Expand ``${key}`` refs in a single string from ``context``.
+
+    The one ``${...}`` grammar shared by env resolution (:func:`resolve_env_split`)
+    and run-spec expansion (argv/volumes/env in a container launch), so a new
+    placeholder only has to be added to the context dict, never to a second engine.
+    Unknown refs — including ``${secret:...}`` — pass through untouched (secrets
+    never belong in argv; they go via ``--env-file``). Write ``$${key}`` to emit a
+    literal ``${key}`` (e.g. a container arg the container's own shell must expand).
+    """
+    if not context:
+        return value
+
+    def replace_var(match: re.Match[str]) -> str:
+        ref = match.group(1)
+        return context.get(ref, match.group(0))
+
+    # Split on the `$$` escape so an escaped `$${x}` never reaches the substitution
+    # regex, then rejoin with a literal `$`.
+    return "$".join(
+        re.sub(r"\$\{([^}]+)\}", replace_var, part) for part in value.split("$$")
+    )
+
+
 def resolve_env_vars(
     env: dict[str, str], context: dict[str, str] | None = None
 ) -> dict[str, str]:
@@ -335,6 +365,7 @@ def load_config(root: Path | None = None) -> CastleConfig:
         acme_dns_provider=gateway_data.get("acme_dns_provider", "cloudflare"),
         public_domain=gateway_data.get("public_domain"),
         tunnel_id=gateway_data.get("tunnel_id"),
+        cert_hook=gateway_data.get("cert_hook", False),
     )
 
     # repo: field points to the git repo for repo-relative sources
@@ -410,7 +441,6 @@ _STRUCTURAL_KEYS = {
     "manage",
     "systemd",
     "expose",
-    "proxy",
 }
 
 
@@ -494,12 +524,17 @@ def save_config(config: CastleConfig) -> None:
     if config.gateway.acme_email:
         gateway_data["acme_email"] = config.gateway.acme_email
     # Only persist the provider when non-default, to keep castle.yaml minimal.
-    if config.gateway.acme_dns_provider and config.gateway.acme_dns_provider != "cloudflare":
+    if (
+        config.gateway.acme_dns_provider
+        and config.gateway.acme_dns_provider != "cloudflare"
+    ):
         gateway_data["acme_dns_provider"] = config.gateway.acme_dns_provider
     if config.gateway.public_domain:
         gateway_data["public_domain"] = config.gateway.public_domain
     if config.gateway.tunnel_id:
         gateway_data["tunnel_id"] = config.gateway.tunnel_id
+    if config.gateway.cert_hook:
+        gateway_data["cert_hook"] = config.gateway.cert_hook
     data: dict = {"gateway": gateway_data}
     if config.repo:
         data["repo"] = str(config.repo)

core/src/castle_core/deploy.py

@@ -22,6 +22,7 @@ from castle_core.config import (
     ensure_dirs,
     load_config,
     resolve_env_split,
+    resolve_placeholders,
 )
 from castle_core.generators.caddyfile import (
     _DNS_TOKEN_ENV,
@@ -47,6 +48,7 @@ from castle_core.manifest import (
     DeploymentSpec,
     PathDeployment,
     RemoteDeployment,
+    TlsMaterial,
     kind_for,
 )
 from castle_core.registry import (
@@ -184,6 +186,7 @@ def _node_config(config: CastleConfig) -> NodeConfig:
         acme_dns_provider=config.gateway.acme_dns_provider,
         public_domain=config.gateway.public_domain,
         tunnel_id=config.gateway.tunnel_id,
+        cert_hook=config.gateway.cert_hook,
     )
 
 
@@ -258,6 +261,19 @@ def apply(
     result.messages = list(deploy_result.messages)
     result.registry = deploy_result.registry
 
+    # Materialize TLS cert files before (re)starting so a TLS service finds them on
+    # start. On a fresh node the gateway reload above only kicks off ACME issuance,
+    # so wait (bounded) for the wildcard first — otherwise the service would start
+    # without its cert and, with cert_hook off (the default), never recover. Scope
+    # materialization to the deployments being applied so a scoped apply doesn't
+    # rewrite an unrelated service's cert without reloading it. No reload here — the
+    # activation loop below starts/restarts as needed; rotation-driven reloads are
+    # the `castle tls reconcile` / cert_obtained path.
+    from castle_core.tls import materialize_all, wait_for_wildcard
+
+    wait_for_wildcard(config, names, result.messages)
+    materialize_all(config, result.messages, only=names)
+
     for name in names:
         after_unit = _unit_bytes(name, is_job[name])
         action = _classify(name, after_unit)
@@ -387,6 +403,28 @@ def _write_tunnel_config(registry: NodeRegistry, messages: list[str]) -> None:
 def _reload_gateway(messages: list[str]) -> None:
     """Reload Caddy if the gateway is running, so new routes take effect."""
     gw_unit = unit_name(_GATEWAY_NAME)
+    # Validate the generated Caddyfile before reloading. An invalid config (most
+    # often gateway.cert_hook enabled while the running Caddy lacks the events-exec
+    # plugin, so the `events {}` block fails to adapt) must not be pushed: a bad
+    # reload leaves stale routing and a later cold start would refuse to load. Skip
+    # the reload and point at the likely cause instead of silently degrading.
+    caddyfile = SPECS_DIR / "Caddyfile"
+    caddy = shutil.which("caddy")
+    if caddy and caddyfile.exists():
+        check = subprocess.run(
+            [caddy, "validate", "--adapter", "caddyfile", "--config", str(caddyfile)],
+            capture_output=True,
+            text=True,
+        )
+        if check.returncode != 0:
+            messages.append(
+                "Warning: generated Caddyfile is invalid — gateway NOT reloaded (the "
+                "running config is left untouched). If gateway.cert_hook is enabled, "
+                "the gateway's Caddy build needs the events-exec plugin; rebuild it "
+                "(install.sh) or set cert_hook: false.\n"
+                + (check.stderr.strip() or check.stdout.strip())
+            )
+            return
     active = subprocess.run(
         ["systemctl", "--user", "is-active", gw_unit],
         capture_output=True,
@@ -459,7 +497,12 @@ def _env_context(
 ) -> dict[str, str]:
     """Placeholder values for defaults.env: ${name}/${data_dir}/${port}/${public_url}/
     ${supabase_app_schemas}."""
-    ctx = {"name": name, "data_dir": str(DATA_DIR / config_key)}
+    ctx = {
+        "name": name,
+        "data_dir": str(DATA_DIR / config_key),
+        "uid": str(os.getuid()),
+        "gid": str(os.getgid()),
+    }
     if port is not None:
         ctx["port"] = str(port)
     if public_url is not None:
@@ -562,13 +605,16 @@ def _build_deployed(
     if dep.manage and dep.manage.systemd and not dep.manage.systemd.enable:
         managed = False
 
-    # `proxy` is the exposure checkbox; the subdomain is the deployment name.
-    expose = bool(dep.proxy)
+    # `http_exposed` is the HTTP-gateway checkbox (reach != off AND an http port);
+    # the subdomain is the deployment name. A raw-TCP service is not http_exposed —
+    # it's reachable at <name>.<domain>:<tcp_port> via bind + wildcard DNS.
+    expose = dep.http_exposed
     port = None
     health_path = None
     if dep.expose and dep.expose.http:
         port = dep.expose.http.internal.port
         health_path = dep.expose.http.health_path
+    tcp_port = dep.tcp_port
 
     # Env is exactly what's in defaults.env — no hidden convention injection.
     # ${port}/${data_dir}/${name}/${public_url} map the program's own env var
@@ -576,10 +622,23 @@ def _build_deployed(
     # mode-0600 file (never in the unit or argv).
     raw_env = dict(dep.defaults.env) if (dep.defaults and dep.defaults.env) else {}
     public_url = _public_url(config, name, expose, port)
-    env, secret_env = resolve_env_split(
-        raw_env,
-        _env_context(name, config_key, port, public_url, _supabase_app_schemas(config)),
-    )
+    ctx = _env_context(name, config_key, port, 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"
+        ctx.update(
+            {
+                "tls_dir": str(tls_dir),
+                "tls_cert": str(tls_dir / "cert.pem"),
+                "tls_key": str(tls_dir / "key.pem"),
+                "tls_pem": str(tls_dir / "combined.pem"),
+                "tls_ca": str(tls_dir / "chain.pem"),
+            }
+        )
+    env, secret_env = resolve_env_split(raw_env, ctx)
     secret_env_file = _write_secret_env_file(name, secret_env)
 
     # `command` launchers resolve a tool on PATH → ensure it's installed.
@@ -588,7 +647,8 @@ def _build_deployed(
         _ensure_python_tool(config, dep.program, messages)
 
     run_cmd = _build_run_cmd(
-        name, run, env, messages, source_dir, secret_env_file=secret_env_file
+        name, run, env, messages, source_dir, secret_env_file=secret_env_file,
+        placeholders=ctx,
     )
     stop_cmd = _build_stop_cmd(name, run, source_dir)
 
@@ -606,6 +666,7 @@ def _build_deployed(
         health_path=health_path,
         subdomain=(name if expose else None),
         public=bool(dep.public and expose),
+        tcp_port=tcp_port,
         schedule=getattr(dep, "schedule", None),
         managed=managed,
         enabled=dep.enabled,
@@ -680,6 +741,15 @@ def _ensure_python_tool(
         messages.append(f"Installed {program}")
 
 
+def _subst(value: str, placeholders: dict[str, str] | None) -> str:
+    """Expand ``${key}`` in a run-spec string field from castle's computed values
+    (``${uid}``/``${gid}``/``${data_dir}``/``${tls_dir}``/…), via the one shared
+    ``${...}`` resolver (:func:`resolve_placeholders`). Unknown refs pass through
+    unchanged (secrets never belong in argv — they go via --env-file); write
+    ``$${key}`` to pass a literal ``${key}`` to a container's own shell/env."""
+    return resolve_placeholders(value, placeholders)
+
+
 def _build_run_cmd(
     name: str,
     run: object,
@@ -687,6 +757,7 @@ def _build_run_cmd(
     messages: list[str],
     source_dir: Path | None = None,
     secret_env_file: Path | None = None,
+    placeholders: dict[str, str] | None = None,
 ) -> list[str]:
     """Build a run command list from a LaunchSpec (a systemd deployment's `run`).
 
@@ -729,24 +800,31 @@ def _build_run_cmd(
             # Container name derives from the SERVICE name (matches the systemd unit),
             # not the image name — so `castle-<service>` is stable and collision-free.
             cmd = [runtime, "run", "--rm", f"--name=castle-{name}"]
+            if run.user:  # type: ignore[union-attr]
+                # Run as the invoking user (uid uniformity → bind-mounted
+                # certs/data/secrets readable with no chown). ${uid}/${gid} expand
+                # to the castle process's own ids.
+                cmd.extend(["--user", _subst(run.user, placeholders)])  # type: ignore[union-attr]
+            for tp in run.tmpfs:  # type: ignore[union-attr]
+                cmd.extend(["--tmpfs", _subst(tp, placeholders)])
             for container_port, host_port in run.ports.items():  # type: ignore[union-attr]
                 cmd.extend(["-p", f"{host_port}:{container_port}"])
             for vol in run.volumes:  # type: ignore[union-attr]
-                cmd.extend(["-v", vol])
+                cmd.extend(["-v", _subst(vol, placeholders)])
             for key, val in run.env.items():  # type: ignore[union-attr]
-                cmd.extend(["-e", f"{key}={val}"])
+                cmd.extend(["-e", f"{key}={_subst(val, placeholders)}"])
             # 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.extend(["-w", _subst(run.workdir, placeholders)])  # type: ignore[union-attr]
             cmd.append(run.image)  # type: ignore[union-attr]
             if run.command:  # type: ignore[union-attr]
-                cmd.extend(run.command)  # type: ignore[union-attr]
+                cmd.extend(_subst(c, placeholders) for c in run.command)  # type: ignore[union-attr]
             if run.args:  # type: ignore[union-attr]
-                cmd.extend(run.args)  # type: ignore[union-attr]
+                cmd.extend(_subst(a, placeholders) for a in run.args)  # type: ignore[union-attr]
             return cmd
         case "compose":
             # A whole docker-compose stack supervised as one unit. `up` runs

core/src/castle_core/generators/caddyfile.py

@@ -54,13 +54,14 @@ def service_proxy_targets(name: str, dep: SystemdDeployment) -> ProxyTargets:
     """Derive a systemd deployment's gateway exposure from its spec.
 
     The single source of truth shared by the registry build (``deploy``) and
-    route computation (``compute_routes``), so they never disagree. ``expose`` is
-    the checkbox (``proxy: true``); the subdomain is always the deployment name.
+    route computation (``compute_routes``), so they never disagree. A gateway
+    route is HTTP-only: ``http_exposed`` requires ``reach != off`` *and* an HTTP
+    port, so a raw-TCP service (``expose.tcp``) never yields a route here.
     """
     port = None
     if dep.expose and dep.expose.http:
         port = dep.expose.http.internal.port
-    return bool(dep.proxy), port, None
+    return dep.http_exposed, port, None
 
 
 def _local_routes(
@@ -214,6 +215,18 @@ def generate_caddyfile_from_registry(
         lines.append(f"    acme_dns {provider} {{env.{token_env}}}")
         if os.environ.get("CASTLE_ACME_STAGING") == "1":
             lines.append(f"    acme_ca {_ACME_STAGING_CA}")
+        # On issuance/renewal, refresh certs materialized onto raw-TCP services and
+        # reload them (idempotent — a no-op when nothing rotated). Requires the
+        # events-exec plugin in the gateway's Caddy build, so it's gated on the
+        # durable `gateway.cert_hook` flag, set only once that Caddy is in place;
+        # false → the block is omitted and a plugin-less gateway parses fine. See
+        # docs/tcp-exposure.md §5.
+        if getattr(node, "cert_hook", False):
+            lines += [
+                "    events {",
+                "        on cert_obtained exec castle tls reconcile",
+                "    }",
+            ]
         lines += ["}", ""]
         # One wildcard site → a single cert covers every subdomain; a new service
         # needs no new cert or challenge.

core/src/castle_core/manifest.py

@@ -16,6 +16,47 @@ class RestartPolicy(str, Enum):
     ALWAYS = "always"
 
 
+class Reach(str, Enum):
+    """How far a deployment is exposed — a protocol-agnostic ladder.
+
+    ``off``      → reachable only at its own host:port (no gateway route).
+    ``internal`` → reachable at ``<name>.<domain>`` (HTTP via the gateway, or TCP
+                   via bind + wildcard DNS).
+    ``public``   → *also* projected to the internet (HTTP via the tunnel origin;
+                   TCP via ``cloudflared access tcp``). Implies ``internal``.
+
+    Replaces the old ``proxy``/``public`` booleans; ``proxy``/``public`` survive as
+    derived read-only accessors and as accepted *legacy input* (normalized below).
+    """
+
+    OFF = "off"
+    INTERNAL = "internal"
+    PUBLIC = "public"
+
+
+def _reach_from_legacy(data: object, default: Reach) -> object:
+    """Map legacy ``proxy``/``public`` booleans on a raw deployment dict to ``reach``.
+
+    Runs as a ``mode="before"`` validator. When ``reach`` is given explicitly it
+    wins (legacy keys are dropped); otherwise ``reach`` is derived from the old
+    booleans: ``public`` → PUBLIC, ``proxy`` → INTERNAL, else ``default``. Non-dict
+    input (e.g. model re-validation) passes through untouched.
+    """
+    if not isinstance(data, dict):
+        return data
+    d = dict(data)
+    proxy = bool(d.pop("proxy", False))
+    public = bool(d.pop("public", False))
+    if "reach" not in d:
+        if public:
+            d["reach"] = Reach.PUBLIC
+        elif proxy:
+            d["reach"] = Reach.INTERNAL
+        else:
+            d["reach"] = default
+    return d
+
+
 # ---------------------
 # Launch specs — how systemd starts a process (discriminated union on `launcher`)
 # ---------------------
@@ -51,6 +92,13 @@ class RunContainer(LaunchBase):
     volumes: list[str] = Field(default_factory=list)
     env: EnvMap = Field(default_factory=dict)
     workdir: str | None = None
+    # Run the container as this uid[:gid] (e.g. "${uid}:${gid}"). Running as the
+    # invoking user makes bind-mounted data/secrets/certs readable with no chown —
+    # see docs/tcp-exposure.md §4. None → the image's own default user.
+    user: str | None = None
+    # tmpfs mounts (e.g. ["/var/run/postgresql"]) for image runtime dirs that must
+    # be writable when the container runs as a non-default uid.
+    tmpfs: list[str] = Field(default_factory=list)
 
 
 class RunNode(LaunchBase):
@@ -119,7 +167,7 @@ class ManageSpec(BaseModel):
 
 
 # ---------------------
-# HTTP exposure + proxy
+# Exposure — HTTP (via the gateway) or raw TCP (bind + DNS)
 # ---------------------
 
 
@@ -134,8 +182,55 @@ class HttpExposeSpec(BaseModel):
     health_path: str | None = None
 
 
+class TlsMaterial(str, Enum):
+    """What cert files castle materializes onto a service from the wildcard cert.
+
+    ``off``      → the service does its own TLS (or none); castle stays out of it.
+    ``pair``     → cert.pem + key.pem (postgres, redis, most daemons).
+    ``combined`` → one file: key+cert concatenated (mongodb, haproxy, …).
+    """
+
+    OFF = "off"
+    PAIR = "pair"
+    COMBINED = "combined"
+
+
+class TlsSpec(BaseModel):
+    """Castle-managed TLS material for a raw-TCP service, cut from the gateway's
+    ACME wildcard cert (valid for ``<name>.<domain>``) and refreshed on renewal.
+    The service consumes the materialized files via the ``${tls_*}`` placeholders.
+    """
+
+    material: TlsMaterial = TlsMaterial.OFF
+    # Optional zero-downtime reload argv (a single command) run after the cert is
+    # re-materialized on renewal — e.g. ["systemctl", "--user", "reload", "castle-postgres"].
+    # Default (None): castle restarts the deployment (fine at a ~60-day cadence).
+    reload: list[str] | None = None
+
+
+class TcpExposeSpec(BaseModel):
+    """A raw-TCP service (postgres, redis, …). It doesn't ride the HTTP gateway:
+    with ``reach: internal`` it's reachable at ``<name>.<domain>:<port>`` via the
+    wildcard DNS record + the bound port (no Caddy route). Publishing the port on
+    the LAN is the deployment's own job (a container's ``run.ports``, or a native
+    service binding ``0.0.0.0``); castle doesn't rebind it, so there's no bind-host
+    field here to imply otherwise. ``tls`` (optional) has castle drop the wildcard
+    cert onto the service so it presents a trusted cert for ``<name>.<domain>``.
+    """
+
+    port: int = Field(ge=1, le=65535)
+    tls: TlsSpec | None = None
+
+
 class ExposeSpec(BaseModel):
     http: HttpExposeSpec | None = None
+    tcp: TcpExposeSpec | None = None
+
+    @model_validator(mode="after")
+    def _one_protocol(self) -> ExposeSpec:
+        if self.http and self.tcp:
+            raise ValueError("a deployment exposes http OR tcp, not both")
+        return self
 
 
 # ---------------------
@@ -327,19 +422,68 @@ class SystemdDeployment(DeploymentBase):
     schedule: str | None = None
     timezone: str = "America/Los_Angeles"
     expose: ExposeSpec | None = None
-    # Route <name>.<gateway.domain> to this process. False → host:port only.
-    proxy: bool = False
-    # Also publish to the public internet via the Cloudflare tunnel, at
-    # <name>.<gateway.public_domain>. Opt-in; requires `proxy`.
-    public: bool = False
+    # How far this process is exposed (off | internal | public). See `Reach`.
+    reach: Reach = Reach.OFF
     manage: ManageSpec | None = None
 
+    @model_validator(mode="before")
+    @classmethod
+    def _normalize_reach(cls, data: object) -> object:
+        return _reach_from_legacy(data, default=Reach.OFF)
+
     @model_validator(mode="after")
-    def _validate(self) -> SystemdDeployment:
-        if self.public and not self.proxy:
-            raise ValueError("public requires proxy (an exposed process).")
+    def _validate_reach(self) -> SystemdDeployment:
+        # An exposed reach needs a port to expose. Without an `expose` block the
+        # reach silently no-ops — no route, no subdomain, no tunnel entry — so a
+        # typo'd/omitted `expose` reads as success while the service is unreachable.
+        # Reject it at load so the mistake surfaces (replaces the old
+        # "public requires proxy" guard, now that reach is the canonical field).
+        # Static frontends (manager: caddy) are inherently exposed and validated
+        # elsewhere; this is a supervised process, which needs an explicit port.
+        if self.reach != Reach.OFF and not self.expose:
+            raise ValueError(
+                f"reach: {self.reach.value} requires an `expose` block "
+                "(expose.http or expose.tcp); a port-only process uses reach: off"
+            )
+        # Public raw-TCP (tunnel + Access) is a later step; guard it explicitly
+        # rather than silently no-op'ing when a TCP service asks for it.
+        if (
+            self.reach == Reach.PUBLIC
+            and self.expose
+            and self.expose.tcp
+            and not self.expose.http
+        ):
+            raise ValueError(
+                "reach: public for a raw-TCP service isn't supported yet "
+                "(see docs/tcp-exposure.md step 5); use reach: internal"
+            )
         return self
 
+    # Derived, read-only back-compat accessors (not serialized) so existing
+    # readers keep working while the stored/authored field is `reach`.
+    @property
+    def proxy(self) -> bool:
+        return self.reach != Reach.OFF
+
+    @property
+    def public(self) -> bool:
+        return self.reach == Reach.PUBLIC
+
+    @property
+    def http_exposed(self) -> bool:
+        """Exposed through the HTTP gateway at ``<name>.<domain>`` — the predicate
+        for a Caddy route / subdomain. Requires ``reach != off`` *and* an HTTP
+        port; a raw-TCP service (``expose.tcp``) is never HTTP-exposed."""
+        return self.reach != Reach.OFF and bool(self.expose and self.expose.http)
+
+    @property
+    def tcp_port(self) -> int | None:
+        """The raw-TCP port this service is exposed on, or None. Reachable at
+        ``<name>.<domain>:<port>`` when ``reach != off`` (bind + wildcard DNS)."""
+        if self.reach != Reach.OFF and self.expose and self.expose.tcp:
+            return self.expose.tcp.port
+        return None
+
 
 class CaddyDeployment(DeploymentBase):
     """A static site served by the gateway (Caddy ``file_server``) — no process.
@@ -350,8 +494,24 @@ class CaddyDeployment(DeploymentBase):
 
     manager: Literal["caddy"]
     root: str = "dist"
-    # Inherently exposed at its subdomain; `public` = also project via the tunnel.
-    public: bool = False
+    # A static site is inherently served at its subdomain, so `reach` is
+    # `internal` or `public` (never `off`). `public` = also project via the tunnel.
+    reach: Reach = Reach.INTERNAL
+
+    @model_validator(mode="before")
+    @classmethod
+    def _normalize_reach(cls, data: object) -> object:
+        return _reach_from_legacy(data, default=Reach.INTERNAL)
+
+    @model_validator(mode="after")
+    def _validate_reach(self) -> CaddyDeployment:
+        if self.reach == Reach.OFF:
+            raise ValueError("a static (caddy) deployment is always served; reach must be internal|public")
+        return self
+
+    @property
+    def public(self) -> bool:
+        return self.reach == Reach.PUBLIC
 
 
 class PathDeployment(DeploymentBase):

core/src/castle_core/registry.py

@@ -30,6 +30,8 @@ class NodeConfig:
     # Cloudflare tunnel: public services publish at <subdomain>.<public_domain>.
     public_domain: str | None = None
     tunnel_id: str | None = None
+    # Emit the cert_obtained → `castle tls reconcile` hook (needs events-exec plugin).
+    cert_hook: bool = False
 
     def __post_init__(self) -> None:
         if not self.hostname:
@@ -66,6 +68,9 @@ class Deployment:
     # Also projected to the public internet via the tunnel at
     # <subdomain>.<gateway.public_domain>. Requires subdomain.
     public: bool = False
+    # Raw-TCP exposure port (postgres, redis, …). Set → reachable at
+    # <name>.<gateway.domain>:<tcp_port> via bind + wildcard DNS (no Caddy route).
+    tcp_port: int | None = None
     # For `static` runner services: the absolute dir the gateway file_servers.
     # Set → the route is `static` (file_server) rather than `proxy` (reverse_proxy).
     static_root: str | None = None
@@ -113,6 +118,7 @@ def load_registry(path: Path | None = None) -> NodeRegistry:
         acme_dns_provider=node_data.get("acme_dns_provider", "cloudflare"),
         public_domain=node_data.get("public_domain"),
         tunnel_id=node_data.get("tunnel_id"),
+        cert_hook=node_data.get("cert_hook", False),
     )
 
     deployed: dict[str, Deployment] = {}
@@ -154,6 +160,7 @@ def load_registry(path: Path | None = None) -> NodeRegistry:
             health_path=comp_data.get("health_path"),
             subdomain=comp_data.get("subdomain"),
             public=comp_data.get("public", False),
+            tcp_port=comp_data.get("tcp_port"),
             static_root=comp_data.get("static_root"),
             base_url=comp_data.get("base_url"),
             schedule=comp_data.get("schedule"),
@@ -194,6 +201,8 @@ def save_registry(registry: NodeRegistry, path: Path | None = None) -> None:
         data["node"]["public_domain"] = registry.node.public_domain
     if registry.node.tunnel_id:
         data["node"]["tunnel_id"] = registry.node.tunnel_id
+    if registry.node.cert_hook:
+        data["node"]["cert_hook"] = registry.node.cert_hook
 
     for name, comp in registry.deployed.items():
         entry: dict = {
@@ -221,6 +230,8 @@ def save_registry(registry: NodeRegistry, path: Path | None = None) -> None:
             entry["subdomain"] = comp.subdomain
         if comp.public:
             entry["public"] = comp.public
+        if comp.tcp_port is not None:
+            entry["tcp_port"] = comp.tcp_port
         if comp.static_root:
             entry["static_root"] = comp.static_root
         if comp.base_url:

core/src/castle_core/tls.py

@@ -0,0 +1,239 @@
+"""Castle-managed TLS material for raw-TCP services.
+
+Cuts the gateway's ACME wildcard cert (valid for ``<name>.<domain>``) onto a
+service so it presents a *trusted* cert on its raw port, and refreshes it on
+renewal. Protocol-agnostic: castle only copies files (in the requested format)
+and signals the service — each deployment declares the format (``pair`` /
+``combined``) and how it consumes the files (via the ``${tls_*}`` placeholders).
+
+Two entry points:
+- ``materialize_all`` — write/refresh the cert files, no reload (used by ``apply``,
+  which (re)starts the service itself).
+- ``reconcile_tls`` — materialize *and* reload the services whose cert changed
+  (used by ``castle tls reconcile`` and the Caddy ``cert_obtained`` hook).
+"""
+
+from __future__ import annotations
+
+import os
+import re
+import subprocess
+import time
+from pathlib import Path
+
+from castle_core.config import DATA_DIR, CastleConfig
+from castle_core.manifest import SystemdDeployment, TlsMaterial
+
+_KEY_MODE_FILES = {"key.pem", "combined.pem"}  # secret → 0600; certs → 0644
+
+
+def _caddy_data_dir() -> Path:
+    xdg = os.environ.get("XDG_DATA_HOME")
+    base = Path(xdg) if xdg else Path.home() / ".local" / "share"
+    return base / "caddy"
+
+
+def wildcard_cert(domain: str) -> tuple[Path, Path] | None:
+    """``(crt, key)`` for ``*.<domain>`` from Caddy's store, or None.
+
+    Caddy stores it at ``certificates/<acme-dir>/wildcard_.<domain>/…``. Prefer a
+    production cert over staging (``CASTLE_ACME_STAGING=1`` yields a staging dir).
+    The ``.crt`` is the full chain (leaf + intermediates).
+    """
+    store = _caddy_data_dir() / "certificates"
+    if not store.is_dir():
+        return None
+    stem = f"wildcard_.{domain}"
+    matches = sorted(
+        store.glob(f"*/{stem}/{stem}.crt"),
+        key=lambda p: 1 if "staging" in str(p) else 0,  # prod (0) before staging (1)
+    )
+    for crt in matches:
+        key = crt.with_suffix(".key")
+        if key.exists():
+            return crt, key
+    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_of(dep: object) -> object | None:
+    """The active TlsSpec for a deployment, or None (not systemd / no tcp / off)."""
+    if not isinstance(dep, SystemdDeployment):
+        return None
+    tcp = dep.expose.tcp if dep.expose else None
+    tls = tcp.tls if tcp else None
+    if not tls or tls.material == TlsMaterial.OFF:
+        return None
+    return tls
+
+
+_PEM_CERT = re.compile(
+    rb"-----BEGIN CERTIFICATE-----.*?-----END CERTIFICATE-----\s*", re.DOTALL
+)
+
+
+def _issuer_chain(crt: bytes) -> bytes:
+    """The issuer chain from a leaf+chain ``.crt``: every cert *after* the leaf
+    (the intermediates), for ``${tls_ca}``. Empty when the ``.crt`` is a bare leaf
+    (a genuine LE cert always ships an intermediate, so this is non-empty in
+    practice)."""
+    blocks = [m.group(0) for m in _PEM_CERT.finditer(crt)]
+    return b"".join(blocks[1:])
+
+
+def _wanted_files(
+    tls_dir: Path, material: TlsMaterial, crt: bytes, key: bytes
+) -> dict[Path, bytes]:
+    """The exact file set for a material choice. ``chain.pem`` (for ``${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``/``combined.pem``, always
+    provided regardless of material."""
+    files: dict[Path, bytes] = {tls_dir / "chain.pem": _issuer_chain(crt)}
+    if material == TlsMaterial.PAIR:
+        files[tls_dir / "cert.pem"] = crt
+        files[tls_dir / "key.pem"] = key
+    elif material == TlsMaterial.COMBINED:
+        files[tls_dir / "combined.pem"] = key + crt
+    return files
+
+
+def materialize_tls(config: CastleConfig, name: str, dep: object) -> bool:
+    """Write ``dep``'s cert files from the wildcard, in its declared format.
+
+    Idempotent: returns ``False`` (no write) when the on-disk copy already matches
+    the source, so it's safe to call on every ``apply`` and every renewal event.
+    Returns ``True`` when files were (re)written. No reload here — see callers.
+    """
+    tls = _tls_of(dep)
+    if tls is None:
+        return False
+    domain = config.gateway.domain
+    if not domain:
+        return False
+    src = wildcard_cert(domain)
+    if src is None:
+        return False
+    crt_path, key_path = src
+    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)
+    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()):
+        return False  # already current
+
+    tls_dir.mkdir(parents=True, exist_ok=True)
+    os.chmod(tls_dir, 0o700)
+    # Drop files left over from a previous material choice.
+    for stale in ("cert.pem", "key.pem", "combined.pem", "chain.pem"):
+        p = tls_dir / stale
+        if p not in wanted and p.exists():
+            p.unlink()
+    for path, content in wanted.items():
+        fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
+        with os.fdopen(fd, "wb") as f:
+            f.write(content)
+        os.chmod(path, 0o600 if path.name in _KEY_MODE_FILES else 0o644)
+    return True
+
+
+def materialize_all(
+    config: CastleConfig,
+    messages: list[str] | None = None,
+    only: list[str] | None = None,
+) -> list[str]:
+    """Materialize certs for TLS-material deployments; no reload. For ``apply``,
+    which starts/restarts the services itself.
+
+    ``only`` scopes materialization to the given deployment names (what a scoped
+    ``castle apply <name>`` is converging). Left None → every deployment. Scoping
+    keeps a scoped apply from rewriting an *unrelated* service's cert on disk
+    without also reloading it (which would leave the file diverged from the running
+    process until the next ``castle tls reconcile``)."""
+    msgs = messages if messages is not None else []
+    scope = set(only) if only is not None else None
+    for name, dep in sorted(config.deployments.items()):
+        if scope is not None and name not in scope:
+            continue
+        if _tls_of(dep) is None:
+            continue
+        if materialize_tls(config, name, dep):
+            msgs.append(f"tls: materialized cert for {name}")
+    return msgs
+
+
+def wait_for_wildcard(
+    config: CastleConfig,
+    names: list[str],
+    messages: list[str] | None = None,
+    timeout: float = 120.0,
+    interval: float = 3.0,
+) -> list[str]:
+    """Block until the ACME wildcard cert exists, when an in-scope deployment needs
+    castle-materialized TLS but the cert isn't issued yet.
+
+    On a fresh node the gateway reload during ``apply`` only *kicks off* DNS-01
+    issuance of ``*.<domain>`` (seconds to a couple minutes); materializing right
+    after would find no cert and start the TLS service pointed at missing files —
+    and with ``gateway.cert_hook`` disabled (the default) nothing would later
+    reconcile it. Waiting here lets ``apply`` bring the service up with its cert in
+    place on first deploy. Bounded: on timeout it warns and returns so ``apply``
+    still proceeds (rerun ``castle tls reconcile`` once the cert lands)."""
+    msgs = messages if messages is not None else []
+    needs = [
+        n
+        for n in names
+        if (dep := config.deployments.get(n)) is not None and _tls_of(dep) is not None
+    ]
+    if not needs:
+        return msgs
+    domain = config.gateway.domain
+    if not domain or wildcard_cert(domain) is not None:
+        return msgs  # no acme domain, or the cert already exists — nothing to wait on
+    msgs.append(f"tls: waiting for ACME wildcard *.{domain} to be issued…")
+    deadline = time.monotonic() + timeout
+    while time.monotonic() < deadline:
+        time.sleep(interval)
+        if wildcard_cert(domain) is not None:
+            msgs.append(f"tls: wildcard *.{domain} issued")
+            return msgs
+    msgs.append(
+        f"tls: wildcard *.{domain} not ready after {int(timeout)}s — "
+        f"{', '.join(needs)} may start without a cert; rerun `castle tls reconcile` "
+        "once it is issued"
+    )
+    return msgs
+
+
+def _reload(name: str, tls: object, msgs: list[str]) -> None:
+    reload_cmd = getattr(tls, "reload", None)
+    if reload_cmd:
+        subprocess.run(reload_cmd, check=False)
+        msgs.append(f"tls: reloaded {name} (reload command)")
+    else:
+        subprocess.run(
+            ["systemctl", "--user", "restart", f"castle-{name}.service"], check=False
+        )
+        msgs.append(f"tls: restarted {name} to pick up rotated cert")
+
+
+def reconcile_tls(config: CastleConfig, messages: list[str] | None = None) -> list[str]:
+    """Materialize certs and reload the services whose cert changed. Idempotent —
+    a no-op when nothing rotated. Invoked by ``castle tls reconcile`` and the Caddy
+    ``cert_obtained`` hook. Logs its own outcome (the hook's ``exec`` swallows it)."""
+    msgs = messages if messages is not None else []
+    for name, dep in sorted(config.deployments.items()):
+        tls = _tls_of(dep)
+        if tls is None:
+            continue
+        if materialize_tls(config, name, dep):
+            msgs.append(f"tls: refreshed cert for {name}")
+            _reload(name, tls, msgs)
+    if not msgs:
+        msgs.append("tls: all materialized certs current — nothing to do")
+    return msgs

core/tests/test_caddyfile.py

@@ -185,6 +185,21 @@ class TestConfigSourceOfTruth:
         routes = compute_routes(_make_registry(), _config({"pg": _service(5432, expose=False)}))
         assert not [r for r in routes if r.name == "pg"]
 
+    def test_tcp_service_has_no_http_route(self) -> None:
+        """A raw-TCP service (reach: internal + expose.tcp) is reachable by
+        name+port via DNS, but must NOT get an HTTP gateway route."""
+        tcp = SystemdDeployment.model_validate(
+            {
+                "manager": "systemd",
+                "run": RunPython(launcher="python", program="pg"),
+                "reach": "internal",
+                "expose": {"tcp": {"port": 5432}},
+            }
+        )
+        assert tcp.tcp_port == 5432 and tcp.http_exposed is False
+        routes = compute_routes(_make_registry(), _config({"pg": tcp}))
+        assert not [r for r in routes if r.name == "pg"]
+
     def test_config_port_overrides_stale_registry(self) -> None:
         registry = _make_registry(
             deployed={"claw": _dep(8001, expose=True, name="claw")}

core/tests/test_deploy_run_cmd.py

@@ -107,6 +107,45 @@ def test_container_without_secrets_has_no_env_file() -> None:
     assert "--env-file" not in cmd
 
 
+def test_container_placeholders_expand_in_run_fields() -> None:
+    """${key} placeholders expand consistently across a container's env, volumes,
+    and args — env used to be the odd one out that passed through literally."""
+    run = RunContainer(
+        launcher="container",
+        image="img:latest",
+        env={"DATA": "${data_dir}/x"},
+        volumes=["${tls_dir}:/tls:ro"],
+        args=["--advertise", "${name}:${port}"],
+    )
+    ph = {
+        "name": "svc",
+        "port": "5432",
+        "data_dir": "/data/castle/svc",
+        "tls_dir": "/data/castle/svc/tls",
+    }
+    with patch("castle_core.deploy.shutil.which", return_value="/usr/bin/docker"):
+        cmd = _build_run_cmd("svc", run, {}, [], placeholders=ph)
+    joined = " ".join(cmd)
+    assert "DATA=/data/castle/svc/x" in joined       # env expanded
+    assert "/data/castle/svc/tls:/tls:ro" in joined  # volume expanded
+    assert "svc:5432" in cmd                          # arg expanded
+
+
+def test_container_double_dollar_escapes_placeholder() -> None:
+    """$${key} passes a literal ${key} through to the container's own shell/env
+    instead of castle expanding it (docker-compose-style escape)."""
+    run = RunContainer(
+        launcher="container",
+        image="img:latest",
+        args=["sh", "-c", "exec myd --advertise $${name}:${port}"],
+    )
+    ph = {"name": "svc", "port": "5432"}
+    with patch("castle_core.deploy.shutil.which", return_value="/usr/bin/docker"):
+        cmd = _build_run_cmd("svc", run, {}, [], placeholders=ph)
+    # castle expands ${port} but leaves $${name} as a literal ${name} for the shell
+    assert "exec myd --advertise ${name}:5432" in cmd
+
+
 def test_compose_runner_up_with_resolved_file(tmp_path: Path) -> None:
     """A compose service runs `docker compose -p <project> -f <abs-file> up`."""
     run = RunCompose(launcher="compose", file="docker-compose.yml")

core/tests/test_manifest.py

@@ -4,6 +4,7 @@ from __future__ import annotations
 
 import pytest
 from castle_core.manifest import (
+    Reach,
     BuildSpec,
     CaddyDeployment,
     ExposeSpec,
@@ -110,14 +111,78 @@ class TestSystemdDeployment:
         )
         assert s.manage.systemd.enable is True
 
-    def test_public_requires_proxy(self) -> None:
-        """public without proxy is invalid (public needs an exposed process)."""
-        with pytest.raises(ValueError, match="public requires proxy"):
-            SystemdDeployment(
-                id="bad",
-                manager="systemd",
-                run=RunPython(launcher="python", program="svc"),
-                public=True,
+    def test_reach_ladder_and_legacy_mapping(self) -> None:
+        """`reach` is canonical; legacy proxy/public map to it, and the derived
+        proxy/public accessors reflect it (public implies internal)."""
+        # An exposed reach needs an expose block (see test_reach_requires_expose),
+        # so give the base one; reach off doesn't, tested separately below.
+        base = dict(
+            id="svc",
+            manager="systemd",
+            run=RunPython(launcher="python", program="svc"),
+            expose={"http": {"internal": {"port": 9001}}},
+        )
+        # legacy input still parses
+        s_proxy = SystemdDeployment.model_validate({**base, "proxy": True})
+        assert s_proxy.reach == Reach.INTERNAL
+        assert s_proxy.proxy is True and s_proxy.public is False
+        s_pub = SystemdDeployment.model_validate({**base, "proxy": True, "public": True})
+        assert s_pub.reach == Reach.PUBLIC
+        assert s_pub.proxy is True and s_pub.public is True
+        # legacy public alone now simply means public (which implies internal)
+        assert SystemdDeployment.model_validate({**base, "public": True}).reach == Reach.PUBLIC
+        # new canonical field (reach off needs no expose block)
+        no_expose = dict(
+            id="svc", manager="systemd", run=RunPython(launcher="python", program="svc")
+        )
+        assert SystemdDeployment(**no_expose, reach=Reach.OFF).reach == Reach.OFF
+        assert SystemdDeployment(**base, reach=Reach.PUBLIC).public is True
+
+    def test_reach_requires_expose(self) -> None:
+        """An exposed reach with no expose block is rejected (it would otherwise
+        silently no-op — no route, no subdomain, no tunnel). Replaces the old
+        'public requires proxy' guard."""
+        base = dict(manager="systemd", run=RunPython(launcher="python", program="svc"))
+        for reach in ("internal", "public"):
+            with pytest.raises(ValueError, match="requires an `expose` block"):
+                SystemdDeployment.model_validate({**base, "reach": reach})
+        # legacy public: true with no expose maps to reach public → same rejection
+        with pytest.raises(ValueError, match="requires an `expose` block"):
+            SystemdDeployment.model_validate({**base, "public": True})
+
+    def test_tcp_exposure_is_not_http_exposed(self) -> None:
+        """A raw-TCP service is reachable by name+port but never HTTP-routed."""
+        base = dict(manager="systemd", run=RunCommand(launcher="command", argv=["pg"]))
+        tcp = SystemdDeployment.model_validate(
+            {**base, "reach": "internal", "expose": {"tcp": {"port": 5432}}}
+        )
+        assert tcp.tcp_port == 5432
+        assert tcp.http_exposed is False  # <-- no Caddy route
+        http = SystemdDeployment.model_validate(
+            {**base, "reach": "internal", "expose": {"http": {"internal": {"port": 9001}}}}
+        )
+        assert http.http_exposed is True and http.tcp_port is None
+
+    def test_expose_is_one_protocol(self) -> None:
+        with pytest.raises(ValueError, match="http OR tcp"):
+            SystemdDeployment.model_validate(
+                {
+                    "manager": "systemd",
+                    "run": RunCommand(launcher="command", argv=["x"]),
+                    "expose": {"http": {"internal": {"port": 1}}, "tcp": {"port": 2}},
+                }
+            )
+
+    def test_public_tcp_guarded(self) -> None:
+        """reach: public on a raw-TCP service is rejected until step 5 lands."""
+        with pytest.raises(ValueError, match="public for a raw-TCP"):
+            SystemdDeployment.model_validate(
+                {
+                    "manager": "systemd",
+                    "run": RunCommand(launcher="command", argv=["x"]),
+                    "reach": "public",
+                    "expose": {"tcp": {"port": 5432}},
+                }
             )
 
     def test_no_run_is_invalid(self) -> None:
@@ -166,11 +231,12 @@ class TestModelSerialization:
                     internal=HttpInternal(port=9001), health_path="/health"
                 )
             ),
-            proxy=True,
+            reach=Reach.INTERNAL,
             manage=ManageSpec(systemd=SystemdSpec()),
         )
         data = s.model_dump(exclude_none=True, exclude={"id"})
         assert data["manager"] == "systemd"
         assert data["run"]["launcher"] == "python"
         assert data["expose"]["http"]["internal"]["port"] == 9001
-        assert data["proxy"] is True
+        assert data["reach"] == "internal"
+        assert "proxy" not in data  # derived accessor, not serialized

core/tests/test_tls.py

@@ -0,0 +1,112 @@
+"""Tests for castle-managed TLS material (core/src/castle_core/tls.py)."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import pytest
+
+import castle_core.config as C
+import castle_core.tls as T
+from castle_core.manifest import SystemdDeployment
+
+
+def _write_wildcard(xdg: Path, domain: str, tag: str, acme_dir: str) -> None:
+    d = xdg / "caddy" / "certificates" / acme_dir / f"wildcard_.{domain}"
+    d.mkdir(parents=True, exist_ok=True)
+    (d / f"wildcard_.{domain}.crt").write_bytes(f"CERT-{tag}\n".encode())
+    (d / f"wildcard_.{domain}.key").write_bytes(f"KEY-{tag}\n".encode())
+
+
+@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."""
+    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")
+    monkeypatch.setenv("XDG_DATA_HOME", str(xdg))
+    monkeypatch.setattr(T, "DATA_DIR", tmp_path / "data")  # tls_dir_for reads this
+    return T, C, domain
+
+
+def _cfg(C, domain, dep):
+    return C.CastleConfig(
+        root=None, gateway=C.GatewayConfig(port=9000, domain=domain), repo=None,
+        programs={}, deployments={"postgres": dep},
+    )
+
+
+def _pg(material: str):
+    return SystemdDeployment.model_validate(
+        {
+            "manager": "systemd",
+            "program": "postgres",
+            "run": {"launcher": "container", "image": "postgres:17"},
+            "reach": "internal",
+            "expose": {"tcp": {"port": 5432, "tls": {"material": material}}},
+        }
+    )
+
+
+def test_prefers_prod_over_staging(tls_env) -> None:
+    T, _, domain = 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
+    pg = _pg("pair")
+    cfg = _cfg(C, domain, pg)
+    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")
+    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 / "cert.pem").stat().st_mode)[-3:] == "644"  # public
+
+
+def test_material_switch_cleans_stale(tls_env) -> None:
+    T, C, domain = tls_env
+    pair = _pg("pair")
+    T.materialize_tls(_cfg(C, domain, pair), "postgres", pair)
+    combined = _pg("combined")
+    assert T.materialize_tls(_cfg(C, domain, combined), "postgres", combined) is True
+    td = T.tls_dir_for("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"
+
+
+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
+    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}"
+    )
+    (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 (td / "cert.pem").read_bytes() == leaf + inter  # server presents leaf+chain
+    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
+    off = SystemdDeployment.model_validate(
+        {"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()