feat(compose): pure renderer for bottle plan -> compose dict

PRD 0018 chunk 1. New module `claude_bottle/backend/docker/compose.py`
exposing `bottle_plan_to_compose(plan) -> dict` — a pure function that
translates a fully-resolved DockerBottlePlan into a Compose v2 spec.

Not wired in yet. Tests cover the conditional-service matrix (git
on/off × egress on/off × supervise on/off) plus per-service shape
(images vs builds, network aliases, bind mounts, env vars, depends_on).

claude_bottle/backend/docker/compose.py

@@ -0,0 +1,385 @@
+"""Compose-spec rendering for a Docker bottle (PRD 0018, chunk 1).
+
+`bottle_plan_to_compose(plan)` returns a Compose v2 spec dict
+describing the per-bottle container topology — one project per
+bottle instance, services for the agent + every applicable sidecar,
+two networks, no named volumes.
+
+Pure function. No I/O, no subprocess. Expects every launch-time
+field (network names, CA host paths, etc.) on the plan's inner
+plans to be populated; chunks 2+3 own that ordering. Chunk 1 just
+encodes the translation so it can be unit-tested in isolation.
+
+Conditional services follow the plan content (matches the
+SDK-call branching in `launch.py` today):
+
+  - pipelock + agent: always.
+  - git-gate:         iff plan.git_gate_plan.upstreams.
+  - egress:           iff plan.egress_plan.routes.
+  - supervise:        iff plan.supervise_plan is not None.
+
+Naming:
+
+  - Compose project: `claude-bottle-<slug>`.
+  - Service names (inside the file): `agent`, `pipelock`,
+    `egress`, `git-gate`, `supervise`.
+  - `container_name:` matches today's pattern
+    (`claude-bottle-<service>-<slug>`) so dashboard/cleanup discovery
+    via the prefix scan keeps working through the transition.
+  - Network aliases preserve the current dial-by-shortname pattern
+    for `egress` / `supervise`, and add the long container-name as
+    an internal-network alias for `pipelock` / `git-gate` so any
+    caller still referencing the long name resolves.
+
+Sidecars that are built (egress, git-gate, supervise) get a
+compose `build:` block pointing at the repo Dockerfile; the
+`image:` tag is set explicitly so cached images on the daemon
+aren't rebuilt on every up.
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+from typing import Any
+
+from ...egress import (
+    EGRESS_HOSTNAME,
+    EGRESS_ROUTES_IN_CONTAINER,
+)
+from ...git_gate import git_gate_aggregate_extra_hosts
+from ...supervise import (
+    CURRENT_CONFIG_DIR_IN_AGENT,
+    QUEUE_DIR_IN_CONTAINER,
+    SUPERVISE_HOSTNAME,
+    SUPERVISE_PORT,
+)
+from ...util import expand_tilde
+from .bottle_plan import DockerBottlePlan
+from .egress import (
+    EGRESS_CA_IN_CONTAINER,
+    EGRESS_DOCKERFILE,
+    EGRESS_IMAGE,
+    EGRESS_PIPELOCK_CA_IN_CONTAINER,
+    egress_container_name,
+)
+from .git_gate import (
+    GIT_GATE_ACCESS_HOOK_IN_CONTAINER,
+    GIT_GATE_CREDS_DIR_IN_CONTAINER,
+    GIT_GATE_DOCKERFILE,
+    GIT_GATE_ENTRYPOINT_IN_CONTAINER,
+    GIT_GATE_HOOK_IN_CONTAINER,
+    GIT_GATE_IMAGE,
+    git_gate_container_name,
+)
+from .pipelock import (
+    PIPELOCK_CA_CERT_IN_CONTAINER,
+    PIPELOCK_CA_KEY_IN_CONTAINER,
+    PIPELOCK_IMAGE,
+    PIPELOCK_PORT,
+    pipelock_container_name,
+)
+from .provision.ca import AGENT_CA_BUNDLE, AGENT_CA_PATH
+from .supervise import (
+    SUPERVISE_DOCKERFILE,
+    SUPERVISE_IMAGE,
+    supervise_container_name,
+)
+
+
+# Repo root, used as the build context for sidecar Dockerfiles.
+# Same derivation as the per-sidecar lifecycle modules.
+_REPO_DIR = str(Path(__file__).resolve().parent.parent.parent.parent)
+
+
+def bottle_plan_to_compose(plan: DockerBottlePlan) -> dict[str, Any]:
+    """Render a Compose v2 spec dict from a fully-resolved
+    DockerBottlePlan.
+
+    The plan must have its inner plans (`proxy_plan`,
+    `git_gate_plan`, `egress_plan`, `supervise_plan`) populated
+    with launch-time fields — network names, CA host paths,
+    pipelock_proxy_url. The renderer doesn't validate; callers
+    feed it a fully-resolved plan or get an incomplete compose
+    spec back.
+    """
+    project = f"claude-bottle-{plan.slug}"
+    services: dict[str, Any] = {}
+
+    services["pipelock"] = _pipelock_service(plan)
+
+    if plan.git_gate_plan.upstreams:
+        services["git-gate"] = _git_gate_service(plan)
+
+    if plan.egress_plan.routes:
+        services["egress"] = _egress_service(plan)
+
+    if plan.supervise_plan is not None:
+        services["supervise"] = _supervise_service(plan)
+
+    services["agent"] = _agent_service(plan)
+
+    return {
+        "name": project,
+        "services": services,
+        "networks": _networks(plan),
+    }
+
+
+def _networks(plan: DockerBottlePlan) -> dict[str, Any]:
+    """Two compose-managed networks with explicit `name:` matching
+    the existing slug-suffixed convention. The internal one is
+    `--internal` (no default gateway); the egress one is a normal
+    user-defined bridge so the upstream-bound sidecars can resolve
+    + reach the outside world."""
+    return {
+        "internal": {
+            "name": plan.proxy_plan.internal_network,
+            "internal": True,
+        },
+        "egress": {
+            "name": plan.proxy_plan.egress_network,
+        },
+    }
+
+
+def _bind(host: str | Path, target: str, *, read_only: bool = True) -> dict[str, Any]:
+    """One bind-mount entry in the long-form `volumes:` shape.
+    Long form is preferred over `host:target:ro` strings because
+    it's easier to inspect in tests and survives whitespace in
+    host paths."""
+    return {
+        "type": "bind",
+        "source": str(host),
+        "target": target,
+        "read_only": read_only,
+    }
+
+
+def _pipelock_service(plan: DockerBottlePlan) -> dict[str, Any]:
+    """Pipelock sidecar. Pinned-digest image (no build). The
+    rendered YAML config + CA cert + key bind-mount in from the
+    paths the prepare step laid down on plan.proxy_plan."""
+    pp = plan.proxy_plan
+    name = pipelock_container_name(plan.slug)
+    return {
+        "image": PIPELOCK_IMAGE,
+        "container_name": name,
+        "command": [
+            "run",
+            "--config", "/etc/pipelock.yaml",
+            "--listen", f"0.0.0.0:{PIPELOCK_PORT}",
+        ],
+        "networks": {
+            "internal": {"aliases": [name]},
+            "egress": None,
+        },
+        "volumes": [
+            _bind(pp.yaml_path, "/etc/pipelock.yaml"),
+            _bind(pp.ca_cert_host_path, PIPELOCK_CA_CERT_IN_CONTAINER),
+            _bind(pp.ca_key_host_path, PIPELOCK_CA_KEY_IN_CONTAINER),
+        ],
+    }
+
+
+def _git_gate_service(plan: DockerBottlePlan) -> dict[str, Any]:
+    """Git-gate sidecar. Built from Dockerfile.git-gate. Entrypoint
+    + pre-receive hook + access-hook bind-mount from the stage
+    paths the prepare step wrote. Per-upstream identity files
+    bind-mount from the user's ssh-key location after `~`
+    expansion. Per-upstream known_hosts files come in via chunk 2 —
+    the GitGatePlan doesn't carry those host paths yet (they're
+    currently materialized at start time by DockerGitGate.start).
+    """
+    gp = plan.git_gate_plan
+    name = git_gate_container_name(plan.slug)
+
+    volumes: list[dict[str, Any]] = [
+        _bind(gp.entrypoint_script, GIT_GATE_ENTRYPOINT_IN_CONTAINER),
+        _bind(gp.hook_script, GIT_GATE_HOOK_IN_CONTAINER),
+        _bind(gp.access_hook_script, GIT_GATE_ACCESS_HOOK_IN_CONTAINER),
+    ]
+    for u in gp.upstreams:
+        keypath = expand_tilde(u.identity_file)
+        volumes.append(_bind(
+            keypath,
+            f"{GIT_GATE_CREDS_DIR_IN_CONTAINER}/{u.name}-key",
+        ))
+
+    service: dict[str, Any] = {
+        "image": GIT_GATE_IMAGE,
+        "build": {
+            "context": _REPO_DIR,
+            "dockerfile": GIT_GATE_DOCKERFILE,
+        },
+        "container_name": name,
+        "networks": {
+            "internal": {"aliases": [name]},
+            "egress": None,
+        },
+        "volumes": volumes,
+    }
+    extra_hosts = git_gate_aggregate_extra_hosts(gp.upstreams)
+    if extra_hosts:
+        service["extra_hosts"] = [
+            f"{host}:{ip}" for host, ip in sorted(extra_hosts.items())
+        ]
+    return service
+
+
+def _egress_service(plan: DockerBottlePlan) -> dict[str, Any]:
+    """Egress sidecar. Built from Dockerfile.egress. Routes
+    through pipelock on its upstream leg via `EGRESS_UPSTREAM_PROXY` +
+    `EGRESS_UPSTREAM_CA`. One env-list entry per upstream-token slot
+    (bare NAME inherits from the compose-up process env, so secret
+    values stay off argv and out of the compose file). routes.yaml +
+    mitmproxy CA + pipelock CA bind-mount from the stage paths."""
+    ep = plan.egress_plan
+    name = egress_container_name(plan.slug)
+
+    env: list[str] = [
+        f"EGRESS_UPSTREAM_PROXY={ep.pipelock_proxy_url}",
+        f"HTTPS_PROXY={ep.pipelock_proxy_url}",
+        f"HTTP_PROXY={ep.pipelock_proxy_url}",
+        "NO_PROXY=localhost,127.0.0.1",
+        f"EGRESS_UPSTREAM_CA={EGRESS_PIPELOCK_CA_IN_CONTAINER}",
+    ]
+    for token_env in sorted(ep.token_env_map.keys()):
+        env.append(token_env)
+
+    return {
+        "image": EGRESS_IMAGE,
+        "build": {
+            "context": _REPO_DIR,
+            "dockerfile": EGRESS_DOCKERFILE,
+        },
+        "container_name": name,
+        "networks": {
+            "internal": {"aliases": [EGRESS_HOSTNAME]},
+            "egress": None,
+        },
+        "environment": env,
+        "volumes": [
+            _bind(ep.routes_path, EGRESS_ROUTES_IN_CONTAINER),
+            _bind(ep.mitmproxy_ca_host_path, EGRESS_CA_IN_CONTAINER),
+            _bind(ep.pipelock_ca_host_path, EGRESS_PIPELOCK_CA_IN_CONTAINER),
+        ],
+        "depends_on": ["pipelock"],
+    }
+
+
+def _supervise_service(plan: DockerBottlePlan) -> dict[str, Any]:
+    """Supervise sidecar. Internal network only — no upstream calls.
+    Queue dir bind-mounts read-write so the sidecar can append audit
+    events and the host-side capability handlers can drop new
+    proposals into it."""
+    sp = plan.supervise_plan
+    assert sp is not None
+    name = supervise_container_name(plan.slug)
+    return {
+        "image": SUPERVISE_IMAGE,
+        "build": {
+            "context": _REPO_DIR,
+            "dockerfile": SUPERVISE_DOCKERFILE,
+        },
+        "container_name": name,
+        "networks": {
+            "internal": {"aliases": [SUPERVISE_HOSTNAME]},
+        },
+        "environment": [
+            f"SUPERVISE_BOTTLE_SLUG={plan.slug}",
+            f"SUPERVISE_QUEUE_DIR={QUEUE_DIR_IN_CONTAINER}",
+            f"SUPERVISE_PORT={SUPERVISE_PORT}",
+        ],
+        "volumes": [
+            {
+                "type": "bind",
+                "source": str(sp.queue_dir),
+                "target": QUEUE_DIR_IN_CONTAINER,
+                "read_only": False,
+            },
+        ],
+    }
+
+
+def _agent_service(plan: DockerBottlePlan) -> dict[str, Any]:
+    """Agent container. Runs `sleep infinity`; claude is `docker
+    exec -it`'d into it later. No TTY at the container level —
+    interactivity is per-exec. HTTP_PROXY/HTTPS_PROXY point at the
+    egress short-alias when an egress is declared, otherwise
+    straight at pipelock's container name. CA trust trio matches
+    the existing launch.py wiring."""
+    proxy_url = _agent_proxy_url(plan)
+    no_proxy = _agent_no_proxy(plan)
+    env: list[str] = [
+        f"HTTPS_PROXY={proxy_url}",
+        f"HTTP_PROXY={proxy_url}",
+        f"https_proxy={proxy_url}",
+        f"http_proxy={proxy_url}",
+        f"NO_PROXY={no_proxy}",
+        f"no_proxy={no_proxy}",
+        f"NODE_EXTRA_CA_CERTS={AGENT_CA_PATH}",
+        f"SSL_CERT_FILE={AGENT_CA_BUNDLE}",
+        f"REQUESTS_CA_BUNDLE={AGENT_CA_BUNDLE}",
+    ]
+    # Forwarded vars (OAuth token, manifest host-interpolations):
+    # bare name → inherits from compose-up process env, value
+    # never lands on argv or in the compose file.
+    for name in sorted(plan.forwarded_env.keys()):
+        env.append(name)
+
+    service: dict[str, Any] = {
+        "image": plan.runtime_image,
+        "container_name": plan.container_name,
+        "command": ["sleep", "infinity"],
+        "networks": {"internal": None},
+        "environment": env,
+    }
+    if plan.use_runsc:
+        service["runtime"] = "runsc"
+    if plan.env_file and plan.env_file.exists() and plan.env_file.stat().st_size > 0:
+        service["env_file"] = [str(plan.env_file)]
+
+    volumes: list[dict[str, Any]] = []
+    if plan.supervise_plan is not None:
+        volumes.append(_bind(
+            plan.supervise_plan.current_config_dir,
+            CURRENT_CONFIG_DIR_IN_AGENT,
+        ))
+    if volumes:
+        service["volumes"] = volumes
+
+    depends_on = ["pipelock"]
+    if plan.git_gate_plan.upstreams:
+        depends_on.append("git-gate")
+    if plan.egress_plan.routes:
+        depends_on.append("egress")
+    if plan.supervise_plan is not None:
+        depends_on.append("supervise")
+    service["depends_on"] = depends_on
+
+    return service
+
+
+def _agent_proxy_url(plan: DockerBottlePlan) -> str:
+    """Pick the agent's HTTP_PROXY. With egress declared, the agent
+    goes through egress (which in turn HTTPS_PROXYs to pipelock on
+    its outbound leg). Without egress, the agent talks straight to
+    pipelock."""
+    if plan.egress_plan.routes:
+        from .egress import EGRESS_PORT
+        return f"http://{EGRESS_HOSTNAME}:{EGRESS_PORT}"
+    return f"http://{pipelock_container_name(plan.slug)}:{PIPELOCK_PORT}"
+
+
+def _agent_no_proxy(plan: DockerBottlePlan) -> str:
+    """NO_PROXY for the agent. Matches the launch.py rules:
+    loopback always, supervise hostname when the supervise sidecar
+    is up (the MCP long-poll pattern needs to bypass pipelock's
+    idle timeout)."""
+    hosts = ["localhost", "127.0.0.1"]
+    if plan.supervise_plan is not None:
+        hosts.append(SUPERVISE_HOSTNAME)
+    return ",".join(hosts)
+
+
+__all__ = ["bottle_plan_to_compose"]

tests/unit/test_compose.py

@@ -0,0 +1,454 @@
+"""Unit: compose-spec renderer (PRD 0018 chunk 1).
+
+Pure-function tests for `bottle_plan_to_compose`. Fixtures build a
+fully-resolved DockerBottlePlan in memory; the renderer just
+translates it to the compose dict. Conditional-service matrix is
+covered via parameterized cases (git on/off × egress on/off ×
+supervise on/off).
+"""
+
+from __future__ import annotations
+
+import unittest
+from pathlib import Path
+
+from claude_bottle.backend import BottleSpec
+from claude_bottle.backend.docker.bottle_plan import DockerBottlePlan
+from claude_bottle.backend.docker.compose import bottle_plan_to_compose
+from claude_bottle.egress import (
+    EgressPlan,
+    EgressRoute,
+)
+from claude_bottle.git_gate import GitGatePlan, GitGateUpstream
+from claude_bottle.manifest import Manifest
+from claude_bottle.pipelock import PipelockProxyPlan
+from claude_bottle.supervise import SupervisePlan
+
+
+SLUG = "demo-abc12"
+STAGE = Path("/tmp/cb-stage")
+STATE = Path("/tmp/cb-state")
+
+
+def _manifest(*, supervise: bool, with_git: bool, with_egress: bool) -> Manifest:
+    """Minimal manifest with the toggles the chunk-1 matrix needs.
+    The renderer only reads from the plan, not the manifest, so this
+    is just here to back BottleSpec."""
+    bottle: dict = {}
+    if supervise:
+        bottle["supervise"] = True
+    if with_git:
+        bottle["git"] = [{
+            "Name": "upstream",
+            "Upstream": "ssh://git@example.com:22/x/y.git",
+            "IdentityFile": "/etc/hostname",  # any existing file
+        }]
+    if with_egress:
+        bottle["egress"] = {
+            "routes": [{
+                "host": "api.example",
+                "auth": {"scheme": "Bearer", "token_ref": "TOK"},
+            }],
+        }
+    return Manifest.from_json_obj({
+        "bottles": {"dev": bottle},
+        "agents": {"demo": {"skills": [], "prompt": "", "bottle": "dev"}},
+    })
+
+
+def _spec(*, supervise: bool, with_git: bool, with_egress: bool) -> BottleSpec:
+    return BottleSpec(
+        manifest=_manifest(
+            supervise=supervise, with_git=with_git, with_egress=with_egress,
+        ),
+        agent_name="demo",
+        copy_cwd=False,
+        user_cwd="/tmp/x",
+    )
+
+
+def _proxy_plan() -> PipelockProxyPlan:
+    return PipelockProxyPlan(
+        yaml_path=STATE / "pipelock.yaml",
+        slug=SLUG,
+        internal_network=f"claude-bottle-net-{SLUG}",
+        internal_network_cidr="10.1.2.0/24",
+        egress_network=f"claude-bottle-egress-{SLUG}",
+        ca_cert_host_path=STATE / "pipelock-ca" / "ca.pem",
+        ca_key_host_path=STATE / "pipelock-ca" / "ca-key.pem",
+    )
+
+
+def _git_gate_plan(upstreams: tuple[GitGateUpstream, ...] = ()) -> GitGatePlan:
+    return GitGatePlan(
+        slug=SLUG,
+        entrypoint_script=STATE / "git-gate" / "entrypoint.sh",
+        hook_script=STATE / "git-gate" / "pre-receive",
+        access_hook_script=STATE / "git-gate" / "access-hook",
+        upstreams=upstreams,
+        internal_network=f"claude-bottle-net-{SLUG}",
+        egress_network=f"claude-bottle-egress-{SLUG}",
+    )
+
+
+def _egress_plan(routes: tuple[EgressRoute, ...] = ()) -> EgressPlan:
+    token_env_map = {
+        r.token_env: r.token_ref
+        for r in routes
+        if r.token_env
+    }
+    return EgressPlan(
+        slug=SLUG,
+        routes_path=STATE / "egress" / "routes.yaml",
+        routes=routes,
+        token_env_map=token_env_map,
+        internal_network=f"claude-bottle-net-{SLUG}",
+        egress_network=f"claude-bottle-egress-{SLUG}",
+        mitmproxy_ca_host_path=STATE / "egress-ca" / "mitmproxy-ca.pem",
+        mitmproxy_ca_cert_only_host_path=STATE / "egress-ca" / "ca.pem",
+        pipelock_ca_host_path=STATE / "pipelock-ca" / "ca.pem",
+        pipelock_proxy_url=f"http://claude-bottle-pipelock-{SLUG}:8888",
+    )
+
+
+def _supervise_plan() -> SupervisePlan:
+    return SupervisePlan(
+        slug=SLUG,
+        queue_dir=STATE / "supervise" / "queue",
+        current_config_dir=STATE / "supervise" / "current-config",
+        internal_network=f"claude-bottle-net-{SLUG}",
+    )
+
+
+def _plan(
+    *,
+    with_git: bool = False,
+    with_egress: bool = False,
+    supervise: bool = False,
+) -> DockerBottlePlan:
+    """Build a fully-resolved DockerBottlePlan. Toggles cover the
+    matrix the renderer's conditional-service logic branches on."""
+    upstreams: tuple[GitGateUpstream, ...] = ()
+    if with_git:
+        upstreams = (GitGateUpstream(
+            name="upstream",
+            upstream_url="ssh://git@example.com:22/x/y.git",
+            upstream_host="example.com",
+            upstream_port="22",
+            identity_file="/etc/hostname",
+            known_host_key="",
+            extra_hosts={"example.com": "10.0.0.1"},
+        ),)
+    routes: tuple[EgressRoute, ...] = ()
+    if with_egress:
+        routes = (EgressRoute(
+            host="api.example",
+            auth_scheme="Bearer",
+            token_env="EGRESS_TOKEN_0",
+            token_ref="TOK",
+            path_allowlist=(),
+            roles=(),
+        ),)
+
+    return DockerBottlePlan(
+        spec=_spec(supervise=supervise, with_git=with_git, with_egress=with_egress),
+        stage_dir=STAGE,
+        slug=SLUG,
+        container_name=f"claude-bottle-{SLUG}",
+        container_name_pinned=False,
+        image="claude-bottle:latest",
+        derived_image="",
+        runtime_image="claude-bottle:latest",
+        dockerfile_path="",
+        env_file=Path("/dev/null"),  # exists, size 0 → renderer skips env_file
+        forwarded_env={"CLAUDE_CODE_OAUTH_TOKEN": "x"},
+        prompt_file=STAGE / "prompt",
+        proxy_plan=_proxy_plan(),
+        git_gate_plan=_git_gate_plan(upstreams),
+        egress_plan=_egress_plan(routes),
+        supervise_plan=_supervise_plan() if supervise else None,
+        use_runsc=False,
+    )
+
+
+class TestProjectAndNetworks(unittest.TestCase):
+    def test_project_name(self):
+        spec = bottle_plan_to_compose(_plan())
+        self.assertEqual(f"claude-bottle-{SLUG}", spec["name"])
+
+    def test_internal_network_is_internal(self):
+        spec = bottle_plan_to_compose(_plan())
+        net = spec["networks"]["internal"]
+        self.assertEqual(f"claude-bottle-net-{SLUG}", net["name"])
+        self.assertTrue(net["internal"])
+
+    def test_egress_network_is_external_bridge(self):
+        spec = bottle_plan_to_compose(_plan())
+        net = spec["networks"]["egress"]
+        self.assertEqual(f"claude-bottle-egress-{SLUG}", net["name"])
+        # No `internal:` key on the egress network — defaults to a
+        # normal user-defined bridge.
+        self.assertNotIn("internal", net)
+
+
+class TestPipelockAlwaysPresent(unittest.TestCase):
+    """Pipelock is unconditional — every bottle has the SSRF guard +
+    body scanner sitting on its upstream leg."""
+
+    def test_minimal_plan_has_pipelock(self):
+        spec = bottle_plan_to_compose(_plan())
+        self.assertIn("pipelock", spec["services"])
+
+    def test_pipelock_pinned_image_no_build(self):
+        s = bottle_plan_to_compose(_plan())["services"]["pipelock"]
+        self.assertTrue(s["image"].startswith("ghcr.io/luckypipewrench/pipelock"))
+        self.assertNotIn("build", s)
+
+    def test_pipelock_container_name(self):
+        s = bottle_plan_to_compose(_plan())["services"]["pipelock"]
+        self.assertEqual(f"claude-bottle-pipelock-{SLUG}", s["container_name"])
+
+    def test_pipelock_on_both_networks(self):
+        s = bottle_plan_to_compose(_plan())["services"]["pipelock"]
+        self.assertIn("internal", s["networks"])
+        self.assertIn("egress", s["networks"])
+
+    def test_pipelock_long_name_alias_on_internal(self):
+        # Backward compat: anything still dialing pipelock by
+        # `claude-bottle-pipelock-<slug>` resolves on the internal
+        # network.
+        s = bottle_plan_to_compose(_plan())["services"]["pipelock"]
+        aliases = s["networks"]["internal"]["aliases"]
+        self.assertIn(f"claude-bottle-pipelock-{SLUG}", aliases)
+
+    def test_pipelock_bind_mounts(self):
+        s = bottle_plan_to_compose(_plan())["services"]["pipelock"]
+        targets = {v["target"] for v in s["volumes"]}
+        self.assertEqual(
+            {"/etc/pipelock.yaml", "/etc/pipelock-ca.pem", "/etc/pipelock-ca-key.pem"},
+            targets,
+        )
+        for v in s["volumes"]:
+            self.assertEqual("bind", v["type"])
+            self.assertTrue(v["read_only"])
+
+    def test_pipelock_command(self):
+        s = bottle_plan_to_compose(_plan())["services"]["pipelock"]
+        self.assertEqual(
+            ["run", "--config", "/etc/pipelock.yaml", "--listen", "0.0.0.0:8888"],
+            s["command"],
+        )
+
+
+class TestAgentAlwaysPresent(unittest.TestCase):
+    def test_agent_in_services(self):
+        s = bottle_plan_to_compose(_plan())["services"]
+        self.assertIn("agent", s)
+
+    def test_agent_command(self):
+        s = bottle_plan_to_compose(_plan())["services"]["agent"]
+        self.assertEqual(["sleep", "infinity"], s["command"])
+
+    def test_agent_image_uses_runtime_image(self):
+        plan = _plan()
+        s = bottle_plan_to_compose(plan)["services"]["agent"]
+        self.assertEqual(plan.runtime_image, s["image"])
+
+    def test_agent_only_on_internal_network(self):
+        s = bottle_plan_to_compose(_plan())["services"]["agent"]
+        self.assertEqual({"internal"}, set(s["networks"].keys()))
+
+    def test_agent_proxy_via_pipelock_when_no_egress(self):
+        s = bottle_plan_to_compose(_plan(with_egress=False))["services"]["agent"]
+        env = s["environment"]
+        # Looking for HTTPS_PROXY pointing at pipelock's container name.
+        proxy_lines = [e for e in env if e.startswith("HTTPS_PROXY=")]
+        self.assertEqual(1, len(proxy_lines))
+        self.assertEqual(
+            f"HTTPS_PROXY=http://claude-bottle-pipelock-{SLUG}:8888",
+            proxy_lines[0],
+        )
+
+    def test_agent_proxy_via_egress_when_egress_present(self):
+        s = bottle_plan_to_compose(_plan(with_egress=True))["services"]["agent"]
+        proxy = [e for e in s["environment"] if e.startswith("HTTPS_PROXY=")][0]
+        self.assertEqual("HTTPS_PROXY=http://egress:9099", proxy)
+
+    def test_agent_no_proxy_adds_supervise_when_enabled(self):
+        s = bottle_plan_to_compose(
+            _plan(supervise=True)
+        )["services"]["agent"]
+        no_proxy = [e for e in s["environment"] if e.startswith("NO_PROXY=")][0]
+        self.assertIn("supervise", no_proxy)
+
+    def test_agent_forwarded_env_uses_bare_names(self):
+        # Bare NAME → compose inherits value from the up-process env,
+        # so secret token values stay out of the file.
+        s = bottle_plan_to_compose(_plan())["services"]["agent"]
+        self.assertIn("CLAUDE_CODE_OAUTH_TOKEN", s["environment"])
+
+    def test_agent_runsc_runtime(self):
+        plan = _plan()
+        plan = type(plan)(**{**vars(plan), "use_runsc": True})
+        s = bottle_plan_to_compose(plan)["services"]["agent"]
+        self.assertEqual("runsc", s["runtime"])
+
+    def test_agent_depends_on_pipelock(self):
+        s = bottle_plan_to_compose(_plan())["services"]["agent"]
+        self.assertIn("pipelock", s["depends_on"])
+
+    def test_agent_depends_on_every_present_sidecar(self):
+        s = bottle_plan_to_compose(
+            _plan(with_git=True, with_egress=True, supervise=True)
+        )["services"]["agent"]
+        self.assertEqual(
+            {"pipelock", "git-gate", "egress", "supervise"},
+            set(s["depends_on"]),
+        )
+
+    def test_agent_current_config_mount_only_with_supervise(self):
+        with_sv = bottle_plan_to_compose(_plan(supervise=True))["services"]["agent"]
+        self.assertTrue(any(
+            v["target"] == "/etc/claude-bottle/current-config"
+            for v in with_sv.get("volumes", [])
+        ))
+        without_sv = bottle_plan_to_compose(_plan(supervise=False))["services"]["agent"]
+        # Either no volumes key at all, or no current-config target.
+        self.assertFalse(any(
+            v["target"] == "/etc/claude-bottle/current-config"
+            for v in without_sv.get("volumes", [])
+        ))
+
+
+class TestConditionalGitGate(unittest.TestCase):
+    def test_absent_when_no_upstreams(self):
+        s = bottle_plan_to_compose(_plan(with_git=False))["services"]
+        self.assertNotIn("git-gate", s)
+
+    def test_present_when_upstreams(self):
+        s = bottle_plan_to_compose(_plan(with_git=True))["services"]
+        self.assertIn("git-gate", s)
+
+    def test_git_gate_built_from_dockerfile(self):
+        s = bottle_plan_to_compose(_plan(with_git=True))["services"]["git-gate"]
+        self.assertEqual("Dockerfile.git-gate", s["build"]["dockerfile"])
+        self.assertEqual("claude-bottle-git-gate:latest", s["image"])
+
+    def test_git_gate_extra_hosts(self):
+        s = bottle_plan_to_compose(_plan(with_git=True))["services"]["git-gate"]
+        self.assertIn("example.com:10.0.0.1", s["extra_hosts"])
+
+    def test_git_gate_identity_file_bind_mount(self):
+        s = bottle_plan_to_compose(_plan(with_git=True))["services"]["git-gate"]
+        # Per-upstream identity file is mounted at /git-gate/creds/<name>-key.
+        self.assertTrue(any(
+            v["target"] == "/git-gate/creds/upstream-key"
+            for v in s["volumes"]
+        ))
+
+
+class TestConditionalEgress(unittest.TestCase):
+    def test_absent_when_no_routes(self):
+        s = bottle_plan_to_compose(_plan(with_egress=False))["services"]
+        self.assertNotIn("egress", s)
+
+    def test_present_when_routes(self):
+        s = bottle_plan_to_compose(_plan(with_egress=True))["services"]
+        self.assertIn("egress", s)
+
+    def test_egress_alias_on_internal(self):
+        s = bottle_plan_to_compose(_plan(with_egress=True))["services"]["egress"]
+        self.assertIn("egress", s["networks"]["internal"]["aliases"])
+
+    def test_egress_upstream_envs(self):
+        s = bottle_plan_to_compose(_plan(with_egress=True))["services"]["egress"]
+        env = s["environment"]
+        self.assertIn(
+            f"EGRESS_UPSTREAM_PROXY=http://claude-bottle-pipelock-{SLUG}:8888",
+            env,
+        )
+        self.assertIn(
+            "EGRESS_UPSTREAM_CA=/home/mitmproxy/.mitmproxy/pipelock-ca.pem",
+            env,
+        )
+
+    def test_egress_token_slot_bare_name(self):
+        # Bare NAME entry in environment list → value inherits from
+        # compose process env, never lands in the rendered file.
+        s = bottle_plan_to_compose(_plan(with_egress=True))["services"]["egress"]
+        self.assertIn("EGRESS_TOKEN_0", s["environment"])
+
+    def test_egress_depends_on_pipelock(self):
+        s = bottle_plan_to_compose(_plan(with_egress=True))["services"]["egress"]
+        self.assertIn("pipelock", s["depends_on"])
+
+    def test_egress_bind_mounts(self):
+        s = bottle_plan_to_compose(_plan(with_egress=True))["services"]["egress"]
+        targets = {v["target"] for v in s["volumes"]}
+        self.assertEqual(
+            {
+                "/etc/egress/routes.yaml",
+                "/home/mitmproxy/.mitmproxy/mitmproxy-ca.pem",
+                "/home/mitmproxy/.mitmproxy/pipelock-ca.pem",
+            },
+            targets,
+        )
+
+
+class TestConditionalSupervise(unittest.TestCase):
+    def test_absent_when_off(self):
+        s = bottle_plan_to_compose(_plan(supervise=False))["services"]
+        self.assertNotIn("supervise", s)
+
+    def test_present_when_on(self):
+        s = bottle_plan_to_compose(_plan(supervise=True))["services"]
+        self.assertIn("supervise", s)
+
+    def test_supervise_internal_only(self):
+        s = bottle_plan_to_compose(_plan(supervise=True))["services"]["supervise"]
+        self.assertEqual({"internal"}, set(s["networks"].keys()))
+
+    def test_supervise_alias_on_internal(self):
+        s = bottle_plan_to_compose(_plan(supervise=True))["services"]["supervise"]
+        self.assertIn("supervise", s["networks"]["internal"]["aliases"])
+
+    def test_supervise_queue_dir_mounted_rw(self):
+        s = bottle_plan_to_compose(_plan(supervise=True))["services"]["supervise"]
+        queue_mount = [v for v in s["volumes"] if v["target"] == "/run/supervise/queue"]
+        self.assertEqual(1, len(queue_mount))
+        self.assertFalse(queue_mount[0]["read_only"])
+
+    def test_supervise_env_vars(self):
+        s = bottle_plan_to_compose(_plan(supervise=True))["services"]["supervise"]
+        self.assertIn(f"SUPERVISE_BOTTLE_SLUG={SLUG}", s["environment"])
+
+
+class TestFullMatrix(unittest.TestCase):
+    """The eight combinations of git/egress/supervise toggles. Just
+    asserts which services appear — content correctness is covered
+    per-service above."""
+
+    def test_matrix(self):
+        cases: list[tuple[bool, bool, bool, set[str]]] = []
+        for g in (False, True):
+            for e in (False, True):
+                for sv in (False, True):
+                    expected = {"pipelock", "agent"}
+                    if g:
+                        expected.add("git-gate")
+                    if e:
+                        expected.add("egress")
+                    if sv:
+                        expected.add("supervise")
+                    cases.append((g, e, sv, expected))
+
+        for g, e, sv, expected in cases:
+            with self.subTest(git=g, egress=e, supervise=sv):
+                s = bottle_plan_to_compose(
+                    _plan(with_git=g, with_egress=e, supervise=sv)
+                )["services"]
+                self.assertEqual(expected, set(s.keys()))
+
+
+if __name__ == "__main__":
+    unittest.main()