fix(backend): fix pyright errors in lazy-load implementation

- Rename _BACKENDS → _backends: pyright treats uppercase module-level
  names as constants and flags the reassignment in _get_backends() as
  reportConstantRedefinition; lowercase avoids this.
- Add TYPE_CHECKING guard importing CommitCancelled/Freezer/get_freezer
  from .freeze: pyright cannot see module-level __getattr__ bindings, so
  reportUnsupportedDunderAll fired for those three __all__ entries; the
  guard makes them visible to the type checker without running at import
  time.
- Update test_backend_selection.py to patch _backends (lowercase).

README.md

@@ -5,7 +5,7 @@
 # bot-bottle
 
 [![test](https://gitea.dideric.is/didericis/bot-bottle/actions/workflows/test.yml/badge.svg?branch=main)](https://gitea.dideric.is/didericis/bot-bottle/actions?workflow=test.yml)
-[![coverage](https://img.shields.io/badge/coverage-82%25-brightgreen)](https://coverage.readthedocs.io/)
+[![coverage](https://img.shields.io/badge/coverage-83%25-brightgreen)](https://coverage.readthedocs.io/)
 [![core coverage](https://img.shields.io/badge/core%20coverage-95%25-brightgreen)](https://gitea.dideric.is/didericis/bot-bottle/src/branch/main/docs/decisions/0004-coverage-policy.md)
 
 **Problem:** Developer wants to run a coding agent without supervision, but they don't want a prompt injected or misbehaving agent wrecking their environment or exfiltrating sensitive data.

bot_bottle/backend/__init__.py

@@ -40,7 +40,7 @@ from abc import ABC, abstractmethod
 from contextlib import AbstractContextManager
 from dataclasses import dataclass
 from pathlib import Path
-from typing import Any, Generic, Sequence, TypeVar
+from typing import TYPE_CHECKING, Any, Generic, Sequence, TypeVar
 
 from ..agent_provider import AgentProvisionPlan, get_provider, build_agent_provision_plan
 from ..egress import EgressPlan
@@ -54,6 +54,9 @@ from ..workspace import WorkspacePlan, workspace_plan
 from .print_util import print_multi, visible_agent_env_names
 from .util import host_skill_dir
 
+if TYPE_CHECKING:
+    from .freeze import CommitCancelled, Freezer, get_freezer
+
 
 @dataclass(frozen=True)
 class BottleSpec:
@@ -572,28 +575,63 @@ class BottleBackend(ABC, Generic[PlanT, CleanupT]):
         Not called by the launch path or the test suite."""
 
 
-# Import concrete backend classes AFTER the base types are defined, so
-# each backend module can pull BottleSpec / BottlePlan / BottleBackend
-# via `from . import ...` without hitting a partially-initialized module.
-from .docker import DockerBottleBackend  # noqa: E402  # pylint: disable=wrong-import-position
-from .firecracker import FirecrackerBottleBackend  # noqa: E402  # pylint: disable=wrong-import-position
-from .macos_container import MacosContainerBottleBackend  # noqa: E402  # pylint: disable=wrong-import-position
-
-# Freezer is imported after the backend classes for the same reason:
-# Freezer.commit_slug constructs ActiveAgent, which must be fully
-# defined first.
-from .freeze import CommitCancelled, Freezer, get_freezer  # noqa: E402  # pylint: disable=wrong-import-position
+# _backends is None until the first call to _get_backends(), at which
+# point all three concrete backend classes are imported and instantiated.
+# Keeping the imports out of module scope means that importing any
+# backend sub-module (e.g. `backend.docker.util`) no longer drags the
+# firecracker and macos-container implementations into memory.
+#
+# Tests may replace _backends with a {name: fake} dict via patch.object;
+# _get_backends() returns the current module-level value as-is when it
+# is not None, so test fakes take effect without triggering real imports.
+_backends: dict[str, BottleBackend[Any, Any]] | None = None
 
 
-# The dict is heterogeneous: each value is a BottleBackend specialized
-# over its own plan type. Concrete plan types are erased here because
-# the registry is selected at runtime and the CLI only needs the
-# unparameterized methods (prepare → plan → launch(plan), cleanup, etc.).
-_BACKENDS: dict[str, BottleBackend[Any, Any]] = {
-    "docker": DockerBottleBackend(),
-    "firecracker": FirecrackerBottleBackend(),
-    "macos-container": MacosContainerBottleBackend(),
-}
+def _get_backends() -> dict[str, BottleBackend[Any, Any]]:
+    """Return the registry of all backend instances, loading lazily on first call."""
+    global _backends  # pylint: disable=global-statement
+    if _backends is None:
+        from .docker import DockerBottleBackend
+        from .firecracker import FirecrackerBottleBackend
+        from .macos_container import MacosContainerBottleBackend
+        _backends = {
+            "docker": DockerBottleBackend(),
+            "firecracker": FirecrackerBottleBackend(),
+            "macos-container": MacosContainerBottleBackend(),
+        }
+    return _backends
+
+
+def __getattr__(name: str) -> Any:
+    """Lazily surface concrete backend classes and freeze symbols at the
+    package level so existing `from bot_bottle.backend import X` and
+    `patch.object(backend_mod, X, ...)` call-sites keep working without
+    forcing an import of every backend at module-init time."""
+    if name == "DockerBottleBackend":
+        from .docker import DockerBottleBackend
+        globals()[name] = DockerBottleBackend
+        return DockerBottleBackend
+    if name == "FirecrackerBottleBackend":
+        from .firecracker import FirecrackerBottleBackend
+        globals()[name] = FirecrackerBottleBackend
+        return FirecrackerBottleBackend
+    if name == "MacosContainerBottleBackend":
+        from .macos_container import MacosContainerBottleBackend
+        globals()[name] = MacosContainerBottleBackend
+        return MacosContainerBottleBackend
+    if name == "CommitCancelled":
+        from .freeze import CommitCancelled
+        globals()[name] = CommitCancelled
+        return CommitCancelled
+    if name == "Freezer":
+        from .freeze import Freezer
+        globals()[name] = Freezer
+        return Freezer
+    if name == "get_freezer":
+        from .freeze import get_freezer
+        globals()[name] = get_freezer
+        return get_freezer
+    raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
 
 
 def get_bottle_backend(
@@ -611,10 +649,11 @@ def get_bottle_backend(
     Dies with a pointer at the known backends if the chosen name
     isn't implemented."""
     resolved = name or os.environ.get("BOT_BOTTLE_BACKEND") or _default_backend_name()
-    if resolved not in _BACKENDS:
-        known = ", ".join(sorted(_BACKENDS))
+    backends = _get_backends()
+    if resolved not in backends:
+        known = ", ".join(sorted(backends))
         die(f"unknown backend {resolved!r}; known backends: {known}")
-    return _BACKENDS[resolved]
+    return backends[resolved]
 
 
 def _default_backend_name() -> str:
@@ -624,16 +663,17 @@ def _default_backend_name() -> str:
     # `firecracker` binary isn't installed yet: selecting it here routes
     # start through firecracker's preflight, which prints an install
     # pointer, instead of silently falling back to docker.
+    from .firecracker import FirecrackerBottleBackend
     if FirecrackerBottleBackend.is_host_capable():
         return "firecracker"
     return "docker"
 
 
 def known_backend_names() -> tuple[str, ...]:
-    """Sorted tuple of all backend keys in `_BACKENDS`. Used by
+    """Sorted tuple of all backend keys in `_get_backends()`. Used by
     argparse (`--backend` choices) and the dashboard's backend
     picker."""
-    return tuple(sorted(_BACKENDS))
+    return tuple(sorted(_get_backends()))
 
 
 def has_backend(name: str) -> bool:
@@ -645,9 +685,10 @@ def has_backend(name: str) -> bool:
 
     Returns False for unknown names so callers can pass
     arbitrary input without separate validation."""
-    if name not in _BACKENDS:
+    backends = _get_backends()
+    if name not in backends:
         return False
-    return _BACKENDS[name].is_available()
+    return backends[name].is_available()
 
 
 def enumerate_active_agents() -> list[ActiveAgent]:
@@ -663,10 +704,11 @@ def enumerate_active_agents() -> list[ActiveAgent]:
     deterministic tiebreaker. Agents with missing metadata
     (`started_at == ""`) sort first."""
     out: list[ActiveAgent] = []
-    for name in known_backend_names():
-        if not has_backend(name):
+    backends = _get_backends()
+    for name in sorted(backends):
+        if not backends[name].is_available():
             continue
-        out.extend(_BACKENDS[name].enumerate_active())
+        out.extend(backends[name].enumerate_active())
     out.sort(key=lambda a: (a.started_at, a.slug))
     return out
 

bot_bottle/backend/docker/backend.py

@@ -111,6 +111,11 @@ class DockerBottleBackend(BottleBackend["DockerBottlePlan", "DockerBottleCleanup
         plumbing needed; the alias resolves inside the bridge."""
         if plan.supervise_plan is None:
             return ""
+        # Consolidated: the supervise daemon lives on the shared gateway, so
+        # the agent registers the gateway address (NO_PROXY bypasses egress),
+        # not the per-bottle `supervise` alias.
+        if plan.agent_supervise_url:
+            return plan.agent_supervise_url
         return f"http://{SUPERVISE_HOSTNAME}:{SUPERVISE_PORT}/"
 
     def prepare_cleanup(self) -> DockerBottleCleanupPlan:

bot_bottle/backend/docker/bottle_plan.py

@@ -28,11 +28,30 @@ class DockerBottlePlan(BottlePlan):
     # accidental log of the plan dataclass.
     forwarded_env: dict[str, str] = field(repr=False)
     use_runsc: bool
+    # Consolidated mode (PRD 0070): the agent reaches git-gate over HTTP at the
+    # shared gateway's address (`http://<gateway_ip>:9420`), set at launch.
+    # Empty → single-tenant defaults (the `git-gate` alias over git://).
+    agent_git_gate_url: str = ""
+    # Likewise the supervise MCP endpoint at the gateway (`http://<gw>:9100/`);
+    # empty → the single-tenant `supervise` alias.
+    agent_supervise_url: str = ""
 
     @property
     def container_name(self) -> str:
         return self.agent_provision.instance_name
 
+    @property
+    def git_gate_insteadof_host(self) -> str:
+        if self.agent_git_gate_url.startswith("http://"):
+            return self.agent_git_gate_url.removeprefix("http://").rstrip("/")
+        return super().git_gate_insteadof_host
+
+    @property
+    def git_gate_insteadof_scheme(self) -> str:
+        if self.agent_git_gate_url.startswith("http://"):
+            return "http"
+        return super().git_gate_insteadof_scheme
+
     @property
     def image(self) -> str:
         return self.agent_provision.image

bot_bottle/backend/docker/consolidated_compose.py

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

bot_bottle/backend/docker/consolidated_launch.py

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

bot_bottle/backend/docker/gateway_provision.py

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

bot_bottle/backend/docker/launch.py

@@ -42,7 +42,6 @@ from ...git_gate import (
     revoke_git_gate_provisioned_keys,
 )
 from ...log import info, warn
-from . import network as network_mod
 from . import util as docker_mod
 from .bottle import DockerBottle
 from .bottle_plan import DockerBottlePlan
@@ -53,7 +52,6 @@ from ...bottle_state import (
     read_committed_image,
 )
 from .compose import (
-    bottle_plan_to_compose,
     compose_down,
     compose_dump_logs,
     compose_file_path,
@@ -62,7 +60,9 @@ from .compose import (
     compose_up,
     write_compose_file,
 )
-from .egress import egress_tls_init
+from .consolidated_compose import consolidated_agent_compose
+from .consolidated_launch import launch_consolidated, teardown_consolidated
+from ...orchestrator.gateway import DockerGateway
 
 
 # Where the repo root lives, for `docker build` context. Computed once.
@@ -97,8 +97,7 @@ def launch(
     try:
         # Step 1: agent image. Use a committed snapshot when one exists
         # and is present in the local daemon; otherwise build from the
-        # Dockerfile. Sidecar images get built lazily by `docker compose
-        # up` via the renderer's `build:` directives.
+        # Dockerfile. (The gateway image is built by the orchestrator.)
         committed = read_committed_image(plan.slug)
         if committed and docker_mod.image_exists(committed):
             info(f"using committed image {committed!r}")
@@ -112,82 +111,82 @@ def launch(
                 dockerfile=plan.dockerfile_path,
             )
 
-        internal_network = network_mod.network_name_for_slug(plan.slug)
-        egress_network = network_mod.network_egress_name_for_slug(plan.slug)
-
-        egress_ca_host, egress_ca_cert_only = egress_tls_init(
-            egress_state_dir(plan.slug),
-        )
-
+        # Step 2: mint the git-gate dynamic (gitea) deploy keys, if any, before
+        # provisioning the bottle's repos into the shared gateway.
         git_gate_plan = plan.git_gate_plan
         if git_gate_plan.upstreams:
             git_gate_plan = provision_git_gate_dynamic_keys(
-                plan.manifest.bottle,
-                git_gate_plan,
-                git_gate_state_dir(plan.slug),
-            )
-            git_gate_plan = dataclasses.replace(
-                git_gate_plan,
-                internal_network=internal_network,
-                egress_network=egress_network,
+                plan.manifest.bottle, git_gate_plan, git_gate_state_dir(plan.slug),
             )
+
+        # Step 3: register on the orchestrator + provision this bottle's
+        # git-gate state into the shared gateway; get the agent's attach
+        # context (pinned source IP, gateway address, shared network). The
+        # per-bottle egress auth tokens are resolved from the host env now and
+        # handed to the orchestrator (in memory) for the gateway to inject —
+        # the agent never sees them.
+        effective_env = {**os.environ, **plan.agent_provision.provisioned_env}
+        token_values = egress_resolve_token_values(
+            plan.egress_plan.token_env_map, effective_env,
+        )
+        ctx = launch_consolidated(
+            plan.egress_plan, git_gate_plan, image_ref=plan.image, tokens=token_values,
+        )
+        stack.callback(
+            teardown_consolidated, ctx.bottle_id, orchestrator_url=ctx.orchestrator_url,
+        )
+
+        # Step 4: install the SHARED gateway CA into the agent (replaces the
+        # per-bottle CA) — read it out of the running gateway.
+        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(DockerGateway(network=ctx.network).ca_cert_pem())
         egress_plan = dataclasses.replace(
             plan.egress_plan,
-            internal_network=internal_network,
-            egress_network=egress_network,
-            mitmproxy_ca_host_path=egress_ca_host,
-            mitmproxy_ca_cert_only_host_path=egress_ca_cert_only,
+            mitmproxy_ca_host_path=ca_file,
+            mitmproxy_ca_cert_only_host_path=ca_file,
+        )
+        # Point the agent's git-gate insteadOf rewrites at the shared gateway's
+        # HTTP git endpoint (9420) instead of the dead per-bottle `git-gate`
+        # alias. git-http + supervise on the gateway bypass the egress proxy
+        # (NO_PROXY includes the gateway address).
+        git_gate_url = (
+            f"http://{ctx.gateway_ip}:9420" if git_gate_plan.upstreams else ""
+        )
+        supervise_url = (
+            f"http://{ctx.gateway_ip}:9100/" if plan.supervise_plan is not None else ""
         )
-        supervise_plan = plan.supervise_plan
-        if supervise_plan is not None:
-            supervise_plan = dataclasses.replace(
-                supervise_plan,
-                internal_network=internal_network,
-            )
         plan = dataclasses.replace(
             plan,
             git_gate_plan=git_gate_plan,
             egress_plan=egress_plan,
-            supervise_plan=supervise_plan,
+            agent_git_gate_url=git_gate_url,
+            agent_supervise_url=supervise_url,
         )
 
-        # Step 6: render + write the compose file. metadata.json
-        # was written at prepare time and already carries
-        # compose_project; nothing to update here.
+        # Step 5: render + up the agent-only compose, pinned on the shared
+        # gateway network and proxied through the gateway's address.
         state_dir = bottle_state_dir(plan.slug)
-        spec = bottle_plan_to_compose(plan)
+        spec = consolidated_agent_compose(
+            plan, gateway_ip=ctx.gateway_ip, source_ip=ctx.source_ip, network=ctx.network,
+        )
         compose_file = write_compose_file(spec, compose_file_path(state_dir))
         project = compose_project_name(plan.slug)
-
-        # Step 7: compose up. Token values + the OAuth placeholder
-        # flow through subprocess env; the compose file holds only
-        # bare names for the secret-carrying entries.
-        effective_env = {**dict(os.environ), **plan.agent_provision.provisioned_env}
-        token_values = egress_resolve_token_values(
-            plan.egress_plan.token_env_map, effective_env,
-        )
-        compose_env: dict[str, str] = {
-            **os.environ,
-            **plan.forwarded_env,
-            **token_values,
-        }
+        # Forwarded vars (OAuth token, host interpolations) flow through the
+        # subprocess env as bare names so values never land in the file.
+        compose_env: dict[str, str] = {**os.environ, **plan.forwarded_env}
         info(
-            f"docker compose up -d  (project {project}, "
-            f"{len(spec['services'])} services)"
+            f"docker compose up -d  (project {project}, agent on shared "
+            f"gateway {ctx.gateway_ip}, ip {ctx.source_ip})"
         )
         compose_up(project, compose_file, env=compose_env)
-
-        # Register teardown in reverse order: log dump first, then
-        # `compose down`. Networks come down last via callbacks
-        # registered in step 2.
         stack.callback(compose_down, project, compose_file)
         stack.callback(
             compose_dump_logs, project, compose_file, compose_log_path(state_dir),
         )
 
-        # Step 8: provision. Create the bottle first so provisioners
-        # can use bottle.exec / bottle.cp_in; set the prompt path
-        # returned by provision_prompt after the fact.
+        # Step 6: provision (CA install now uses the gateway CA) + yield.
         bottle = DockerBottle(
             plan.container_name,
             teardown,
@@ -200,10 +199,6 @@ def launch(
             agent_workdir=plan.workspace_plan.workdir,
         )
         bottle.prompt_path = provision(plan, bottle)
-
-        # Step 9: yield. exec_agent continues to use `docker exec -it`
-        # — the agent runs `sleep infinity` per the renderer's
-        # service spec.
         yield bottle
     finally:
         teardown()

bot_bottle/backend/docker/util.py

@@ -7,8 +7,9 @@ from __future__ import annotations
 import re
 import shutil
 import subprocess
-from typing import Iterable, Iterator
+from typing import Iterator
 
+from ...docker_cmd import run_docker
 from ...log import die, info
 # from ...workspace import WorkspacePlan
 
@@ -30,12 +31,7 @@ def container_name_candidates(base: str) -> Iterator[str]:
 def runsc_available() -> bool:
     """Return True if the Docker daemon has the gVisor (`runsc`) runtime
     registered. Called once per prepare; the result lives on the plan."""
-    r = subprocess.run(
-        ["docker", "info", "--format", "{{json .Runtimes}}"],
-        capture_output=True,
-        text=True,
-        check=False,
-    )
+    r = run_docker(["docker", "info", "--format", "{{json .Runtimes}}"])
     return r.returncode == 0 and "runsc" in r.stdout
 
 
@@ -49,20 +45,15 @@ def require_docker() -> None:
 
 
 def image_exists(ref: str) -> bool:
-    return _silent_run(["docker", "image", "inspect", ref]) == 0
+    return run_docker(["docker", "image", "inspect", ref]).returncode == 0
 
 
 def container_exists(name: str) -> bool:
     """Returns True if a container (running or stopped) with the given
     name exists. Uses `docker ps -a -q -f name=^<name>$` so substring
     matches don't false-positive."""
-    result = subprocess.run(
-        ["docker", "ps", "-a", "-q", "-f", f"name=^{name}$"],
-        capture_output=True,
-        text=True,
-        check=True,
-    )
-    return bool(result.stdout.strip())
+    result = run_docker(["docker", "ps", "-a", "-q", "-f", f"name=^{name}$"])
+    return result.returncode == 0 and bool(result.stdout.strip())
 
 
 def force_remove_container(name: str) -> None:
@@ -70,12 +61,7 @@ def force_remove_container(name: str) -> None:
     doesn't — and the rm itself is best-effort (errors swallowed) so
     this is safe to register as a teardown callback."""
     if container_exists(name):
-        subprocess.run(
-            ["docker", "rm", "-f", name],
-            stdout=subprocess.DEVNULL,
-            stderr=subprocess.DEVNULL,
-            check=False,
-        )
+        run_docker(["docker", "rm", "-f", name])
 
 
 def docker_exec_root(container: str, argv: list[str]) -> None:
@@ -155,22 +141,10 @@ def build_image(ref: str, context: str, *, dockerfile: str = "") -> None:
 def commit_container(container_name: str, image_tag: str) -> None:
     """Run `docker commit <container_name> <image_tag>` to snapshot the
     running container's filesystem state as a local Docker image."""
-    result = subprocess.run(
-        ["docker", "commit", container_name, image_tag],
-        capture_output=True, text=True, check=False,
-    )
+    result = run_docker(["docker", "commit", container_name, image_tag])
     if result.returncode != 0:
         die(
             f"docker commit {container_name!r} → {image_tag!r} failed: "
             f"{(result.stderr or '').strip() or '<no stderr>'}"
         )
     info(f"committed {container_name!r} → {image_tag!r}")
-
-
-def _silent_run(cmd: Iterable[str]) -> int:
-    return subprocess.run(
-        list(cmd),
-        stdout=subprocess.DEVNULL,
-        stderr=subprocess.DEVNULL,
-        check=False,
-    ).returncode

bot_bottle/egress_addon.py

@@ -10,6 +10,7 @@ import json
 import os
 import signal
 import sys
+import typing
 from pathlib import Path
 
 from mitmproxy import http  # type: ignore[import-not-found]  # pylint: disable=import-error
@@ -230,21 +231,27 @@ class EgressAddon:
             + "\n"
         )
 
-    def _resolve_flow(self, flow: http.HTTPFlow) -> "tuple[Config, str]":
-        """The `(Config, supervise slug)` to apply to this request. Single-tenant
-        → the static `self.config` and the env slug. Consolidated → the calling
-        bottle's Config and its bottle id, resolved by source IP in one
-        round-trip (fail-closed to deny-all + empty slug if unattributed). The
-        identity token, if the agent injected one, is read then stripped so it
-        never leaks upstream. The slug keys both the proposal queue and the
-        per-bottle safelist, so an approval only ever affects its own bottle."""
+    def _resolve_flow(
+        self, flow: http.HTTPFlow,
+    ) -> "tuple[Config, str, typing.Mapping[str, str]]":
+        """The `(Config, supervise slug, env)` to apply to this request.
+        Single-tenant → the static `self.config`, the env slug, and the process
+        env. Consolidated → the calling bottle's Config + bottle id + auth
+        tokens, resolved by source IP in one round-trip (fail-closed to deny-all
+        + empty slug if unattributed); `env` is the process env overlaid with
+        the bottle's tokens, so upstream-auth injection (and DLP) use *this*
+        bottle's credentials — exactly what the per-bottle sidecar's env did.
+        The identity token, if the agent injected one, is read then stripped so
+        it never leaks upstream."""
         if self._resolver is None:
-            return self.config, self._supervise_slug
+            return self.config, self._supervise_slug, os.environ
         conn = flow.client_conn
         client_ip = conn.peername[0] if conn and conn.peername else ""
         token = flow.request.headers.get(IDENTITY_HEADER, "")
         flow.request.headers.pop(IDENTITY_HEADER, None)
-        return resolve_client_context(self._resolver, client_ip, token)
+        config, slug, tokens = resolve_client_context(self._resolver, client_ip, token)
+        env = {**os.environ, **tokens} if tokens else os.environ
+        return config, slug, env
 
     async def request(self, flow: http.HTTPFlow) -> None:
         request_path, _, query = flow.request.path.partition("?")
@@ -253,14 +260,14 @@ class EgressAddon:
             self._serve_introspection(flow, request_path)
             return
 
-        config, slug = self._resolve_flow(flow)
+        config, slug, env = self._resolve_flow(flow)
 
         # DLP outbound scan BEFORE stripping auth — catches tokens the
         # agent tried to smuggle in any header, path, query param, or body.
         # Hostname is included to catch DNS-tunnelling exfiltration attempts.
         route = match_route(config.routes, flow.request.pretty_host)
         if route is not None:
-            if not await self._handle_outbound_dlp(flow, route, slug):
+            if not await self._handle_outbound_dlp(flow, route, slug, env):
                 return
             # The redact policy may have rewritten the request line; recompute
             # the path/query the git checks below rely on.
@@ -299,7 +306,7 @@ class EgressAddon:
             config.routes,
             flow.request.pretty_host,
             request_path,
-            os.environ,
+            env,
             request_method=flow.request.method,
             request_headers=req_headers,
         )
@@ -325,10 +332,12 @@ class EgressAddon:
         flow: http.HTTPFlow,
         route: Route,
         slug: str,
+        env: "typing.Mapping[str, str]",
     ) -> bool:
         """Scan the outbound request and apply the route's on-match policy
-        (PRD 0062). Returns True if the request may be forwarded, False if a
-        403 response has been written to `flow`.
+        (PRD 0062). `env` is the per-bottle env overlay (process env + this
+        bottle's tokens) used for DLP detection. Returns True if the request may
+        be forwarded, False if a 403 response has been written to `flow`.
 
         Loops so the supervise policy can re-scan after each approval — a
         second, un-approved token in the same request is still caught."""
@@ -345,7 +354,7 @@ class EgressAddon:
                 flow.request.pretty_host, request_path, query, headers, "",
             )
             result = scan_outbound(
-                route, scan_text, os.environ,
+                route, scan_text, env,
                 safe_tokens=self._safe_tokens_for(slug), crlf_text=crlf_text,
             )
             if result is None or result.severity != "block":
@@ -356,7 +365,7 @@ class EgressAddon:
             # redact scrubs every detection (tokens and structural CRLF) and
             # forwards; it fails closed only if a match survives the scrub.
             if policy == ON_MATCH_REDACT:
-                if self._redact_outbound(flow, route):
+                if self._redact_outbound(flow, route, env):
                     if self.config.log >= LOG_BLOCKS:
                         sys.stderr.write(json.dumps({
                             "event": "egress_redacted",
@@ -384,29 +393,31 @@ class EgressAddon:
             if not self._supervise_available(slug):
                 self._block_dlp(flow, result)
                 return False
-            approved = await self._supervise_token_block(flow, request_path, result, slug)
+            approved = await self._supervise_token_block(flow, request_path, result, slug, env)
             if not approved:
                 return False  # _supervise_token_block wrote the 403 response
             # loop: the approved value is now in safe_tokens; re-scan.
 
-    def _redact_outbound(self, flow: http.HTTPFlow, route: Route) -> bool:
+    def _redact_outbound(
+        self, flow: http.HTTPFlow, route: Route, env: "typing.Mapping[str, str]",
+    ) -> bool:
         """Scrub detected tokens (and CRLF injection sequences) from the mutable
-        request surfaces (body, headers, path/query) and re-scan. Returns True
-        if the request is now clean; False if a block-severity match remains on
-        a surface redaction cannot rewrite (the hostname) so the caller fails
-        closed."""
+        request surfaces (body, headers, path/query) and re-scan. `env` is the
+        per-bottle env overlay. Returns True if the request is now clean; False
+        if a block-severity match remains on a surface redaction cannot rewrite
+        (the hostname) so the caller fails closed."""
         body = flow.request.get_text(strict=False)
         if body:
-            redacted_body = redact_tokens(body, env=os.environ)
+            redacted_body = redact_tokens(body, env=env)
             if redacted_body != body:
                 flow.request.text = redacted_body
         for name, value in list(flow.request.headers.items()):
             if name.lower() == "host":
                 continue  # routing-critical; never a legitimate token
-            redacted = strip_crlf(redact_tokens(value, env=os.environ))
+            redacted = strip_crlf(redact_tokens(value, env=env))
             if redacted != value:
                 flow.request.headers[name] = redacted
-        redacted_path = strip_crlf(redact_tokens(flow.request.path, env=os.environ))
+        redacted_path = strip_crlf(redact_tokens(flow.request.path, env=env))
         if redacted_path != flow.request.path:
             flow.request.path = redacted_path
 
@@ -419,7 +430,7 @@ class EgressAddon:
         crlf_text = build_outbound_scan_text(
             flow.request.pretty_host, request_path, query, headers, "",
         )
-        result = scan_outbound(route, scan_text, os.environ, crlf_text=crlf_text)
+        result = scan_outbound(route, scan_text, env, crlf_text=crlf_text)
         return result is None or result.severity != "block"
 
     async def _supervise_token_block(
@@ -428,20 +439,22 @@ class EgressAddon:
         request_path: str,
         result: ScanResult,
         slug: str,
+        env: "typing.Mapping[str, str]",
     ) -> bool:
         """Route a token DLP block to the operator's supervisor queue and wait.
 
         `slug` attributes the proposal to the calling bottle (its own queue +
         safelist) — in the shared gateway this is what keeps one bottle's
-        approval from unblocking another's request. Returns True if the operator
-        approved (the matched value is added to that bottle's safelist and the
-        caller re-scans); False if the request must be blocked (a 403 response
-        has been written to `flow`)."""
+        approval from unblocking another's request. `env` is the per-bottle env
+        overlay used to redact secrets from the proposal text. Returns True if
+        the operator approved (the matched value is added to that bottle's
+        safelist and the caller re-scans); False if the request must be blocked
+        (a 403 response has been written to `flow`)."""
         host = flow.request.pretty_host
         payload = build_token_allow_payload(
-            redact_tokens(host, env=os.environ),
+            redact_tokens(host, env=env),
             flow.request.method,
-            redact_tokens(request_path, env=os.environ),
+            redact_tokens(request_path, env=env),
             result,
         )
         proposal = _sv.Proposal.new(

bot_bottle/egress_addon_core.py

@@ -450,28 +450,31 @@ def resolve_client_config(
 
 class ContextResolverLike(typing.Protocol):
     """The bit of `policy_resolver.PolicyResolver` `resolve_client_context`
-    needs — one round-trip returning both policy and bottle id."""
+    needs — one round-trip returning policy, bottle id, and auth tokens."""
 
     def resolve_policy_and_bottle_id(
         self, source_ip: str, identity_token: str = ...,
-    ) -> "tuple[str | None, str | None]":
+    ) -> "tuple[str | None, str | None, dict[str, str]]":
         ...
 
 
 def resolve_client_context(
     resolver: ContextResolverLike, client_ip: str, identity_token: str = "",
-) -> "tuple[Config, str]":
-    """The calling client's `(Config, bottle_id)` in one round-trip —
+) -> "tuple[Config, str, dict[str, str]]":
+    """The calling client's `(Config, bottle_id, tokens)` in one round-trip —
     **fail-closed**. The Config follows `resolve_client_config`'s deny-all
     rules; the bottle id is `""` whenever unattributed or the orchestrator
-    errored, which the caller treats as "supervise unavailable for this
-    bottle" (never another bottle's queue). One `/resolve` keys both the
-    per-request egress policy and the per-bottle supervise queue + safelist."""
+    errored (caller treats as "supervise unavailable", never another bottle's
+    queue); `tokens` are the per-bottle upstream auth values the addon injects.
+    One `/resolve` keys the egress policy, the supervise queue + safelist, and
+    auth injection."""
     try:
-        policy, bottle_id = resolver.resolve_policy_and_bottle_id(client_ip, identity_token)
+        policy, bottle_id, tokens = resolver.resolve_policy_and_bottle_id(
+            client_ip, identity_token,
+        )
     except Exception:  # noqa: BLE001  # pylint: disable=broad-exception-caught
-        return Config(routes=()), ""  # orchestrator unreachable/errored → deny
-    return _config_from_policy(policy), (bottle_id or "")
+        return Config(routes=()), "", {}  # orchestrator unreachable/errored → deny
+    return _config_from_policy(policy), (bottle_id or ""), tokens
 
 
 # ---------------------------------------------------------------------------

bot_bottle/orchestrator/__main__.py

@@ -51,7 +51,13 @@ def main(argv: list[str] | None = None) -> int:
     # anything; 'docker' runs real containers (firecracker drops in later).
     secret = secrets.token_bytes(32)
     broker: LaunchBroker = DockerBroker(secret) if args.broker == "docker" else StubBroker(secret)
-    gateway: Gateway | None = DockerGateway() if args.gateway else None
+    # Standalone `--gateway` (not the consolidated flow, where the host
+    # lifecycle runs the gateway). The gateway resolves against this same
+    # process; the URL is only reachable when they share a docker network.
+    gateway: Gateway | None = (
+        DockerGateway(orchestrator_url=f"http://{args.host}:{args.port}")
+        if args.gateway else None
+    )
     orchestrator = Orchestrator(registry, broker, secret, gateway)
 
     # One persistent per-host gateway, shared by every bottle: build the

bot_bottle/orchestrator/client.py

@@ -90,14 +90,18 @@ class OrchestratorClient:
         image_ref: str = "",
         metadata: str = "",
         policy: str = "",
+        tokens: dict[str, str] | None = None,
     ) -> RegisteredBottle:
-        """Register a bottle and broker its launch (`POST /bottles`). Returns
-        its minted id + identity token."""
+        """Register a bottle and broker its launch (`POST /bottles`). `tokens`
+        are the per-bottle egress auth values (env_name -> value) the
+        orchestrator holds in memory for the gateway to inject. Returns the
+        minted id + identity token."""
         payload = self._ok("POST", "/bottles", {
             "source_ip": source_ip,
             "image_ref": image_ref,
             "metadata": metadata,
             "policy": policy,
+            "tokens": tokens or {},
         })
         bottle_id = payload.get("bottle_id")
         token = payload.get("identity_token")

bot_bottle/orchestrator/control_plane.py

@@ -34,6 +34,7 @@ import http.server
 import json
 import os
 import socketserver
+import sys
 import typing
 from urllib.parse import urlsplit
 
@@ -80,11 +81,16 @@ def dispatch(  # pylint: disable=too-many-return-statements,too-many-branches
         image_ref = data.get("image_ref")
         metadata = data.get("metadata")
         policy = data.get("policy")
+        raw_tokens = data.get("tokens")
+        tokens = {
+            k: v for k, v in raw_tokens.items() if isinstance(k, str) and isinstance(v, str)
+        } if isinstance(raw_tokens, dict) else {}
         rec = orch.launch_bottle(
             source_ip,
             image_ref=image_ref if isinstance(image_ref, str) else "",
             metadata=metadata if isinstance(metadata, str) else "",
             policy=policy if isinstance(policy, str) else "",
+            tokens=tokens,
         )
         return 201, {"bottle_id": rec.bottle_id, "identity_token": rec.identity_token}
 
@@ -136,7 +142,13 @@ def dispatch(  # pylint: disable=too-many-return-statements,too-many-branches
         rec = orch.resolve(source_ip, token if isinstance(token, str) else "")
         if rec is None:
             return 403, {"error": "unattributed"}
-        return 200, {"bottle_id": rec.bottle_id, "policy": rec.policy}
+        # tokens are the in-memory per-bottle egress auth values the gateway
+        # injects; served here, never persisted.
+        return 200, {
+            "bottle_id": rec.bottle_id,
+            "policy": rec.policy,
+            "tokens": orch.tokens_for(rec.bottle_id),
+        }
 
     return 404, {"error": "not found"}
 
@@ -151,12 +163,20 @@ class Handler(http.server.BaseHTTPRequestHandler):
             super().log_message(format, *args)
 
     def _serve(self, method: str) -> None:
-        """Read the request body, dispatch it, and write the JSON reply."""
+        """Read the request body, dispatch it, and write the JSON reply. A
+        dispatch failure (e.g. a broker error) returns a 500 rather than
+        crashing the connection, so one bad request can't take the control
+        plane down for the caller."""
         server = self.server
         assert isinstance(server, ControlPlaneServer)
         length = int(self.headers.get("Content-Length") or 0)
         body = self.rfile.read(length) if length > 0 else b""
-        status, payload = dispatch(server.orchestrator, method, self.path, body)
+        try:
+            status, payload = dispatch(server.orchestrator, method, self.path, body)
+        except Exception as e:  # noqa: BLE001 — the control plane must stay up
+            sys.stderr.write(f"orchestrator: {method} {self.path} failed: {e!r}\n")
+            sys.stderr.flush()
+            status, payload = 500, {"error": f"internal error: {e}"}
         data = json.dumps(payload).encode()
         self.send_response(status)
         self.send_header("Content-Type", "application/json")

bot_bottle/orchestrator/gateway.py

@@ -19,10 +19,16 @@ from __future__ import annotations
 
 import abc
 import os
+import time
 from pathlib import Path
 
 from ..docker_cmd import run_docker
 
+# The gateway's mitmproxy writes its CA a beat after the container starts, so
+# reads poll for it rather than assuming it's there on a fresh launch.
+_CA_POLL_SECONDS = 0.5
+DEFAULT_CA_TIMEOUT_SECONDS = 30.0
+
 GATEWAY_NAME = "bot-bottle-orch-gateway"
 GATEWAY_LABEL = "bot-bottle-orch-gateway=1"
 # The single user-defined network the gateway and every agent bottle share.
@@ -31,6 +37,14 @@ GATEWAY_LABEL = "bot-bottle-orch-gateway=1"
 # the source IP the gateway attributes by is the address on this network.
 GATEWAY_NETWORK = "bot-bottle-gateway"
 
+# mitmproxy's CA dir in the bundle. A persistent named volume here keeps the
+# gateway's self-generated CA STABLE across container recreation — every agent
+# installs this one CA to trust the shared gateway's TLS interception, so it
+# must not rotate when the gateway restarts.
+MITMPROXY_HOME = "/home/mitmproxy/.mitmproxy"
+GATEWAY_CA_VOLUME = "bot-bottle-gateway-mitmproxy"
+GATEWAY_CA_CERT = f"{MITMPROXY_HOME}/mitmproxy-ca-cert.pem"
+
 # The real sidecar-bundle image + its Dockerfile. Kept as a local constant
 # rather than imported from backend.docker.sidecar_bundle, which would drag
 # the whole backend layer into the lean orchestrator (see #359); unify when
@@ -83,12 +97,17 @@ class DockerGateway(Gateway):
         *,
         name: str = GATEWAY_NAME,
         network: str = GATEWAY_NETWORK,
+        orchestrator_url: str = "",
         build_context: Path | None = None,
         dockerfile: str | None = GATEWAY_DOCKERFILE,
     ) -> None:
         self.image_ref = image_ref
         self.name = name
         self.network = network
+        # The control-plane URL the gateway's data plane resolves per bottle
+        # against — reached by container name over docker DNS on the shared
+        # network (container↔container, no host firewall). Empty → single-tenant.
+        self._orchestrator_url = orchestrator_url
         self._build_context = build_context or _REPO_ROOT
         self._dockerfile = dockerfile
 
@@ -96,17 +115,23 @@ class DockerGateway(Gateway):
         return run_docker(["docker", "image", "inspect", self.image_ref]).returncode == 0
 
     def ensure_built(self) -> None:
-        """Build the bundle image from its Dockerfile when it's missing.
-        No-op when the image is already present, or when no dockerfile is
-        configured (e.g. a pre-pulled image)."""
-        if self._dockerfile is None or self.image_exists():
+        """Build the bundle image from its Dockerfile, **cache-aware** — cheap
+        (a cache check) when nothing changed, a real rebuild when the flat
+        sources (egress addon / git-http / policy_resolver / supervise) moved.
+
+        This deliberately builds every time rather than build-if-missing: the
+        per-bottle model kept the image fresh via compose's `build:` on up, and
+        a stale image silently runs the OLD single-tenant daemons. No-op only
+        when no dockerfile is configured (a pre-pulled image). BOT_BOTTLE_NO_CACHE
+        forces a full rebuild (parity with `start --no-cache`)."""
+        if self._dockerfile is None:
             return
-        proc = run_docker([
-            "docker", "build",
-            "-t", self.image_ref,
-            "-f", str(self._build_context / self._dockerfile),
-            str(self._build_context),
-        ])
+        argv = ["docker", "build", "-t", self.image_ref,
+                "-f", str(self._build_context / self._dockerfile),
+                str(self._build_context)]
+        if os.environ.get("BOT_BOTTLE_NO_CACHE"):
+            argv.insert(2, "--no-cache")
+        proc = run_docker(argv)
         if proc.returncode != 0:
             raise GatewayError(f"gateway image build failed: {proc.stderr.strip()}")
 
@@ -119,6 +144,18 @@ class DockerGateway(Gateway):
         ])
         return self.name in proc.stdout.split()
 
+    def _running_image_is_current(self) -> bool:
+        """True iff the running gateway was created from the *current*
+        `image_ref`. When `ensure_built` rebuilds the image (a source change),
+        the running container is still the OLD image running the OLD flat
+        daemons — so this is how a rebuild actually takes effect: a mismatch
+        means recreate."""
+        running = run_docker(["docker", "inspect", "--format", "{{.Image}}", self.name])
+        current = run_docker(["docker", "image", "inspect", "--format", "{{.Id}}", self.image_ref])
+        if running.returncode != 0 or current.returncode != 0:
+            return True  # can't compare → don't churn a working container
+        return running.stdout.strip() == current.stdout.strip()
+
     def _ensure_network(self) -> None:
         """Create the shared gateway network if it doesn't exist. Idempotent —
         a concurrent create loses harmlessly (the loser sees 'already exists').
@@ -132,22 +169,52 @@ class DockerGateway(Gateway):
             )
 
     def ensure_running(self) -> None:
-        if self.is_running():
+        # Recreate when the running container's image is stale (a rebuild),
+        # so source changes to the gateway's flat daemons take effect — not
+        # just when the container is absent.
+        if self.is_running() and self._running_image_is_current():
             return
         self._ensure_network()
-        # Clear any stale (stopped) container holding the fixed name, then
-        # start fresh. `rm --force` on an absent name is a tolerated no-op.
+        # Clear any stale (stopped OR outdated-image) container holding the
+        # fixed name, then start fresh. `rm --force` on an absent name is a
+        # tolerated no-op.
         run_docker(["docker", "rm", "--force", self.name])
-        proc = run_docker([
+        argv = [
             "docker", "run", "--detach",
             "--name", self.name,
             "--label", GATEWAY_LABEL,
             "--network", self.network,
-            self.image_ref,
-        ])
+            # Persist the self-generated CA so it survives restarts (agents
+            # trust it) — see GATEWAY_CA_VOLUME.
+            "--volume", f"{GATEWAY_CA_VOLUME}:{MITMPROXY_HOME}",
+        ]
+        if self._orchestrator_url:
+            # Makes the gateway's egress / git / supervise daemons multi-tenant:
+            # each request resolves source-IP -> policy against the control plane.
+            argv += ["--env", f"BOT_BOTTLE_ORCHESTRATOR_URL={self._orchestrator_url}"]
+        argv.append(self.image_ref)
+        proc = run_docker(argv)
         if proc.returncode != 0:
             raise GatewayError(f"gateway failed to start: {proc.stderr.strip()}")
 
+    def ca_cert_pem(self, *, timeout: float = DEFAULT_CA_TIMEOUT_SECONDS) -> str:
+        """The gateway's CA certificate (PEM) that agents install to trust its
+        TLS interception. mitmproxy generates it a moment after the container
+        starts, so this **polls** for it (up to `timeout`) rather than assuming
+        it's already there on a fresh gateway — raising only if it never
+        appears."""
+        deadline = time.monotonic() + timeout
+        while True:
+            proc = run_docker(["docker", "exec", self.name, "cat", GATEWAY_CA_CERT])
+            if proc.returncode == 0 and proc.stdout.strip():
+                return proc.stdout
+            if time.monotonic() >= deadline:
+                raise GatewayError(
+                    f"gateway CA cert not available after {timeout:g}s: "
+                    f"{proc.stderr.strip() or 'empty'}"
+                )
+            time.sleep(_CA_POLL_SECONDS)
+
     def stop(self) -> None:
         proc = run_docker(["docker", "rm", "--force", self.name])
         if proc.returncode != 0 and "No such container" not in proc.stderr:
@@ -157,4 +224,5 @@ class DockerGateway(Gateway):
 __all__ = [
     "Gateway", "DockerGateway", "GatewayError",
     "GATEWAY_NAME", "GATEWAY_LABEL", "GATEWAY_IMAGE", "GATEWAY_NETWORK",
+    "GATEWAY_CA_VOLUME", "GATEWAY_CA_CERT",
 ]

bot_bottle/orchestrator/lifecycle.py

@@ -1,91 +1,145 @@
-"""Orchestrator process lifecycle (PRD 0070, docker slice).
+"""Orchestrator + gateway lifecycle (PRD 0070, docker slice).
 
-Before the CLI can register or launch bottles against the consolidated
-model, exactly one orchestrator control plane — and the single per-host
-gateway it manages — must be running. This starts the orchestrator
-dev-harness (`python -m bot_bottle.orchestrator`) as a background host
-process and health-checks it.
+Runs the orchestrator control plane **as a container** on the shared gateway
+network, alongside the gateway container. This is the PRD's "virtualize the
+orchestrator": container↔container between the gateway and the orchestrator
+avoids the host firewall (which drops container→host traffic), and the gateway
+reaches the control plane by container name over docker DNS. The host CLI
+reaches it via a published loopback port.
 
-It is an **idempotent singleton**: `ensure_running` returns immediately if a
-healthy control plane already answers on the port, and otherwise spawns one
-and waits for it to come up. The control-plane port is the singleton key —
-a second orchestrator can't bind it, so a stray double-start fails fast
-rather than forking a rival.
-
-Host-process (not container) on purpose: the PRD sequences the orchestrator
-as a plain-process dev-harness first (fast iteration, and it already has the
-host user's docker access to broker launches), while the data-plane
-*gateway* it manages runs as a container. Wrapping the orchestrator itself
-in a backend-native unit is a later step.
+The orchestrator runs with the **register-only broker** — the *backend*
+launches agent containers (compose), so the orchestrator needs no docker
+socket. That keeps this control-plane container unprivileged; the host manages
+both containers. `ensure_running` is an idempotent singleton (fixed container
+names + the published port).
 """
 
 from __future__ import annotations
 
-import subprocess
-import sys
 import time
 import urllib.error
 import urllib.request
+from pathlib import Path
 
 from .. import log
+from ..docker_cmd import run_docker
 from ..paths import bot_bottle_root
+from .gateway import GATEWAY_IMAGE, GATEWAY_NETWORK, DockerGateway
+
+DEFAULT_PORT = 8099
+ORCHESTRATOR_NAME = "bot-bottle-orchestrator"
+ORCHESTRATOR_LABEL = "bot-bottle-orchestrator=1"
+
+# The repo root is bind-mounted into the control-plane container so
+# `python -m bot_bottle.orchestrator` resolves the package (the orchestrator
+# is stdlib-only, so the bundle image's python is enough).
+_REPO_ROOT = Path(__file__).resolve().parents[2]
+_APP_DIR = "/app"
+_ROOT_IN_CONTAINER = "/bot-bottle-root"
 
-DEFAULT_HOST = "127.0.0.1"
-DEFAULT_PORT = 8080
-# Poll cadence + default ceiling while waiting for a freshly-spawned control
-# plane to answer /health (the first start also builds/boots the gateway).
 _HEALTH_POLL_SECONDS = 0.25
 DEFAULT_STARTUP_TIMEOUT_SECONDS = 45.0
 _HEALTH_REQUEST_TIMEOUT_SECONDS = 1.0
 
 
 class OrchestratorStartError(RuntimeError):
-    """The orchestrator process did not become healthy within the timeout."""
+    """The orchestrator container did not become healthy within the timeout."""
 
 
-class OrchestratorProcess:
-    """Manages the local orchestrator control-plane process for the docker
-    backend. Backend-neutral callers only need `ensure_running()` + `url`."""
+class OrchestratorService:
+    """Manages the orchestrator control-plane container + the shared gateway.
+    Callers only need `ensure_running()` + `url`."""
 
     def __init__(
         self,
-        host: str = DEFAULT_HOST,
-        port: int = DEFAULT_PORT,
         *,
-        broker: str = "docker",
-        gateway: bool = True,
+        port: int = DEFAULT_PORT,
+        network: str = GATEWAY_NETWORK,
+        image: str = GATEWAY_IMAGE,
+        repo_root: Path = _REPO_ROOT,
+        host_root: Path | None = None,
     ) -> None:
-        self.host = host
         self.port = port
-        self._broker = broker
-        self._gateway = gateway
+        self.network = network
+        self.image = image
+        self._repo_root = repo_root
+        self._host_root = host_root or bot_bottle_root()
 
     @property
     def url(self) -> str:
-        """The control-plane base URL — also what the data plane's
-        BOT_BOTTLE_ORCHESTRATOR_URL points at."""
-        return f"http://{self.host}:{self.port}"
+        """Host-side control-plane URL (published loopback port)."""
+        return f"http://127.0.0.1:{self.port}"
+
+    @property
+    def internal_url(self) -> str:
+        """Control-plane URL as the gateway container reaches it — by name over
+        docker DNS on the shared network. This is the gateway's
+        BOT_BOTTLE_ORCHESTRATOR_URL."""
+        return f"http://{ORCHESTRATOR_NAME}:{self.port}"
 
     def is_healthy(self, *, timeout: float = _HEALTH_REQUEST_TIMEOUT_SECONDS) -> bool:
-        """True iff a control plane answers `GET /health` with 200 — the
-        singleton liveness check."""
         try:
             with urllib.request.urlopen(f"{self.url}/health", timeout=timeout) as resp:
                 return resp.status == 200
         except (urllib.error.URLError, TimeoutError, OSError):
             return False
 
+    def _container_running(self, name: str) -> bool:
+        proc = run_docker(["docker", "ps", "--filter", f"name=^/{name}$", "--format", "{{.Names}}"])
+        return name in proc.stdout.split()
+
+    def _run_orchestrator_container(self) -> None:
+        """Start the control-plane container (idempotent: clears a stale
+        fixed-name container first). Register-only broker → no docker socket."""
+        run_docker(["docker", "rm", "--force", ORCHESTRATOR_NAME])
+        proc = run_docker([
+            "docker", "run", "--detach",
+            "--name", ORCHESTRATOR_NAME,
+            "--label", ORCHESTRATOR_LABEL,
+            "--network", self.network,
+            # Host CLI reaches the control plane here; bound to loopback so it
+            # is not exposed on the host's external interfaces.
+            "--publish", f"127.0.0.1:{self.port}:{self.port}",
+            "--volume", f"{self._repo_root}:{_APP_DIR}:ro",
+            "--workdir", _APP_DIR,
+            # Persist the registry DB on the host (sole-owner: only the
+            # orchestrator opens bot-bottle.db).
+            "--volume", f"{self._host_root}:{_ROOT_IN_CONTAINER}",
+            "--env", f"BOT_BOTTLE_ROOT={_ROOT_IN_CONTAINER}",
+            "--entrypoint", "python3",
+            self.image,
+            "-m", "bot_bottle.orchestrator",
+            "--host", "0.0.0.0", "--port", str(self.port), "--broker", "stub",
+        ])
+        if proc.returncode != 0:
+            raise OrchestratorStartError(
+                f"orchestrator container failed to start: {proc.stderr.strip()}"
+            )
+
+    def _gateway(self) -> DockerGateway:
+        return DockerGateway(self.image, network=self.network, orchestrator_url=self.internal_url)
+
     def ensure_running(
         self, *, startup_timeout: float = DEFAULT_STARTUP_TIMEOUT_SECONDS,
     ) -> str:
-        """Return the control-plane URL, starting the orchestrator first if it
-        isn't already healthy. Idempotent — a healthy control plane is left
-        untouched. Raises `OrchestratorStartError` if a freshly-spawned one
-        doesn't answer within `startup_timeout`."""
-        if self.is_healthy():
-            return self.url
-        log.info("starting orchestrator", context={"url": self.url, "broker": self._broker})
-        self._spawn()
+        """Ensure the control plane + shared gateway are up; return the host
+        control-plane URL. Idempotent — a healthy control plane and a running
+        gateway are left untouched. Raises `OrchestratorStartError` on
+        timeout."""
+        gateway = self._gateway()
+        gateway.ensure_built()  # rebuild the bundle image on a source change
+        gateway.ensure_running()  # creates the shared network + (re)starts gateway
+
+        # Always (re)create the orchestrator container. It runs the repo's code
+        # bind-mounted, but the Python process loaded that code at startup and
+        # won't reload — so reusing a healthy-but-stale container would keep
+        # running OLD control-plane code (e.g. dropping the tokens field). Cheap
+        # (~seconds); the registry DB persists and the current launch
+        # re-registers its own in-memory state. (The dedicated orchestrator
+        # image follow-up replaces this with image-staleness detection.)
+        log.info("starting orchestrator container", context={"name": ORCHESTRATOR_NAME})
+        self._run_orchestrator_container()
+
         deadline = time.monotonic() + startup_timeout
         while time.monotonic() < deadline:
             if self.is_healthy():
@@ -93,49 +147,19 @@ class OrchestratorProcess:
                 return self.url
             time.sleep(_HEALTH_POLL_SECONDS)
         raise OrchestratorStartError(
-            f"orchestrator at {self.url} did not become healthy within "
-            f"{startup_timeout:g}s"
+            f"orchestrator at {self.url} did not become healthy within {startup_timeout:g}s"
         )
 
-    def _argv(self) -> list[str]:
-        """`python -m bot_bottle.orchestrator ...` — static flags only."""
-        argv = [
-            sys.executable, "-m", "bot_bottle.orchestrator",
-            "--host", self.host, "--port", str(self.port),
-            "--broker", self._broker,
-        ]
-        if self._gateway:
-            argv.append("--gateway")
-        return argv
-
-    def _log_path(self) -> str:
-        """Where the detached orchestrator's stdout/stderr goes so a failed
-        start is diagnosable after the CLI has moved on."""
-        return str(bot_bottle_root() / "orchestrator.log")
-
-    def _spawn(self) -> None:
-        """Launch the orchestrator detached so it outlives this CLI process,
-        with its output tee'd to a log file under the bot-bottle root."""
-        root = bot_bottle_root()
-        root.mkdir(parents=True, exist_ok=True)
-        logfile = open(self._log_path(), "a", encoding="utf-8")  # noqa: SIM115  # pylint: disable=consider-using-with
-        try:
-            subprocess.Popen(  # noqa: S603  # pylint: disable=consider-using-with
-                self._argv(),
-                stdout=logfile,
-                stderr=subprocess.STDOUT,
-                stdin=subprocess.DEVNULL,
-                start_new_session=True,
-            )
-        finally:
-            # The child inherits its own dup'd fd; this handle is ours to drop.
-            logfile.close()
+    def stop(self) -> None:
+        """Remove the orchestrator + gateway containers (idempotent)."""
+        run_docker(["docker", "rm", "--force", ORCHESTRATOR_NAME])
+        self._gateway().stop()
 
 
 __all__ = [
-    "OrchestratorProcess",
+    "OrchestratorService",
     "OrchestratorStartError",
-    "DEFAULT_HOST",
+    "ORCHESTRATOR_NAME",
     "DEFAULT_PORT",
     "DEFAULT_STARTUP_TIMEOUT_SECONDS",
 ]

bot_bottle/orchestrator/registry.py

@@ -149,7 +149,15 @@ class RegistryStore(DbStore):
         policy: str = "",
     ) -> BottleRecord:
         """Register (or replace) a bottle. Mints a bottle_id / identity token
-        when not supplied. Live — this is the control-plane reload path."""
+        when not supplied. Live — this is the control-plane reload path.
+
+        Supersedes any *other* active bottle at the same source IP first:
+        `by_source_ip` fail-closes on ambiguity (>1 active row → None), so a
+        leftover row at a reused IP — from an untorn-down bottle, crash
+        recovery, or a persisted DB — would otherwise *brick* the new bottle's
+        attribution. The new registration is authoritative for its IP; the
+        stale rows go. INSERT OR REPLACE handles a same-`bottle_id` reload in
+        place (its own row is exempt from the sweep)."""
         rec = BottleRecord(
             bottle_id=bottle_id or secrets.token_hex(8),
             source_ip=source_ip,
@@ -160,6 +168,11 @@ class RegistryStore(DbStore):
             policy=policy,
         )
         with self._connect() as conn:
+            conn.execute(
+                "DELETE FROM orchestrator_bottles "
+                "WHERE source_ip = ? AND state = 'active' AND bottle_id != ?",
+                (rec.source_ip, rec.bottle_id),
+            )
             conn.execute(
                 "INSERT OR REPLACE INTO orchestrator_bottles "
                 "(bottle_id, source_ip, identity_token, state, created_at, metadata, policy) "

bot_bottle/orchestrator/service.py

@@ -38,6 +38,12 @@ class Orchestrator:
         self._broker = broker
         self._secret = sign_secret
         self._gateway = gateway
+        # Per-bottle egress auth tokens (env_name -> value), keyed by bottle_id.
+        # Held **in memory only** — never written to the registry DB — so the
+        # gateway can inject each bottle's upstream credential without secrets
+        # at rest. Lost on restart (re-launch re-registers them); the future
+        # SecretProvider (#355) replaces this with per-request minting.
+        self._tokens: dict[str, dict[str, str]] = {}
 
     def launch_bottle(
         self,
@@ -47,11 +53,14 @@ class Orchestrator:
         slot: int | None = None,
         metadata: str = "",
         policy: str = "",
+        tokens: dict[str, str] | None = None,
     ) -> BottleRecord:
-        """Register a bottle (with its gateway policy) and broker its launch.
-        Rolls the registry entry back if the launch doesn't take, so a
-        failure leaves no orphan."""
+        """Register a bottle (with its gateway policy + in-memory egress auth
+        tokens) and broker its launch. Rolls the registry entry back if the
+        launch doesn't take, so a failure leaves no orphan."""
         rec = self.registry.register(source_ip, metadata=metadata, policy=policy)
+        if tokens:
+            self._tokens[rec.bottle_id] = dict(tokens)
         req = LaunchRequest(
             op="launch",
             bottle_id=rec.bottle_id,
@@ -66,6 +75,7 @@ class Orchestrator:
         finally:
             if not launched:
                 self.registry.deregister(rec.bottle_id)
+                self._tokens.pop(rec.bottle_id, None)
         return rec
 
     def teardown_bottle(self, bottle_id: str) -> bool:
@@ -76,8 +86,15 @@ class Orchestrator:
         req = LaunchRequest(op="teardown", bottle_id=bottle_id, source_ip=rec.source_ip)
         self._broker.submit(sign_request(req, self._secret))
         self.registry.deregister(bottle_id)
+        self._tokens.pop(bottle_id, None)
         return True
 
+    def tokens_for(self, bottle_id: str) -> dict[str, str]:
+        """The bottle's in-memory egress auth tokens (env_name -> value), or
+        empty. The gateway injects these per request; they are never
+        persisted."""
+        return dict(self._tokens.get(bottle_id, {}))
+
     def attribute(self, source_ip: str, identity_token: str) -> BottleRecord | None:
         """Fail-closed attribution (delegates to the registry)."""
         return self.registry.attribute(source_ip, identity_token)

bot_bottle/policy_resolver.py

@@ -102,21 +102,26 @@ class PolicyResolver:
 
     def resolve_policy_and_bottle_id(
         self, source_ip: str, identity_token: str = "",
-    ) -> tuple[str | None, str | None]:
-        """Both the policy blob and the bottle id in a single `/resolve` — so
-        a caller that needs each (the egress addon: policy for routing, bottle
-        id to key the calling bottle's supervise queue + safelist) makes one
-        round-trip, not two. Returns `(None, None)` when unattributed (a clean
-        `403`). Raises `PolicyResolveError` if the orchestrator can't be
-        reached, so the caller still fails closed."""
+    ) -> tuple[str | None, str | None, dict[str, str]]:
+        """The policy blob, bottle id, **and per-bottle egress auth tokens** in
+        a single `/resolve` — so the egress addon gets everything it needs
+        (policy for routing, bottle id for the supervise queue/safelist, tokens
+        to inject upstream auth) in one round-trip. Returns `(None, None, {})`
+        when unattributed (a clean `403`). Raises `PolicyResolveError` if the
+        orchestrator can't be reached, so the caller still fails closed."""
         payload = self._post_resolve(source_ip, identity_token)
         if payload is None:
-            return None, None
+            return None, None, {}
         policy = payload.get("policy")
         bottle_id = payload.get("bottle_id")
+        raw = payload.get("tokens")
+        tokens = {
+            k: v for k, v in raw.items() if isinstance(k, str) and isinstance(v, str)
+        } if isinstance(raw, dict) else {}
         return (
             policy if isinstance(policy, str) else "",
             bottle_id if isinstance(bottle_id, str) and bottle_id else None,
+            tokens,
         )
 
 

docs/research/agent-credential-proxy-landscape.md

@@ -15,6 +15,25 @@ the biggest credential risk).
 
 ## Summary
 
+> **Status — implemented (2026-07-14).** This note's recommendation
+> shipped, so the "today" / "in-flight" tenses below are now historical
+> (kept as the reasoning that led here). Current reality: bot-bottle no
+> longer injects raw provider tokens into the agent. The agent's environ
+> carries only a **placeholder** (e.g.
+> `CLAUDE_CODE_OAUTH_TOKEN=egress-placeholder`); the real token is held
+> in the egress sidecar and injected as the upstream `Authorization`
+> header. Rather than adding a *separate* in-container reverse proxy (as
+> proposed below), credential injection was folded into the **existing
+> pipelock/mitmproxy egress firewall**: an `EgressRoute` carries
+> `auth_scheme` + `token_ref`, the host token is forwarded into the
+> sidecar addon under an `EGRESS_TOKEN_N` slot, and the addon rewrites
+> `Authorization` on matching routes — **one MITM, not two** (this
+> resolves the Infisical-doubling concern noted later). The same
+> mechanism now covers Codex and Pi provider tokens and git-host PATs.
+> A `printenv` inside the bottle yields the placeholder, not the secret.
+> For current wiring see `bot_bottle/egress.py` and
+> `bot_bottle/contrib/*/agent_provider.py`.
+
 Today every bot-bottle agent gets `CLAUDE_CODE_OAUTH_TOKEN` (and
 any `bottle.env` secrets like a Gitea PAT) injected as env vars,
 which means the agent process can read them with `printenv` or
@@ -36,6 +55,24 @@ small bot-bottle-specific reverse proxy modeled on the
 phantom-token shape is probably the right call. For Gitea / GitHub /
 GitLab, the same proxy generalizes by config.
 
+**Updated 2026-07-14:** OneCLI ([onecli.sh](https://onecli.sh/)) —
+already listed below — has matured into a GA, YC-backed Rust
+credential gateway ("The Identity Gateway for AI Agents", ~2.5k⭐,
+300k+ downloads) that implements the **same phantom-token pattern**
+this note recommends: the agent holds a placeholder token, the
+gateway swaps it for the real (AES-256-GCM-encrypted) credential at
+request time. It's now the most mature open-source realization of the
+exact design proposed here — a production-ready alternative to
+alpha-stage nono — at the cost of being a broader product (built-in
+vault + management dashboard + hosted cloud tier + 50+ app
+integrations) rather than a minimal proxy. Its managed/cloud tier and
+per-agent dashboard also overlap bot-bottle's own planned paid control
+plane (bot-bottle-console, issue #327), so it's worth tracking as both
+a build-vs-adopt option *and* a product-level competitor. The
+build-first recommendation still stands (see synthesis below), but
+adopting OneCLI's OSS core is now a credible alternative where nono was
+too green.
+
 ## The shared problem
 
 Linux has no per-env-var ACL. Once a var is in a process's
@@ -77,7 +114,9 @@ The remaining credible designs reduce to three:
 
 ### Anthropic / Claude Code
 
-**Today's wiring** (`bot_bottle/cli/start.py`): the host's
+**Original wiring** (pre-gateway; **superseded 2026-07-14** — see the
+Status note in the Summary; the token now lives in the egress sidecar
+and the agent gets a placeholder): the host's
 `BOT_BOTTLE_CLAUDE_OAUTH_TOKEN` is forwarded into the bottle as
 `CLAUDE_CODE_OAUTH_TOKEN` via `docker run -e CLAUDE_CODE_OAUTH_TOKEN`
 (no `=value`, so the value never lands on argv — good). Inside the
@@ -223,7 +262,7 @@ Two categories:
 | **Infisical Agent Vault** | B | MIT (EE carve-out) | In-process HTTPS_PROXY forward proxy | TLS MITM, dummy-to-real swap | No — HTTPS_PROXY model | Service-level | Active; v0.19.0 May 2026, ~1k⭐ |
 | **nono** | B | Apache-2.0 | In-process reverse proxy | Phantom token, explicit URL routing | **Yes** — `BASE_URL=http://127.0.0.1:PORT/…` | Host + endpoint | Early alpha; v0.53.0 May 2026, 2.4k⭐ |
 | **Aegis** | B | Apache-2.0 | In-process reverse proxy | Path routing (`localhost:3100/{svc}/…`) | Configurable, undocumented for Anthropic | Method/path/rate/time | Very new, 10⭐ |
-| **OneCLI** | B | Apache-2.0 | Reverse proxy + management UI | Host/path matching, Bitwarden integration | Configurable | Per-agent scoping | Active; v1.23.0 May 2026, 2.1k⭐ |
+| **OneCLI** | B | Apache-2.0 | Reverse proxy + built-in vault + dashboard | **Phantom token** (placeholder→real swap at request time), AES-256-GCM vault | Yes (host match) | Per-agent + endpoint/method + rate limits | GA; Rust; YC-backed; ~2.5k⭐, 300k+ downloads (Jul 2026) |
 | **Aembit** | B | Proprietary | Sidecar + cloud control plane | TLS intercept, SPIFFE, JIT creds | No — intercepts by destination | Policy-based | GA (Apr 2026) |
 | **LiteLLM Proxy** | A | MIT | Reverse proxy | Virtual key → upstream key | Yes — set base URL to LiteLLM | Route-level | 45k⭐; **CVE-2026-42208 exploited Apr 2026**, patch v1.83.7 |
 | **Portkey Gateway** | A | MIT (OSS core) | Reverse proxy | Virtual key vault (cloud or Enterprise self-host) | Yes — documented for Claude Code | Config-based | Production; virtual-key vault needs Enterprise for self-host |
@@ -242,6 +281,18 @@ Two categories:
   `ANTHROPIC_BASE_URL`. **Blocker:** nono is explicitly
   "early alpha, not security audited."
 
+  **Update (2026-07-14):** OneCLI ([onecli.sh](https://onecli.sh/))
+  ships the same phantom-token shape but is GA and YC-backed (Rust,
+  ~2.5k⭐, 300k+ downloads) — the maturity nono lacks. Trade-offs: it's
+  a full identity-gateway *product* (encrypted vault, management
+  dashboard, 50+ one-click app integrations, hosted cloud tier), much
+  heavier than the ~100-line proxy proposed below, and its hosted tier
+  is a third-party credential custodian — precisely the trust
+  dependency bot-bottle's isolation model exists to avoid. So: its
+  Apache-2.0 OSS core is a credible *adopt* candidate for the
+  phantom-token slice; its managed offering is a *competitor*, not a
+  dependency to lean on.
+
 - **TLS-MITM forward proxies** (Infisical Agent Vault, Cloudflare
   Sandbox Auth, Aembit, the existing pipelock) all double up on
   the CA-trust machinery PRD 0006 already built for pipelock.
@@ -273,6 +324,15 @@ routing matches the design recommended here exactly; zero TLS
 work. But "not security audited" + "early alpha" means adopting it
 is a bet on the project rather than a buy-vs-build win.
 
+**Mature phantom-token option (added 2026-07-14):** OneCLI — same
+architecture as nono, but GA, Rust, and YC-backed. Its Apache-2.0 OSS
+core is now the strongest *adopt* candidate for the phantom-token slice
+if building is undesirable; the caveats are product surface you don't
+need (bundled vault + dashboard) and that its hosted tier is a
+competitor rather than a dependency. Doesn't change the build-first
+recommendation for the narrow Anthropic-token slice, but it does mean
+"is there a mature drop-in?" now has a real answer.
+
 **Most mature OSS purpose-built:** Infisical Agent Vault. MIT,
 v0.19.0 active, v0.17.0 added a containerized agent mode that
 maps directly to bot-bottle. Friction is the TLS-MITM topology
@@ -293,6 +353,12 @@ exposed.
 
 ## Recommended path forward
 
+> **Mostly implemented (2026-07-14).** Steps 1–3 shipped — the token
+> injection moved into the egress sidecar (folded into pipelock rather
+> than a standalone reverse proxy) and generalized to Codex/Pi/git-host
+> tokens. Steps 4–5 (issuance-side scope narrowing, per-route allowlists)
+> remain good hygiene. See the Status note in the Summary.
+
 In priority order:
 
 1. **In-container reverse proxy holding `CLAUDE_CODE_OAUTH_TOKEN`.**
@@ -376,6 +442,9 @@ already gives us for upstream push credentials.
 - [nono — phantom token blog](https://nono.sh/blog/blog-credential-injection)
 - [Aegis — GitHub](https://github.com/getaegis/aegis)
 - [OneCLI — GitHub](https://github.com/onecli/onecli)
+- [OneCLI — homepage](https://onecli.sh/)
+- [OneCLI — Y Combinator company page](https://www.ycombinator.com/companies/onecli)
+- [Show HN: OneCLI – Vault for AI Agents in Rust](https://news.ycombinator.com/item?id=47353558)
 - [Sandbox0 — GitHub](https://github.com/sandbox0-ai/sandbox0)
 - [Buildkite Cleanroom — GitHub](https://github.com/buildkite/cleanroom)
 - [Aembit IAM for Agentic AI — GA](https://aembit.io/blog/aembit-iam-for-agentic-ai-is-now-generally-available/)

docs/research/landscape-containerized-claude.md

@@ -71,6 +71,56 @@ manifest merge.
   network egress is logged by pipelock/mitmproxy, and per-run op-log/audit state
   is persisted to SQLite.
 
+- **OneCLI** ([onecli.sh](https://onecli.sh/)) — YC-backed, GA, open-source
+  (Apache-2.0, Rust) "identity gateway for AI agents": a credential/secret
+  broker that holds API keys and OAuth tokens out of the agent's reach and
+  injects them at the network layer (phantom-token — the agent sees a
+  placeholder, the gateway swaps in the real, AES-256-GCM-encrypted credential
+  at request time). Framework-agnostic and drop-in for any HTTP-calling agent,
+  50+ app integrations, plus a hosted cloud tier with a per-agent dashboard and
+  audit logs. Full technical breakdown in
+  [`agent-credential-proxy-landscape.md`](agent-credential-proxy-landscape.md).
+
+  **How close a competitor:** near-exact on the *single axis of agent secret
+  custody* — the exact thing bot-bottle sells as "the agent never sees real
+  credentials, even via `printenv`." OneCLI does that one job well, is mature
+  and funded, and is *more portable* (it sits in front of anything; bot-bottle
+  only helps agents launched through bot-bottle). Takeaway: bot-bottle should
+  stop treating secret custody as a *unique* differentiator. But OneCLI is
+  **not** a competitor to bot-bottle's actual product — it does no agent
+  sandboxing (containers/microVMs), no fleet/manifest layer, no named agents /
+  skills / per-agent system prompts, no multi-provider launching, no egress
+  firewall.
+
+  **Our edge:** (1) *Isolation is the product, not a proxy.* OneCLI keeps the
+  key out of reach at the network layer, but the agent itself still runs
+  unsandboxed — a hijacked agent behind OneCLI has full run of its host and can
+  exfil captured data through any allowed host. bot-bottle runs the agent inside
+  a kernel/VM-enforced sandbox, injects credentials across that same
+  out-of-process boundary, *and* clamps egress with pipelock — defense in depth
+  vs. a single network layer. (2) *Fleet + manifest model* with named agents,
+  skills, per-agent system prompts, multi-provider and multi-backend — OneCLI
+  has no equivalent. (3) *Trust posture:* OneCLI's managed tier reintroduces a
+  third-party credential custodian, whereas bot-bottle's OSS-runtime +
+  paid-control-plane split keeps custody inside the operator's own boundary —
+  the stronger story for the security-minded self-hoster. (4) *Runs inside your
+  network boundary — local/internal reach.* Because bot-bottle executes the
+  agent on your own host (homelab, corporate LAN, a Tailnet) and egress is a
+  manifest field, giving an agent *scoped* access to **internal** resources — a
+  private Gitea, a LAN database, a Tailscale node — is just another egress-route
+  line, not a networking project (the same move an operator already makes to
+  reach their Tailscale services). OneCLI's OSS core can self-host too, but it's
+  a credential *broker* for outbound API calls, not an agent runtime, and its
+  managed tier + 50+ integrations are oriented at public SaaS — it doesn't put
+  the agent behind your firewall for you. This is a reach advantage, distinct
+  from the isolation ones above, and it's a wedge cloud-first agent products
+  (Devin, Copilot Workspace, OneCLI Cloud) structurally can't match. **Tactical
+  read:**
+  adopt OneCLI's OSS core for the credential slice if building is undesirable
+  (it's mature now); don't build atop its managed tier (competitor, not
+  dependency); re-position bot-bottle on isolation + fleet + self-hosted custody
+  rather than "we hide your secrets."
+
 ## What no found project does
 
 None combine:

tests/integration/test_multitenant_isolation.py

@@ -0,0 +1,161 @@
+"""Integration: two bottles share one gateway; source-IP attribution keeps
+their egress tokens *and* allowlists isolated (PRD 0070 multi-tenancy).
+
+The whole premise of the consolidation is one shared gateway for every
+bottle, isolated only by the unspoofable source IP. A single-bottle test
+can't catch cross-bottle token bleed or an allowlist leak — this brings up
+the real orchestrator + gateway and drives two bottles through the one
+gateway to prove each gets exactly its own token and its own routes.
+
+Gated on a reachable Docker daemon and builds the bundle image, so it runs
+with the other docker integration tests, not in unit CI. Uses the real
+fixed gateway/orchestrator names (that's what it's testing), so it can't run
+concurrently with a live per-host gateway; it points the orchestrator at a
+throwaway BOT_BOTTLE_ROOT for a clean registry and tears everything down.
+"""
+
+from __future__ import annotations
+
+import os
+import subprocess
+import tempfile
+import unittest
+from pathlib import Path
+
+from bot_bottle.backend.docker.consolidated_launch import (
+    _network_cidr,
+    _network_container_ips,
+)
+from bot_bottle.backend.docker.egress import EGRESS_PORT
+from bot_bottle.backend.docker.gateway_net import next_free_ip
+from bot_bottle.orchestrator.client import OrchestratorClient
+from bot_bottle.orchestrator.gateway import GATEWAY_IMAGE, GATEWAY_NAME, GATEWAY_NETWORK
+from bot_bottle.orchestrator.lifecycle import OrchestratorService
+from tests._docker import skip_unless_docker
+
+# One upstream reachable under two names; reflects the Authorization header so
+# the probe can see exactly what the gateway injected (or didn't).
+_ECHO_NAME = "bot-bottle-mtitest-echo"
+_ECHO_ALIASES = ("echo-shared", "echo-bonly")
+_ECHO_SRC = (
+    "from http.server import BaseHTTPRequestHandler, HTTPServer\n"
+    "class H(BaseHTTPRequestHandler):\n"
+    " def do_GET(s):\n"
+    "  a=s.headers.get('Authorization','NONE'); s.send_response(200); s.end_headers()\n"
+    "  s.wfile.write(('AUTH='+a).encode())\n"
+    " def log_message(s,*a): pass\n"
+    "HTTPServer(('0.0.0.0',80),H).serve_forever()\n"
+)
+
+# A: echo-shared only, authed.  B: echo-shared authed + echo-bonly open.
+_POLICY_A = (
+    'routes:\n'
+    '  - host: echo-shared\n'
+    '    auth_scheme: "Bearer"\n'
+    '    token_env: "EGRESS_TOKEN_0"\n'
+)
+_POLICY_B = _POLICY_A + '  - host: echo-bonly\n'
+_TOKEN_A = "sk-AAA-alice"
+_TOKEN_B = "sk-BBB-bob"
+
+# Client probe: open http://<host>/ through the gateway proxy from a container
+# pinned to the bottle's source IP, printing "<status> <body>". Uses only the
+# bundle image's python3 — no dependency on the agent image.
+_PROBE_SRC = (
+    "import sys,urllib.request,urllib.error\n"
+    "op=urllib.request.build_opener(urllib.request.ProxyHandler({'http':sys.argv[1]}))\n"
+    "try:\n"
+    " r=op.open('http://'+sys.argv[2]+'/',timeout=8)\n"
+    " sys.stdout.write(str(r.status)+' '+r.read().decode())\n"
+    "except urllib.error.HTTPError as e:\n"
+    " sys.stdout.write(str(e.code)+' '+e.read().decode())\n"
+)
+
+
+@skip_unless_docker()
+@unittest.skipIf(
+    os.environ.get("GITEA_ACTIONS") == "true",
+    "skipped under act_runner: the orchestrator container bind-mounts the repo "
+    "path into a container on the socket-shared host daemon, which can't see the "
+    "runner's /workspace — same host-bind-mount constraint as the other "
+    "bottle-bringup integration tests",
+)
+class TestMultitenantIsolation(unittest.TestCase):
+    def setUp(self) -> None:
+        self._tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(self._tmp.cleanup)
+        # Throwaway root → a clean registry DB, independent of the host's.
+        self.svc = OrchestratorService(host_root=Path(self._tmp.name))
+        self.addCleanup(self._teardown_docker)
+        # ensure_running builds the bundle image (slow on a cold cache) and
+        # brings up the shared network + gateway + orchestrator.
+        url = self.svc.ensure_running(startup_timeout=300)
+        self.client = OrchestratorClient(url)
+        self.gw_ip = self._container_ip(GATEWAY_NAME)
+        self._run_echo()
+
+    def _teardown_docker(self) -> None:
+        subprocess.run(["docker", "rm", "--force", _ECHO_NAME],
+                       stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=False)
+        self.svc.stop()
+        subprocess.run(["docker", "network", "rm", GATEWAY_NETWORK],
+                       stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=False)
+        # The orchestrator container wrote the registry DB as root into the
+        # throwaway root; chown it back so the (non-root) tempdir cleanup can
+        # remove it.
+        subprocess.run(
+            ["docker", "run", "--rm", "-v", f"{self._tmp.name}:/r",
+             "--entrypoint", "chown", GATEWAY_IMAGE, "-R",
+             f"{os.getuid()}:{os.getgid()}", "/r"],
+            stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=False)
+
+    @staticmethod
+    def _container_ip(name: str) -> str:
+        proc = subprocess.run(
+            ["docker", "inspect", "-f",
+             '{{(index .NetworkSettings.Networks "' + GATEWAY_NETWORK + '").IPAddress}}', name],
+            stdout=subprocess.PIPE, text=True, check=True,
+        )
+        return proc.stdout.strip()
+
+    def _run_echo(self) -> None:
+        argv = ["docker", "run", "--detach", "--name", _ECHO_NAME,
+                "--network", GATEWAY_NETWORK]
+        for alias in _ECHO_ALIASES:
+            argv += ["--network-alias", alias]
+        argv += ["--entrypoint", "python3", GATEWAY_IMAGE, "-c", _ECHO_SRC]
+        proc = subprocess.run(argv, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
+                              text=True, check=False)
+        self.assertEqual(0, proc.returncode, f"echo upstream failed: {proc.stderr}")
+
+    def _free_ip(self, extra_taken: list[str]) -> str:
+        taken = _network_container_ips(GATEWAY_NETWORK) + extra_taken
+        return next_free_ip(_network_cidr(GATEWAY_NETWORK), taken)
+
+    def _probe(self, source_ip: str, host: str) -> str:
+        proc = subprocess.run(
+            ["docker", "run", "--rm", "--network", GATEWAY_NETWORK, "--ip", source_ip,
+             "--entrypoint", "python3", GATEWAY_IMAGE, "-c", _PROBE_SRC,
+             f"http://{self.gw_ip}:{EGRESS_PORT}", host],
+            stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, check=False, timeout=90,
+        )
+        return proc.stdout.strip()
+
+    def test_two_bottles_share_gateway_with_isolated_tokens_and_allowlists(self) -> None:
+        ip_a = self._free_ip([])
+        ip_b = self._free_ip([ip_a])
+        self.client.register_bottle(ip_a, policy=_POLICY_A, tokens={"EGRESS_TOKEN_0": _TOKEN_A})
+        self.client.register_bottle(ip_b, policy=_POLICY_B, tokens={"EGRESS_TOKEN_0": _TOKEN_B})
+
+        # Each bottle gets its OWN token injected on the shared route — no bleed.
+        self.assertEqual(f"200 AUTH=Bearer {_TOKEN_A}", self._probe(ip_a, "echo-shared"))
+        self.assertEqual(f"200 AUTH=Bearer {_TOKEN_B}", self._probe(ip_b, "echo-shared"))
+
+        # Allowlist is per-bottle: echo-bonly is only in B's policy.
+        self.assertTrue(self._probe(ip_a, "echo-bonly").startswith("403"),  # fail-closed for A
+                        "A reached a host outside its allowlist")
+        self.assertEqual("200 AUTH=NONE", self._probe(ip_b, "echo-bonly"))  # allowed, unauthed for B
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/unit/test_backend_selection.py

@@ -40,7 +40,7 @@ class TestGetBottleBackend(unittest.TestCase):
                 return True
 
         with patch.dict(os.environ, {}, clear=True), \
-             patch.object(backend_mod, "_BACKENDS", {
+             patch.object(backend_mod, "_backends", {
                  "macos-container": _FakeBackend(),
                  "docker": _FakeBackend(),
              }):
@@ -61,7 +61,7 @@ class TestGetBottleBackend(unittest.TestCase):
         with patch.dict(os.environ, {}, clear=True), \
              patch.object(backend_mod.FirecrackerBottleBackend,
                           "is_host_capable", classmethod(lambda cls: False)), \
-             patch.object(backend_mod, "_BACKENDS", {
+             patch.object(backend_mod, "_backends", {
                  "macos-container": _FakeBackend("macos-container", False),
                  "docker": _FakeBackend("docker", True),
              }):
@@ -83,7 +83,7 @@ class TestGetBottleBackend(unittest.TestCase):
         with patch.dict(os.environ, {}, clear=True), \
              patch.object(backend_mod.FirecrackerBottleBackend,
                           "is_host_capable", classmethod(lambda cls: True)), \
-             patch.object(backend_mod, "_BACKENDS", {
+             patch.object(backend_mod, "_backends", {
                  "macos-container": _FakeBackend("macos-container", False),
                  "firecracker": _FakeBackend("firecracker", False),
                  "docker": _FakeBackend("docker", True),
@@ -133,7 +133,7 @@ class TestEnumerateActiveAgents(unittest.TestCase):
                 return self._items
 
         with patch.object(
-            backend_mod, "_BACKENDS",
+            backend_mod, "_backends",
             {"docker": _FakeBackend([a]), "firecracker": _FakeBackend([b])},
         ):
             self.assertEqual([a, b], enumerate_active_agents())
@@ -167,7 +167,7 @@ class TestEnumerateActiveAgents(unittest.TestCase):
                 return self._items
 
         with patch.object(
-            backend_mod, "_BACKENDS",
+            backend_mod, "_backends",
             {
                 "docker": _FakeBackend([newer, tie_b]),
                 "firecracker": _FakeBackend([missing_metadata, tie_a]),
@@ -187,7 +187,7 @@ class TestEnumerateActiveAgents(unittest.TestCase):
                 return []
 
         with patch.object(
-            backend_mod, "_BACKENDS",
+            backend_mod, "_backends",
             {"docker": _FakeBackend(), "firecracker": _FakeBackend()},
         ):
             self.assertEqual([], enumerate_active_agents())
@@ -218,7 +218,7 @@ class TestEnumerateActiveAgents(unittest.TestCase):
                 return self._items
 
         with patch.object(
-            backend_mod, "_BACKENDS",
+            backend_mod, "_backends",
             {
                 "docker": _FakeBackend([present], available=True),
                 "firecracker": _FakeBackend([hidden], available=False),
@@ -234,7 +234,7 @@ class TestHasBackend(unittest.TestCase):
                 return False
 
         with patch.object(
-            backend_mod, "_BACKENDS", {"docker": _FakeBackend()},
+            backend_mod, "_backends", {"docker": _FakeBackend()},
         ):
             from bot_bottle.backend import has_backend
             self.assertFalse(has_backend("docker"))

tests/unit/test_consolidated_compose.py

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

tests/unit/test_consolidated_launch.py

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

tests/unit/test_docker_launch_committed_image.py

@@ -1,4 +1,4 @@
-"""Unit: Docker launch step uses committed image when available."""
+"""Unit: Docker launch step uses committed image when available (consolidated)."""
 
 from __future__ import annotations
 
@@ -6,6 +6,7 @@ import contextlib
 import io
 import tempfile
 import unittest
+from collections.abc import Callable, Generator
 from pathlib import Path
 from typing import Any
 from unittest import mock
@@ -14,9 +15,11 @@ from bot_bottle.agent_provider import AgentProvisionPlan
 from bot_bottle.backend import BottleSpec
 from bot_bottle.backend.docker import launch as launch_mod
 from bot_bottle.backend.docker.bottle_plan import DockerBottlePlan
+from bot_bottle.backend.docker.consolidated_launch import LaunchContext
 from bot_bottle.egress import EgressPlan
 from bot_bottle.git_gate import GitGatePlan
 from bot_bottle.manifest import ManifestIndex
+from tests.unit import use_bottle_root
 
 
 _SLUG = "dev-abc12"
@@ -28,163 +31,111 @@ _IDX = ManifestIndex.from_json_obj({
     "agents": {"demo": {"skills": [], "prompt": "", "bottle": "dev"}},
 })
 
+_CTX = LaunchContext(
+    bottle_id="b1", identity_token="t", source_ip="172.20.0.4",
+    network="bot-bottle-gateway", gateway_ip="172.20.0.2",
+    orchestrator_url="http://orch:8099",
+)
+
 
 def _plan(tmp: str) -> DockerBottlePlan:
     stage = Path(tmp)
     spec = BottleSpec(
-        manifest=_IDX,
-        agent_name="demo",
-        copy_cwd=False,
-        user_cwd=tmp,
-        identity=_SLUG,
+        manifest=_IDX, agent_name="demo", copy_cwd=False, user_cwd=tmp, identity=_SLUG,
     )
     return DockerBottlePlan(
         spec=spec,
         manifest=_IDX.load_for_agent("demo"),
         stage_dir=stage,
         git_gate_plan=GitGatePlan(
-            slug=_SLUG,
-            entrypoint_script=stage / "entrypoint.sh",
-            hook_script=stage / "hook.sh",
-            access_hook_script=stage / "access-hook.sh",
-            upstreams=(),
+            slug=_SLUG, entrypoint_script=stage / "e.sh", hook_script=stage / "h.sh",
+            access_hook_script=stage / "a.sh", upstreams=(),
         ),
         egress_plan=EgressPlan(
-            slug=_SLUG,
-            routes_path=stage / "egress.yaml",
-            routes=(),
-            token_env_map={},
+            slug=_SLUG, routes_path=stage / "egress.yaml", routes=(), token_env_map={},
         ),
         supervise_plan=None,
         agent_provision=AgentProvisionPlan(
-            template="claude",
-            command="claude",
-            prompt_mode="append_file",
-            image=_DEFAULT_IMAGE,
-            dockerfile="",
-            guest_home="/home/node",
-            instance_name=f"bot-bottle-{_SLUG}",
-            prompt_file=stage / "prompt.txt",
+            template="claude", command="claude", prompt_mode="append_file",
+            image=_DEFAULT_IMAGE, dockerfile="", guest_home="/home/node",
+            instance_name=f"bot-bottle-{_SLUG}", prompt_file=stage / "prompt.txt",
             guest_env={},
         ),
-        slug=_SLUG,
-        forwarded_env={},
-        use_runsc=False,
+        slug=_SLUG, forwarded_env={}, use_runsc=False,
     )
 
 
 class TestLaunchCommittedImage(unittest.TestCase):
     def setUp(self) -> None:
         self._tmp = tempfile.mkdtemp(prefix="launch-committed-test.")
+        self.addCleanup(lambda: __import__("shutil").rmtree(self._tmp, ignore_errors=True))
+        # The launch writes the gateway CA under the bottle root — sandbox it.
+        self.addCleanup(use_bottle_root(Path(self._tmp)))
 
-    def tearDown(self) -> None:
-        import shutil
-        shutil.rmtree(self._tmp, ignore_errors=True)
-
-    def _run_launch(
+    @contextlib.contextmanager
+    def _patched(
         self,
-        plan: DockerBottlePlan,
         *,
-        committed_tag: str | None = None,
-        image_present: bool = True,
-    ) -> list[str]:
-        """Drive launch() through its full sequence with the committed-image
-        behaviour controlled by the arguments. Returns the images that were
-        passed to `build_image` (empty list if it was never called)."""
+        committed_tag: str | None,
+        image_present: bool,
+        compose: Callable[..., dict[str, Any]],
+    ) -> Generator[list[str], None, None]:
         built: list[str] = []
+        gw = mock.Mock()
+        gw.ca_cert_pem.return_value = "-----BEGIN CERTIFICATE-----\nx\n-----END CERTIFICATE-----\n"
 
-        def fake_build(image: str, ctx: str, *, dockerfile: str = "") -> None:
+        def _build(image: str, ctx: str, *, dockerfile: str = "") -> None:
             del ctx, dockerfile
             built.append(image)
 
-        with mock.patch.object(
-            launch_mod, "read_committed_image", return_value=committed_tag,
-        ), mock.patch.object(
-            launch_mod.docker_mod, "image_exists", return_value=image_present,
-        ), mock.patch.object(
-            launch_mod.docker_mod, "build_image", side_effect=fake_build,
-        ), mock.patch.object(
-            launch_mod, "egress_tls_init",
-            return_value=(Path("/egress_ca"), Path("/egress_cert")),
-        ), mock.patch.object(
-            launch_mod.network_mod, "network_name_for_slug",
-            return_value="bb-internal",
-        ), mock.patch.object(
-            launch_mod.network_mod, "network_egress_name_for_slug",
-            return_value="bb-egress",
-        ), mock.patch.object(
-            launch_mod, "bottle_plan_to_compose",
-            return_value={"services": {"agent": {}}},
-        ), mock.patch.object(
-            launch_mod, "write_compose_file",
-            return_value=Path("/tmp/compose.yml"),
-        ), mock.patch.object(launch_mod, "compose_up"), \
-           mock.patch.object(launch_mod, "compose_dump_logs"), \
-           mock.patch.object(launch_mod, "compose_down"), \
-           contextlib.redirect_stderr(io.StringIO()):
-            provision = mock.Mock(return_value=None)
-            with launch_mod.launch(plan, provision=provision):
-                pass
+        with mock.patch.object(launch_mod, "read_committed_image", return_value=committed_tag), \
+                mock.patch.object(launch_mod.docker_mod, "image_exists", return_value=image_present), \
+                mock.patch.object(launch_mod.docker_mod, "build_image", side_effect=_build), \
+                mock.patch.object(launch_mod, "launch_consolidated", return_value=_CTX), \
+                mock.patch.object(launch_mod, "teardown_consolidated"), \
+                mock.patch.object(launch_mod, "DockerGateway", return_value=gw), \
+                mock.patch.object(launch_mod, "consolidated_agent_compose", side_effect=compose), \
+                mock.patch.object(launch_mod, "write_compose_file", return_value=Path("/tmp/c.yml")), \
+                mock.patch.object(launch_mod, "compose_up"), \
+                mock.patch.object(launch_mod, "compose_dump_logs"), \
+                mock.patch.object(launch_mod, "compose_down"), \
+                contextlib.redirect_stderr(io.StringIO()):
+            yield built
 
+    def _run_launch(
+        self, plan: DockerBottlePlan, *,
+        committed_tag: str | None = None, image_present: bool = True,
+    ) -> list[str]:
+        def compose(_p: DockerBottlePlan, **_kw: Any) -> dict[str, Any]:
+            return {"services": {"agent": {}}}
+        with self._patched(
+            committed_tag=committed_tag, image_present=image_present, compose=compose,
+        ) as built:
+            with launch_mod.launch(plan, provision=mock.Mock(return_value=None)):
+                pass
         return built
 
     def test_skips_build_when_committed_image_present(self) -> None:
-        plan = _plan(self._tmp)
-        built = self._run_launch(plan, committed_tag=_COMMITTED_TAG, image_present=True)
-        self.assertEqual([], built, "build_image should not be called when committed image exists")
+        built = self._run_launch(_plan(self._tmp), committed_tag=_COMMITTED_TAG, image_present=True)
+        self.assertEqual([], built)  # committed image reused, no build
 
     def test_uses_committed_image_in_compose_spec(self) -> None:
-        """The compose spec renderer receives the committed image tag via
-        plan.image — captured here by checking what bottle_plan_to_compose
-        was called with."""
-        plan = _plan(self._tmp)
-        captured_plans: list[DockerBottlePlan] = []
+        captured: list[DockerBottlePlan] = []
 
-        def fake_compose(p: DockerBottlePlan) -> dict[str, Any]:
-            captured_plans.append(p)
+        def compose(p: DockerBottlePlan, **_kw: Any) -> dict[str, Any]:
+            captured.append(p)
             return {"services": {"agent": {}}}
 
-        with mock.patch.object(
-            launch_mod, "read_committed_image", return_value=_COMMITTED_TAG,
-        ), mock.patch.object(
-            launch_mod.docker_mod, "image_exists", return_value=True,
-        ), mock.patch.object(
-            launch_mod.docker_mod, "build_image",
-        ), mock.patch.object(
-            launch_mod, "egress_tls_init",
-            return_value=(Path("/egress_ca"), Path("/egress_cert")),
-        ), mock.patch.object(
-            launch_mod.network_mod, "network_name_for_slug",
-            return_value="bb-internal",
-        ), mock.patch.object(
-            launch_mod.network_mod, "network_egress_name_for_slug",
-            return_value="bb-egress",
-        ), mock.patch.object(
-            launch_mod, "bottle_plan_to_compose", side_effect=fake_compose,
-        ), mock.patch.object(
-            launch_mod, "write_compose_file",
-            return_value=Path("/tmp/compose.yml"),
-        ), mock.patch.object(launch_mod, "compose_up"), \
-           mock.patch.object(launch_mod, "compose_dump_logs"), \
-           mock.patch.object(launch_mod, "compose_down"), \
-           contextlib.redirect_stderr(io.StringIO()):
-            provision = mock.Mock(return_value=None)
-            with launch_mod.launch(plan, provision=provision):
+        with self._patched(committed_tag=_COMMITTED_TAG, image_present=True, compose=compose):
+            with launch_mod.launch(_plan(self._tmp), provision=mock.Mock(return_value=None)):
                 pass
-
-        self.assertEqual(1, len(captured_plans))
-        self.assertEqual(_COMMITTED_TAG, captured_plans[0].image)
+        self.assertEqual(_COMMITTED_TAG, captured[0].image)
 
     def test_falls_back_to_build_when_no_committed_image(self) -> None:
-        plan = _plan(self._tmp)
-        built = self._run_launch(plan, committed_tag=None)
-        self.assertEqual([_DEFAULT_IMAGE], built)
+        self.assertEqual([_DEFAULT_IMAGE], self._run_launch(_plan(self._tmp), committed_tag=None))
 
     def test_falls_back_to_build_when_committed_image_missing_from_daemon(self) -> None:
-        plan = _plan(self._tmp)
-        built = self._run_launch(
-            plan, committed_tag=_COMMITTED_TAG, image_present=False,
-        )
+        built = self._run_launch(_plan(self._tmp), committed_tag=_COMMITTED_TAG, image_present=False)
         self.assertEqual([_DEFAULT_IMAGE], built)
 
 

tests/unit/test_docker_launch_teardown.py

@@ -19,9 +19,11 @@ from bot_bottle.agent_provider import AgentProvisionPlan
 from bot_bottle.backend import BottleSpec
 from bot_bottle.backend.docker import launch as launch_mod
 from bot_bottle.backend.docker.bottle_plan import DockerBottlePlan
+from bot_bottle.backend.docker.consolidated_launch import LaunchContext
 from bot_bottle.egress import EgressPlan
 from bot_bottle.git_gate import GitGatePlan
 from bot_bottle.manifest import ManifestIndex
+from tests.unit import use_bottle_root
 
 _INDEX = ManifestIndex.from_json_obj({
     "bottles": {"dev": {}},
@@ -77,41 +79,36 @@ def _plan(tmp: str) -> DockerBottlePlan:
 class TestTeardownWarning(unittest.TestCase):
     def setUp(self) -> None:
         self._tmp = tempfile.mkdtemp(prefix="docker-launch-teardown-test.")
-
-    def tearDown(self) -> None:
-        import shutil
-        shutil.rmtree(self._tmp, ignore_errors=True)
+        self.addCleanup(lambda: __import__("shutil").rmtree(self._tmp, ignore_errors=True))
+        self.addCleanup(use_bottle_root(Path(self._tmp)))  # sandbox the gateway CA write
 
     def test_teardown_failure_emits_warning_with_container_and_operation(self):
         plan = _plan(self._tmp)
         buf = io.StringIO()
+        gw = mock.Mock()
+        gw.ca_cert_pem.return_value = "-----BEGIN CERTIFICATE-----\nx\n-----END CERTIFICATE-----\n"
+        ctx = LaunchContext(
+            bottle_id="b1", identity_token="t", source_ip="172.20.0.4",
+            network="bot-bottle-gateway", gateway_ip="172.20.0.2",
+            orchestrator_url="http://orch:8099",
+        )
 
         with mock.patch.object(launch_mod.docker_mod, "build_image"), \
+             mock.patch.object(launch_mod, "launch_consolidated", return_value=ctx), \
+             mock.patch.object(launch_mod, "teardown_consolidated"), \
+             mock.patch.object(launch_mod, "DockerGateway", return_value=gw), \
              mock.patch.object(
-                 launch_mod, "egress_tls_init",
-                 return_value=(Path("/egress_ca"), Path("/egress_cert")),
-             ), \
-             mock.patch.object(
-                 launch_mod.network_mod, "network_name_for_slug",
-                 return_value="bb-internal-test",
-             ), \
-             mock.patch.object(
-                 launch_mod.network_mod, "network_egress_name_for_slug",
-                 return_value="bb-egress-test",
-             ), \
-             mock.patch.object(
-                 launch_mod, "bottle_plan_to_compose",
+                 launch_mod, "consolidated_agent_compose",
                  return_value={"services": {"agent": {}}},
              ), \
              mock.patch.object(
-                 launch_mod, "write_compose_file",
-                 return_value=Path("/tmp/compose.yml"),
+                 launch_mod, "write_compose_file", return_value=Path("/tmp/compose.yml"),
              ), \
              mock.patch.object(launch_mod, "compose_up"), \
              mock.patch.object(launch_mod, "compose_dump_logs"), \
              mock.patch.object(
                  launch_mod, "compose_down",
-                 side_effect=RuntimeError("network remove failed"),
+                 side_effect=RuntimeError("compose down failed"),
              ), \
              contextlib.redirect_stderr(buf):
             provision = mock.Mock(return_value=None)

tests/unit/test_docker_util_image.py

@@ -29,7 +29,7 @@ def _fail(stderr: str = "boom") -> subprocess.CompletedProcess:  # type: ignore
 class TestCommitContainer(unittest.TestCase):
     def test_runs_docker_commit(self):
         with patch.object(
-            docker_mod.subprocess, "run", return_value=_ok(),
+            docker_mod, "run_docker", return_value=_ok(),
         ) as run, patch.object(docker_mod, "info"):
             docker_mod.commit_container(
                 "bot-bottle-dev-abc12",
@@ -47,7 +47,7 @@ class TestCommitContainer(unittest.TestCase):
 
     def test_dies_on_docker_commit_failure(self):
         with patch.object(
-            docker_mod.subprocess, "run", return_value=_fail("No such container"),
+            docker_mod, "run_docker", return_value=_fail("No such container"),
         ), patch.object(
             docker_mod, "die", side_effect=SystemExit("die"),
         ) as die:
@@ -58,7 +58,7 @@ class TestCommitContainer(unittest.TestCase):
 
     def test_die_message_includes_image_tag(self):
         with patch.object(
-            docker_mod.subprocess, "run", return_value=_fail("boom"),
+            docker_mod, "run_docker", return_value=_fail("boom"),
         ), patch.object(
             docker_mod, "die", side_effect=SystemExit("die"),
         ) as die:

tests/unit/test_egress_addon_request_flow.py

@@ -233,16 +233,19 @@ class _CtxResolver:
     """Fake orchestrator resolver: maps source IP -> bottle id, and grants the
     same allow-list to any attributed bottle (unattributed -> deny)."""
 
-    def __init__(self, ip_to_bottle: dict[str, str]) -> None:
+    def __init__(
+        self, ip_to_bottle: dict[str, str], tokens: dict[str, str] | None = None,
+    ) -> None:
         self._map = ip_to_bottle
+        self._tokens = tokens or {}
 
     def resolve_policy_and_bottle_id(
         self, source_ip: str, identity_token: str = "",
-    ) -> tuple[str | None, str | None]:
+    ) -> tuple[str | None, str | None, dict[str, str]]:
         del identity_token
         bottle_id = self._map.get(source_ip)
         policy = "routes:\n  - host: api.example.com\n" if bottle_id else None
-        return policy, bottle_id
+        return policy, bottle_id, (self._tokens if bottle_id else {})
 
 
 # ---------------------------------------------------------------------------
@@ -803,6 +806,46 @@ class TestSuperviseMultiTenant(unittest.TestCase):
             _run_request(addon, flow)
         self.assertEqual(["bottle-b"], seen)  # proposal keyed by the resolved bottle
 
+    def test_auth_token_injected_from_resolved_tokens(self) -> None:
+        # The bottle's upstream token comes from /resolve (in-memory on the
+        # orchestrator), NOT the gateway's env — the request is forwarded with
+        # Authorization injected. This is the token-injection cut-over fix.
+        policy = (
+            'routes:\n  - host: api.example.com\n'
+            '    auth_scheme: "Bearer"\n    token_env: "EGRESS_TOKEN_0"\n'
+        )
+
+        class _AuthResolver:
+            def resolve_policy_and_bottle_id(self, ip: str, identity_token: str = ""):
+                del ip, identity_token
+                return policy, "bottle-a", {"EGRESS_TOKEN_0": "sk-secret"}
+
+        addon = _addon(Config(routes=()))
+        addon._resolver = cast(Any, _AuthResolver())
+        flow = _with_client_ip(_Flow(_Request(host="api.example.com", method="GET")), "10.0.0.1")
+        _run_request(addon, flow)
+        self.assertIsNone(flow.response)  # forwarded, not blocked
+        self.assertEqual("Bearer sk-secret", flow.request.headers.get("authorization"))
+
+    def test_authed_route_without_token_is_blocked(self) -> None:
+        # No token resolved for the bottle → the authed route can't inject, so
+        # the request is blocked (the error the cut-over originally produced).
+        policy = (
+            'routes:\n  - host: api.example.com\n'
+            '    auth_scheme: "Bearer"\n    token_env: "EGRESS_TOKEN_0"\n'
+        )
+
+        class _NoTokenResolver:
+            def resolve_policy_and_bottle_id(self, ip: str, identity_token: str = ""):
+                del ip, identity_token
+                return policy, "bottle-a", {}  # no tokens
+
+        addon = _addon(Config(routes=()))
+        addon._resolver = cast(Any, _NoTokenResolver())
+        flow = _with_client_ip(_Flow(_Request(host="api.example.com", method="GET")), "10.0.0.1")
+        _run_request(addon, flow)
+        self.assertIsNotNone(flow.response)  # blocked — token unset
+
     def test_unattributed_source_ip_cannot_supervise(self) -> None:
         addon = self._consolidated_addon()
         # 10.9.9.9 is not in the resolver map -> deny-all config, empty slug.

tests/unit/test_egress_multitenant.py

@@ -51,45 +51,55 @@ class TestResolveClientConfig(unittest.TestCase):
 
 class _FakeContextResolver:
     def __init__(
-        self, policy: str | None = None, bottle_id: str | None = None, raises: bool = False,
+        self, policy: str | None = None, bottle_id: str | None = None,
+        raises: bool = False, tokens: dict[str, str] | None = None,
     ) -> None:
         self._policy = policy
         self._bottle_id = bottle_id
         self._raises = raises
+        self._tokens = tokens or {}
 
     def resolve_policy_and_bottle_id(
         self, source_ip: str, identity_token: str = "",
-    ) -> tuple[str | None, str | None]:
+    ) -> tuple[str | None, str | None, dict[str, str]]:
         if self._raises:
             raise PolicyResolveError("orchestrator down")
-        return self._policy, self._bottle_id
+        return self._policy, self._bottle_id, self._tokens
 
 
 class TestResolveClientContext(unittest.TestCase):
-    def test_returns_config_and_bottle_id(self) -> None:
-        cfg, slug = resolve_client_context(
-            _FakeContextResolver(policy="routes:\n  - host: example.com\n", bottle_id="b1"),
+    def test_returns_config_bottle_id_and_tokens(self) -> None:
+        cfg, slug, tokens = resolve_client_context(
+            _FakeContextResolver(
+                policy="routes:\n  - host: example.com\n", bottle_id="b1",
+                tokens={"EGRESS_TOKEN_0": "sekret"},
+            ),
             "10.243.0.1",
         )
         self.assertEqual(("example.com",), tuple(r.host for r in cfg.routes))
         self.assertEqual("b1", slug)
+        self.assertEqual({"EGRESS_TOKEN_0": "sekret"}, tokens)  # for auth injection
 
     def test_unattributed_denies_and_empty_slug(self) -> None:
-        cfg, slug = resolve_client_context(
+        cfg, slug, tokens = resolve_client_context(
             _FakeContextResolver(policy=None, bottle_id=None), "10.243.0.9",
         )
         self.assertEqual((), cfg.routes)
         self.assertEqual("", slug)  # no bottle → supervise unavailable
+        self.assertEqual({}, tokens)
 
-    def test_resolver_error_denies_and_empty_slug(self) -> None:
-        cfg, slug = resolve_client_context(_FakeContextResolver(raises=True), "10.243.0.1")
+    def test_resolver_error_denies_empty_slug_no_tokens(self) -> None:
+        cfg, slug, tokens = resolve_client_context(
+            _FakeContextResolver(raises=True), "10.243.0.1",
+        )
         self.assertEqual((), cfg.routes)
         self.assertEqual("", slug)
+        self.assertEqual({}, tokens)
 
     def test_unparseable_policy_denies_but_keeps_slug(self) -> None:
         # A bad policy denies egress, but the bottle is still attributed (its
         # supervise queue is keyed by the id, independent of route parsing).
-        cfg, slug = resolve_client_context(
+        cfg, slug, _tokens = resolve_client_context(
             _FakeContextResolver(policy="routes: notalist\n", bottle_id="b2"), "10.243.0.1",
         )
         self.assertEqual((), cfg.routes)

tests/unit/test_gateway_provision.py

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

tests/unit/test_orchestrator_gateway.py

@@ -6,11 +6,15 @@ import unittest
 from unittest.mock import Mock, patch
 
 from bot_bottle.orchestrator.gateway import (
+    GATEWAY_CA_CERT,
     GATEWAY_NAME,
     DockerGateway,
     GatewayError,
 )
 
+
+_CA_PEM = "-----BEGIN CERTIFICATE-----\nMII...\n-----END CERTIFICATE-----\n"
+
 _RUN_DOCKER = "bot_bottle.orchestrator.gateway.run_docker"
 
 
@@ -31,11 +35,43 @@ class TestDockerGateway(unittest.TestCase):
         with patch(_RUN_DOCKER, return_value=_proc(stdout="")):
             self.assertFalse(self.sc.is_running())
 
-    def test_ensure_running_is_noop_when_already_up(self) -> None:
-        with patch(_RUN_DOCKER, return_value=_proc(stdout=self.sc.name)) as m:
+    def test_ensure_running_noop_when_up_and_image_current(self) -> None:
+        calls: list[list[str]] = []
+
+        def fake(argv: list[str]) -> Mock:
+            calls.append(argv)
+            if argv[:2] == ["docker", "ps"]:
+                return _proc(stdout=self.sc.name)          # running
+            if argv[:3] == ["docker", "image", "inspect"]:
+                return _proc(stdout="img-A")               # current image id
+            if argv[:2] == ["docker", "inspect"]:
+                return _proc(stdout="img-A")               # container's image (same)
+            return _proc()
+
+        with patch(_RUN_DOCKER, side_effect=fake):
             self.sc.ensure_running()
-        # Only the is_running() ps probe — no rm / run.
-        self.assertEqual(1, m.call_count)
+        self.assertEqual([], [c for c in calls if c[:2] == ["docker", "run"]])
+        self.assertEqual([], [c for c in calls if c[:2] == ["docker", "rm"]])
+
+    def test_ensure_running_recreates_when_image_is_stale(self) -> None:
+        # Running, but the container was built from an OLD image → recreate so
+        # a rebuild's new flat daemons take effect.
+        calls: list[list[str]] = []
+
+        def fake(argv: list[str]) -> Mock:
+            calls.append(argv)
+            if argv[:2] == ["docker", "ps"]:
+                return _proc(stdout=self.sc.name)
+            if argv[:3] == ["docker", "image", "inspect"]:
+                return _proc(stdout="img-NEW")
+            if argv[:2] == ["docker", "inspect"]:
+                return _proc(stdout="img-OLD")
+            return _proc()
+
+        with patch(_RUN_DOCKER, side_effect=fake):
+            self.sc.ensure_running()
+        self.assertEqual(1, len([c for c in calls if c[:2] == ["docker", "run"]]))
+        self.assertTrue(any(c[:2] == ["docker", "rm"] for c in calls))
 
     def test_ensure_running_starts_the_singleton_when_absent(self) -> None:
         calls: list[list[str]] = []
@@ -53,6 +89,8 @@ class TestDockerGateway(unittest.TestCase):
         self.assertIn("bot-bottle-sidecars:latest", runs[0])
         # Runs on the shared gateway network so agents can reach it by IP.
         self.assertEqual(self.sc.network, runs[0][runs[0].index("--network") + 1])
+        # Persists its CA on a named volume so agents keep trusting it.
+        self.assertTrue(any("mitmproxy" in a for a in runs[0]))
 
     def test_ensure_running_creates_network_when_missing(self) -> None:
         calls: list[list[str]] = []
@@ -68,6 +106,25 @@ class TestDockerGateway(unittest.TestCase):
         creates = [c for c in calls if c[:3] == ["docker", "network", "create"]]
         self.assertEqual([["docker", "network", "create", self.sc.network]], creates)
 
+    def test_ca_cert_pem_reads_from_container(self) -> None:
+        with patch(_RUN_DOCKER, return_value=_proc(stdout=_CA_PEM)) as m:
+            self.assertEqual(_CA_PEM, self.sc.ca_cert_pem())
+        argv = m.call_args.args[0]
+        self.assertEqual(["docker", "exec", self.sc.name, "cat", GATEWAY_CA_CERT], argv)
+
+    def test_ca_cert_pem_raises_when_absent(self) -> None:
+        # timeout=0 → one probe then give up (no polling delay in the test).
+        with patch(_RUN_DOCKER, return_value=_proc(returncode=1, stderr="No such file")):
+            with self.assertRaises(GatewayError):
+                self.sc.ca_cert_pem(timeout=0)
+
+    def test_ca_cert_pem_polls_until_mitmproxy_writes_it(self) -> None:
+        # First read: CA not there yet; second read: present.
+        seq = [_proc(returncode=1, stderr="No such file"), _proc(stdout=_CA_PEM)]
+        with patch(_RUN_DOCKER, side_effect=seq), \
+                patch("bot_bottle.orchestrator.gateway.time.sleep"):
+            self.assertEqual(_CA_PEM, self.sc.ca_cert_pem(timeout=5))
+
     def test_ensure_running_reuses_existing_network(self) -> None:
         calls: list[list[str]] = []
 
@@ -112,28 +169,35 @@ class TestDockerGatewayBuild(unittest.TestCase):
         with patch(_RUN_DOCKER, return_value=_proc(returncode=1)):
             self.assertFalse(self.sc.image_exists())
 
-    def test_ensure_built_is_noop_when_image_present(self) -> None:
-        with patch(_RUN_DOCKER, return_value=_proc(returncode=0)) as m:
-            self.sc.ensure_built()
-        self.assertEqual(1, m.call_count)  # only the image-inspect probe
-
-    def test_ensure_built_builds_from_dockerfile_when_missing(self) -> None:
+    def test_ensure_built_builds_even_when_image_present(self) -> None:
+        # Always build (cache-aware) so a flat-source change rebuilds; the old
+        # build-if-missing silently ran a stale single-tenant image.
         calls: list[list[str]] = []
 
-        def fake(argv: list[str]) -> Mock:
+        def rec(argv: list[str]) -> Mock:
             calls.append(argv)
-            if argv[:3] == ["docker", "image", "inspect"]:
-                return _proc(returncode=1)  # missing
-            return _proc()
+            return _proc()  # image present, build succeeds
 
-        with patch(_RUN_DOCKER, side_effect=fake):
+        with patch(_RUN_DOCKER, side_effect=rec):
             self.sc.ensure_built()
-
         builds = [c for c in calls if c[:2] == ["docker", "build"]]
         self.assertEqual(1, len(builds))
         self.assertIn(self.sc.image_ref, builds[0])
-        self.assertIn("-f", builds[0])
         self.assertTrue(any(a.endswith("Dockerfile.sidecars") for a in builds[0]))
+        self.assertNotIn("--no-cache", builds[0])
+
+    def test_ensure_built_no_cache_env_forces_full_rebuild(self) -> None:
+        calls: list[list[str]] = []
+
+        def rec(argv: list[str]) -> Mock:
+            calls.append(argv)
+            return _proc()
+
+        with patch(_RUN_DOCKER, side_effect=rec), \
+                patch.dict("os.environ", {"BOT_BOTTLE_NO_CACHE": "1"}):
+            self.sc.ensure_built()
+        builds = [c for c in calls if c[:2] == ["docker", "build"]]
+        self.assertIn("--no-cache", builds[0])
 
     def test_ensure_built_noop_when_no_dockerfile(self) -> None:
         sc = DockerGateway("busybox", dockerfile=None)
@@ -142,12 +206,7 @@ class TestDockerGatewayBuild(unittest.TestCase):
         m.assert_not_called()
 
     def test_ensure_built_raises_on_build_failure(self) -> None:
-        def fake(argv: list[str]) -> Mock:
-            if argv[:3] == ["docker", "image", "inspect"]:
-                return _proc(returncode=1)
-            return _proc(returncode=1, stderr="build boom")
-
-        with patch(_RUN_DOCKER, side_effect=fake):
+        with patch(_RUN_DOCKER, return_value=_proc(returncode=1, stderr="build boom")):
             with self.assertRaises(GatewayError):
                 self.sc.ensure_built()
 

tests/unit/test_orchestrator_lifecycle.py

@@ -1,4 +1,4 @@
-"""Unit: orchestrator process lifecycle — idempotent singleton (PRD 0070)."""
+"""Unit: orchestrator+gateway container lifecycle — idempotent singleton (PRD 0070)."""
 
 from __future__ import annotations
 
@@ -6,81 +6,86 @@ import tempfile
 import unittest
 import urllib.error
 from pathlib import Path
-from unittest.mock import MagicMock, patch
+from unittest.mock import MagicMock, Mock, patch
 
 from bot_bottle.orchestrator.lifecycle import (
-    OrchestratorProcess,
+    ORCHESTRATOR_NAME,
+    OrchestratorService,
     OrchestratorStartError,
 )
 from tests.unit import use_bottle_root
 
 _URLOPEN = "bot_bottle.orchestrator.lifecycle.urllib.request.urlopen"
-_POPEN = "bot_bottle.orchestrator.lifecycle.subprocess.Popen"
+_RUN = "bot_bottle.orchestrator.lifecycle.run_docker"
+_GATEWAY = "bot_bottle.orchestrator.lifecycle.DockerGateway"
 _SLEEP = "bot_bottle.orchestrator.lifecycle.time.sleep"
 _MONOTONIC = "bot_bottle.orchestrator.lifecycle.time.monotonic"
 
 
 def _health(status: int) -> MagicMock:
-    """A urlopen() context-manager whose `.status` is `status`."""
     m = MagicMock()
     m.__enter__.return_value.status = status
     return m
 
 
-class TestOrchestratorProcess(unittest.TestCase):
+class TestOrchestratorService(unittest.TestCase):
     def setUp(self) -> None:
         self._tmp = tempfile.TemporaryDirectory()
         self.addCleanup(self._tmp.cleanup)
         self.addCleanup(use_bottle_root(Path(self._tmp.name)))
-        self.p = OrchestratorProcess(port=8099)
+        self.svc = OrchestratorService(port=8099)
 
-    def test_url(self) -> None:
-        self.assertEqual("http://127.0.0.1:8099", self.p.url)
+    def test_urls(self) -> None:
+        self.assertEqual("http://127.0.0.1:8099", self.svc.url)
+        # The gateway reaches the control plane by container name over docker DNS.
+        self.assertEqual(f"http://{ORCHESTRATOR_NAME}:8099", self.svc.internal_url)
 
-    def test_is_healthy_true_on_200(self) -> None:
+    def test_is_healthy(self) -> None:
         with patch(_URLOPEN, return_value=_health(200)):
-            self.assertTrue(self.p.is_healthy())
-
-    def test_is_healthy_false_on_error(self) -> None:
+            self.assertTrue(self.svc.is_healthy())
         with patch(_URLOPEN, side_effect=urllib.error.URLError("refused")):
-            self.assertFalse(self.p.is_healthy())
+            self.assertFalse(self.svc.is_healthy())
 
-    def test_ensure_running_noop_when_already_healthy(self) -> None:
-        with patch(_URLOPEN, return_value=_health(200)), patch(_POPEN) as popen:
-            self.assertEqual(self.p.url, self.p.ensure_running())
-        popen.assert_not_called()  # a live control plane is left untouched
+    def test_ensure_running_always_recreates_orchestrator(self) -> None:
+        # Even when a control plane is already healthy, the orchestrator is
+        # recreated so bind-mounted code changes take effect (its process
+        # won't reload). The gateway is ensured too.
+        run = Mock(return_value=Mock(returncode=0, stderr=""))
+        with patch(_URLOPEN, return_value=_health(200)), \
+                patch(_GATEWAY) as gw_cls, patch(_RUN, run), patch(_SLEEP):
+            self.assertEqual(self.svc.url, self.svc.ensure_running())
+        gw_cls.return_value.ensure_running.assert_called()   # gateway kept up
+        runs = [c.args[0] for c in run.call_args_list if c.args[0][:2] == ["docker", "run"]]
+        self.assertEqual(1, len(runs))                       # orchestrator recreated
+        self.assertIn(ORCHESTRATOR_NAME, runs[0])
 
-    def test_ensure_running_spawns_then_waits_for_health(self) -> None:
-        # First check (before spawn) fails; after spawn the poll succeeds.
+    def test_ensure_running_starts_orchestrator_container_when_absent(self) -> None:
+        run = Mock(return_value=Mock(returncode=0, stderr=""))
         with patch(_URLOPEN, side_effect=[urllib.error.URLError("down"), _health(200)]), \
-                patch(_POPEN) as popen, patch(_SLEEP):
-            url = self.p.ensure_running()
-        self.assertEqual(self.p.url, url)
-        popen.assert_called_once()
-
-    def test_ensure_running_raises_on_startup_timeout(self) -> None:
-        with patch(_URLOPEN, side_effect=urllib.error.URLError("down")), \
-                patch(_POPEN), patch(_SLEEP), \
-                patch(_MONOTONIC, side_effect=[0.0, 0.5, 2.0]):
-            with self.assertRaises(OrchestratorStartError):
-                self.p.ensure_running(startup_timeout=1.0)
-
-    def test_argv_includes_gateway_and_broker(self) -> None:
-        argv = OrchestratorProcess(port=8099, broker="docker", gateway=True)._argv()
-        self.assertIn("--gateway", argv)
+                patch(_GATEWAY), patch(_RUN, run), patch(_SLEEP):
+            self.assertEqual(self.svc.url, self.svc.ensure_running())
+        runs = [c.args[0] for c in run.call_args_list if c.args[0][:2] == ["docker", "run"]]
+        self.assertEqual(1, len(runs))
+        argv = runs[0]
+        self.assertIn(ORCHESTRATOR_NAME, argv)
+        self.assertIn("--broker", argv)
+        self.assertEqual("stub", argv[argv.index("--broker") + 1])  # register-only, no socket
         self.assertIn("bot_bottle.orchestrator", argv)
-        self.assertEqual("docker", argv[argv.index("--broker") + 1])
+        self.assertEqual("127.0.0.1:8099:8099", argv[argv.index("--publish") + 1])
 
-    def test_argv_omits_gateway_when_disabled(self) -> None:
-        self.assertNotIn("--gateway", OrchestratorProcess(gateway=False)._argv())
+    def test_ensure_running_raises_on_timeout(self) -> None:
+        with patch(_URLOPEN, side_effect=urllib.error.URLError("down")), \
+                patch(_GATEWAY), patch(_RUN, return_value=Mock(returncode=0, stderr="")), \
+                patch(_SLEEP), patch(_MONOTONIC, side_effect=[0.0, 0.5, 2.0]):
+            with self.assertRaises(OrchestratorStartError):
+                self.svc.ensure_running(startup_timeout=1.0)
 
-    def test_spawn_launches_detached_and_logs(self) -> None:
-        with patch(_POPEN) as popen:
-            self.p._spawn()
-        popen.assert_called_once()
-        kwargs = popen.call_args.kwargs
-        self.assertTrue(kwargs["start_new_session"])  # outlives the CLI
-        self.assertTrue((Path(self._tmp.name) / "orchestrator.log").exists())
+    def test_stop_removes_orchestrator_and_gateway(self) -> None:
+        with patch(_RUN) as run, patch(_GATEWAY) as gw_cls:
+            self.svc.stop()
+        rms = [c.args[0] for c in run.call_args_list if c.args[0][:3] == ["docker", "rm", "--force"]]
+        self.assertTrue(any(ORCHESTRATOR_NAME in a for a in rms))
+        gw_cls.return_value.stop.assert_called_once()
 
 
 if __name__ == "__main__":

tests/unit/test_orchestrator_registry.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import sqlite3
 import tempfile
 import unittest
 from pathlib import Path
@@ -23,6 +24,20 @@ class TestRegistryStore(unittest.TestCase):
     def tearDown(self) -> None:
         self._tmp.cleanup()
 
+    def _force_active_row(self, source_ip: str, bottle_id: str) -> None:
+        """Insert a raw active row, bypassing `register`'s same-IP supersede —
+        the only way to manufacture the ambiguity the fail-closed guard exists
+        for (crash-orphaned / externally-injected duplicate)."""
+        conn = sqlite3.connect(self.db)
+        conn.execute(
+            "INSERT INTO orchestrator_bottles "
+            "(bottle_id, source_ip, identity_token, state, created_at, metadata, policy) "
+            "VALUES (?, ?, ?, 'active', 0.0, '', '')",
+            (bottle_id, source_ip, "tok-" + bottle_id),
+        )
+        conn.commit()
+        conn.close()
+
     def test_register_mints_id_and_token(self) -> None:
         rec = self.store.register("10.243.0.1")
         self.assertTrue(rec.bottle_id)
@@ -76,11 +91,29 @@ class TestRegistryStore(unittest.TestCase):
 
     def test_attribute_ambiguous_ip_denied(self) -> None:
         # Two active bottles on one source IP is a misconfiguration — deny
-        # rather than guess (fail-closed), even with a valid token.
+        # rather than guess (fail-closed), even with a valid token. (Forced
+        # directly: `register` now supersedes, so this can only arise from an
+        # orphaned/injected row.)
         a = self.store.register("10.243.0.1", bottle_id="a")
-        self.store.register("10.243.0.1", bottle_id="b")
+        self._force_active_row("10.243.0.1", "b")
         self.assertIsNone(self.store.attribute("10.243.0.1", a.identity_token))
 
+    def test_reregister_same_source_ip_supersedes(self) -> None:
+        # Re-registering a source IP (reused IP / crash recovery / persisted
+        # DB) must supersede the prior active bottle, not add a second active
+        # row that would fail-closed the whole IP to 403.
+        a = self.store.register("10.243.0.1", bottle_id="a", policy="A")
+        b = self.store.register("10.243.0.1", bottle_id="b", policy="B")
+        # exactly one active row at that IP, and it's the new bottle
+        got = self.store.by_source_ip("10.243.0.1")
+        assert got is not None
+        self.assertEqual("b", got.bottle_id)
+        self.assertEqual("B", got.policy)
+        # the superseded bottle is gone and its token no longer attributes
+        self.assertIsNone(self.store.get("a"))
+        self.assertIsNone(self.store.attribute("10.243.0.1", a.identity_token))
+        self.assertIsNotNone(self.store.attribute("10.243.0.1", b.identity_token))
+
     def test_state_persists_across_reopen(self) -> None:
         rec = self.store.register("10.243.0.1")
         reopened = RegistryStore(self.db)
@@ -130,7 +163,7 @@ class TestRegistryStore(unittest.TestCase):
     def test_by_source_ip_unknown_or_ambiguous_denied(self) -> None:
         self.assertIsNone(self.store.by_source_ip("10.243.9.9"))  # unknown
         self.store.register("10.243.0.1", bottle_id="a")
-        self.store.register("10.243.0.1", bottle_id="b")
+        self._force_active_row("10.243.0.1", "b")  # orphaned duplicate
         self.assertIsNone(self.store.by_source_ip("10.243.0.1"))  # ambiguous
 
 

tests/unit/test_orchestrator_service.py

@@ -92,6 +92,18 @@ class TestOrchestrator(unittest.TestCase):
         assert got is not None
         self.assertEqual('{"routes":[]}', got.policy)
 
+    def test_tokens_held_in_memory_and_cleared_on_teardown(self) -> None:
+        rec = self.orch.launch_bottle("10.243.0.5", tokens={"EGRESS_TOKEN_0": "sk"})
+        self.assertEqual({"EGRESS_TOKEN_0": "sk"}, self.orch.tokens_for(rec.bottle_id))
+        # not persisted in the registry (redacted record has no tokens field)
+        self.assertNotIn("tokens", rec.redacted())
+        self.orch.teardown_bottle(rec.bottle_id)
+        self.assertEqual({}, self.orch.tokens_for(rec.bottle_id))
+
+    def test_tokens_default_empty(self) -> None:
+        rec = self.orch.launch_bottle("10.243.0.6")
+        self.assertEqual({}, self.orch.tokens_for(rec.bottle_id))
+
     def test_set_policy_live_reload(self) -> None:
         rec = self.orch.launch_bottle("10.243.0.3")
         self.assertTrue(self.orch.set_policy(rec.bottle_id, '{"x":1}'))

tests/unit/test_policy_resolver.py

@@ -89,13 +89,17 @@ class TestPolicyResolver(unittest.TestCase):
                 self.r.resolve_bottle_id("10.243.0.1", "tok")
 
     def test_resolve_policy_and_bottle_id_one_call(self) -> None:
-        with patch(_URLOPEN, return_value=_resp({"bottle_id": "b1", "policy": "P"})) as m:
-            self.assertEqual(("P", "b1"), self.r.resolve_policy_and_bottle_id("10.243.0.1", "t"))
-        self.assertEqual(1, m.call_count)  # both from a single /resolve
+        payload = {"bottle_id": "b1", "policy": "P", "tokens": {"EGRESS_TOKEN_0": "s"}}
+        with patch(_URLOPEN, return_value=_resp(payload)) as m:
+            self.assertEqual(
+                ("P", "b1", {"EGRESS_TOKEN_0": "s"}),
+                self.r.resolve_policy_and_bottle_id("10.243.0.1", "t"),
+            )
+        self.assertEqual(1, m.call_count)  # policy + id + tokens from a single /resolve
 
-    def test_resolve_policy_and_bottle_id_403_is_none_none(self) -> None:
+    def test_resolve_policy_and_bottle_id_403_is_none_none_empty(self) -> None:
         with patch(_URLOPEN, side_effect=_http_error(403)):
-            self.assertEqual((None, None), self.r.resolve_policy_and_bottle_id("10.243.0.9"))
+            self.assertEqual((None, None, {}), self.r.resolve_policy_and_bottle_id("10.243.0.9"))
 
     def test_resolve_policy_and_bottle_id_error_raises(self) -> None:
         with patch(_URLOPEN, side_effect=urllib.error.URLError("refused")):