feat(backend): reconcile running bottles' CA on gateway bring-up (0081)

Add `attach_bottled_agents_to_gateway()` as an `@abc.abstractmethod` on
`BottleBackend` and implement the first reconcile step across all three
backends: on a gateway cold boot, replace every already-running agent's
trusted CA with the freshly-minted gateway CA. Reconciling running bottles
is a backend responsibility (only the backend can enumerate its agents and
reach them — firecracker over SSH, docker/macOS over exec/cp), so making it
an abstract method keeps the fix cross-backend by construction.

The gateway rootfs is ephemeral, so a rebuild mints a new CA that every
running bottle distrusts (SSL verification fails — #510). This reconcile is
what distributes the fresh CA, so a routine gateway rebuild now doubles as a
free CA rotation. Per-bottle steps are best-effort: one unreachable or
malformed bottle is logged and skipped, never blocking the others.

Trigger — `Gateway.connect_to_orchestrator` now returns a cold-boot bool
(True when it actually (re)brought the gateway up, False when a healthy
current gateway was left untouched); each infra `ensure_running` gates the
reconcile on it, so it fires exactly on cold boot and never on an adopt.

Git-gate re-provision and egress-token restore fold into this same reconcile
on the same trigger in follow-up PRs (#516).

Refs #516. Addresses #510.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

bot_bottle/backend/docker/backend.py

@@ -140,3 +140,7 @@ class DockerBottleBackend(BottleBackend["DockerBottlePlan", "DockerBottleCleanup
 
     def enumerate_active(self) -> Sequence[ActiveAgent]:
         return _enumerate.enumerate_active()
+
+    def attach_bottled_agents_to_gateway(self) -> None:
+        from .consolidated_launch import attach_bottled_agents_to_gateway
+        attach_bottled_agents_to_gateway()

bot_bottle/backend/docker/consolidated_launch.py

@@ -13,6 +13,7 @@ orchestrator-facing wiring so that sequence stays testable in isolation.
 
 from __future__ import annotations
 
+import tempfile
 from dataclasses import dataclass
 
 from ... import log
@@ -25,6 +26,7 @@ from .infra import INFRA_NAME, DockerInfraService
 from ...orchestrator.store.secret_store import ENV_VAR_SECRET_NAME
 from ...orchestrator.reprovision import reprovision_bottles
 from ..provision_bottle import deprovision_bottle, provision_bottle
+from ..util import AGENT_CA_PATH
 from .gateway_transport import DockerGatewayTransport
 from .gateway_net import next_free_ip
 
@@ -74,6 +76,77 @@ def _network_container_ips(network: str) -> list[str]:
     return ips
 
 
+def _push_ca_to_container(container: str, ca_pem: str) -> None:
+    """Replace one running agent container's trusted gateway CA with `ca_pem`
+    and rebuild its trust store via `docker cp` + `docker exec`. Unconditional
+    install — there is one gateway, so no fingerprint match is needed. Raises
+    `ConsolidatedLaunchError` on failure."""
+    mkdir = run_docker(
+        ["docker", "exec", container, "mkdir", "-p",
+         "/usr/local/share/ca-certificates"]
+    )
+    if mkdir.returncode != 0:
+        raise ConsolidatedLaunchError(
+            f"CA push to {container} failed (mkdir): {mkdir.stderr.strip()}"
+        )
+    with tempfile.NamedTemporaryFile("w", suffix=".crt") as tmp:
+        tmp.write(ca_pem)
+        tmp.flush()
+        cp = run_docker(["docker", "cp", tmp.name, f"{container}:{AGENT_CA_PATH}"])
+    if cp.returncode != 0:
+        raise ConsolidatedLaunchError(
+            f"CA push to {container} failed (cp): {cp.stderr.strip()}"
+        )
+    ex = run_docker(
+        ["docker", "exec", container, "sh", "-c",
+         f"chmod 644 {AGENT_CA_PATH} && update-ca-certificates"]
+    )
+    if ex.returncode != 0:
+        raise ConsolidatedLaunchError(
+            f"CA push to {container} failed (update-ca-certificates): "
+            f"{ex.stderr.strip()}"
+        )
+
+
+def attach_bottled_agents_to_gateway(
+    ca_pem: str | None = None,
+    *,
+    network: str = GATEWAY_NETWORK,
+    gateway_name: str = INFRA_NAME,
+) -> None:
+    """Reconcile every running agent container against the freshly-(re)created
+    gateway (PRD 0081): replace each agent's trusted CA with the current gateway
+    CA, so a gateway rebuild doesn't silently break their TLS egress (#510).
+
+    Per-container steps are best-effort: one failure is logged and skipped so it
+    never blocks the others or the gateway coming up."""
+    if ca_pem is None:
+        ca_pem = DockerInfraService().gateway().ca_cert_pem()
+    try:
+        proc = run_docker([
+            "docker", "network", "inspect", "--format",
+            "{{range .Containers}}{{.Name}}\n{{end}}", network,
+        ])
+    except OSError as exc:
+        log.info(f"CA reconcile skipped: {exc}")
+        return
+    reconciled = 0
+    for line in proc.stdout.splitlines():
+        name = line.strip()
+        if not name or name == gateway_name:
+            continue
+        try:
+            _push_ca_to_container(name, ca_pem)
+            reconciled += 1
+        except (OSError, ConsolidatedLaunchError) as exc:
+            log.info(f"CA reconcile skipped for {name}: {exc}")
+    if reconciled:
+        log.info(
+            "reconciled gateway CA into running docker bottle(s)",
+            context={"count": reconciled},
+        )
+
+
 def _reprovision_running_bottles(
     orchestrator_url: str,
     network: str = GATEWAY_NETWORK,
@@ -180,6 +253,7 @@ def deprovision_consolidated(
 __all__ = [
     "LaunchContext",
     "launch_consolidated",
+    "attach_bottled_agents_to_gateway",
     "deprovision_consolidated",
     "ConsolidatedLaunchError",
 ]

bot_bottle/backend/docker/gateway.py

@@ -157,7 +157,7 @@ class DockerGateway(Gateway):
                 f"gateway network {self.network} failed to create: {proc.stderr.strip()}"
             )
 
-    def connect_to_orchestrator(self, orchestrator_url: str, gateway_token: str) -> None:
+    def connect_to_orchestrator(self, orchestrator_url: str, gateway_token: str) -> bool:
         # Bind to this orchestrator (PRD 0070): stash the URL its daemons resolve
         # policy against + the pre-minted `gateway` token they present, then bring
         # the container up. The gateway never mints, so it holds no signing key —
@@ -184,7 +184,7 @@ class DockerGateway(Gateway):
         # just when the container is absent.
         self._ensure_network()
         if self.is_running() and self._running_image_is_current():
-            return
+            return False
         # Clear any stale (stopped OR outdated-image) container holding the
         # fixed name, then start fresh. `rm --force` on an absent name is a
         # tolerated no-op.
@@ -228,6 +228,9 @@ class DockerGateway(Gateway):
         # pre-connect window (they retry /resolve per request).
         if self._control_network:
             self._connect_control_network()
+        # (Re)created a fresh container — signal a cold boot so the caller
+        # reconciles running bottles against it (PRD 0081).
+        return True
 
     def _connect_control_network(self) -> None:
         """Ensure the `--internal` control network exists and attach the gateway

bot_bottle/backend/docker/infra.py

@@ -142,9 +142,16 @@ class DockerInfraService(InfraService):
         # mints the role-scoped `gateway` JWT here and hands it to the gateway,
         # which never sees the key (#469). `connect_to_orchestrator` is
         # idempotent.
-        gateway.connect_to_orchestrator(
+        cold_booted = gateway.connect_to_orchestrator(
             orchestrator.gateway_url(), orchestrator.mint_gateway_token(),
         )
+        # A (re)created gateway container minted/mounted its CA afresh and lost
+        # every bottle's per-bottle state; reconcile the already-running bottles
+        # against it (PRD 0081). Skipped when a healthy current gateway was left
+        # untouched — no cold boot, nothing to reconcile.
+        if cold_booted:
+            from .backend import DockerBottleBackend
+            DockerBottleBackend().attach_bottled_agents_to_gateway()
         return orchestrator.url()
 
     def stop(self) -> None: