refactor(macos): one infra container (control plane + gateway), fixes shared-DB races

Adopts the firecracker infra-VM pattern for macOS: the orchestrator control
plane and the gateway data plane now run in a SINGLE Apple container instead of
two. Apple Containers are lightweight VMs with separate kernels, so the prior
two-container design had both guests writing one bot-bottle.db over virtiofs,
where fcntl locks are not coherent across kernels — concurrent writes (the
orchestrator's registry vs the gateway supervise daemon's queue) could corrupt
it. One container = one kernel = coherent locking.

The DB moves onto a container-only Apple volume (bot-bottle-mac-db), never
bind-mounted from the host, so no host process opens the live file either. The
host CLI already reaches registry + supervise state over the control-plane HTTP
surface (cli/supervise.py uses OrchestratorClient), exactly as firecracker's
VM-only DB requires.

Two simplifications fall out of the single container:
- No DNS dance: the control plane and gateway daemons reach each other over
  127.0.0.1, so the orchestrator-before-gateway ordering (a workaround for
  Apple having no container DNS) is gone, along with the moved-IP recreate
  logic it needed.
- Net -243 lines.

Mechanics: the infra container runs from the gateway image with the
control-plane source bind-mounted read-only (like the docker orchestrator, so a
code change needs no rebuild) and a small sh -c init that starts both processes
(mirrors firecracker's _infra_init). Also implements the macOS backend's
ensure_orchestrator() and adds it to discover_orchestrator_url, so operator
tools (supervise) can bring up / find the control plane on demand — previously
the macOS backend died with "no orchestrator control plane".

Verified end-to-end on real Apple Container 1.0.0: the single infra container
comes up healthy (one address for control plane + gateway), both processes run,
the DB is written on the container-only volume, host-side supervise works over
HTTP, and a registered agent gets 200 for an allowed host / 403 for a denied
one. 1824 unit tests pass with `container` absent (CI parity), pyright clean,
pylint 9.89.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

bot_bottle/backend/macos_container/backend.py

@@ -89,6 +89,14 @@ class MacosContainerBottleBackend(
         with _launch.launch(plan, provision=self.provision) as bottle:
             yield bottle
 
+    def ensure_orchestrator(self) -> str:
+        """Bring up the per-host infra container (control plane + gateway) and
+        return its control-plane URL — the on-demand entry point operator tools
+        (`supervise`) call when no control plane is running yet. Mirrors
+        firecracker's infra-VM bring-up."""
+        from .infra import MacosInfraService
+        return MacosInfraService().ensure_running().control_plane_url
+
     def prepare_cleanup(self) -> MacosContainerBottleCleanupPlan:
         return _cleanup.prepare_cleanup()
 

bot_bottle/backend/macos_container/bottle.py

@@ -52,6 +52,7 @@ class MacosContainerBottle(Bottle):
         terminal_title: str = "",
         terminal_color: str = "",
         agent_workdir: str = "/home/node",
+        exec_env: dict[str, str] | None = None,
     ):
         self.name = container
         self._teardown = teardown
@@ -62,6 +63,15 @@ class MacosContainerBottle(Bottle):
         self.terminal_color = terminal_color
         self.agent_provider_template = agent_provider_template
         self.agent_workdir = agent_workdir
+        # Env applied to the agent process at `container exec` time, on top of
+        # what the container was run with. This is how the identity token
+        # reaches the agent (PRD 0070): registration mints it *after* the
+        # container exists — its source IP is the registration key and Apple
+        # Container assigns that by DHCP — so it cannot be in the run-time env
+        # the way docker's compose spec does it. `container exec --env` wins
+        # over the run-time value, so the token-bearing proxy URL set here
+        # supersedes the token-less one baked in at launch.
+        self._exec_env = dict(exec_env or {})
         self._closed = False
 
     def agent_argv(self, argv: list[str], *, tty: bool = True) -> list[str]:
@@ -74,6 +84,12 @@ class MacosContainerBottle(Bottle):
             )
         )
         container_exec = ["container", "exec"]
+        # Bare env names, same rule as the terminal hints below: the value
+        # stays in the child env `exec_agent` builds and never reaches argv —
+        # the proxy URL here carries the identity token, which `ps` would
+        # otherwise expose to every process on the host.
+        for name in sorted(self._exec_env):
+            container_exec.extend(["--env", name])
         if tty:
             container_exec.extend(["--interactive", "--tty"])
             # Forward terminal capability hints so TUIs can enable modified-key
@@ -94,21 +110,33 @@ class MacosContainerBottle(Bottle):
 
     def exec_agent(self, argv: list[str], *, tty: bool = True) -> int:
         agent_argv = self.agent_argv(argv, tty=tty)
+        # The values behind the bare `--env` names in `agent_argv`. `sh -lc`
+        # below is in this process tree, so the child env reaches `container
+        # exec` either way.
+        env = {**os.environ, **self._exec_env} if self._exec_env else None
         script = (
             exec_shell_script(agent_argv, self.terminal_title, self.terminal_color)
             if tty else None
         )
         if script is None:
-            return subprocess.run(agent_argv, check=False).returncode
-        return subprocess.run(["sh", "-lc", script], check=False).returncode
+            return subprocess.run(agent_argv, env=env, check=False).returncode
+        return subprocess.run(["sh", "-lc", script], env=env, check=False).returncode
 
     def exec(self, script: str, *, user: str = "node") -> ExecResult:
+        # Carry the same exec env the agent gets: provisioning steps run
+        # through here, and a provider whose provision step fetches anything
+        # would egress without the identity token and be denied by /resolve.
+        # Bare `--env NAME` again, so the token stays off argv.
+        argv = ["container", "exec", "--user", user, "--interactive"]
+        for name in sorted(self._exec_env):
+            argv.extend(["--env", name])
+        argv.extend([self.name, "sh", "-s"])
         result = subprocess.run(
-            ["container", "exec", "--user", user, "--interactive",
-             self.name, "sh", "-s"],
+            argv,
             input=script,
             capture_output=True,
             text=True,
+            env={**os.environ, **self._exec_env} if self._exec_env else None,
             check=False,
         )
         return ExecResult(

bot_bottle/backend/macos_container/bottle_plan.py

@@ -13,9 +13,13 @@ from .. import BottlePlan
 class MacosContainerBottlePlan(BottlePlan):
     slug: str
     forwarded_env: dict[str, str] = field(repr=False)
-    agent_proxy_url: str = ""
     agent_git_gate_url: str = ""
     agent_supervise_url: str = ""
+    # Read by provision-time consumers (git extraHeader, supervise MCP header)
+    # via getattr(plan, "identity_token", ""); stamped in launch after the
+    # bottle is registered. See launch.py's stamp for why it lives here and not
+    # only in the exec-time proxy env.
+    identity_token: str = ""
 
     @property
     def container_name(self) -> str:

bot_bottle/backend/macos_container/consolidated_launch.py

@@ -0,0 +1,144 @@
+"""Consolidated bottle launch sequence for the macOS backend (PRD 0070).
+
+The docker backend allocates a free address, pins the agent to it with
+`--ip`, registers it, *then* starts the agent — registration precedes launch
+because the pinned address is known up front.
+
+**Apple Container 1.0.0 has no `--ip`.** The `--network` flag takes only
+`<name>[,mac=…][,mtu=…]`; the address is assigned by vmnet's DHCP and is
+knowable only once the container is running. So the macOS order inverts:
+
+    ensure_gateway()  ->  caller starts the agent  ->  register_agent(source_ip)
+
+That is why this module exposes two functions where docker has one — the
+caller has to start the agent in between. `ensure_gateway` runs first because
+the agent's proxy env needs the gateway's address at `container run` time; the
+agent's *own* address (the attribution key) only exists afterwards.
+
+The control plane and the gateway are one **infra container** here (see
+`infra`), so `gateway_ip` and the control-plane host are the same address.
+
+The consequence for the identity token: it is minted by registration, i.e.
+*after* the agent container exists, so it cannot be baked into the run-time
+env the way docker's compose spec does. It is delivered at `container exec`
+time instead — see `bottle.MacosContainerBottle`.
+
+That delivery is load-bearing, not a nicety: `/resolve` requires a matching
+`(source_ip, identity_token)` pair and fail-closes with no source-IP-only
+fallback (#366). So egress that does not carry the token is denied — which is
+the safe direction, and is why the agent's init process is a bare `sleep` and
+every real command arrives through `container exec`.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+from ...egress import EgressPlan
+from ...git_gate import GitGatePlan
+from ...orchestrator.client import OrchestratorClient
+from ...orchestrator.registration import registration_inputs
+from ..docker.gateway_provision import deprovision_git_gate, provision_git_gate
+from .gateway import GATEWAY_NETWORK
+from .gateway_provision import AppleGatewayTransport
+from .infra import MacosInfraService, OrchestratorStartError
+
+
+class ConsolidatedLaunchError(RuntimeError):
+    """The consolidated register/provision sequence could not complete."""
+
+
+@dataclass(frozen=True)
+class GatewayEndpoint:
+    """What the agent `container run` needs to reach the shared gateway (the
+    infra container). `gateway_ip` is that container's host-only address, the
+    same host the control-plane URL points at."""
+
+    orchestrator_url: str
+    gateway_ip: str         # the gateway's address — the agent's proxy target
+    gateway_ca_pem: str     # the shared CA the provisioner installs
+    network: str            # the shared host-only network to attach to
+
+
+@dataclass(frozen=True)
+class LaunchContext:
+    """What the running agent needs once it has been registered."""
+
+    bottle_id: str
+    identity_token: str
+    source_ip: str          # the agent's DHCP-assigned address (attribution key)
+    gateway_ip: str
+    network: str
+    orchestrator_url: str
+
+
+def ensure_gateway(
+    *, service: MacosInfraService | None = None,
+) -> GatewayEndpoint:
+    """Ensure the per-host infra container (control plane + gateway) is up and
+    report how to reach it. Idempotent — one singleton, so N bottle launches
+    share it. Call before starting the agent container: the agent's proxy env
+    needs `gateway_ip` at run time."""
+    service = service or MacosInfraService()
+    infra = service.ensure_running()
+    return GatewayEndpoint(
+        orchestrator_url=infra.control_plane_url,
+        gateway_ip=infra.gateway_ip,
+        gateway_ca_pem=service.ca_cert_pem(),
+        network=service.network,
+    )
+
+
+def register_agent(
+    egress_plan: EgressPlan,
+    git_gate_plan: GitGatePlan,
+    *,
+    source_ip: str,
+    endpoint: GatewayEndpoint,
+    image_ref: str = "",
+    tokens: dict[str, str] | None = None,
+) -> LaunchContext:
+    """Register the (already running) agent by its address and provision its
+    git-gate state into the gateway. `source_ip` must be read from the live
+    container — it is the attribution key the gateway resolves policy by.
+    Raises on failure; the caller tears down."""
+    client = OrchestratorClient(endpoint.orchestrator_url)
+    inputs = registration_inputs(egress_plan)
+    reg = client.register_bottle(
+        source_ip, image_ref=image_ref, policy=inputs.policy,
+        metadata=inputs.metadata, tokens=tokens,
+    )
+    try:
+        provision_git_gate(AppleGatewayTransport(), 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,
+        gateway_ip=endpoint.gateway_ip,
+        network=endpoint.network,
+        orchestrator_url=endpoint.orchestrator_url,
+    )
+
+
+def teardown_consolidated(bottle_id: str, *, orchestrator_url: str) -> 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. Does NOT
+    stop the gateway — it's a persistent per-host singleton."""
+    OrchestratorClient(orchestrator_url).teardown_bottle(bottle_id)
+    deprovision_git_gate(AppleGatewayTransport(), bottle_id)
+
+
+__all__ = [
+    "GatewayEndpoint",
+    "LaunchContext",
+    "ensure_gateway",
+    "register_agent",
+    "teardown_consolidated",
+    "ConsolidatedLaunchError",
+    "OrchestratorStartError",
+    "GATEWAY_NETWORK",
+]

bot_bottle/backend/macos_container/egress_apply.py

@@ -1,9 +1,11 @@
 """Host-side egress route-apply for the macos-container backend.
 
 The per-bottle companion container this used to signal (`container kill
---signal HUP <container>`) was removed in the companion-container removal (#385),
-along with the disabled macOS launch path. Fails closed until the macOS
-backend grows the consolidated gateway.
+--signal HUP <container>`) was removed in the companion-container removal
+(#385). In the consolidated model the shared gateway resolves egress policy
+per-request against the orchestrator rather than reloading a per-bottle routes
+file, so the live per-bottle reload is not supported here and fails closed
+until the gateway-side apply lands — same posture as the docker backend.
 """
 
 from __future__ import annotations
@@ -16,8 +18,8 @@ class MacOSContainerEgressApplicator(EgressApplicator):
         del slug
         raise EgressApplyError(
             "live egress route-apply was removed with the per-bottle "
-            "companion container (#385); the macos-container backend is "
-            "disabled until it uses the consolidated gateway."
+            "companion container (#385); route changes will flow through "
+            "the consolidated gateway in a follow-up."
         )
 
 

bot_bottle/backend/macos_container/enumerate.py

@@ -1,14 +1,42 @@
-"""Active-agent enumeration for the macOS Apple Container backend.
-
-The backend is disabled during the companion-container removal (#385) — it can't
-launch bottles, so there are none to enumerate. Enumeration returns when
-the backend grows the consolidated gateway.
-"""
+"""Active-agent enumeration for the macOS Apple Container backend."""
 
 from __future__ import annotations
 
+import subprocess
+
+from ...bottle_state import read_metadata
 from .. import ActiveAgent
+from .infra import INFRA_NAME
+
+_PREFIX = "bot-bottle-"
+# The shared per-host infra container carries the same prefix as agent
+# containers but is infrastructure, not a bottle — one control plane + gateway
+# serves every agent, so listing it as an agent would invent one per host.
+_INFRA_NAMES = frozenset({INFRA_NAME})
 
 
 def enumerate_active() -> list[ActiveAgent]:
-    return []
+    result = subprocess.run(
+        ["container", "list", "--quiet"],
+        capture_output=True,
+        text=True,
+        check=False,
+    )
+    if result.returncode != 0:
+        return []
+    out: list[ActiveAgent] = []
+    for name in sorted(line.strip() for line in result.stdout.splitlines()):
+        if not name.startswith(_PREFIX) or name in _INFRA_NAMES:
+            continue
+        slug = name[len(_PREFIX):]
+        metadata = read_metadata(slug)
+        out.append(ActiveAgent(
+            backend_name="macos-container",
+            slug=slug,
+            agent_name=metadata.agent_name if metadata else "?",
+            started_at=metadata.started_at if metadata else "",
+            services=(),
+            label=metadata.label if metadata else "",
+            color=metadata.color if metadata else "",
+        ))
+    return out

bot_bottle/backend/macos_container/gateway.py

@@ -0,0 +1,46 @@
+"""Shared network/image constants for the macOS consolidated infra container.
+
+The gateway data plane no longer runs as its own Apple container — it shares a
+single per-host **infra container** with the control plane (see `infra`),
+because two Apple-Container guests writing one `bot-bottle.db` over virtiofs
+would race incoherent `fcntl` locks. This module holds the pieces both the
+infra service and the launch/provision glue need: the network names, the
+gateway image, and the network-creation helper.
+"""
+
+from __future__ import annotations
+
+import os
+
+from ...orchestrator.gateway import GatewayError
+from . import util as container_mod
+
+# The shared host-only network the infra container and every agent bottle sit
+# on. The agent's address here is the attribution key. Distinct from the docker
+# names so both backends can coexist on one host.
+GATEWAY_NETWORK = "bot-bottle-mac-gateway"
+# The NAT network that gives the infra container (and only it) a route out.
+GATEWAY_EGRESS_NETWORK = "bot-bottle-mac-egress"
+
+GATEWAY_IMAGE = os.environ.get("BOT_BOTTLE_GATEWAY_IMAGE", "bot-bottle-gateway:latest")
+
+DEFAULT_CA_TIMEOUT_SECONDS = 30.0
+
+
+def ensure_networks(
+    network: str = GATEWAY_NETWORK, egress_network: str = GATEWAY_EGRESS_NETWORK,
+) -> None:
+    """Create the shared host-only network + the NAT egress network. Idempotent
+    — `create_network` tolerates 'already exists'."""
+    container_mod.create_network(egress_network)
+    container_mod.create_network(network, internal=True)
+
+
+__all__ = [
+    "GATEWAY_NETWORK",
+    "GATEWAY_EGRESS_NETWORK",
+    "GATEWAY_IMAGE",
+    "GatewayError",
+    "DEFAULT_CA_TIMEOUT_SECONDS",
+    "ensure_networks",
+]

bot_bottle/backend/macos_container/gateway_provision.py

@@ -0,0 +1,44 @@
+"""`GatewayTransport` for the Apple infra container (PRD 0070).
+
+The provisioning *logic* (per-bottle creds dirs, namespaced repo init) is
+backend-neutral and lives in `backend.docker.gateway_provision`; this is only
+the transport — how files and commands reach the running gateway. Docker uses
+`docker exec`/`docker cp` and Firecracker uses SSH; Apple uses the `container`
+CLI's equivalents against the infra container that hosts the gateway daemons.
+"""
+
+from __future__ import annotations
+
+from ..docker.gateway_provision import GatewayProvisionError
+from . import util as container_mod
+from .infra import INFRA_NAME
+
+
+class AppleGatewayTransport:
+    """`GatewayTransport` for the gateway daemons in the Apple infra container."""
+
+    def __init__(self, gateway: str = INFRA_NAME) -> None:
+        self.gateway = gateway
+
+    def exec(self, argv: list[str]) -> None:
+        result = container_mod.run_container_argv(
+            ["container", "exec", self.gateway, *argv]
+        )
+        if result.returncode != 0:
+            raise GatewayProvisionError(
+                f"gateway exec {argv!r} failed: "
+                f"{(result.stderr or '').strip() or '<no stderr>'}"
+            )
+
+    def cp_into(self, src: str, dest: str) -> None:
+        result = container_mod.run_container_argv(
+            ["container", "cp", src, f"{self.gateway}:{dest}"]
+        )
+        if result.returncode != 0:
+            raise GatewayProvisionError(
+                f"gateway cp {src} -> {dest} failed: "
+                f"{(result.stderr or '').strip() or '<no stderr>'}"
+            )
+
+
+__all__ = ["AppleGatewayTransport", "GatewayProvisionError"]

bot_bottle/backend/macos_container/infra.py

@@ -0,0 +1,292 @@
+"""The per-host infra container for the macOS backend (PRD 0070).
+
+A single persistent Apple container that runs BOTH the orchestrator control
+plane and the gateway data plane — the macOS analogue of the Firecracker infra
+VM (`backend/firecracker/infra_vm.py`), not the docker backend's two separate
+containers.
+
+Why one container, not two: Apple Containers are lightweight VMs, each with its
+own kernel. The docker backend runs the orchestrator and gateway as two
+containers safely because they share the host kernel, so their concurrent
+writes to the one `bot-bottle.db` (the orchestrator's registry + the gateway
+supervise daemon's queue) are serialized by coherent `fcntl` locks. Across two
+*guest* kernels sharing a virtiofs-mounted DB those locks are not coherent, and
+concurrent writers can corrupt the file. Firecracker solved this by putting
+both services in one guest with the DB on a device only that guest mounts; this
+does the same with Apple primitives.
+
+Two consequences fall out of the single container, both simplifications:
+
+- **No DNS dance.** The control plane and the gateway daemons reach each other
+  over `127.0.0.1`, so nothing depends on Apple's (absent) container DNS and
+  there is no orchestrator-before-gateway ordering to get right.
+- **The DB is never host-shared.** It lives on a container-only volume, so no
+  host process opens the live file. The host CLI reaches registry + supervise
+  state through the control-plane HTTP surface (`cli/supervise.py` already uses
+  `OrchestratorClient`), exactly as it does for firecracker.
+
+The control-plane source is bind-mounted (like the docker orchestrator), so a
+code change takes effect on the next launch without an image rebuild; the
+gateway daemons are baked in the gateway image and rebuild through its own
+digest check.
+"""
+
+from __future__ import annotations
+
+import time
+import urllib.error
+import urllib.request
+from dataclasses import dataclass
+from pathlib import Path
+
+from ... import log
+from ...orchestrator.gateway import GATEWAY_CA_CERT
+from ...orchestrator.lifecycle import (
+    DEFAULT_PORT,
+    DEFAULT_STARTUP_TIMEOUT_SECONDS,
+    OrchestratorStartError,
+    source_hash,
+)
+from ...paths import HOST_DB_FILENAME
+from . import util as container_mod
+from .gateway import (
+    DEFAULT_CA_TIMEOUT_SECONDS,
+    GATEWAY_EGRESS_NETWORK,
+    GATEWAY_IMAGE,
+    GATEWAY_NETWORK,
+    GatewayError,
+    ensure_networks,
+)
+
+# The one per-host infra container: control plane + gateway data plane.
+INFRA_NAME = "bot-bottle-mac-infra"
+INFRA_LABEL = "bot-bottle-mac-infra=1"
+# Container-only volume holding bot-bottle.db. No host bind-mount, so the DB is
+# written by exactly one kernel (this container's). Survives recreation.
+INFRA_DB_VOLUME = "bot-bottle-mac-db"
+
+# BOT_BOTTLE_ROOT inside the container; host_db_path() resolves the DB to
+# <root>/db/<filename> and the supervise daemon writes the same file.
+_DB_ROOT_IN_CONTAINER = "/var/lib/bot-bottle"
+_DB_PATH_IN_CONTAINER = f"{_DB_ROOT_IN_CONTAINER}/db/{HOST_DB_FILENAME}"
+_SRC_IN_CONTAINER = "/bot-bottle-src"
+
+_REPO_ROOT = Path(__file__).resolve().parents[3]
+
+_HEALTH_POLL_SECONDS = 0.25
+_HEALTH_REQUEST_TIMEOUT_SECONDS = 1.0
+_CA_POLL_SECONDS = 0.5
+
+# The gateway subset the consolidated model runs (no per-bottle git:// daemon).
+_GATEWAY_DAEMONS = "egress,git-http,supervise"
+
+
+def _init_script(port: int) -> str:
+    """PID-1 init: start the control plane and the gateway daemons, both in
+    this container, reaching each other over loopback. Backgrounded so `wait`
+    reaps as PID 1. No `set -e` — a transient daemon failure must not kill the
+    whole container (gateway_init applies the same 'stay up' policy)."""
+    return (
+        "export PATH=/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin\n"
+        f"mkdir -p $(dirname {_DB_PATH_IN_CONTAINER})\n"
+        # Control plane, from the bind-mounted source (stdlib-only package).
+        f"( cd {_SRC_IN_CONTAINER} && BOT_BOTTLE_ROOT={_DB_ROOT_IN_CONTAINER} "
+        f"python3 -m bot_bottle.orchestrator --host 0.0.0.0 --port {port} "
+        "--broker stub ) &\n"
+        # Gateway data plane, multi-tenant against the local control plane.
+        f"( cd /app && BOT_BOTTLE_GATEWAY_DAEMONS={_GATEWAY_DAEMONS} "
+        f"BOT_BOTTLE_ORCHESTRATOR_URL=http://127.0.0.1:{port} "
+        f"SUPERVISE_DB_PATH={_DB_PATH_IN_CONTAINER} python3 /app/gateway_init.py ) &\n"
+        "while : ; do wait ; done\n"
+    )
+
+
+@dataclass(frozen=True)
+class InfraEndpoint:
+    """How to reach the running infra container. The control plane and the
+    gateway are the same container, so one address serves both."""
+
+    control_plane_url: str   # http://<infra ip>:8099 — host CLI + registration
+    gateway_ip: str          # same container; agents' proxy / git-http / MCP target
+
+
+class MacosInfraService:
+    """Manages the single per-host infra container. Callers use
+    `ensure_running()` (returns the endpoint) and `ca_cert_pem()`."""
+
+    def __init__(
+        self,
+        *,
+        port: int = DEFAULT_PORT,
+        network: str = GATEWAY_NETWORK,
+        egress_network: str = GATEWAY_EGRESS_NETWORK,
+        image: str = GATEWAY_IMAGE,
+        repo_root: Path = _REPO_ROOT,
+        name: str = INFRA_NAME,
+        db_volume: str = INFRA_DB_VOLUME,
+    ) -> None:
+        self.port = port
+        self.network = network
+        self.egress_network = egress_network
+        self.image = image
+        self._repo_root = repo_root
+        self._name = name
+        self._db_volume = db_volume
+
+    def _resolve_url(self) -> str:
+        """The control-plane URL, or "" while the container has no address."""
+        ip = container_mod.try_container_ipv4_on_network(self._name, self.network)
+        return f"http://{ip}:{self.port}" if ip else ""
+
+    def is_healthy(
+        self, url: str, *, timeout: float = _HEALTH_REQUEST_TIMEOUT_SECONDS,
+    ) -> bool:
+        if not url:
+            return False
+        try:
+            with urllib.request.urlopen(f"{url}/health", timeout=timeout) as resp:
+                return resp.status == 200
+        except (urllib.error.URLError, TimeoutError, OSError):
+            return False
+
+    def _source_current(self, current_hash: str) -> bool:
+        """True iff the running infra container was created from the current
+        bind-mounted control-plane source. The control-plane process loads that
+        code at startup and won't reload it, so a stale container keeps serving
+        OLD code."""
+        if not container_mod.container_is_running(self._name):
+            return False
+        env = container_mod.container_env(self._name)
+        if not env:
+            return True  # can't compare → don't churn a working container
+        return env.get("BOT_BOTTLE_SOURCE_HASH") == current_hash
+
+    def _running_healthy_endpoint(self, current_hash: str) -> InfraEndpoint | None:
+        """The endpoint if the running container is BOTH source-current and
+        answering /health, else None (→ recreate). Health, not just the source
+        label, is what lets a wedged-but-current container self-heal instead of
+        being polled to death forever."""
+        if not self._source_current(current_hash):
+            return None
+        url = self._resolve_url()
+        if url and self.is_healthy(url):
+            return InfraEndpoint(control_plane_url=url, gateway_ip=_ip_of(url))
+        return None
+
+    def ensure_built(self) -> None:
+        """Ensure the gateway data-plane image exists. The control-plane source
+        is bind-mounted, not baked, so only the gateway image needs building."""
+        container_mod.build_image(
+            self.image, str(self._repo_root), dockerfile="Dockerfile.gateway",
+        )
+
+    def ensure_running(
+        self, *, startup_timeout: float = DEFAULT_STARTUP_TIMEOUT_SECONDS,
+    ) -> InfraEndpoint:
+        """Ensure the single infra container is up; return how to reach it.
+        Idempotent per-host singleton — a healthy container on current source
+        is left untouched, so N launches share the one control plane + gateway.
+        Raises `OrchestratorStartError` on startup timeout."""
+        current_hash = source_hash(self._repo_root)
+        endpoint = self._running_healthy_endpoint(current_hash)
+        if endpoint is not None:
+            return endpoint
+        self.ensure_built()
+        log.info("starting infra container", context={"name": self._name})
+        self._run_container(current_hash)
+        return self._wait_healthy(startup_timeout)
+
+    def _run_container(self, current_hash: str) -> None:
+        ensure_networks(self.network, self.egress_network)
+        container_mod.force_remove_container(self._name)
+        argv = [
+            "container", "run", "--detach",
+            "--name", self._name,
+            "--label", "bot-bottle.backend=macos-container",
+            "--label", INFRA_LABEL,
+            # NAT network FIRST so the gateway's egress has a default route;
+            # the host-only network is where agents (and the host CLI) reach it.
+            "--network", self.egress_network,
+            "--network", self.network,
+            "--dns", container_mod.dns_server(),
+            # Container-only DB volume: one kernel writes bot-bottle.db, never
+            # shared with the host or another guest.
+            "--volume", f"{self._db_volume}:{_DB_ROOT_IN_CONTAINER}",
+            # Bind-mount the control-plane source (read-only); a code change
+            # takes effect on relaunch with no image rebuild.
+            "--mount",
+            container_mod.bind_mount_spec(
+                str(self._repo_root), _SRC_IN_CONTAINER, readonly=True),
+            # Baked onto the container so `_source_current` can detect a real
+            # control-plane code change and recreate.
+            "--env", f"BOT_BOTTLE_SOURCE_HASH={current_hash}",
+            "--entrypoint", "sh",
+            self.image,
+            "-c", _init_script(self.port),
+        ]
+        result = container_mod.run_container_argv(argv)
+        if result.returncode != 0:
+            raise OrchestratorStartError(
+                f"infra container failed to start: "
+                f"{(result.stderr or '').strip() or '<no stderr>'}"
+            )
+
+    def _wait_healthy(self, startup_timeout: float) -> InfraEndpoint:
+        deadline = time.monotonic() + startup_timeout
+        while True:
+            url = self._resolve_url()
+            if url and self.is_healthy(url):
+                log.info("infra container healthy", context={"url": url})
+                return InfraEndpoint(control_plane_url=url, gateway_ip=_ip_of(url))
+            if time.monotonic() >= deadline:
+                raise OrchestratorStartError(
+                    f"infra container did not become healthy within "
+                    f"{startup_timeout:g}s"
+                )
+            time.sleep(_HEALTH_POLL_SECONDS)
+
+    def ca_cert_pem(self, *, timeout: float = DEFAULT_CA_TIMEOUT_SECONDS) -> str:
+        """The gateway's mitmproxy CA (PEM) agents install to trust its TLS
+        interception. Read out of the container (the CA lives on a
+        container-internal path, not a host mount); polls because mitmproxy
+        writes it a beat after start."""
+        deadline = time.monotonic() + timeout
+        while True:
+            result = container_mod.run_container_argv(
+                ["container", "exec", self._name, "cat", GATEWAY_CA_CERT])
+            if result.returncode == 0 and result.stdout.strip():
+                return result.stdout
+            if time.monotonic() >= deadline:
+                raise GatewayError(
+                    f"gateway CA not available in {self._name} after {timeout:g}s: "
+                    f"{(result.stderr or '').strip() or 'empty'}"
+                )
+            time.sleep(_CA_POLL_SECONDS)
+
+    def stop(self) -> None:
+        """Remove the infra container (idempotent). The DB volume persists."""
+        container_mod.force_remove_container(self._name)
+
+
+def _ip_of(url: str) -> str:
+    """The host from an http://host:port URL."""
+    return url.split("://", 1)[-1].rsplit(":", 1)[0]
+
+
+def probe_control_plane_url(port: int = DEFAULT_PORT) -> str:
+    """The running infra container's control-plane URL, or "" if it isn't up.
+    Used by host-side control-plane discovery (`discover_orchestrator_url`);
+    safe to call on any host — returns "" when the container or the `container`
+    CLI isn't present."""
+    ip = container_mod.try_container_ipv4_on_network(INFRA_NAME, GATEWAY_NETWORK)
+    return f"http://{ip}:{port}" if ip else ""
+
+
+__all__ = [
+    "MacosInfraService",
+    "InfraEndpoint",
+    "OrchestratorStartError",
+    "GatewayError",
+    "INFRA_NAME",
+    "INFRA_DB_VOLUME",
+]

bot_bottle/backend/macos_container/launch.py

@@ -1,24 +1,74 @@
-"""Launch flow for the macOS Apple Container backend — disabled (#385).
+"""Launch flow for the macOS Apple Container backend (PRD 0070).
 
-This backend launched a per-bottle companion container (the egress /
-git-gate / supervise data plane) alongside the agent container, with the
-agent's proxy env pointed at the companion's host-only IP. That
-per-bottle-companion architecture was removed in the companion-container removal;
-the macOS backend will be re-enabled once it grows the consolidated
-per-host gateway the docker backend already uses.
+The agent container attaches to the **shared host-only gateway network** and
+proxies egress through the one per-host gateway, replacing the per-bottle
+companion container removed in #385.
 
-Until then, launching a macOS bottle fails closed. `prepare` / `status`
-/ cleanup still work.
+The order differs from docker's, forced by Apple Container 1.0.0 having no
+`--ip` (see `consolidated_launch`): the agent is started *before* it is
+registered, because its DHCP-assigned address — the attribution key — does not
+exist until then.
+
+    gateway up -> run agent -> read its IP -> register it -> provision
+
+Two things follow from that inversion:
+
+- The **identity token** is minted by registration and so cannot be in the
+  agent's run-time env; it rides the proxy URL applied at `container exec`
+  time (`bottle.MacosContainerBottle`). `/resolve` requires it (#366), so
+  egress without it is denied — hence the bare `sleep` init: every real agent
+  command goes through exec and therefore carries the token.
+- The agent is run with `--cap-drop CAP_NET_RAW`. Apple Container grants
+  NET_RAW by default, which would let an agent open a raw socket and forge a
+  neighbour's source address on the shared segment. NET_ADMIN is already
+  absent (the agent cannot change its own address or route), so dropping
+  NET_RAW is what closes the source-address half of PRD 0070's invariant:
+  "a packet's source address, as seen by the orchestrator, provably identifies
+  the originating bottle." The identity token is the other half — an attacker
+  would need to forge the address *and* steal the token — but the invariant is
+  a stated precondition of consolidation, so it is enforced on its own terms
+  rather than left to the token.
 """
 
 from __future__ import annotations
 
-from contextlib import contextmanager
+import dataclasses
+import os
+import subprocess
+from contextlib import ExitStack, contextmanager
+from pathlib import Path
 from typing import Callable, Generator
 
-from ...log import die
+from ...bottle_state import (
+    egress_state_dir,
+    git_gate_state_dir,
+    read_committed_image,
+)
+from ...egress import (
+    egress_agent_env_entries,
+    egress_resolve_token_values,
+)
+from ...git_gate import (
+    provision_git_gate_dynamic_keys,
+    revoke_git_gate_provisioned_keys,
+)
+from ...git_http_backend import DEFAULT_PORT as _GIT_HTTP_PORT
+from ...log import die, info, warn
+from ...supervise import SUPERVISE_PORT
+from ..docker.egress import EGRESS_PORT
+from ..util import AGENT_CA_BUNDLE, AGENT_CA_PATH
+from . import util as container_mod
 from .bottle import MacosContainerBottle
 from .bottle_plan import MacosContainerBottlePlan
+from .consolidated_launch import (
+    GatewayEndpoint,
+    ensure_gateway,
+    register_agent,
+    teardown_consolidated,
+)
+
+_REPO_DIR = str(Path(__file__).resolve().parent.parent.parent.parent)
+_AGENT_SLEEP_SECONDS = "2147483647"
 
 
 @contextmanager
@@ -27,13 +77,274 @@ def launch(
     *,
     provision: Callable[[MacosContainerBottlePlan, "MacosContainerBottle"], str | None],
 ) -> Generator[MacosContainerBottle, None, None]:
-    """Fail closed: the macOS backend is disabled until it grows the
-    consolidated per-host gateway (the companion-container path it used
-    was removed in #385)."""
-    del plan, provision
-    die(
-        "the macos-container backend is temporarily disabled during the "
-        "companion-container removal (#385); it will return once it uses "
-        "the consolidated gateway. Use --backend=docker for now."
+    """Build, run, register, provision, and yield an Apple Container bottle on
+    the shared per-host gateway."""
+    stack = ExitStack()
+    bottle_for_revoke = plan.manifest.bottle
+    git_gate_dir_for_revoke = git_gate_state_dir(plan.slug)
+
+    def teardown() -> None:
+        teardown_exc: BaseException | None = None
+        try:
+            stack.close()
+        except BaseException as exc:  # noqa: W0718 - teardown must continue
+            teardown_exc = exc
+            warn(f"macos-container teardown failed: {exc!r}")
+        revoke_git_gate_provisioned_keys(bottle_for_revoke, git_gate_dir_for_revoke)
+        if teardown_exc is not None:
+            raise teardown_exc
+
+    try:
+        plan = _build_images(plan)
+
+        # Step 1: the per-host singletons. Must precede the agent run — its
+        # proxy env needs the gateway's address at `container run` time.
+        endpoint = ensure_gateway()
+
+        # Step 2: mint this bottle's deploy keys, then point it at the SHARED
+        # gateway's CA + git-http/supervise ports.
+        plan = _provision_git_gate_keys(plan)
+        plan = _install_gateway_ca(plan, endpoint)
+        plan = _stamp_agent_urls(plan, endpoint)
+
+        # Step 3: run the agent. It has no identity token yet — registration
+        # needs the address this run assigns.
+        container_mod.force_remove_container(plan.container_name)
+        _start_agent(plan, endpoint)
+        stack.callback(container_mod.force_remove_container, plan.container_name)
+
+        # Step 4: read the assigned address and register by it. This is the
+        # attribution key; `--cap-drop CAP_NET_RAW` at run is what makes it
+        # unforgeable. Poll: `container run --detach` can return before vmnet's
+        # DHCP has assigned the address.
+        source_ip = container_mod.wait_container_ipv4_on_network(
+            plan.container_name, endpoint.network,
+        )
+        if not source_ip:
+            die(
+                f"agent {plan.container_name} never got an address on "
+                f"{endpoint.network}"
+            )
+        effective_env = {**os.environ, **plan.agent_provision.provisioned_env}
+        token_values = egress_resolve_token_values(
+            plan.egress_plan.token_env_map, effective_env,
+        )
+        ctx = register_agent(
+            plan.egress_plan,
+            plan.git_gate_plan,
+            source_ip=source_ip,
+            endpoint=endpoint,
+            image_ref=plan.image,
+            tokens=token_values,
+        )
+        stack.callback(
+            teardown_consolidated, ctx.bottle_id,
+            orchestrator_url=ctx.orchestrator_url,
+        )
+        info(
+            f"agent {plan.container_name} registered "
+            f"(gateway {endpoint.gateway_ip}, ip {source_ip})"
+        )
+
+        # Stamp the token onto the plan so provision-time consumers can read it,
+        # not only the exec-time egress proxy. git-gate's gitconfig extraHeader
+        # and the supervise MCP --header both reach the gateway on NO_PROXY (they
+        # bypass the egress proxy that carries the token), so without this the
+        # gateway's /resolve fail-closes and every git fetch/push and supervise
+        # call from the bottle is denied. Registration already produced the
+        # token above, so — unlike the run-time env — the plan CAN carry it.
+        plan = dataclasses.replace(plan, identity_token=ctx.identity_token)
+
+        bottle = MacosContainerBottle(
+            plan.container_name,
+            teardown,
+            None,
+            agent_command=plan.agent_command,
+            agent_prompt_mode=plan.agent_prompt_mode,
+            agent_provider_template=plan.agent_provider_template,
+            terminal_title=(
+                f"{plan.spec.label} ({plan.spec.agent_name})"
+                if plan.spec.label else plan.spec.agent_name
+            ),
+            terminal_color=plan.spec.color,
+            agent_workdir=plan.workspace_plan.workdir,
+            exec_env=_identity_proxy_env(endpoint, ctx.identity_token),
+        )
+        bottle.prompt_path = provision(plan, bottle)
+
+        yield bottle
+    finally:
+        teardown()
+
+
+def _build_images(plan: MacosContainerBottlePlan) -> MacosContainerBottlePlan:
+    """Build the agent image. The gateway's own image is built by
+    `ensure_gateway` — it belongs to the shared singleton, not to a bottle."""
+    committed = read_committed_image(plan.slug)
+    if committed and container_mod.image_exists(committed):
+        info(f"using committed image {committed!r}")
+        return dataclasses.replace(
+            plan,
+            agent_provision=dataclasses.replace(
+                plan.agent_provision, image=committed,
+            ),
+        )
+    container_mod.build_image(
+        plan.image, _REPO_DIR, dockerfile=plan.dockerfile_path,
     )
-    yield  # unreachable — `die` raises; keeps this a generator/contextmanager
+    return plan
+
+
+def _provision_git_gate_keys(
+    plan: MacosContainerBottlePlan,
+) -> MacosContainerBottlePlan:
+    if not plan.git_gate_plan.upstreams:
+        return plan
+    git_gate_plan = provision_git_gate_dynamic_keys(
+        plan.manifest.bottle,
+        plan.git_gate_plan,
+        git_gate_state_dir(plan.slug),
+    )
+    return dataclasses.replace(plan, git_gate_plan=git_gate_plan)
+
+
+def _install_gateway_ca(
+    plan: MacosContainerBottlePlan, endpoint: GatewayEndpoint,
+) -> MacosContainerBottlePlan:
+    """Stage the SHARED gateway's CA for the provisioner to install, replacing
+    the per-bottle CA the companion container used to mint. Every bottle on
+    this host trusts this one CA."""
+    ca_dir = egress_state_dir(plan.slug) / "gateway-ca"
+    ca_dir.mkdir(parents=True, exist_ok=True)
+    ca_file = ca_dir / "gateway-ca.pem"
+    ca_file.write_text(endpoint.gateway_ca_pem)
+    egress_plan = dataclasses.replace(
+        plan.egress_plan,
+        mitmproxy_ca_host_path=ca_file,
+        mitmproxy_ca_cert_only_host_path=ca_file,
+    )
+    return dataclasses.replace(plan, egress_plan=egress_plan)
+
+
+def _stamp_agent_urls(
+    plan: MacosContainerBottlePlan, endpoint: GatewayEndpoint,
+) -> MacosContainerBottlePlan:
+    """Point the agent's git-gate insteadOf rewrites + supervise MCP at the
+    shared gateway's ports. Both bypass the egress proxy (NO_PROXY covers the
+    gateway address)."""
+    git_gate_url = (
+        f"http://{endpoint.gateway_ip}:{_GIT_HTTP_PORT}"
+        if plan.git_gate_plan.upstreams else ""
+    )
+    supervise_url = (
+        f"http://{endpoint.gateway_ip}:{SUPERVISE_PORT}/"
+        if plan.supervise_plan is not None else ""
+    )
+    return dataclasses.replace(
+        plan,
+        agent_git_gate_url=git_gate_url,
+        agent_supervise_url=supervise_url,
+    )
+
+
+def _proxy_url(gateway_ip: str, identity_token: str = "") -> str:
+    """The agent's egress proxy URL. The identity token rides as proxy
+    credentials — the gateway reads Proxy-Authorization, resolves the
+    (source_ip, token) pair against the control plane, and strips it before
+    upstream. Without a valid pair `/resolve` denies the request (#366)."""
+    cred = f"bottle:{identity_token}@" if identity_token else ""
+    return f"http://{cred}{gateway_ip}:{EGRESS_PORT}"
+
+
+def _no_proxy(gateway_ip: str) -> str:
+    # git-http + supervise live on the gateway and must NOT go through the
+    # egress proxy — the agent reaches them directly by its address.
+    return f"localhost,127.0.0.1,{gateway_ip}"
+
+
+def _identity_proxy_env(
+    endpoint: GatewayEndpoint, identity_token: str,
+) -> dict[str, str]:
+    """The token-bearing proxy env applied at `container exec`. It supersedes
+    the token-less run-time value (exec `--env` wins), which is the only way to
+    get the token in: it does not exist until after the container runs."""
+    if not identity_token:
+        return {}
+    url = _proxy_url(endpoint.gateway_ip, identity_token)
+    return {
+        "HTTPS_PROXY": url, "HTTP_PROXY": url,
+        "https_proxy": url, "http_proxy": url,
+    }
+
+
+def _start_agent(plan: MacosContainerBottlePlan, endpoint: GatewayEndpoint) -> None:
+    argv = _agent_run_argv(plan, endpoint)
+    env = {**os.environ, **plan.forwarded_env}
+    info(f"container run agent {plan.container_name}")
+    result = subprocess.run(
+        argv, capture_output=True, text=True, env=env, check=False,
+    )
+    if result.returncode != 0:
+        die(
+            f"container run for agent {plan.container_name} failed: "
+            f"{(result.stderr or '').strip() or '<no stderr>'}"
+        )
+
+
+def _agent_run_argv(
+    plan: MacosContainerBottlePlan, endpoint: GatewayEndpoint,
+) -> list[str]:
+    argv = [
+        "container", "run",
+        "--name", plan.container_name,
+        "--detach",
+        "--label", "bot-bottle.backend=macos-container",
+        "--network", endpoint.network,
+        # The attribution invariant: without NET_RAW the agent cannot open a
+        # raw socket, so it cannot source-IP-spoof its neighbours on the shared
+        # segment. NET_ADMIN is not granted by default, so its address and
+        # route are already fixed. See the module docstring.
+        "--cap-drop", "CAP_NET_RAW",
+    ]
+    for entry in _agent_env_entries(plan, endpoint):
+        argv += ["--env", entry]
+    # The init process is a no-op: every agent command arrives via
+    # `container exec`, which is also how the identity token gets in.
+    argv += [plan.image, "sleep", _AGENT_SLEEP_SECONDS]
+    return argv
+
+
+def _agent_env_entries(
+    plan: MacosContainerBottlePlan, endpoint: GatewayEndpoint,
+) -> tuple[str, ...]:
+    # Token-less at run time — the token does not exist yet (see
+    # `_identity_proxy_env`). Anything egressing before the exec-time override
+    # is denied by `/resolve`, which is the safe direction.
+    proxy_url = _proxy_url(endpoint.gateway_ip)
+    no_proxy = _no_proxy(endpoint.gateway_ip)
+    env = [
+        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}",
+    ]
+    if plan.agent_git_gate_url:
+        env.append(f"GIT_GATE_URL={plan.agent_git_gate_url}")
+    if plan.agent_supervise_url:
+        env.append(f"MCP_SUPERVISE_URL={plan.agent_supervise_url}")
+    for name, value in sorted(plan.agent_provision.guest_env.items()):
+        env.append(f"{name}={value}")
+    # Forwarded vars: bare name → inherits from the `container run` process env
+    # so the secret value never lands on argv.
+    for name in sorted(plan.forwarded_env.keys()):
+        env.append(name)
+    env.extend(egress_agent_env_entries(plan.egress_plan))
+    return tuple(env)
+
+
+__all__ = ["launch"]

bot_bottle/backend/macos_container/util.py

@@ -437,22 +437,137 @@ def inspect_container(name: str) -> dict[str, object]:
 
 
 def container_ipv4_on_network(name: str, network: str) -> str:
-    data = inspect_container(name)
-    status = data.get("status")
+    """The container's IPv4 address on `network`. Fatal if absent — callers
+    that can tolerate "not yet" want `try_container_ipv4_on_network`."""
+    ip = try_container_ipv4_on_network(name, network)
+    if not ip:
+        die(f"container {name} has no IPv4 address on {network}")
+    return ip
+
+
+def run_container_argv(argv: list[str]) -> subprocess.CompletedProcess[str]:
+    """Run a `container` command, returning the result for the caller to
+    interpret. Unlike the `die`-on-failure helpers above, this lets callers
+    that raise their own typed errors (the gateway / orchestrator lifecycle)
+    keep control of the failure path."""
+    return subprocess.run(argv, capture_output=True, text=True, check=False)
+
+
+def bind_mount_spec(source: str, target: str, *, readonly: bool = False) -> str:
+    """A `container run --mount` bind spec. One definition so the gateway and
+    orchestrator emit an identical string — a divergence here would silently
+    break one backend's mounts while the other kept working."""
+    spec = f"type=bind,source={source},target={target}"
+    if readonly:
+        spec += ",readonly"
+    return spec
+
+
+def _normalize_digest(value: str) -> str:
+    return value.split(":", 1)[1] if ":" in value else value
+
+
+def _inspect_first(argv: list[str]) -> dict[str, object]:
+    """Run an inspect command and return its first JSON object, or {} on any
+    failure (non-zero exit, malformed JSON, unexpected shape). {} is the shared
+    'don't know' signal all the non-fatal inspect readers below build on — a
+    caller comparing against it treats it as 'leave the working container
+    alone', never as a mismatch."""
+    result = run_container_argv(argv)
+    if result.returncode != 0:
+        return {}
+    try:
+        data = json.loads(result.stdout or "[]")
+    except json.JSONDecodeError:
+        return {}
+    if isinstance(data, list):
+        data = data[0] if data else {}
+    return data if isinstance(data, dict) else {}
+
+
+def _descriptor_digest(node: object) -> str:
+    """The normalized digest under a `{... "descriptor": {"digest": ...}}`
+    node, or "". Both the image and container inspect shapes nest the image's
+    identity this way, so the digest readers stay symmetric — a difference
+    between them is what would spuriously recreate a container."""
+    if not isinstance(node, dict):
+        return ""
+    descriptor = node.get("descriptor")
+    if isinstance(descriptor, dict) and descriptor.get("digest"):
+        return _normalize_digest(str(descriptor["digest"]))
+    return ""
+
+
+def image_digest(ref: str) -> str:
+    """The digest of image `ref`, or "" if it can't be read. Reads exactly the
+    field `container_image_digest` reads (`configuration.descriptor.digest`) so
+    the two are comparable; "" means 'don't know' → callers don't churn."""
+    data = _inspect_first([_CONTAINER, "image", "inspect", ref])
+    return _descriptor_digest(data.get("configuration"))
+
+
+def container_image_digest(name: str) -> str:
+    """The digest of the image container `name` was created from, or "" if it
+    can't be read. Compare with `image_digest(ref)` to tell whether a running
+    container predates an image rebuild."""
+    config = _inspect_first([_CONTAINER, "inspect", name]).get("configuration")
+    image = config.get("image") if isinstance(config, dict) else None
+    return _descriptor_digest(image)
+
+
+def container_env(name: str) -> dict[str, str]:
+    """The env container `name` was started with, or {} if unreadable. Lets a
+    caller tell whether a running container's baked-in configuration still
+    matches what it would pass today."""
+    config = _inspect_first([_CONTAINER, "inspect", name]).get("configuration")
+    init = config.get("initProcess") if isinstance(config, dict) else None
+    entries = init.get("environment") if isinstance(init, dict) else None
+    if not isinstance(entries, list):
+        return {}
+    env: dict[str, str] = {}
+    for entry in entries:
+        if isinstance(entry, str) and "=" in entry:
+            key, value = entry.split("=", 1)
+            env[key] = value
+    return env
+
+
+def try_container_ipv4_on_network(name: str, network: str) -> str:
+    """`container_ipv4_on_network` without the fatal exit: "" when the address
+    isn't readable yet. For pollers — a container is created before it has an
+    address, so "not yet" is an expected state there, not an error."""
+    status = _inspect_first([_CONTAINER, "inspect", name]).get("status")
     networks = status.get("networks") if isinstance(status, dict) else None
     if not isinstance(networks, list):
-        die(f"container inspect {name} did not include status.networks")
+        return ""
     for entry in networks:
-        if not isinstance(entry, dict):
-            continue
-        if entry.get("network") != network:
+        if not isinstance(entry, dict) or entry.get("network") != network:
             continue
         raw = entry.get("ipv4Address")
-        if not isinstance(raw, str) or not raw:
-            die(f"container {name} has no IPv4 address on {network}")
-        return raw.split("/", 1)[0]
-    die(f"container {name} is not attached to network {network}")
-    raise AssertionError("unreachable")
+        if isinstance(raw, str) and raw:
+            return raw.split("/", 1)[0]
+    return ""
+
+
+def wait_container_ipv4_on_network(
+    name: str, network: str, *, timeout: float = 15.0, poll: float = 0.25,
+) -> str:
+    """Poll for the container's DHCP-assigned address on `network`, returning
+    it once available or "" on timeout.
+
+    Apple Container has no `--ip`: `container run --detach` can return before
+    vmnet's DHCP has populated `status.networks[].ipv4Address`, so a bare read
+    right after start races the assignment. Callers that need the address (the
+    attribution key, the gateway's proxy target) poll through here instead of
+    the fatal `container_ipv4_on_network`."""
+    deadline = time.monotonic() + timeout
+    while True:
+        ip = try_container_ipv4_on_network(name, network)
+        if ip:
+            return ip
+        if time.monotonic() >= deadline:
+            return ""
+        time.sleep(poll)
 
 
 def image_id(ref: str) -> str:

bot_bottle/orchestrator/client.py

@@ -186,6 +186,13 @@ def discover_orchestrator_url(*, timeout: float = 2.0) -> str:
             f"http://{netpool.orch_slot().guest_ip}:{CONTROL_PLANE_PORT}")
     except Exception:  # noqa: BLE001 — backend optional / not firecracker
         pass
+    try:  # macOS: infra container control plane on its host-only address
+        from ..backend.macos_container.infra import probe_control_plane_url
+        url = probe_control_plane_url()
+        if url:
+            candidates.append(url)
+    except Exception:  # noqa: BLE001 — backend optional / not macOS
+        pass
     for url in candidates:
         if OrchestratorClient(url, timeout=timeout).health():
             return url

bot_bottle/orchestrator/lifecycle.py

@@ -41,7 +41,7 @@ ORCHESTRATOR_IMAGE = os.environ.get(
 ORCHESTRATOR_DOCKERFILE = "Dockerfile.orchestrator"
 # Baked onto the container as a label so `ensure_running` can tell whether the
 # running process is executing the *current* bind-mounted source — see
-# `_source_hash`.
+# `source_hash`.
 ORCHESTRATOR_SOURCE_HASH_LABEL = "bot-bottle-orchestrator-source-hash"
 
 # The repo root is bind-mounted into the control-plane container so
@@ -60,7 +60,7 @@ class OrchestratorStartError(RuntimeError):
     """The orchestrator container did not become healthy within the timeout."""
 
 
-def _source_hash(repo_root: Path) -> str:
+def source_hash(repo_root: Path) -> str:
     """Content hash of the orchestrator's bind-mounted Python source (the
     `bot_bottle` package the control-plane process imports). This only
     changes when the code that would actually run inside the container
@@ -134,17 +134,17 @@ class OrchestratorService:
         proc = run_docker(["docker", "ps", "--filter", f"name=^/{name}$", "--format", "{{.Names}}"])
         return name in proc.stdout.split()
 
-    def _run_orchestrator_container(self, source_hash: str) -> None:
+    def _run_orchestrator_container(self, current_hash: str) -> None:
         """Start the control-plane container (idempotent: clears a stale
         fixed-name container first). Register-only broker → no docker socket.
-        Labels the container with `source_hash` so a later `ensure_running`
-        can detect a real code change (see `_source_hash`)."""
+        Labels the container with `current_hash` so a later `ensure_running`
+        can detect a real code change (see `source_hash`)."""
         run_docker(["docker", "rm", "--force", self._orchestrator_name])
         proc = run_docker([
             "docker", "run", "--detach",
             "--name", self._orchestrator_name,
             "--label", self._orchestrator_label,
-            "--label", f"{ORCHESTRATOR_SOURCE_HASH_LABEL}={source_hash}",
+            "--label", f"{ORCHESTRATOR_SOURCE_HASH_LABEL}={current_hash}",
             "--network", self.network,
             # Host CLI reaches the control plane here; bound to loopback so it
             # is not exposed on the host's external interfaces.
@@ -224,7 +224,7 @@ class OrchestratorService:
         # launch (the prior behaviour) would drop every other active
         # bottle's in-memory egress tokens each time a new bottle starts,
         # since the orchestrator process holds them only in memory (#381).
-        current_hash = _source_hash(self._repo_root)
+        current_hash = source_hash(self._repo_root)
         if self.is_healthy() and self._orchestrator_source_current(current_hash):
             return self.url
 

bot_bottle/paths.py

@@ -40,4 +40,13 @@ def host_db_path() -> Path:
     return bot_bottle_root() / "db" / HOST_DB_FILENAME
 
 
-__all__ = ["HOST_DB_FILENAME", "bot_bottle_root", "host_db_path"]
+def host_db_dir() -> Path:
+    """The directory holding the shared host state DB, created if missing.
+    Backends bind-mount this into their gateway so the supervise daemon writes
+    to the one DB the orchestrator (and the operator over HTTP) reads."""
+    db_dir = host_db_path().parent
+    db_dir.mkdir(parents=True, exist_ok=True)
+    return db_dir
+
+
+__all__ = ["HOST_DB_FILENAME", "bot_bottle_root", "host_db_path", "host_db_dir"]

docs/research/apple-container-networking-spike.md

@@ -358,3 +358,111 @@ the Apple Container-specific constraints directly:
 
 Do not implement the backend as a direct clone of Docker Compose
 service aliases. That assumption failed in this run.
+
+## Addendum: consolidated-gateway findings (2026-07-17, PRD 0070)
+
+Re-tested on Apple Container 1.0.0 while porting the backend to the
+per-host consolidated gateway (#351). The two-network shape above still
+holds; these are the additional constraints that shaped the port, each
+verified against the live CLI on this host.
+
+### No static IP for a container
+
+`container run --network` accepts only
+`<name>[,mac=XX:XX:XX:XX:XX:XX][,mtu=VALUE]`. There is no `--ip`. The
+address comes from vmnet's DHCP and is knowable only after the container
+is running:
+
+```console
+$ container run --name a --network bb-net --detach alpine sleep 900
+$ container inspect a | jq -r '.[0].status.networks[0].ipv4Address'
+192.168.128.3/24
+```
+
+Consequence: the docker backend's "allocate a free IP -> pin it with
+`--ip` -> register -> launch" order cannot be reproduced. macOS inverts
+it to "launch -> read the assigned address -> register". The identity
+token therefore cannot be in the agent's run-time env (registration mints
+it after the container exists) and is delivered at `container exec` time.
+
+### Networks are fixed at run time
+
+There is no `container network connect`; `container network` exposes only
+`create`, `delete`, `list`, `inspect`, `prune`. A network cannot be
+attached to a running container, so a *persistent* shared gateway rules
+out per-bottle networks — they would force a gateway restart per launch.
+One shared host-only network, created up front, is the only shape that
+keeps the gateway a singleton.
+
+### No container DNS
+
+Containers cannot resolve each other by name; the host-only network's
+resolver refuses the query:
+
+```console
+$ container exec agent nslookup gw
+;; connection timed out; no servers could be reached
+$ container exec agent cat /etc/resolv.conf
+nameserver 192.168.128.1
+```
+
+Consequence: the gateway is handed the control plane's **IP**, not a
+container name as on docker. That forces the startup order
+orchestrator -> read its address -> gateway.
+
+### The host can reach the host-only network directly
+
+```console
+$ container inspect c | jq -r '.[0].status.networks[0].ipv4Address'
+192.168.128.2/24
+$ curl -s http://192.168.128.2:8099/i
+ok
+```
+
+So no `--publish` hop is needed: the host CLI and the gateway use the
+same control-plane URL. Docker needs `--publish 127.0.0.1:...` plus a
+separate internal URL for the same job.
+
+### CAP_NET_RAW is granted by default — and matters for attribution
+
+Apple grants NET_RAW but not NET_ADMIN. The agent therefore cannot change
+its own address or route:
+
+```console
+$ container exec agent ip addr add 192.168.128.99/24 dev eth0
+ip: RTNETLINK answers: Operation not permitted
+$ container exec agent ip route replace default via 192.168.128.2 dev eth0
+ip: RTNETLINK answers: Operation not permitted
+$ container exec agent grep CapEff /proc/self/status
+CapEff:	00000000a80425fb        # bit 13 (NET_RAW) set, bit 12 (NET_ADMIN) clear
+```
+
+But NET_RAW permits raw sockets, i.e. source-address forgery against
+neighbours on the shared segment — directly against PRD 0070's invariant
+("a packet's source address, as seen by the orchestrator, provably
+identifies the originating bottle"). `--cap-drop CAP_NET_RAW` closes it:
+
+```console
+$ container run --cap-drop CAP_NET_RAW ... alpine
+$ container exec nr grep CapEff /proc/self/status
+CapEff:	00000000a80405fb        # bit 13 cleared
+$ container exec nr ping -c1 192.168.128.2
+ping: permission denied (are you root?)
+```
+
+The agent is run with `--cap-drop CAP_NET_RAW` for this reason.
+
+### `container exec` inherits run-time env, and `--env` overrides it
+
+```console
+$ container run --name e --env FOO=from_run --detach alpine sleep 120
+$ container exec e sh -c 'echo $FOO'
+from_run
+$ container exec --env FOO=from_exec e sh -c 'echo $FOO'
+from_exec
+```
+
+This is what makes exec-time identity-token delivery work: the token-less
+proxy URL baked in at launch is superseded by the token-bearing one at
+exec. Bare `--env NAME` (inherit from the parent process) keeps the token
+value off argv.

tests/unit/test_backend_selection.py

@@ -247,7 +247,7 @@ class TestHasBackend(unittest.TestCase):
 class TestEnsureOrchestrator(unittest.TestCase):
     """The backend-agnostic orchestrator bring-up entry point. Docker starts
     the orchestrator + gateway containers; firecracker boots the infra VM;
-    backends without one (macos-container) die with a pointer."""
+    macos-container starts the infra container."""
 
     def test_docker_delegates_to_orchestrator_service(self):
         b = get_bottle_backend("docker")
@@ -272,11 +272,16 @@ class TestEnsureOrchestrator(unittest.TestCase):
             url = b.ensure_orchestrator()
         self.assertEqual(url, "http://10.243.255.1:8099")
 
-    def test_macos_default_dies(self):
-        from bot_bottle.log import Die
+    def test_macos_delegates_to_infra_container(self):
         b = get_bottle_backend("macos-container")
-        with self.assertRaises(Die):
-            b.ensure_orchestrator()
+        with patch(
+            "bot_bottle.backend.macos_container.infra.MacosInfraService"
+        ) as service_cls:
+            service_cls.return_value.ensure_running.return_value.control_plane_url = (
+                "http://192.168.128.2:8099"
+            )
+            url = b.ensure_orchestrator()
+        self.assertEqual(url, "http://192.168.128.2:8099")
 
 
 if __name__ == "__main__":

tests/unit/test_macos_consolidated_launch.py

@@ -0,0 +1,136 @@
+"""Unit: macOS consolidated launch — register-after-start (PRD 0070)."""
+
+from __future__ import annotations
+
+import unittest
+from pathlib import Path
+from unittest.mock import MagicMock, Mock, patch
+
+from bot_bottle.backend.macos_container.consolidated_launch import (
+    GatewayEndpoint,
+    ensure_gateway,
+    register_agent,
+    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.macos_container.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 _endpoint() -> GatewayEndpoint:
+    return GatewayEndpoint(
+        orchestrator_url="http://192.168.128.2:8099",
+        gateway_ip="192.168.128.3",
+        gateway_ca_pem="-----BEGIN CERTIFICATE-----\n",
+        network="bot-bottle-mac-gateway",
+    )
+
+
+def _client() -> Mock:
+    c = Mock()
+    c.register_bottle.return_value = RegisteredBottle("b1", "tok")
+    return c
+
+
+class TestEnsureGateway(unittest.TestCase):
+    def _run(self, service: MagicMock) -> GatewayEndpoint:
+        with patch(f"{_MOD}.MacosInfraService", return_value=service):
+            return ensure_gateway()
+
+    def _service(self) -> MagicMock:
+        from bot_bottle.backend.macos_container.infra import InfraEndpoint
+        service = MagicMock()
+        service.ensure_running.return_value = InfraEndpoint(
+            control_plane_url="http://192.168.128.2:8099",
+            gateway_ip="192.168.128.2",
+        )
+        service.network = "bot-bottle-mac-gateway"
+        service.ca_cert_pem.return_value = "PEM"
+        return service
+
+    def test_reports_gateway_endpoint(self) -> None:
+        endpoint = self._run(self._service())
+        self.assertEqual("http://192.168.128.2:8099", endpoint.orchestrator_url)
+        self.assertEqual("192.168.128.2", endpoint.gateway_ip)
+        self.assertEqual("PEM", endpoint.gateway_ca_pem)
+        self.assertEqual("bot-bottle-mac-gateway", endpoint.network)
+
+    def test_control_plane_and_gateway_share_one_address(self) -> None:
+        """One infra container hosts both, so the gateway IP and the
+        control-plane host are the same."""
+        endpoint = self._run(self._service())
+        self.assertEqual(
+            endpoint.gateway_ip,
+            endpoint.orchestrator_url.split("://")[1].split(":")[0],
+        )
+
+
+class TestRegisterAgent(unittest.TestCase):
+    def _run(
+        self, client: Mock, provision: Mock | None = None,
+        *, source_ip: str = "192.168.128.9",
+    ):
+        with patch(f"{_MOD}.OrchestratorClient", return_value=client), \
+                patch(f"{_MOD}.provision_git_gate", provision or Mock()):
+            return register_agent(
+                _egress_plan(), _git_plan(),
+                source_ip=source_ip, endpoint=_endpoint(), image_ref="img:1",
+            )
+
+    def test_registers_by_the_address_read_from_the_live_container(self) -> None:
+        """The attribution key is the DHCP-assigned address the caller read
+        back — there is no --ip to pin it up front."""
+        client = _client()
+        ctx = self._run(client, source_ip="192.168.128.9")
+        self.assertEqual("192.168.128.9", ctx.source_ip)
+        self.assertEqual("192.168.128.9", client.register_bottle.call_args.args[0])
+
+    def test_returns_identity_token_and_bottle_id(self) -> None:
+        ctx = self._run(_client())
+        self.assertEqual("b1", ctx.bottle_id)
+        self.assertEqual("tok", ctx.identity_token)
+
+    def test_provisions_git_gate_for_the_registered_bottle(self) -> None:
+        provision = Mock()
+        self._run(_client(), provision)
+        self.assertEqual("b1", provision.call_args.args[1])
+
+    def test_rolls_registration_back_when_provisioning_fails(self) -> None:
+        """A provisioning failure must not leave an orphan registration
+        holding the source IP."""
+        client = _client()
+        provision = Mock(side_effect=RuntimeError("boom"))
+        with self.assertRaises(RuntimeError):
+            self._run(client, provision)
+        client.teardown_bottle.assert_called_once_with("b1")
+
+
+class TestTeardown(unittest.TestCase):
+    def test_deregisters_and_deprovisions(self) -> None:
+        client = Mock()
+        deprovision = Mock()
+        with patch(f"{_MOD}.OrchestratorClient", return_value=client), \
+                patch(f"{_MOD}.deprovision_git_gate", deprovision):
+            teardown_consolidated("b1", orchestrator_url="http://o:8099")
+        client.teardown_bottle.assert_called_once_with("b1")
+        self.assertEqual("b1", deprovision.call_args.args[1])
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/unit/test_macos_container_cleanup.py

@@ -43,10 +43,31 @@ class TestMacosContainerCleanup(unittest.TestCase):
 
 
 class TestMacosContainerEnumerate(unittest.TestCase):
-    def test_enumerate_active_is_empty_while_disabled(self):
-        # The macOS backend is disabled during the companion-container removal cleanup
-        # (#385); it launches nothing, so there is nothing to enumerate.
-        self.assertEqual([], enum_mod.enumerate_active())
+    """The backend launches bottles again (PRD 0070), so enumeration is real
+    rather than the disabled-era stub. These must not shell out: `container`
+    does not exist on the Linux CI host."""
+
+    def _enumerate(self, stdout: str, returncode: int = 0):
+        completed = enum_mod.subprocess.CompletedProcess(
+            args=[], returncode=returncode, stdout=stdout, stderr="",
+        )
+        with patch.object(enum_mod.subprocess, "run", return_value=completed), \
+                patch.object(enum_mod, "read_metadata", return_value=None):
+            return enum_mod.enumerate_active()
+
+    def test_lists_agent_containers_by_slug(self):
+        agents = self._enumerate("bot-bottle-dev-abc\nunrelated\n")
+        self.assertEqual(["dev-abc"], [a.slug for a in agents])
+        self.assertEqual(["macos-container"], [a.backend_name for a in agents])
+
+    def test_excludes_the_infra_singleton(self):
+        """The infra container shares the bot-bottle- prefix but is
+        infrastructure — listing it would invent an agent per host."""
+        agents = self._enumerate("bot-bottle-mac-infra\nbot-bottle-dev-abc\n")
+        self.assertEqual(["dev-abc"], [a.slug for a in agents])
+
+    def test_empty_when_the_cli_fails(self):
+        self.assertEqual([], self._enumerate("", returncode=1))
 
 
 if __name__ == "__main__":

tests/unit/test_macos_container_launch_wiring.py

@@ -0,0 +1,210 @@
+"""Unit: macOS agent run wiring — the attribution invariant + token delivery
+(PRD 0070).
+
+Replaces the argv coverage from the per-bottle companion-container era
+(`test_macos_container_launch.py`, removed with that architecture in #385).
+"""
+
+from __future__ import annotations
+
+import tempfile
+import unittest
+from pathlib import Path
+from types import SimpleNamespace
+from typing import cast
+from unittest.mock import patch
+
+from bot_bottle.backend.macos_container.bottle import MacosContainerBottle
+from bot_bottle.backend.macos_container.bottle_plan import MacosContainerBottlePlan
+from bot_bottle.backend.macos_container.consolidated_launch import GatewayEndpoint
+from bot_bottle.backend.macos_container.launch import (
+    _agent_run_argv,
+    _identity_proxy_env,
+    _proxy_url,
+)
+from bot_bottle.manifest import ManifestIndex
+
+_BOTTLE = "bot_bottle.backend.macos_container.bottle"
+
+_MANIFEST = ManifestIndex.from_json_obj({
+    "bottles": {"dev": {}},
+    "agents": {"demo": {"skills": [], "prompt": "", "bottle": "dev"}},
+}).load_for_agent("demo")
+
+
+def _endpoint() -> GatewayEndpoint:
+    return GatewayEndpoint(
+        orchestrator_url="http://192.168.128.2:8099",
+        gateway_ip="192.168.128.3",
+        gateway_ca_pem="PEM",
+        network="bot-bottle-mac-gateway",
+    )
+
+
+def _plan(
+    stage_dir: Path,
+    *,
+    agent_git_gate_url: str = "",
+    agent_supervise_url: str = "",
+) -> MacosContainerBottlePlan:
+    routes_path = stage_dir / "routes.yaml"
+    routes_path.write_text("routes: []\n", encoding="utf-8")
+    ca_path = stage_dir / "gateway-ca.pem"
+    ca_path.write_text("ca\n", encoding="utf-8")
+    egress_plan = SimpleNamespace(
+        mitmproxy_ca_host_path=ca_path,
+        routes_path=routes_path,
+        routes=("route",),
+        token_env_map={"EGRESS_TOKEN_0": "HOST_TOKEN"},
+        canary="",
+        canary_env="",
+    )
+    return cast(MacosContainerBottlePlan, SimpleNamespace(
+        spec=SimpleNamespace(),
+        manifest=_MANIFEST,
+        stage_dir=stage_dir,
+        slug="dev-abc",
+        container_name="bot-bottle-dev-abc",
+        image="bot-bottle-agent:latest",
+        forwarded_env={"OAUTH_TOKEN": "host-value"},
+        egress_plan=egress_plan,
+        git_gate_plan=SimpleNamespace(upstreams=()),
+        supervise_plan=None,
+        agent_provision=SimpleNamespace(
+            guest_env={"LITERAL": "value"},
+            provisioned_env={},
+        ),
+        agent_git_gate_url=agent_git_gate_url,
+        agent_supervise_url=agent_supervise_url,
+    ))
+
+
+class TestAgentRunArgv(unittest.TestCase):
+    def setUp(self) -> None:
+        self._tmp = tempfile.TemporaryDirectory()
+        self.argv = _agent_run_argv(_plan(Path(self._tmp.name)), _endpoint())
+
+    def tearDown(self) -> None:
+        self._tmp.cleanup()
+
+    def test_drops_net_raw(self) -> None:
+        """The attribution invariant: Apple grants CAP_NET_RAW by default,
+        which would let an agent forge a neighbour's source address with a raw
+        socket and be attributed as that bottle."""
+        self.assertIn("--cap-drop", self.argv)
+        self.assertEqual("CAP_NET_RAW", self.argv[self.argv.index("--cap-drop") + 1])
+
+    def test_attaches_to_the_shared_gateway_network(self) -> None:
+        self.assertEqual(
+            "bot-bottle-mac-gateway", self.argv[self.argv.index("--network") + 1],
+        )
+
+    def test_never_pins_an_ip(self) -> None:
+        """Apple Container 1.0.0 has no --ip: the address is DHCP-assigned and
+        read back after start."""
+        self.assertNotIn("--ip", self.argv)
+
+    def test_run_time_proxy_carries_no_identity_token(self) -> None:
+        """The token is minted by registration, which happens after this run —
+        so it cannot be here. `/resolve` denies the token-less pair (#366),
+        which is the safe direction; the real value arrives at exec time."""
+        joined = " ".join(self.argv)
+        self.assertIn(f"HTTP_PROXY={_proxy_url('192.168.128.3')}", joined)
+        self.assertNotIn("bottle:", joined)
+
+    def test_gateway_bypasses_the_proxy(self) -> None:
+        """git-http + supervise live on the gateway and must be reached
+        directly, not through its own egress proxy."""
+        entry = next(a for a in self.argv if a.startswith("NO_PROXY="))
+        self.assertIn("192.168.128.3", entry)
+
+    def test_forwarded_secrets_stay_off_argv(self) -> None:
+        """Bare name → inherited from the run process env, so the value never
+        lands on the command line."""
+        self.assertIn("OAUTH_TOKEN", self.argv)
+        self.assertNotIn("host-value", " ".join(self.argv))
+
+    def test_agent_init_is_a_no_op(self) -> None:
+        """Every agent command arrives via `container exec`; the init process
+        just holds the container open."""
+        self.assertEqual("sleep", self.argv[-2])
+
+
+class TestIdentityTokenDelivery(unittest.TestCase):
+    def test_exec_env_carries_the_token_as_proxy_credentials(self) -> None:
+        env = _identity_proxy_env(_endpoint(), "s3cret")
+        self.assertEqual(
+            "http://bottle:s3cret@192.168.128.3:9099", env["HTTP_PROXY"],
+        )
+        self.assertEqual(env["HTTP_PROXY"], env["https_proxy"])
+
+    def test_no_token_means_no_override(self) -> None:
+        self.assertEqual({}, _identity_proxy_env(_endpoint(), ""))
+
+    def test_token_value_never_reaches_argv(self) -> None:
+        """`ps` is world-readable: the token rides the child env behind a bare
+        `--env` name, never the command line."""
+        bottle = MacosContainerBottle(
+            "bot-bottle-demo", lambda: None, None,
+            exec_env=_identity_proxy_env(_endpoint(), "s3cret"),
+        )
+        argv = bottle.agent_argv(["--help"], tty=False)
+        self.assertNotIn("s3cret", " ".join(argv))
+        self.assertIn("HTTP_PROXY", argv)
+        self.assertEqual("--env", argv[argv.index("HTTP_PROXY") - 1])
+
+    def test_bottle_without_exec_env_is_unchanged(self) -> None:
+        bottle = MacosContainerBottle("bot-bottle-demo", lambda: None, None)
+        argv = bottle.agent_argv(["--help"], tty=False)
+        self.assertNotIn("--env", argv)
+
+
+class TestPlanIdentityToken(unittest.TestCase):
+    """git-gate's gitconfig extraHeader and the supervise MCP --header read
+    `getattr(plan, "identity_token", "")` at provision time and both bypass the
+    egress proxy (NO_PROXY), so the exec-time proxy token never reaches them —
+    the plan must carry the token or /resolve fail-closes and git + supervise
+    are denied on macOS."""
+
+    def test_macos_plan_has_the_identity_token_field(self) -> None:
+        from dataclasses import fields
+
+        from bot_bottle.backend.macos_container.bottle_plan import (
+            MacosContainerBottlePlan,
+        )
+        names = {f.name for f in fields(MacosContainerBottlePlan)}
+        self.assertIn("identity_token", names)
+
+    def test_matches_the_docker_plan(self) -> None:
+        """Both consolidated backends must expose identity_token so a shared
+        provision-time consumer degrades to neither backend silently."""
+        from dataclasses import fields
+
+        from bot_bottle.backend.docker.bottle_plan import DockerBottlePlan
+        from bot_bottle.backend.macos_container.bottle_plan import (
+            MacosContainerBottlePlan,
+        )
+        docker = {f.name for f in fields(DockerBottlePlan)}
+        macos = {f.name for f in fields(MacosContainerBottlePlan)}
+        self.assertIn("identity_token", docker & macos)
+
+    def test_provisioning_exec_also_carries_the_token(self) -> None:
+        """`provision` runs through `exec`; a provider whose provision step
+        fetches anything would otherwise egress token-less and be denied."""
+        bottle = MacosContainerBottle(
+            "bot-bottle-demo", lambda: None, None,
+            exec_env=_identity_proxy_env(_endpoint(), "s3cret"),
+        )
+        with patch(f"{_BOTTLE}.subprocess.run") as run:
+            run.return_value = SimpleNamespace(returncode=0, stdout="", stderr="")
+            bottle.exec("echo hi")
+        argv, kwargs = run.call_args.args[0], run.call_args.kwargs
+        self.assertIn("HTTP_PROXY", argv)
+        self.assertNotIn("s3cret", " ".join(argv))
+        self.assertEqual(
+            "http://bottle:s3cret@192.168.128.3:9099", kwargs["env"]["HTTP_PROXY"],
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/unit/test_macos_container_util.py

@@ -273,5 +273,55 @@ resolver #2
             )
 
 
+def _completed(stdout: str, returncode: int = 0):
+    return util.subprocess.CompletedProcess(args=[], returncode=returncode, stdout=stdout, stderr="")
+
+
+class TestInspectDigests(unittest.TestCase):
+    """image_digest and container_image_digest must read the SAME field shape
+    (a `descriptor.digest`) so a running container and its image are
+    comparable. An asymmetry recreates the gateway on every launch."""
+
+    _IMAGE = '{"configuration": {"descriptor": {"digest": "sha256:abc123"}}, "id": "abc123"}'
+    _CONTAINER = '[{"configuration": {"image": {"descriptor": {"digest": "sha256:abc123"}}}}]'
+
+    def test_image_and_container_digests_agree_for_the_same_image(self):
+        with patch.object(util, "run_container_argv") as run:
+            run.return_value = _completed(self._IMAGE)
+            img = util.image_digest("bot-bottle-gateway:latest")
+            run.return_value = _completed(self._CONTAINER)
+            ctr = util.container_image_digest("bot-bottle-mac-gateway")
+        self.assertEqual("abc123", img)
+        self.assertEqual(img, ctr)
+
+    def test_image_digest_never_falls_back_to_a_tag_or_id(self):
+        """The old `id` fallback could yield a value container_image_digest
+        can't produce, permanently mismatching. Missing descriptor → '' (don't
+        churn), never a stray id/tag."""
+        with patch.object(util, "run_container_argv") as run:
+            run.return_value = _completed('{"id": "bot-bottle-gateway:latest"}')
+            self.assertEqual("", util.image_digest("bot-bottle-gateway:latest"))
+
+    def test_unreadable_inspect_returns_empty(self):
+        with patch.object(util, "run_container_argv") as run:
+            run.return_value = _completed("", returncode=1)
+            self.assertEqual("", util.image_digest("x"))
+            self.assertEqual("", util.container_image_digest("x"))
+            self.assertEqual({}, util.container_env("x"))
+
+
+class TestWaitContainerIpv4(unittest.TestCase):
+    def test_returns_address_once_dhcp_assigns_it(self):
+        with patch.object(util, "try_container_ipv4_on_network", side_effect=["", "", "192.168.128.4"]), \
+                patch.object(util.time, "sleep"):
+            ip = util.wait_container_ipv4_on_network("c", "net", timeout=5, poll=0)
+        self.assertEqual("192.168.128.4", ip)
+
+    def test_returns_empty_on_timeout(self):
+        with patch.object(util, "try_container_ipv4_on_network", return_value=""), \
+                patch.object(util.time, "sleep"):
+            self.assertEqual("", util.wait_container_ipv4_on_network("c", "net", timeout=-1))
+
+
 if __name__ == "__main__":
     unittest.main()

tests/unit/test_macos_infra.py

@@ -0,0 +1,168 @@
+"""Unit: the single macOS infra container (control plane + gateway, PRD 0070)."""
+
+from __future__ import annotations
+
+import unittest
+from pathlib import Path
+from unittest.mock import Mock, patch
+
+from bot_bottle.backend.macos_container.infra import (
+    INFRA_DB_VOLUME,
+    MacosInfraService,
+    OrchestratorStartError,
+    probe_control_plane_url,
+)
+
+_INFRA = "bot_bottle.backend.macos_container.infra"
+
+
+def _ok(stdout: str = "") -> Mock:
+    return Mock(returncode=0, stdout=stdout, stderr="")
+
+
+def _fail(stderr: str = "boom") -> Mock:
+    return Mock(returncode=1, stdout="", stderr=stderr)
+
+
+class TestInfraRun(unittest.TestCase):
+    def _run_container(self, svc: MacosInfraService) -> list[str]:
+        run = Mock(return_value=_ok())
+
+        def _spec(src: str, tgt: str, readonly: bool = False) -> str:
+            return f"type=bind,source={src},target={tgt}" + (
+                ",readonly" if readonly else "")
+
+        with patch(f"{_INFRA}.container_mod") as mod, \
+                patch(f"{_INFRA}.ensure_networks"):
+            mod.dns_server.return_value = "1.1.1.1"
+            mod.bind_mount_spec.side_effect = _spec
+            mod.run_container_argv = run
+            svc._run_container("h1")
+        return run.call_args.args[0]
+
+    def test_single_container_runs_both_processes(self) -> None:
+        """The whole point: one container starts the control plane AND the
+        gateway daemons, so one kernel owns the DB."""
+        argv = self._run_container(MacosInfraService(repo_root=Path("/r")))
+        script = argv[-1]
+        self.assertIn("bot_bottle.orchestrator", script)
+        self.assertIn("gateway_init.py", script)
+        self.assertIn("127.0.0.1", script)  # they reach each other on loopback
+
+    def test_db_is_a_container_only_volume(self) -> None:
+        """No host bind-mount of the DB — a named volume only this container
+        mounts, so the DB is never written by two kernels."""
+        argv = self._run_container(MacosInfraService(repo_root=Path("/r")))
+        vols = [argv[i + 1] for i, a in enumerate(argv) if a == "--volume"]
+        self.assertTrue(any(v.startswith(f"{INFRA_DB_VOLUME}:") for v in vols))
+        # The repo source is bind-mounted read-only; the DB is not a bind mount.
+        mounts = [argv[i + 1] for i, a in enumerate(argv) if a == "--mount"]
+        self.assertTrue(all("bot-bottle.db" not in m for m in mounts))
+
+    def test_nat_network_precedes_the_host_only_network(self) -> None:
+        argv = self._run_container(MacosInfraService(repo_root=Path("/r")))
+        nets = [argv[i + 1] for i, a in enumerate(argv) if a == "--network"]
+        self.assertEqual(["bot-bottle-mac-egress", "bot-bottle-mac-gateway"], nets)
+
+    def test_source_hash_is_labelled_for_recreate(self) -> None:
+        argv = self._run_container(MacosInfraService(repo_root=Path("/r")))
+        self.assertIn("BOT_BOTTLE_SOURCE_HASH=h1", argv)
+
+    def test_start_failure_raises(self) -> None:
+        svc = MacosInfraService(repo_root=Path("/r"))
+        with patch(f"{_INFRA}.container_mod") as mod, \
+                patch(f"{_INFRA}.ensure_networks"):
+            mod.dns_server.return_value = "1.1.1.1"
+            mod.bind_mount_spec.return_value = "m"
+            mod.run_container_argv = Mock(return_value=_fail())
+            with self.assertRaises(OrchestratorStartError):
+                svc._run_container("h1")
+
+
+class TestInfraEnsureRunning(unittest.TestCase):
+    def test_current_healthy_container_is_left_alone(self) -> None:
+        """Idempotent singleton: N launches must not churn the infra container
+        and drop every live bottle's control plane."""
+        svc = MacosInfraService(repo_root=Path("/r"))
+        run = Mock()
+        with patch(f"{_INFRA}.container_mod") as mod, \
+                patch(f"{_INFRA}.source_hash", return_value="h1"), \
+                patch.object(svc, "_run_container", run), \
+                patch.object(svc, "is_healthy", return_value=True):
+            mod.container_is_running.return_value = True
+            mod.container_env.return_value = {"BOT_BOTTLE_SOURCE_HASH": "h1"}
+            mod.try_container_ipv4_on_network.return_value = "192.168.128.2"
+            endpoint = svc.ensure_running()
+        run.assert_not_called()
+        self.assertEqual("http://192.168.128.2:8099", endpoint.control_plane_url)
+        self.assertEqual("192.168.128.2", endpoint.gateway_ip)
+
+    def test_changed_source_recreates(self) -> None:
+        svc = MacosInfraService(repo_root=Path("/r"))
+        run = Mock()
+        with patch(f"{_INFRA}.container_mod") as mod, \
+                patch(f"{_INFRA}.source_hash", return_value="h2"), \
+                patch.object(svc, "ensure_built"), \
+                patch.object(svc, "_run_container", run), \
+                patch.object(svc, "is_healthy", return_value=True):
+            mod.container_is_running.return_value = True
+            mod.container_env.return_value = {"BOT_BOTTLE_SOURCE_HASH": "h1"}
+            mod.try_container_ipv4_on_network.return_value = "192.168.128.2"
+            svc.ensure_running()
+        run.assert_called_once()
+
+    def test_wedged_but_current_container_is_recreated(self) -> None:
+        """Current source but a dead HTTP server must be recreated, not polled
+        to death forever — health, not just the source label, gates reuse."""
+        svc = MacosInfraService(repo_root=Path("/r"))
+        run = Mock()
+        health = Mock(side_effect=[False, True])
+        with patch(f"{_INFRA}.container_mod") as mod, \
+                patch(f"{_INFRA}.source_hash", return_value="h1"), \
+                patch.object(svc, "ensure_built"), \
+                patch.object(svc, "_run_container", run), \
+                patch.object(svc, "is_healthy", health):
+            mod.container_is_running.return_value = True
+            mod.container_env.return_value = {"BOT_BOTTLE_SOURCE_HASH": "h1"}
+            mod.try_container_ipv4_on_network.return_value = "192.168.128.2"
+            svc.ensure_running()
+        run.assert_called_once()
+
+    def test_never_healthy_raises(self) -> None:
+        svc = MacosInfraService(repo_root=Path("/r"))
+        with patch(f"{_INFRA}.container_mod") as mod, \
+                patch(f"{_INFRA}.source_hash", return_value="h1"), \
+                patch.object(svc, "ensure_built"), \
+                patch.object(svc, "_run_container"), \
+                patch.object(svc, "is_healthy", return_value=False):
+            mod.container_is_running.return_value = False
+            mod.try_container_ipv4_on_network.return_value = "192.168.128.2"
+            with self.assertRaises(OrchestratorStartError):
+                svc.ensure_running(startup_timeout=0.01)
+
+
+class TestCaCertPem(unittest.TestCase):
+    def test_reads_ca_out_of_the_container(self) -> None:
+        svc = MacosInfraService(repo_root=Path("/r"))
+        with patch(f"{_INFRA}.container_mod") as mod:
+            mod.run_container_argv.return_value = _ok("-----BEGIN CERTIFICATE-----\n")
+            pem = svc.ca_cert_pem()
+        self.assertTrue(pem.startswith("-----BEGIN CERTIFICATE-----"))
+        argv = mod.run_container_argv.call_args.args[0]
+        self.assertEqual(["container", "exec", "bot-bottle-mac-infra", "cat"], argv[:4])
+
+
+class TestProbeControlPlane(unittest.TestCase):
+    def test_returns_url_when_running(self) -> None:
+        with patch(f"{_INFRA}.container_mod") as mod:
+            mod.try_container_ipv4_on_network.return_value = "192.168.128.2"
+            self.assertEqual("http://192.168.128.2:8099", probe_control_plane_url())
+
+    def test_empty_when_absent(self) -> None:
+        with patch(f"{_INFRA}.container_mod") as mod:
+            mod.try_container_ipv4_on_network.return_value = ""
+            self.assertEqual("", probe_control_plane_url())
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/unit/test_orchestrator_lifecycle.py

@@ -14,7 +14,7 @@ from bot_bottle.orchestrator.lifecycle import (
     ORCHESTRATOR_SOURCE_HASH_LABEL,
     OrchestratorService,
     OrchestratorStartError,
-    _source_hash,
+    source_hash,
 )
 from tests.unit import use_bottle_root
 
@@ -57,7 +57,7 @@ class TestOrchestratorService(unittest.TestCase):
         # A healthy control plane already running the *current* bind-mounted
         # source is left alone — recreating it on every launch would drop
         # every other active bottle's in-memory egress tokens (#381).
-        current = _source_hash(self.svc._repo_root)
+        current = source_hash(self.svc._repo_root)
         calls: list[list[str]] = []
 
         def fake(argv: list[str]) -> Mock:
@@ -98,7 +98,7 @@ class TestOrchestratorService(unittest.TestCase):
         self.assertEqual(1, len(runs))
         self.assertIn(ORCHESTRATOR_NAME, runs[0])
         # the fresh container is labeled with the current hash, not the stale one
-        current = _source_hash(self.svc._repo_root)
+        current = source_hash(self.svc._repo_root)
         self.assertIn(f"{ORCHESTRATOR_SOURCE_HASH_LABEL}={current}", runs[0])
 
     def test_ensure_running_starts_orchestrator_container_when_absent(self) -> None: