feat(ssh-gate): wire gate into DockerBottlePlan, prepare, launch

PRD 0007: thread the DockerSSHGate through the bottle lifecycle.

- DockerBottlePlan gains gate_plan: SSHGatePlan.
- prepare.resolve_plan accepts a gate and renders its entrypoint
  script next to the pipelock yaml.
- launch.launch starts the gate sidecar after pipelock (so it's on
  the same internal + egress networks) and registers its stop in
  the ExitStack. Skipped when the bottle has no ssh entries.
- DockerBottleBackend instantiates DockerSSHGate alongside the
  pipelock proxy.
- bottle_plan.print + to_dict surface the upstream table so
  --dry-run shows the per-host listen-port mapping.

ssh_config provisioning still points at pipelock; that swap lands
in the next commit so this one stays a pure wiring change.

claude_bottle/backend/docker/backend.py

@@ -29,6 +29,7 @@ from .provision import git as _git
 from .provision import prompt as _prompt
 from .provision import skills as _skills
 from .provision import ssh as _ssh
+from .ssh_gate import DockerSSHGate
 
 
 class DockerBottleBackend(BottleBackend["DockerBottlePlan", "DockerBottleCleanupPlan"]):
@@ -39,13 +40,18 @@ class DockerBottleBackend(BottleBackend["DockerBottlePlan", "DockerBottleCleanup
 
     def __init__(self) -> None:
         self._proxy = DockerPipelockProxy()
+        self._gate = DockerSSHGate()
 
     def _resolve_plan(self, spec: BottleSpec, *, stage_dir: Path) -> DockerBottlePlan:
-        return _prepare.resolve_plan(spec, stage_dir=stage_dir, proxy=self._proxy)
+        return _prepare.resolve_plan(
+            spec, stage_dir=stage_dir, proxy=self._proxy, gate=self._gate
+        )
 
     @contextmanager
     def launch(self, plan: DockerBottlePlan) -> Generator[DockerBottle, None, None]:
-        with _launch.launch(plan, proxy=self._proxy, provision=self.provision) as bottle:
+        with _launch.launch(
+            plan, proxy=self._proxy, gate=self._gate, provision=self.provision
+        ) as bottle:
             yield bottle
 
     def provision_ca(self, plan: DockerBottlePlan, target: str) -> None:

claude_bottle/backend/docker/bottle_plan.py

@@ -14,6 +14,7 @@ from pathlib import Path
 from ...log import info
 from ...manifest import Agent, Bottle
 from ...pipelock import PipelockProxyPlan, pipelock_effective_allowlist
+from ...ssh_gate import SSHGatePlan
 from .. import BottlePlan
 
 
@@ -49,6 +50,7 @@ class DockerBottlePlan(BottlePlan):
     forwarded_env: dict[str, str] = field(repr=False)
     prompt_file: Path
     proxy_plan: PipelockProxyPlan
+    gate_plan: SSHGatePlan
     allowlist_summary: str
     use_runsc: bool
 
@@ -90,6 +92,12 @@ class DockerBottlePlan(BottlePlan):
         info(f"bottle             : {v.agent.bottle}")
         if v.ssh_hosts:
             info(f"  ssh hosts     : {', '.join(v.ssh_hosts)}")
+            gate_lines = [
+                f"{u.bottle_host_alias} -> {u.upstream_host}:{u.upstream_port} "
+                f"(listen {u.listen_port})"
+                for u in self.gate_plan.upstreams
+            ]
+            info(f"  ssh gate      : {'; '.join(gate_lines)}")
         else:
             info("  ssh hosts     : (none)")
         info(f"  egress        : {self.allowlist_summary}")
@@ -115,6 +123,14 @@ class DockerBottlePlan(BottlePlan):
             "env_names": v.env_names,
             "skills": list(v.agent.skills),
             "ssh_hosts": v.ssh_hosts,
+            "ssh_gate": [
+                {
+                    "host": u.bottle_host_alias,
+                    "upstream": f"{u.upstream_host}:{u.upstream_port}",
+                    "listen_port": u.listen_port,
+                }
+                for u in self.gate_plan.upstreams
+            ],
             "egress": {
                 "host_count": len(hosts),
                 "hosts": hosts,

claude_bottle/backend/docker/launch.py

@@ -24,6 +24,7 @@ from .bottle import DockerBottle
 from .bottle_plan import DockerBottlePlan
 from .pipelock import DockerPipelockProxy, pipelock_proxy_url, pipelock_tls_init
 from .provision.ca import AGENT_CA_BUNDLE, AGENT_CA_PATH
+from .ssh_gate import DockerSSHGate
 
 
 # Where the repo root lives, for `docker build` context. Computed once.
@@ -35,6 +36,7 @@ def launch(
     plan: DockerBottlePlan,
     *,
     proxy: DockerPipelockProxy,
+    gate: DockerSSHGate,
     provision: Callable[[DockerBottlePlan, str], str | None],
 ) -> Generator[DockerBottle, None, None]:
     """Build, launch, and provision a Docker bottle. Teardown on exit.
@@ -85,6 +87,21 @@ def launch(
         pipelock_name = proxy.start(plan.proxy_plan)
         stack.callback(proxy.stop, pipelock_name)
 
+        # SSH egress gate (PRD 0007). One sidecar per agent, only
+        # brought up when the bottle has ssh entries. Lives on the
+        # same internal + egress networks pipelock straddles; the
+        # agent dials it by container name (DNS works on --internal,
+        # confirmed by the PRD 0007 spike).
+        if plan.gate_plan.upstreams:
+            gate_plan = dataclasses.replace(
+                plan.gate_plan,
+                internal_network=internal_network,
+                egress_network=egress_network,
+            )
+            plan = dataclasses.replace(plan, gate_plan=gate_plan)
+            gate_name = gate.start(plan.gate_plan)
+            stack.callback(gate.stop, gate_name)
+
         container = _run_agent_container(plan, internal_network)
         stack.callback(docker_mod.force_remove_container, container)
 

claude_bottle/backend/docker/prepare.py

@@ -20,6 +20,7 @@ from .. import BottleSpec
 from . import util as docker_mod
 from .bottle_plan import DockerBottlePlan
 from .pipelock import DockerPipelockProxy
+from .ssh_gate import DockerSSHGate
 
 
 def resolve_plan(
@@ -27,6 +28,7 @@ def resolve_plan(
     *,
     stage_dir: Path,
     proxy: DockerPipelockProxy,
+    gate: DockerSSHGate,
 ) -> DockerBottlePlan:
     """Resolve Docker-specific names and write scratch files. Trusts
     that the agent and its skills/SSH keys are present — validation
@@ -78,6 +80,7 @@ def resolve_plan(
     prompt_file.chmod(0o600)
 
     proxy_plan = proxy.prepare(bottle, slug, stage_dir)
+    gate_plan = gate.prepare(bottle, slug, stage_dir)
     resolved = resolve_env(manifest, spec.agent_name)
     # Everything that should reach the bottle by-name (so its value
     # never lands on argv or in env_file) goes into one dict. The
@@ -105,6 +108,7 @@ def resolve_plan(
         forwarded_env=forwarded_env,
         prompt_file=prompt_file,
         proxy_plan=proxy_plan,
+        gate_plan=gate_plan,
         allowlist_summary=allowlist_summary,
         use_runsc=use_runsc,
     )

tests/integration/test_dry_run_plan.py

@@ -80,6 +80,7 @@ class TestDryRunPlan(unittest.TestCase):
                              "runsc isn't available on the CI runner")
             self.assertEqual([], plan["skills"])
             self.assertEqual([], plan["ssh_hosts"])
+            self.assertEqual([], plan["ssh_gate"])
             self.assertEqual(False, plan["remote_control"])
             self.assertEqual(0, plan["prompt"]["length"])