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/base.py

@@ -512,6 +512,27 @@ class BottleBackend(ABC, Generic[PlanT, CleanupT]):
         compose ls`; firecracker cross-references its running gateway
         containers against per-bottle metadata."""
 
+    @abstractmethod
+    def attach_bottled_agents_to_gateway(self) -> None:
+        """Reconcile every already-running bottle against the freshly-(re)booted
+        gateway (PRD 0081). The host calls this on the gateway bring-up path —
+        the cold-boot branch only, never on an adopt of a healthy current
+        gateway — so a gateway rebuild/restart doesn't silently break running
+        bottles' gateway-dependent state.
+
+        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 every backend must implement it —
+        cross-backend by construction. Each per-bottle step is best-effort: one
+        unreachable or malformed bottle is logged and skipped so it never blocks
+        the others or the gateway coming up.
+
+        Currently reconciles the **CA**: it replaces each agent's trusted CA
+        with the current gateway CA (unconditional — there is one gateway, so no
+        fingerprint match is needed), which rotates the CA for free on every
+        bring-up. Git-gate re-provision and egress-token restore fold into this
+        same reconcile on the same trigger (see #516)."""
+
     @classmethod
     @abstractmethod
     def is_available(cls) -> bool:

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
 
@@ -79,6 +81,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,
@@ -185,6 +258,7 @@ def deprovision_consolidated(
 __all__ = [
     "LaunchContext",
     "launch_consolidated",
+    "attach_bottled_agents_to_gateway",
     "deprovision_consolidated",
     "ConsolidatedLaunchError",
 ]

bot_bottle/backend/docker/gateway.py

@@ -200,7 +200,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 —
@@ -227,7 +227,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.
@@ -271,6 +271,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:

bot_bottle/backend/firecracker/backend.py

@@ -119,6 +119,10 @@ class FirecrackerBottleBackend(
     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()
+
     def supervise_mcp_url(self, plan: FirecrackerBottlePlan) -> str:
         return plan.agent_supervise_url
 

bot_bottle/backend/firecracker/consolidated_launch.py

@@ -32,6 +32,7 @@ from dataclasses import dataclass
 from pathlib import Path
 
 from ...egress import EgressPlan
+from ...gateway import GatewayError
 from ...git_gate import GitGatePlan
 from ...log import info
 from ...orchestrator.client import OrchestratorClient, OrchestratorClientError
@@ -41,6 +42,7 @@ from ...orchestrator.lifecycle import (
 from ...orchestrator.reprovision import reprovision_bottles
 from ...orchestrator.store.secret_store import ENV_VAR_SECRET_NAME
 from ..provision_bottle import deprovision_bottle, provision_bottle
+from ..util import AGENT_CA_PATH
 from . import cleanup, util
 from .gateway import FirecrackerGateway
 from .infra import FirecrackerInfraService
@@ -89,6 +91,56 @@ def persist_env_var_secret(private_key: Path, guest_ip: str, secret: str) -> Non
         )
 
 
+def _push_gateway_ca_to_agent(private_key: Path, guest_ip: str, ca_pem: str) -> None:
+    """Replace one running agent VM's trusted gateway CA with `ca_pem` and
+    rebuild its trust store over SSH. Unconditional install — there is one
+    gateway, so no fingerprint match is needed. Bare-pipe input keeps the cert
+    off argv. Raises `ConsolidatedLaunchError` on failure."""
+    install = (
+        "umask 022; mkdir -p /usr/local/share/ca-certificates && "
+        f"cat > {AGENT_CA_PATH} && chmod 644 {AGENT_CA_PATH} && "
+        "update-ca-certificates"
+    )
+    proc = subprocess.run(
+        util.ssh_base_argv(private_key, guest_ip) + [install],
+        input=ca_pem, capture_output=True, text=True, check=False,
+    )
+    if proc.returncode != 0:
+        raise ConsolidatedLaunchError(
+            f"CA push to {guest_ip} failed: "
+            f"{proc.stderr.strip() or '<no stderr>'}"
+        )
+
+
+def attach_bottled_agents_to_gateway() -> None:
+    """Reconcile every running agent VM against the freshly-booted gateway
+    (PRD 0081). Replaces each agent's trusted CA with the current gateway CA —
+    the gateway rootfs is ephemeral, so a cold boot mints a fresh CA and this is
+    what distributes it (a gateway rebuild doubles as a free CA rotation, #510).
+
+    Per-bottle steps are best-effort: one unreachable or malformed VM is logged
+    and skipped so it never blocks the others or the gateway coming up."""
+    gateway = FirecrackerGateway()
+    try:
+        ca_pem = gateway.ca_cert_pem()
+    except GatewayError as exc:
+        info(f"bring-up reconcile skipped: gateway CA unavailable: {exc}")
+        return
+    reconciled = 0
+    for run_dir in cleanup.live_run_dirs():
+        guest_ip = _guest_ip_from_config(run_dir / "config.json")
+        private_key = run_dir / "bottle_id_ed25519"
+        if not guest_ip or not private_key.is_file():
+            continue
+        try:
+            _push_gateway_ca_to_agent(private_key, guest_ip, ca_pem)
+            reconciled += 1
+        except (OSError, ConsolidatedLaunchError) as exc:
+            info(f"CA reconcile skipped for {guest_ip}: {exc}")
+    if reconciled:
+        info(f"reconciled gateway CA into {reconciled} running Firecracker bottle(s)")
+
+
 def _reprovision_running_bottles(client: OrchestratorClient) -> None:
     """Read keys from live agent VMs and restore the restarted gateway."""
     try:
@@ -161,6 +213,7 @@ def deprovision_consolidated(
 __all__ = [
     "LaunchContext",
     "launch_consolidated",
+    "attach_bottled_agents_to_gateway",
     "deprovision_consolidated",
     "ConsolidatedLaunchError",
     "OrchestratorStartError",

bot_bottle/backend/firecracker/gateway.py

@@ -69,13 +69,17 @@ class FirecrackerGateway(Gateway):
         self._orchestrator_url = ""
         self._gateway_token = ""
 
-    def connect_to_orchestrator(self, orchestrator_url: str, gateway_token: str) -> None:
+    def connect_to_orchestrator(self, orchestrator_url: str, gateway_token: str) -> bool:
         """Boot the gateway VM on its link resolving policy against
         `orchestrator_url`, then seed `gateway_token` over SSH. The orchestrator's
         guest IP is parsed from the URL and passed on the cmdline (`bb_orch=`), so
         the baked init stays IP-independent. Boot the orchestrator first — the
         gateway daemons reach it at startup. The gateway never mints, so it holds
-        no signing key, only this token (#469)."""
+        no signing key, only this token (#469).
+
+        Always returns True: the firecracker gateway VM is booted fresh here
+        (this runs only on the infra cold-boot path), so it is always a cold
+        boot that minted a new CA — the caller reconciles running bottles."""
         self._orchestrator_url = orchestrator_url
         self._gateway_token = gateway_token
         if not self._orchestrator_url:
@@ -105,6 +109,7 @@ class FirecrackerGateway(Gateway):
             "the gateway JWT to the gateway VM (its data plane will not start)",
         )
         self._vm = vm
+        return True
 
     def is_running(self) -> bool:
         return infra_vm._pidfile_alive(infra_vm._gw_dir())

bot_bottle/backend/firecracker/infra.py

@@ -67,9 +67,17 @@ class FirecrackerInfraService(InfraService):
             # holds the signing key) mints the role-scoped `gateway` JWT for the
             # gateway, which never sees the key (#469).
             orchestrator.ensure_running(startup_timeout=startup_timeout)
-            self.gateway().connect_to_orchestrator(
+            cold_booted = self.gateway().connect_to_orchestrator(
                 orchestrator.gateway_url(), orchestrator.mint_gateway_token())
             infra_vm.record_booted_version(want)
+            # The fresh gateway rootfs minted a new CA and lost every bottle's
+            # git-gate/token state; reconcile the already-running bottles against
+            # it so a gateway rebuild doesn't silently break their egress
+            # (PRD 0081). Cold-boot only — the adopt fast-paths above never reach
+            # here, so a healthy current gateway is left untouched.
+            if cold_booted:
+                from .backend import FirecrackerBottleBackend
+                FirecrackerBottleBackend().attach_bottled_agents_to_gateway()
             return url
 
     def stop(self) -> None:

bot_bottle/backend/macos_container/backend.py

@@ -117,5 +117,9 @@ class MacosContainerBottleBackend(
     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()
+
     def supervise_mcp_url(self, plan: MacosContainerBottlePlan) -> str:
         return plan.agent_supervise_url

bot_bottle/backend/macos_container/consolidated_launch.py

@@ -34,6 +34,7 @@ every real command arrives through `container exec`.
 
 from __future__ import annotations
 
+import tempfile
 from dataclasses import dataclass
 
 from ...egress import EgressPlan
@@ -43,6 +44,7 @@ from ...orchestrator.client import OrchestratorClient, OrchestratorClientError
 from ...orchestrator.reprovision import reprovision_bottles
 from ...orchestrator.store.secret_store import ENV_VAR_SECRET_NAME
 from ..provision_bottle import deprovision_bottle, provision_bottle
+from ..util import AGENT_CA_PATH
 from . import util as container_mod
 from .enumerate import CONTAINER_NAME_PREFIX, EnumerationError, enumerate_active
 from .gateway import GATEWAY_NETWORK
@@ -98,6 +100,66 @@ def ensure_gateway(
     return endpoint
 
 
+def _push_ca_to_container(name: str, ca_pem: str) -> None:
+    """Replace one running agent container's trusted gateway CA with `ca_pem`
+    and rebuild its trust store via `container cp` + `container exec`.
+    Unconditional install — there is one gateway, so no fingerprint match is
+    needed. Raises `ConsolidatedLaunchError` on failure."""
+    mkdir = container_mod.run_container_argv(
+        ["container", "exec", name, "mkdir", "-p",
+         "/usr/local/share/ca-certificates"]
+    )
+    if mkdir.returncode != 0:
+        raise ConsolidatedLaunchError(
+            f"CA push to {name} failed (mkdir): {(mkdir.stderr or '').strip()}"
+        )
+    with tempfile.NamedTemporaryFile("w", suffix=".crt") as tmp:
+        tmp.write(ca_pem)
+        tmp.flush()
+        cp = container_mod.run_container_argv(
+            ["container", "cp", tmp.name, f"{name}:{AGENT_CA_PATH}"]
+        )
+    if cp.returncode != 0:
+        raise ConsolidatedLaunchError(
+            f"CA push to {name} failed (cp): {(cp.stderr or '').strip()}"
+        )
+    ex = container_mod.run_container_argv(
+        ["container", "exec", name, "sh", "-c",
+         f"chmod 644 {AGENT_CA_PATH} && update-ca-certificates"]
+    )
+    if ex.returncode != 0:
+        raise ConsolidatedLaunchError(
+            f"CA push to {name} failed (update-ca-certificates): "
+            f"{(ex.stderr or '').strip()}"
+        )
+
+
+def attach_bottled_agents_to_gateway(ca_pem: str | None = None) -> 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 = MacosInfraService().ca_cert_pem()
+    try:
+        agents = list(enumerate_active())
+    except (EnumerationError, OSError) as exc:
+        info(f"CA reconcile skipped: {exc}")
+        return
+    reconciled = 0
+    for agent in agents:
+        name = f"{CONTAINER_NAME_PREFIX}{agent.slug}"
+        try:
+            _push_ca_to_container(name, ca_pem)
+            reconciled += 1
+        except (OSError, ConsolidatedLaunchError) as exc:
+            info(f"CA reconcile skipped for {name}: {exc}")
+    if reconciled:
+        info(f"reconciled gateway CA into {reconciled} running macOS bottle(s)")
+
+
 def _reprovision_running_bottles(endpoint: GatewayEndpoint) -> None:
     """Recover keys from live Apple containers and restore gateway tokens."""
     try:
@@ -200,6 +262,7 @@ __all__ = [
     "GatewayEndpoint",
     "LaunchContext",
     "ensure_gateway",
+    "attach_bottled_agents_to_gateway",
     "live_source_ips",
     "register_agent",
     "deprovision_consolidated",

bot_bottle/backend/macos_container/gateway.py

@@ -104,11 +104,15 @@ class MacosGateway(Gateway):
         container_mod.build_image(
             self.image_ref, str(self._repo_root), dockerfile="Dockerfile.gateway")
 
-    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 the gateway to this orchestrator and (re)start it, dual-homed on
         the agent + control networks, resolving policy against `orchestrator_url`
         (the orchestrator's control-network address — Apple has no container
-        DNS) and presenting `gateway_token`."""
+        DNS) and presenting `gateway_token`.
+
+        Always returns True: this recreates the gateway container unconditionally
+        (a cold boot), so the caller reconciles running bottles against it
+        (PRD 0081)."""
         self._orchestrator_url = orchestrator_url
         self._gateway_token = gateway_token
         # Fail closed on a missing policy source or token: the resolver-only data
@@ -156,6 +160,7 @@ class MacosGateway(Gateway):
                 f"gateway container failed to start: "
                 f"{(result.stderr or '').strip() or '<no stderr>'}"
             )
+        return True
 
     def is_running(self) -> bool:
         return container_mod.container_is_running(self.name)

bot_bottle/backend/macos_container/infra.py

@@ -138,7 +138,15 @@ class MacosInfraService(InfraService):
         # `gateway` JWT and hands it to the gateway, which never sees the key
         # (#469).
         url = orchestrator.url()
-        gateway.connect_to_orchestrator(url, orchestrator.mint_gateway_token())
+        cold_booted = gateway.connect_to_orchestrator(
+            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 MacosContainerBottleBackend
+            MacosContainerBottleBackend().attach_bottled_agents_to_gateway()
         return url
 
     def ca_cert_pem(self, *, timeout: float = DEFAULT_CA_TIMEOUT_SECONDS) -> str:

bot_bottle/gateway/__init__.py

@@ -122,12 +122,18 @@ class Gateway(abc.ABC):
         return
 
     @abc.abstractmethod
-    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 the gateway to this orchestrator and bring it up: store the URL +
         the pre-minted `gateway` token as instance state, then (re)start the
         gateway unit carrying the mitmproxy CA + that token, resolving policy
         against `orchestrator_url`. Idempotent — a healthy, current gateway on
-        the same binding is left alone; a changed binding reconciles it."""
+        the same binding is left alone; a changed binding reconciles it.
+
+        Returns True when the gateway was actually (re)brought up (a cold boot,
+        so its mitmproxy minted a fresh CA and lost its per-bottle state), False
+        when a healthy current gateway was left untouched. The bring-up flow
+        uses this as the cold-boot signal that gates the running-bottle
+        reconcile (`attach_bottled_agents_to_gateway`, PRD 0081)."""
 
     @abc.abstractmethod
     def is_running(self) -> bool:

tests/unit/test_backend_secret_reprovision.py

@@ -159,5 +159,126 @@ class TestFirecrackerReprovision(unittest.TestCase):
         restore.assert_called_once_with(client, {})
 
 
+class TestFirecrackerCaReconcile(unittest.TestCase):
+    """PRD 0081: attach_bottled_agents_to_gateway pushes the current gateway CA
+    into every running agent VM over SSH."""
+
+    def _run_dir(self, root: Path, ip: str = "10.243.0.3") -> Path:
+        run_dir = root / "demo"
+        run_dir.mkdir()
+        (run_dir / "bottle_id_ed25519").write_text("key")
+        (run_dir / "config.json").write_text(json.dumps({
+            "boot-source": {"boot_args": f"root=/dev/vda ip={ip}::gw:mask::eth0:off"}
+        }))
+        return run_dir
+
+    def test_pushes_current_ca_over_stdin_and_rebuilds_trust_store(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            run_dir = self._run_dir(Path(tmp))
+            gw = Mock()
+            gw.ca_cert_pem.return_value = "PEM-DATA"
+            with patch.object(fc, "FirecrackerGateway", return_value=gw), \
+                 patch.object(fc.cleanup, "live_run_dirs", return_value=(run_dir,)), \
+                 patch.object(fc.util, "ssh_base_argv", return_value=["ssh", "guest"]), \
+                 patch.object(fc.subprocess, "run", return_value=_proc()) as run, \
+                 patch.object(fc, "info"):
+                fc.attach_bottled_agents_to_gateway()
+        # The cert goes over stdin (off argv), and the trust store is rebuilt.
+        self.assertEqual("PEM-DATA", run.call_args.kwargs["input"])
+        script = run.call_args.args[0][-1]
+        self.assertIn(fc.AGENT_CA_PATH, script)
+        self.assertIn("update-ca-certificates", script)
+
+    def test_unreachable_vm_is_skipped_not_fatal(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            run_dir = self._run_dir(Path(tmp))
+            gw = Mock()
+            gw.ca_cert_pem.return_value = "PEM"
+            with patch.object(fc, "FirecrackerGateway", return_value=gw), \
+                 patch.object(fc.cleanup, "live_run_dirs", return_value=(run_dir,)), \
+                 patch.object(fc.util, "ssh_base_argv", return_value=["ssh", "guest"]), \
+                 patch.object(fc.subprocess, "run", return_value=_proc(1, stderr="down")), \
+                 patch.object(fc, "info") as info:
+                fc.attach_bottled_agents_to_gateway()  # does not raise
+        self.assertTrue(any("skipped" in c.args[0] for c in info.call_args_list))
+
+    def test_missing_gateway_ca_skips_the_whole_reconcile(self) -> None:
+        gw = Mock()
+        gw.ca_cert_pem.side_effect = fc.GatewayError("no CA yet")
+        with patch.object(fc, "FirecrackerGateway", return_value=gw), \
+             patch.object(fc.cleanup, "live_run_dirs") as live, \
+             patch.object(fc, "info"):
+            fc.attach_bottled_agents_to_gateway()
+        live.assert_not_called()  # bailed before enumerating agents
+
+
+class TestDockerCaReconcile(unittest.TestCase):
+    def test_pushes_ca_to_each_agent_container_excluding_gateway(self) -> None:
+        # network inspect lists the gateway (excluded) + one agent + a blank line.
+        inspect = _proc(stdout="bot-bottle-orch-gateway\nbot-bottle-a\n\n")
+        with patch.object(
+            docker, "run_docker",
+            side_effect=[inspect, _proc(), _proc(), _proc()],
+        ) as run, patch.object(docker.log, "info"):
+            docker.attach_bottled_agents_to_gateway(
+                "PEM", gateway_name="bot-bottle-orch-gateway")
+        argvs = [c.args[0] for c in run.call_args_list]
+        # First call enumerates; the agent then gets mkdir + cp + exec — never
+        # the gateway container.
+        self.assertNotIn(
+            "bot-bottle-orch-gateway",
+            " ".join(tok for a in argvs[1:] for tok in a),
+        )
+        cp = next(a for a in argvs if a[:2] == ["docker", "cp"])
+        self.assertEqual(f"bot-bottle-a:{docker.AGENT_CA_PATH}", cp[-1])
+        exec_argv = next(a for a in argvs if "update-ca-certificates" in a[-1])
+        self.assertIn(docker.AGENT_CA_PATH, exec_argv[-1])
+
+    def test_one_container_failure_does_not_block_others(self) -> None:
+        inspect = _proc(stdout="a\nb\n")
+        # a: mkdir fails (skip); b: mkdir/cp/exec all succeed.
+        with patch.object(
+            docker, "run_docker",
+            side_effect=[inspect, _proc(1, stderr="gone"), _proc(), _proc(), _proc()],
+        ), patch.object(docker.log, "info") as info:
+            docker.attach_bottled_agents_to_gateway("PEM", gateway_name="gw")
+        self.assertTrue(
+            any("skipped for a" in str(c.args[0]) for c in info.call_args_list)
+        )
+
+    def test_fetches_ca_from_gateway_when_not_supplied(self) -> None:
+        svc = Mock()
+        svc.gateway.return_value.ca_cert_pem.return_value = "FETCHED"
+        with patch.object(docker, "DockerInfraService", return_value=svc), \
+             patch.object(docker, "run_docker", return_value=_proc(stdout="")), \
+             patch.object(docker.log, "info"):
+            docker.attach_bottled_agents_to_gateway(network="net", gateway_name="gw")
+        svc.gateway.return_value.ca_cert_pem.assert_called_once()
+
+
+class TestMacosCaReconcile(unittest.TestCase):
+    def test_pushes_ca_to_each_running_agent_container(self) -> None:
+        agent = SimpleNamespace(slug="demo")
+        with patch.object(mac, "enumerate_active", return_value=[agent]), \
+             patch.object(mac.container_mod, "run_container_argv",
+                          return_value=_proc()) as run, \
+             patch.object(mac, "info"):
+            mac.attach_bottled_agents_to_gateway("PEM")
+        argvs = [c.args[0] for c in run.call_args_list]
+        name = f"{mac.CONTAINER_NAME_PREFIX}demo"
+        self.assertTrue(any(name in a for a in argvs))
+        self.assertTrue(
+            any("update-ca-certificates" in a[-1] for a in argvs if a[-2:-1] == ["-c"])
+        )
+        self.assertTrue(any(mac.AGENT_CA_PATH in " ".join(a) for a in argvs))
+
+    def test_enumeration_failure_is_best_effort(self) -> None:
+        with patch.object(mac, "enumerate_active",
+                          side_effect=mac.EnumerationError("failed")), \
+             patch.object(mac, "info") as info:
+            mac.attach_bottled_agents_to_gateway("PEM")  # does not raise
+        self.assertIn("skipped", info.call_args.args[0])
+
+
 if __name__ == "__main__":
     unittest.main()

tests/unit/test_docker_infra.py

@@ -29,10 +29,20 @@ class TestDockerInfraService(unittest.TestCase):
         self.orch.gateway_url.return_value = f"http://{ORCHESTRATOR_NAME}:8099"
         self.orch.mint_gateway_token.return_value = "gw.jwt"
         self.gw = MagicMock()
+        # Default to a cold boot (gateway (re)created) so the reconcile branch
+        # runs; the actual reconcile is stubbed so these composition tests stay
+        # isolated from docker.
+        self.gw.connect_to_orchestrator.return_value = True
         for name, mock in (("orchestrator", self.orch), ("gateway", self.gw)):
             p = patch.object(self.svc, name, return_value=mock)
             p.start()
             self.addCleanup(p.stop)
+        ap = patch(
+            "bot_bottle.backend.docker.backend.DockerBottleBackend"
+            ".attach_bottled_agents_to_gateway"
+        )
+        self.attach = ap.start()
+        self.addCleanup(ap.stop)
 
     def test_url_delegates_to_the_orchestrator(self) -> None:
         self.assertEqual("http://127.0.0.1:8099", self.svc.url())
@@ -58,6 +68,20 @@ class TestDockerInfraService(unittest.TestCase):
             f"http://{ORCHESTRATOR_NAME}:8099", "gw.jwt")
         self.orch.mint_gateway_token.assert_called_once()
 
+    def test_cold_boot_reconciles_running_bottles(self) -> None:
+        # A (re)created gateway (connect returns True) triggers the bring-up
+        # reconcile so running bottles re-trust the fresh gateway (PRD 0081).
+        self.gw.connect_to_orchestrator.return_value = True
+        self.svc.ensure_running()
+        self.attach.assert_called_once_with()
+
+    def test_no_reconcile_when_gateway_left_untouched(self) -> None:
+        # A healthy current gateway (connect returns False) is not a cold boot,
+        # so there is nothing to reconcile.
+        self.gw.connect_to_orchestrator.return_value = False
+        self.svc.ensure_running()
+        self.attach.assert_not_called()
+
     def test_stop_removes_both_containers(self) -> None:
         with patch(_RUN) as run:
             self.svc.stop()

tests/unit/test_firecracker_gateway.py

@@ -61,7 +61,10 @@ class TestFirecrackerGatewayConnect(unittest.TestCase):
         booted = infra_vm.InfraVm(guest_ip="10.243.255.3", private_key=Path("/k"))
         with patch.object(infra_vm, "boot_vm", return_value=booted) as boot, \
                 patch.object(infra_vm, "push_secret") as push:
-            gw.connect_to_orchestrator(_ORCH_URL, _TOKEN)
+            cold_booted = gw.connect_to_orchestrator(_ORCH_URL, _TOKEN)
+        # The VM is booted fresh here (cold-boot path only), so it always
+        # signals a cold boot → the caller reconciles running bottles (PRD 0081).
+        self.assertTrue(cold_booted)
         # Booted on the gateway link with the gateway role; the orchestrator's
         # guest IP (parsed off the URL) rides the cmdline as bb_orch.
         kw = boot.call_args.kwargs

tests/unit/test_firecracker_infra.py

@@ -40,7 +40,21 @@ class TestAccessors(unittest.TestCase):
         self.assertEqual(f"http://{ip}:{infra_vm.ORCHESTRATOR_PORT}", svc.url())
 
 
+_ATTACH = (
+    "bot_bottle.backend.firecracker.backend.FirecrackerBottleBackend"
+    ".attach_bottled_agents_to_gateway"
+)
+
+
 class TestEnsureRunning(unittest.TestCase):
+    def setUp(self) -> None:
+        # The cold-boot branch reconciles running bottles against the fresh
+        # gateway (PRD 0081). Stub it so these composition tests stay isolated
+        # from SSH; the wiring itself is asserted in test_cold_boot_reconciles*.
+        ap = patch(_ATTACH)
+        self.attach = ap.start()
+        self.addCleanup(ap.stop)
+
     def test_adopts_when_healthy_alive_and_version_matches(self):
         # Healthy control plane + live gateway + existing key + matching version
         # marker -> adopt (no stop/build/boot); returns the orchestrator URL.
@@ -59,6 +73,8 @@ class TestEnsureRunning(unittest.TestCase):
             built.assert_not_called()
             ip = infra_vm.netpool.orch_slot().guest_ip
             self.assertEqual(f"http://{ip}:{infra_vm.ORCHESTRATOR_PORT}", url)
+            # Adopt path — no cold boot, so no running-bottle reconcile.
+            self.attach.assert_not_called()
 
     def test_reboots_both_when_version_stale(self):
         # Healthy control plane but the running pair booted an OLDER image
@@ -120,6 +136,8 @@ class TestEnsureRunning(unittest.TestCase):
         # reach the control plane at startup).
         orch_cls.return_value.ensure_running.assert_called_once()
         gw_cls.return_value.connect_to_orchestrator.assert_called_once()
+        # A cold boot minted a fresh gateway CA — reconcile running bottles.
+        self.attach.assert_called_once_with()
 
 
 if __name__ == "__main__":

tests/unit/test_macos_gateway.py

@@ -44,6 +44,17 @@ class TestMacosGatewayConnect(unittest.TestCase):
     def test_default_name(self) -> None:
         self.assertEqual(GATEWAY_NAME, self.gw.name)
 
+    def test_connect_signals_cold_boot(self) -> None:
+        # The container is recreated unconditionally, so connect always signals
+        # a cold boot → the caller reconciles running bottles (PRD 0081).
+        run = Mock(return_value=_proc())
+        with patch(f"{_GW}.container_mod") as mod, \
+                patch(f"{_GW}.host_gateway_ca_dir", return_value=Path("/ca")):
+            mod.dns_server.return_value = "1.1.1.1"
+            mod.bind_mount_spec.side_effect = _spec
+            mod.run_container_argv = run
+            self.assertTrue(self.gw.connect_to_orchestrator(_ORCH_URL, _TOKEN))
+
     def test_refuses_without_orchestrator_url(self) -> None:
         with patch(f"{_GW}.container_mod") as mod:
             with self.assertRaises(GatewayError):

tests/unit/test_macos_infra.py

@@ -24,10 +24,20 @@ class TestInfraEnsureRunning(unittest.TestCase):
         self.orch.url.return_value = "http://192.168.128.2:8099"
         self.orch.mint_gateway_token.return_value = "gw.jwt"
         self.gw = MagicMock()
+        # Default to a cold boot (gateway (re)created) so the reconcile branch
+        # runs; the actual reconcile is stubbed so these composition tests stay
+        # isolated from the `container` CLI.
+        self.gw.connect_to_orchestrator.return_value = True
         for name, mock in (("orchestrator", self.orch), ("gateway", self.gw)):
             p = patch.object(self.svc, name, return_value=mock)
             p.start()
             self.addCleanup(p.stop)
+        ap = patch(
+            "bot_bottle.backend.macos_container.backend"
+            ".MacosContainerBottleBackend.attach_bottled_agents_to_gateway"
+        )
+        self.attach = ap.start()
+        self.addCleanup(ap.stop)
 
     def test_composes_networks_builds_and_brings_up_orchestrator_first(self) -> None:
         with patch(f"{_INFRA}.ensure_networks") as nets:
@@ -46,6 +56,20 @@ class TestInfraEnsureRunning(unittest.TestCase):
             "http://192.168.128.2:8099", "gw.jwt")
         self.orch.mint_gateway_token.assert_called_once()
 
+    def test_cold_boot_reconciles_running_bottles(self) -> None:
+        # A (re)created gateway (connect returns True) triggers the bring-up
+        # reconcile so running bottles re-trust the fresh gateway (PRD 0081).
+        self.gw.connect_to_orchestrator.return_value = True
+        with patch(f"{_INFRA}.ensure_networks"):
+            self.svc.ensure_running()
+        self.attach.assert_called_once_with()
+
+    def test_no_reconcile_when_gateway_left_untouched(self) -> None:
+        self.gw.connect_to_orchestrator.return_value = False
+        with patch(f"{_INFRA}.ensure_networks"):
+            self.svc.ensure_running()
+        self.attach.assert_not_called()
+
     def test_is_healthy_delegates_to_the_orchestrator(self) -> None:
         self.orch.is_healthy.return_value = True
         self.assertTrue(self.svc.is_healthy())

tests/unit/test_orchestrator_gateway.py

@@ -86,9 +86,12 @@ class TestDockerGateway(unittest.TestCase):
             return _proc()
 
         with patch(_RUN_DOCKER, side_effect=fake):
-            self.sc.connect_to_orchestrator(_ORCH_URL, _TOKEN)
+            cold_booted = self.sc.connect_to_orchestrator(_ORCH_URL, _TOKEN)
         self.assertEqual([], [c for c in calls if c[:2] == ["docker", "run"]])
         self.assertEqual([], [c for c in calls if c[:2] == ["docker", "rm"]])
+        # A healthy current gateway left untouched is not a cold boot — the
+        # bring-up flow skips the running-bottle reconcile (PRD 0081).
+        self.assertFalse(cold_booted)
 
     def test_ensure_running_recreates_when_image_is_stale(self) -> None:
         # Running, but the container was built from an OLD image → recreate so
@@ -106,9 +109,12 @@ class TestDockerGateway(unittest.TestCase):
             return _proc()
 
         with patch(_RUN_DOCKER, side_effect=fake):
-            self.sc.connect_to_orchestrator(_ORCH_URL, _TOKEN)
+            cold_booted = self.sc.connect_to_orchestrator(_ORCH_URL, _TOKEN)
         self.assertEqual(1, len([c for c in calls if c[:2] == ["docker", "run"]]))
         self.assertTrue(any(c[:2] == ["docker", "rm"] for c in calls))
+        # A recreated container minted/mounted its CA afresh — a cold boot that
+        # triggers the running-bottle reconcile (PRD 0081).
+        self.assertTrue(cold_booted)
 
     def test_ensure_running_starts_the_singleton_when_absent(self) -> None:
         calls: list[list[str]] = []
@@ -118,7 +124,8 @@ class TestDockerGateway(unittest.TestCase):
             return _proc(stdout="") if argv[:2] == ["docker", "ps"] else _proc()
 
         with patch(_RUN_DOCKER, side_effect=fake):
-            self.sc.connect_to_orchestrator(_ORCH_URL, _TOKEN)
+            cold_booted = self.sc.connect_to_orchestrator(_ORCH_URL, _TOKEN)
+        self.assertTrue(cold_booted)  # started fresh → reconcile running bottles
 
         runs = [c for c in calls if c[:2] == ["docker", "run"]]
         self.assertEqual(1, len(runs))