feat(orchestrator): slice 13c(viii) — agent-only consolidated compose render

The per-bottle model rendered a compose project with the agent + a sidecar
bundle on two per-bottle networks. Consolidated has no sidecars — one shared
gateway serves everyone — so this renders JUST the agent, attached to the
external shared gateway network at the orchestrator-allocated pinned IP, and
pointed at the gateway's address for egress.

- consolidated_agent_compose(plan, *, gateway_ip, source_ip, network):
  agent service only; networks {<gateway>: {ipv4_address: source_ip}} on the
  external gateway network; HTTPS_PROXY -> http://<gateway_ip>:9099; NO_PROXY
  includes the gateway address so git-http + supervise (also on the gateway)
  bypass the proxy and are reached directly. No depends_on/sidecars.
- Pure (takes the launch-time LaunchContext values), so it's unit-testable
  without docker.

pyright 0 errors; pylint 9.83/10; unit suite green (the 13 test_sidecar_init
/bin/sleep errors are pre-existing NixOS-local noise).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01WBMWTEtQdJ4W5UrWuLHCck

bot_bottle/backend/docker/consolidated_compose.py

@@ -0,0 +1,78 @@
+"""Agent-only compose for the consolidated docker backend (PRD 0070).
+
+The per-bottle model rendered a compose project with the agent *and* a
+sidecar bundle on two per-bottle networks. In the consolidated model the
+sidecars are gone — one shared gateway serves every bottle — so this renders
+just the agent, attached to the **external shared gateway network** with the
+pinned source IP the orchestrator allocated, and pointed at the gateway's
+address for egress (and, around the proxy, for git-http / supervise).
+
+Pure: it takes the launch-time `LaunchContext` values (gateway address,
+source IP, network) and the prepared plan, and returns a compose dict — no
+docker, so it's testable in isolation.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+from ...egress import egress_agent_env_entries
+from ..util import AGENT_CA_BUNDLE, AGENT_CA_PATH
+from .bottle_plan import DockerBottlePlan
+from .egress import EGRESS_PORT
+
+
+def consolidated_agent_compose(
+    plan: DockerBottlePlan,
+    *,
+    gateway_ip: str,
+    source_ip: str,
+    network: str,
+) -> dict[str, Any]:
+    """A compose spec with only the agent service, on the external gateway
+    network at `source_ip`, proxying egress through `gateway_ip`."""
+    proxy_url = f"http://{gateway_ip}:{EGRESS_PORT}"
+    # git-http + supervise live on the gateway too and must NOT go through the
+    # egress proxy — the agent reaches them directly by the gateway address.
+    no_proxy = f"localhost,127.0.0.1,{gateway_ip}"
+    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}",
+    ]
+    for name, value in sorted(plan.agent_provision.guest_env.items()):
+        env.append(f"{name}={value}")
+    # Forwarded vars: bare name → inherits from the compose-up process env so
+    # the secret value never lands on argv or in the compose file.
+    for name in sorted(plan.forwarded_env.keys()):
+        env.append(name)
+    env.extend(egress_agent_env_entries(plan.egress_plan))
+
+    service: dict[str, Any] = {
+        "image": plan.image,
+        "container_name": plan.container_name,
+        "command": ["sleep", "infinity"],
+        # Pinned address on the shared gateway network — the orchestrator
+        # registered this IP, and the gateway attributes the bottle by it.
+        "networks": {network: {"ipv4_address": source_ip}},
+        "environment": env,
+    }
+    if plan.use_runsc:
+        service["runtime"] = "runsc"
+
+    return {
+        "name": f"bot-bottle-{plan.slug}",
+        "services": {"agent": service},
+        # The gateway network is created + owned by the orchestrator; compose
+        # attaches to it (external) and must not create or destroy it.
+        "networks": {network: {"external": True}},
+    }
+
+
+__all__ = ["consolidated_agent_compose"]

bot_bottle/backend/docker/consolidated_launch.py

@@ -0,0 +1,145 @@
+"""Consolidated bottle launch sequence for the docker backend (PRD 0070).
+
+Composes the orchestrator primitives into the register/teardown sequence that
+replaces the per-bottle sidecar bundle:
+
+  1. ensure the orchestrator control plane + shared gateway are up;
+  2. allocate the bottle a pinned source IP on the gateway network (the
+     attribution key), skipping the gateway's own address + live bottles;
+  3. register it (egress policy blob + slug metadata) → bottle id + identity
+     token;
+  4. provision its git-gate repos/creds into the running gateway.
+
+It returns a `LaunchContext` with everything the agent container needs to
+attach — network, pinned IP, the gateway's address (its proxy target), the
+orchestrator URL, and the identity token. The agent `docker run` itself is
+the backend's job (it owns provider provisioning); this owns the
+orchestrator-facing wiring so that sequence stays testable in isolation.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+from ...docker_cmd import run_docker
+from ...egress import EgressPlan
+from ...git_gate import GitGatePlan
+from ...orchestrator.client import OrchestratorClient
+from ...orchestrator.gateway import GATEWAY_NAME, GATEWAY_NETWORK
+from ...orchestrator.lifecycle import OrchestratorProcess
+from ...orchestrator.registration import registration_inputs
+from .gateway_net import next_free_ip
+from .gateway_provision import deprovision_git_gate, provision_git_gate
+
+
+class ConsolidatedLaunchError(RuntimeError):
+    """The consolidated register/provision sequence could not complete."""
+
+
+@dataclass(frozen=True)
+class LaunchContext:
+    """What the agent container needs to join the shared gateway."""
+
+    bottle_id: str
+    identity_token: str
+    source_ip: str          # the agent's pinned address (attribution key)
+    network: str            # the shared gateway network to attach to
+    gateway_ip: str         # the gateway's address — the agent's proxy target
+    orchestrator_url: str
+
+
+def _network_cidr(network: str) -> str:
+    """The gateway network's IPv4 subnet, or raise."""
+    proc = run_docker([
+        "docker", "network", "inspect",
+        "--format", "{{range .IPAM.Config}}{{.Subnet}}{{end}}", network,
+    ])
+    cidr = proc.stdout.strip()
+    if proc.returncode != 0 or not cidr:
+        raise ConsolidatedLaunchError(
+            f"gateway network {network} has no subnet: {proc.stderr.strip()}"
+        )
+    return cidr
+
+
+def _container_ip(name: str, network: str) -> str:
+    """A container's IPv4 address on `network`, or raise."""
+    proc = run_docker([
+        "docker", "inspect", "--format",
+        f'{{{{(index .NetworkSettings.Networks "{network}").IPAddress}}}}', name,
+    ])
+    ip = proc.stdout.strip()
+    if proc.returncode != 0 or not ip:
+        raise ConsolidatedLaunchError(
+            f"gateway {name} has no address on {network}: {proc.stderr.strip()}"
+        )
+    return ip
+
+
+def _taken_ips(client: OrchestratorClient, gateway_ip: str) -> list[str]:
+    """Every address already in use on the gateway network: the gateway
+    container plus every live bottle the registry knows."""
+    taken = [gateway_ip]
+    for rec in client.list_bottles():
+        src = rec.get("source_ip")
+        if isinstance(src, str) and src:
+            taken.append(src)
+    return taken
+
+
+def launch_consolidated(
+    egress_plan: EgressPlan,
+    git_gate_plan: GitGatePlan,
+    *,
+    image_ref: str = "",
+    process: OrchestratorProcess | None = None,
+    gateway_name: str = GATEWAY_NAME,
+    network: str = GATEWAY_NETWORK,
+) -> LaunchContext:
+    """Ensure the orchestrator + gateway are up, allocate + register the
+    bottle, and provision its git-gate state. Returns the agent's attach
+    context. Raises `ConsolidatedLaunchError` (or the primitives' own errors)
+    if any step fails — the caller tears down on failure."""
+    process = process or OrchestratorProcess()
+    url = process.ensure_running()
+    client = OrchestratorClient(url)
+
+    cidr = _network_cidr(network)
+    gateway_ip = _container_ip(gateway_name, network)
+    source_ip = next_free_ip(cidr, _taken_ips(client, gateway_ip))
+
+    inputs = registration_inputs(egress_plan)
+    reg = client.register_bottle(
+        source_ip, image_ref=image_ref, policy=inputs.policy, metadata=inputs.metadata,
+    )
+    try:
+        provision_git_gate(gateway_name, reg.bottle_id, git_gate_plan)
+    except Exception:
+        # Roll the registration back so a provisioning failure leaves no orphan.
+        client.teardown_bottle(reg.bottle_id)
+        raise
+    return LaunchContext(
+        bottle_id=reg.bottle_id,
+        identity_token=reg.identity_token,
+        source_ip=source_ip,
+        network=network,
+        gateway_ip=gateway_ip,
+        orchestrator_url=url,
+    )
+
+
+def teardown_consolidated(
+    bottle_id: str, *, orchestrator_url: str, gateway_name: str = GATEWAY_NAME,
+) -> None:
+    """Deregister the bottle and remove its git-gate state from the gateway.
+    Both steps are idempotent so this is safe from a cleanup trap."""
+    OrchestratorClient(orchestrator_url).teardown_bottle(bottle_id)
+    deprovision_git_gate(gateway_name, bottle_id)
+
+
+__all__ = [
+    "LaunchContext",
+    "launch_consolidated",
+    "teardown_consolidated",
+    "ConsolidatedLaunchError",
+]

bot_bottle/backend/docker/gateway_provision.py

@@ -0,0 +1,94 @@
+"""Provision one bottle's git-gate state into the running shared gateway
+(PRD 0070, docker slice).
+
+The consolidated gateway serves every bottle's repos under `/git/<bottle_id>/`
+with per-repo credentials under `/git-gate/creds/<bottle_id>/`. When a bottle
+is registered the launcher must place *its* deploy keys + known_hosts into
+that per-bottle creds dir and init its bare repos there — so this copies the
+credential files into the live gateway container and runs the (namespaced,
+init-only) provisioning script produced by `git_gate_render_provision`.
+
+Isolating each bottle's creds dir + repo root by id is what keeps one
+bottle's push credentials out of another's repos on the shared gateway.
+"""
+
+from __future__ import annotations
+
+import re
+
+from ...docker_cmd import run_docker
+from ...git_gate import GitGatePlan, git_gate_render_provision
+
+# bottle ids index the gateway's per-bottle repo + creds dirs; they land in
+# `docker cp`/`rm` path arguments, so validate before any path is built (a
+# traversal id like "../etc" must never reach the container). Registry ids are
+# token_hex — this is defense in depth at the docker boundary.
+_SAFE_BOTTLE_ID = re.compile(r"[A-Za-z0-9_-]+")
+
+
+class GatewayProvisionError(RuntimeError):
+    """A git-gate provisioning step against the running gateway failed."""
+
+
+def _require_safe(bottle_id: str) -> None:
+    if not _SAFE_BOTTLE_ID.fullmatch(bottle_id):
+        raise GatewayProvisionError(f"unsafe bottle id {bottle_id!r}")
+
+
+def _creds_dir(bottle_id: str) -> str:
+    return f"/git-gate/creds/{bottle_id}"
+
+
+def _exec(gateway: str, argv: list[str]) -> None:
+    """`docker exec` a command in the gateway, raising on non-zero exit."""
+    proc = run_docker(["docker", "exec", gateway, *argv])
+    if proc.returncode != 0:
+        raise GatewayProvisionError(
+            f"gateway exec {argv!r} failed: {proc.stderr.strip()}"
+        )
+
+
+def _cp_into(gateway: str, src: str, dest: str) -> None:
+    """`docker cp` a host file into the gateway, raising on non-zero exit."""
+    proc = run_docker(["docker", "cp", src, f"{gateway}:{dest}"])
+    if proc.returncode != 0:
+        raise GatewayProvisionError(
+            f"gateway cp {src} -> {dest} failed: {proc.stderr.strip()}"
+        )
+
+
+def provision_git_gate(gateway: str, bottle_id: str, plan: GitGatePlan) -> None:
+    """Place `bottle_id`'s git-gate credentials into the running `gateway`
+    container and init its bare repos under `/git/<bottle_id>/`.
+
+    Copies each upstream's identity key (and known_hosts, when present) into
+    `/git-gate/creds/<bottle_id>/`, then runs the namespaced provisioning
+    script. No-op for a bottle with no git upstreams."""
+    _require_safe(bottle_id)
+    if not plan.upstreams:
+        return
+    creds = _creds_dir(bottle_id)
+    _exec(gateway, ["mkdir", "-p", creds])
+    for u in plan.upstreams:
+        if u.identity_file:
+            _cp_into(gateway, u.identity_file, f"{creds}/{u.name}-key")
+        known_hosts = str(u.known_hosts_file)
+        if known_hosts and known_hosts != ".":
+            _cp_into(gateway, known_hosts, f"{creds}/{u.name}-known_hosts")
+    # Init the bare repos + per-repo credential config for this namespace.
+    script = git_gate_render_provision(bottle_id, plan.upstreams)
+    _exec(gateway, ["sh", "-c", script])
+
+
+def deprovision_git_gate(gateway: str, bottle_id: str) -> None:
+    """Remove a bottle's repos + creds from the gateway on teardown. Idempotent
+    — an already-absent namespace is a clean no-op (best effort; a stray dir
+    can't leak, since attribution is by source IP and the bottle is gone)."""
+    _require_safe(bottle_id)
+    run_docker([
+        "docker", "exec", gateway, "rm", "-rf",
+        f"/git/{bottle_id}", _creds_dir(bottle_id),
+    ])
+
+
+__all__ = ["provision_git_gate", "deprovision_git_gate", "GatewayProvisionError"]

bot_bottle/orchestrator/gateway.py

@@ -25,6 +25,19 @@ from ..docker_cmd import run_docker
 
 GATEWAY_NAME = "bot-bottle-orch-gateway"
 GATEWAY_LABEL = "bot-bottle-orch-gateway=1"
+# The single user-defined network the gateway and every agent bottle share.
+# Agents attach here with a pinned IP and reach the gateway's egress /
+# git-http / supervise ports by its address — no host port publishing, and
+# the source IP the gateway attributes by is the address on this network.
+GATEWAY_NETWORK = "bot-bottle-gateway"
+
+# mitmproxy's CA dir in the bundle. A persistent named volume here keeps the
+# gateway's self-generated CA STABLE across container recreation — every agent
+# installs this one CA to trust the shared gateway's TLS interception, so it
+# must not rotate when the gateway restarts.
+MITMPROXY_HOME = "/home/mitmproxy/.mitmproxy"
+GATEWAY_CA_VOLUME = "bot-bottle-gateway-mitmproxy"
+GATEWAY_CA_CERT = f"{MITMPROXY_HOME}/mitmproxy-ca-cert.pem"
 
 # The real sidecar-bundle image + its Dockerfile. Kept as a local constant
 # rather than imported from backend.docker.sidecar_bundle, which would drag
@@ -77,11 +90,13 @@ class DockerGateway(Gateway):
         image_ref: str = GATEWAY_IMAGE,
         *,
         name: str = GATEWAY_NAME,
+        network: str = GATEWAY_NETWORK,
         build_context: Path | None = None,
         dockerfile: str | None = GATEWAY_DOCKERFILE,
     ) -> None:
         self.image_ref = image_ref
         self.name = name
+        self.network = network
         self._build_context = build_context or _REPO_ROOT
         self._dockerfile = dockerfile
 
@@ -112,9 +127,22 @@ class DockerGateway(Gateway):
         ])
         return self.name in proc.stdout.split()
 
+    def _ensure_network(self) -> None:
+        """Create the shared gateway network if it doesn't exist. Idempotent —
+        a concurrent create loses harmlessly (the loser sees 'already exists').
+        Docker picks the subnet; the launcher reads it back to allocate IPs."""
+        if run_docker(["docker", "network", "inspect", self.network]).returncode == 0:
+            return
+        proc = run_docker(["docker", "network", "create", self.network])
+        if proc.returncode != 0 and "already exists" not in proc.stderr:
+            raise GatewayError(
+                f"gateway network {self.network} failed to create: {proc.stderr.strip()}"
+            )
+
     def ensure_running(self) -> None:
         if self.is_running():
             return
+        self._ensure_network()
         # Clear any stale (stopped) container holding the fixed name, then
         # start fresh. `rm --force` on an absent name is a tolerated no-op.
         run_docker(["docker", "rm", "--force", self.name])
@@ -122,11 +150,27 @@ class DockerGateway(Gateway):
             "docker", "run", "--detach",
             "--name", self.name,
             "--label", GATEWAY_LABEL,
+            "--network", self.network,
+            # Persist the self-generated CA so it survives restarts (agents
+            # trust it) — see GATEWAY_CA_VOLUME.
+            "--volume", f"{GATEWAY_CA_VOLUME}:{MITMPROXY_HOME}",
             self.image_ref,
         ])
         if proc.returncode != 0:
             raise GatewayError(f"gateway failed to start: {proc.stderr.strip()}")
 
+    def ca_cert_pem(self) -> str:
+        """The gateway's CA certificate (PEM) that agents install to trust its
+        TLS interception. Read from the running container; raises if the
+        gateway hasn't generated it yet (it's created on first mitmproxy
+        start)."""
+        proc = run_docker(["docker", "exec", self.name, "cat", GATEWAY_CA_CERT])
+        if proc.returncode != 0 or not proc.stdout.strip():
+            raise GatewayError(
+                f"gateway CA cert not available: {proc.stderr.strip() or 'empty'}"
+            )
+        return proc.stdout
+
     def stop(self) -> None:
         proc = run_docker(["docker", "rm", "--force", self.name])
         if proc.returncode != 0 and "No such container" not in proc.stderr:
@@ -135,5 +179,6 @@ class DockerGateway(Gateway):
 
 __all__ = [
     "Gateway", "DockerGateway", "GatewayError",
-    "GATEWAY_NAME", "GATEWAY_LABEL", "GATEWAY_IMAGE",
+    "GATEWAY_NAME", "GATEWAY_LABEL", "GATEWAY_IMAGE", "GATEWAY_NETWORK",
+    "GATEWAY_CA_VOLUME", "GATEWAY_CA_CERT",
 ]

tests/unit/test_consolidated_compose.py

@@ -0,0 +1,51 @@
+"""Unit: agent-only consolidated compose render (PRD 0070)."""
+
+from __future__ import annotations
+
+import unittest
+
+from bot_bottle.backend.docker.consolidated_compose import consolidated_agent_compose
+from tests.unit.test_compose import _plan
+
+_GW = "172.18.0.2"
+_IP = "172.18.0.5"
+_NET = "bot-bottle-gateway"
+
+
+class TestConsolidatedAgentCompose(unittest.TestCase):
+    def _spec(self, *, runsc: bool = False):
+        plan = _plan(with_egress=True, supervise=True, with_git=True)
+        if runsc:
+            plan = type(plan)(**{**vars(plan), "use_runsc": True})  # type: ignore[arg-type]
+        return consolidated_agent_compose(plan, gateway_ip=_GW, source_ip=_IP, network=_NET)
+
+    def test_only_agent_service_no_sidecars(self) -> None:
+        # The whole point of consolidation: no per-bottle sidecar bundle.
+        self.assertEqual(["agent"], list(self._spec()["services"]))
+
+    def test_agent_pinned_on_external_gateway_network(self) -> None:
+        spec = self._spec()
+        self.assertEqual({"external": True}, spec["networks"][_NET])
+        agent_net = spec["services"]["agent"]["networks"][_NET]
+        self.assertEqual(_IP, agent_net["ipv4_address"])
+
+    def test_proxy_and_ca_point_at_gateway(self) -> None:
+        env = self._spec()["services"]["agent"]["environment"]
+        self.assertIn(f"HTTPS_PROXY=http://{_GW}:9099", env)
+        # git-http + supervise on the gateway must bypass the egress proxy.
+        self.assertTrue(any(e.startswith("NO_PROXY=") and _GW in e for e in env))
+
+    def test_no_sidecar_dependency(self) -> None:
+        self.assertNotIn("depends_on", self._spec()["services"]["agent"])
+
+    def test_runsc_runtime_when_enabled(self) -> None:
+        self.assertEqual("runsc", self._spec(runsc=True)["services"]["agent"]["runtime"])
+
+    def test_forwarded_env_stays_bare_names(self) -> None:
+        env = self._spec()["services"]["agent"]["environment"]
+        # forwarded secrets are bare names (value inherited from process env).
+        self.assertIn("CLAUDE_CODE_OAUTH_TOKEN", env)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/unit/test_consolidated_launch.py

@@ -0,0 +1,91 @@
+"""Unit: consolidated launch sequence — compose the orchestrator primitives (PRD 0070)."""
+
+from __future__ import annotations
+
+import unittest
+from pathlib import Path
+from unittest.mock import MagicMock, Mock, patch
+
+from bot_bottle.backend.docker.consolidated_launch import (
+    launch_consolidated,
+    teardown_consolidated,
+)
+from bot_bottle.egress import EgressPlan, EgressRoute
+from bot_bottle.git_gate import GitGatePlan
+from bot_bottle.orchestrator.client import RegisteredBottle
+
+_MOD = "bot_bottle.backend.docker.consolidated_launch"
+
+
+def _egress_plan() -> EgressPlan:
+    return EgressPlan(
+        slug="demo", routes_path=Path("/x"), routes=(EgressRoute(host="api.example.com"),),
+        token_env_map={},
+    )
+
+
+def _git_plan() -> GitGatePlan:
+    return GitGatePlan(
+        slug="demo", entrypoint_script=Path(), hook_script=Path(),
+        access_hook_script=Path(), upstreams=(),
+    )
+
+
+def _client(*, bottles: list[dict[str, object]] | None = None) -> Mock:
+    c = Mock()
+    c.list_bottles.return_value = bottles or []
+    c.register_bottle.return_value = RegisteredBottle("b1", "tok")
+    return c
+
+
+class TestLaunchConsolidated(unittest.TestCase):
+    def _run(self, client: Mock, provision: Mock | None = None):
+        process = MagicMock()
+        process.ensure_running.return_value = "http://orch:8080"
+        with patch(f"{_MOD}._network_cidr", return_value="172.18.0.0/16"), \
+                patch(f"{_MOD}._container_ip", return_value="172.18.0.2"), \
+                patch(f"{_MOD}.OrchestratorClient", return_value=client), \
+                patch(f"{_MOD}.provision_git_gate", provision or Mock()):
+            return launch_consolidated(_egress_plan(), _git_plan(), process=process)
+
+    def test_allocates_ip_registers_and_provisions(self) -> None:
+        client = _client()
+        provision = Mock()
+        ctx = self._run(client, provision)
+        # .1 is the router, .2 is the gateway → first bottle gets .3.
+        self.assertEqual("172.18.0.3", ctx.source_ip)
+        self.assertEqual("172.18.0.2", ctx.gateway_ip)
+        self.assertEqual("b1", ctx.bottle_id)
+        self.assertEqual("tok", ctx.identity_token)
+        self.assertEqual("http://orch:8080", ctx.orchestrator_url)
+        # Registered with the source IP + the egress policy blob.
+        kwargs = client.register_bottle.call_args
+        self.assertEqual("172.18.0.3", kwargs.args[0])
+        self.assertIn("api.example.com", kwargs.kwargs["policy"])
+        provision.assert_called_once()
+
+    def test_skips_gateway_and_live_bottle_addresses(self) -> None:
+        client = _client(bottles=[{"source_ip": "172.18.0.3"}])
+        ctx = self._run(client)
+        self.assertEqual("172.18.0.4", ctx.source_ip)  # .2 gw, .3 taken → .4
+
+    def test_provision_failure_rolls_back_registration(self) -> None:
+        client = _client()
+        provision = Mock(side_effect=RuntimeError("provision boom"))
+        with self.assertRaises(RuntimeError):
+            self._run(client, provision)
+        client.teardown_bottle.assert_called_once_with("b1")  # no orphan left
+
+
+class TestTeardownConsolidated(unittest.TestCase):
+    def test_deregisters_and_deprovisions(self) -> None:
+        client = Mock()
+        with patch(f"{_MOD}.OrchestratorClient", return_value=client), \
+                patch(f"{_MOD}.deprovision_git_gate") as deprov:
+            teardown_consolidated("b1", orchestrator_url="http://orch:8080")
+        client.teardown_bottle.assert_called_once_with("b1")
+        deprov.assert_called_once()
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/unit/test_gateway_provision.py

@@ -0,0 +1,111 @@
+"""Unit: git-gate provisioning into the running shared gateway (PRD 0070)."""
+
+from __future__ import annotations
+
+import unittest
+from pathlib import Path
+from unittest.mock import Mock, patch
+
+from bot_bottle.backend.docker.gateway_provision import (
+    GatewayProvisionError,
+    deprovision_git_gate,
+    provision_git_gate,
+)
+from bot_bottle.git_gate import GitGatePlan, GitGateUpstream
+
+_RUN = "bot_bottle.backend.docker.gateway_provision.run_docker"
+
+
+def _proc(returncode: int = 0, stderr: str = "") -> Mock:
+    return Mock(returncode=returncode, stderr=stderr)
+
+
+def _recorder(calls: list[list[str]]):
+    """A run_docker side_effect that records argv and returns success."""
+    def fake(argv: list[str]) -> Mock:
+        calls.append(argv)
+        return _proc()
+    return fake
+
+
+def _plan(*upstreams: GitGateUpstream) -> GitGatePlan:
+    return GitGatePlan(
+        slug="demo",
+        entrypoint_script=Path(),
+        hook_script=Path(),
+        access_hook_script=Path(),
+        upstreams=tuple(upstreams),
+    )
+
+
+def _up(name: str, *, key: str = "/host/keys/id", known_hosts: str = "") -> GitGateUpstream:
+    return GitGateUpstream(
+        name=name,
+        upstream_url=f"ssh://git@github.com/x/{name}.git",
+        upstream_host="github.com",
+        upstream_port="22",
+        identity_file=key,
+        known_host_key="",
+        known_hosts_file=Path(known_hosts) if known_hosts else Path(),
+    )
+
+
+class TestProvisionGitGate(unittest.TestCase):
+    def test_copies_creds_and_runs_namespaced_init(self) -> None:
+        calls: list[list[str]] = []
+        with patch(_RUN, side_effect=_recorder(calls)):
+            provision_git_gate("gw", "bottle1", _plan(_up("foo", known_hosts="/host/kh")))
+
+        cps = [c for c in calls if c[:2] == ["docker", "cp"]]
+        self.assertIn(["docker", "cp", "/host/keys/id", "gw:/git-gate/creds/bottle1/foo-key"], cps)
+        self.assertIn(
+            ["docker", "cp", "/host/kh", "gw:/git-gate/creds/bottle1/foo-known_hosts"], cps,
+        )
+        # The init script runs in the gateway, namespaced under the bottle id.
+        exec_scripts = [c for c in calls if c[:3] == ["docker", "exec", "gw"] and c[3] == "sh"]
+        self.assertEqual(1, len(exec_scripts))
+        self.assertIn("repo=/git/bottle1/${name}.git", exec_scripts[0][-1])
+
+    def test_omits_known_hosts_copy_when_absent(self) -> None:
+        calls: list[list[str]] = []
+        with patch(_RUN, side_effect=_recorder(calls)):
+            provision_git_gate("gw", "b1", _plan(_up("foo")))  # no known_hosts
+        cps = [c for c in calls if c[:2] == ["docker", "cp"]]
+        self.assertEqual(1, len(cps))  # only the key, not known_hosts
+        self.assertTrue(cps[0][3].endswith("/foo-key"))
+
+    def test_no_upstreams_is_noop(self) -> None:
+        with patch(_RUN) as m:
+            provision_git_gate("gw", "b1", _plan())
+        m.assert_not_called()
+
+    def test_raises_on_docker_failure(self) -> None:
+        with patch(_RUN, return_value=_proc(returncode=1, stderr="boom")):
+            with self.assertRaises(GatewayProvisionError):
+                provision_git_gate("gw", "b1", _plan(_up("foo")))
+
+    def test_rejects_unsafe_bottle_id_before_any_docker(self) -> None:
+        with patch(_RUN) as m:
+            with self.assertRaises(GatewayProvisionError):
+                provision_git_gate("gw", "../etc", _plan(_up("foo")))
+        m.assert_not_called()  # rejected before a single docker call
+
+
+class TestDeprovision(unittest.TestCase):
+    def test_removes_repo_and_creds(self) -> None:
+        with patch(_RUN, return_value=_proc()) as m:
+            deprovision_git_gate("gw", "b1")
+        argv = m.call_args.args[0]
+        self.assertEqual(["docker", "exec", "gw", "rm", "-rf"], argv[:5])
+        self.assertIn("/git/b1", argv)
+        self.assertIn("/git-gate/creds/b1", argv)
+
+    def test_rejects_unsafe_bottle_id(self) -> None:
+        with patch(_RUN) as m:
+            with self.assertRaises(GatewayProvisionError):
+                deprovision_git_gate("gw", "a/b")
+        m.assert_not_called()
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/unit/test_orchestrator_gateway.py

@@ -6,11 +6,15 @@ import unittest
 from unittest.mock import Mock, patch
 
 from bot_bottle.orchestrator.gateway import (
+    GATEWAY_CA_CERT,
     GATEWAY_NAME,
     DockerGateway,
     GatewayError,
 )
 
+
+_CA_PEM = "-----BEGIN CERTIFICATE-----\nMII...\n-----END CERTIFICATE-----\n"
+
 _RUN_DOCKER = "bot_bottle.orchestrator.gateway.run_docker"
 
 
@@ -51,6 +55,46 @@ class TestDockerGateway(unittest.TestCase):
         self.assertEqual(1, len(runs))
         self.assertIn(self.sc.name, runs[0])
         self.assertIn("bot-bottle-sidecars:latest", runs[0])
+        # Runs on the shared gateway network so agents can reach it by IP.
+        self.assertEqual(self.sc.network, runs[0][runs[0].index("--network") + 1])
+        # Persists its CA on a named volume so agents keep trusting it.
+        self.assertTrue(any("mitmproxy" in a for a in runs[0]))
+
+    def test_ensure_running_creates_network_when_missing(self) -> None:
+        calls: list[list[str]] = []
+
+        def fake(argv: list[str]) -> Mock:
+            calls.append(argv)
+            if argv[:3] == ["docker", "network", "inspect"]:
+                return _proc(returncode=1, stderr="No such network")
+            return _proc(stdout="") if argv[:2] == ["docker", "ps"] else _proc()
+
+        with patch(_RUN_DOCKER, side_effect=fake):
+            self.sc.ensure_running()
+        creates = [c for c in calls if c[:3] == ["docker", "network", "create"]]
+        self.assertEqual([["docker", "network", "create", self.sc.network]], creates)
+
+    def test_ca_cert_pem_reads_from_container(self) -> None:
+        with patch(_RUN_DOCKER, return_value=_proc(stdout=_CA_PEM)) as m:
+            self.assertEqual(_CA_PEM, self.sc.ca_cert_pem())
+        argv = m.call_args.args[0]
+        self.assertEqual(["docker", "exec", self.sc.name, "cat", GATEWAY_CA_CERT], argv)
+
+    def test_ca_cert_pem_raises_when_absent(self) -> None:
+        with patch(_RUN_DOCKER, return_value=_proc(returncode=1, stderr="No such file")):
+            with self.assertRaises(GatewayError):
+                self.sc.ca_cert_pem()
+
+    def test_ensure_running_reuses_existing_network(self) -> None:
+        calls: list[list[str]] = []
+
+        def fake(argv: list[str]) -> Mock:
+            calls.append(argv)
+            return _proc(stdout="") if argv[:2] == ["docker", "ps"] else _proc()
+
+        with patch(_RUN_DOCKER, side_effect=fake):
+            self.sc.ensure_running()  # network inspect returns 0 → exists
+        self.assertEqual([], [c for c in calls if c[:3] == ["docker", "network", "create"]])
 
     def test_ensure_running_raises_on_docker_failure(self) -> None:
         def fake(argv: list[str]) -> Mock: