feat(orchestrator): slice 13c(iv) — gateway joins a shared network

The consolidated gateway ran on the default bridge; the model needs it on a
dedicated user-defined network that every agent bottle also joins, reaching
the gateway's egress / git-http / supervise ports by its address (no host
port publishing) — and the source IP the gateway attributes by is the
bottle's address on this network.

- GATEWAY_NETWORK = "bot-bottle-gateway"; DockerGateway gains a `network`
  param.
- ensure_running now ensures the network exists (idempotent create,
  tolerating a concurrent 'already exists') then runs with `--network`.
- Docker picks the subnet; the launcher reads it back (13c source-IP
  allocator) to hand each bottle a pinned address.

pyright 0 errors; pylint 9.83/10; unit suite green (the 13 test_sidecar_init
/bin/sleep errors are pre-existing NixOS-local noise).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01WBMWTEtQdJ4W5UrWuLHCck

bot_bottle/backend/docker/gateway_net.py

@@ -0,0 +1,47 @@
+"""Shared-gateway source-IP allocation for the consolidated docker backend
+(PRD 0070).
+
+In the consolidated model one gateway container serves every bottle over a
+single shared docker network, and each agent bottle attaches with a pinned,
+deterministic address that the gateway uses as its **attribution key**. This
+allocates those addresses from the network's subnet, skipping the reserved
+ones — the network address and broadcast (excluded by `hosts()`), docker's
+router `.1`, and everything already in use (`taken`: the gateway container
+plus every live bottle, which the caller reads from the registry).
+
+Pure `ipaddress` logic — the docker-specific bits (the subnet CIDR, the
+gateway container's own address) are gathered by the caller and passed in, so
+this stays testable without docker.
+"""
+
+from __future__ import annotations
+
+import ipaddress
+from collections.abc import Iterable
+
+
+class NoFreeAddressError(RuntimeError):
+    """The shared gateway network's subnet is exhausted — every host address
+    is reserved or already assigned to a bottle."""
+
+
+def next_free_ip(cidr: str, taken: Iterable[str]) -> str:
+    """The lowest host address in `cidr` not in `taken` and not docker's
+    router (`.1`). `taken` must include the gateway container's own address
+    and every live bottle's. Raises `NoFreeAddressError` if the subnet is
+    full."""
+    net = ipaddress.ip_network(cidr, strict=False)
+    reserved = {str(a) for a in taken}
+    # Docker assigns the network's first host (.1) to the bridge router; a
+    # bottle must never be handed that address.
+    reserved.add(str(net.network_address + 1))
+    for host in net.hosts():  # hosts() already excludes network + broadcast
+        candidate = str(host)
+        if candidate not in reserved:
+            return candidate
+    raise NoFreeAddressError(
+        f"no free address in {cidr} ({len(reserved)} reserved/assigned)"
+    )
+
+
+__all__ = ["next_free_ip", "NoFreeAddressError"]

bot_bottle/git_gate.py

@@ -46,6 +46,7 @@ from .git_gate_render import (
     git_gate_known_hosts_line,
     git_gate_render_access_hook,
     git_gate_render_entrypoint,
+    git_gate_render_provision,
     git_gate_render_gitconfig,
     git_gate_render_hook,
     git_gate_upstreams_for_bottle,
@@ -155,6 +156,7 @@ __all__ = [
     "git_gate_render_gitconfig",
     "git_gate_known_hosts_line",
     "git_gate_render_entrypoint",
+    "git_gate_render_provision",
     "git_gate_render_hook",
     "git_gate_render_access_hook",
     "provision_git_gate_dynamic_keys",

bot_bottle/git_gate_render.py

@@ -9,6 +9,7 @@ own; `git_gate` re-exports these names for API stability."""
 
 from __future__ import annotations
 
+import re
 import shlex
 from dataclasses import dataclass
 from pathlib import Path
@@ -125,22 +126,18 @@ def git_gate_known_hosts_line(host: str, port: str, key: str) -> str:
     return f"{target} {key}\n"
 
 
-def git_gate_render_entrypoint(upstreams: tuple[GitGateUpstream, ...]) -> str:
-    """Posix-sh entrypoint. One `init_repo` call per upstream, then
-    `exec git daemon`. The function reads
-    `/git-gate/creds/<name>-{key,known_hosts}` (bind-mounted into
-    the bundle by the renderer) and wires them into each bare repo's
-    config; the access-hook + pre-receive hook pick those paths up
-    at fetch / push time."""
-    lines = [
-        "#!/bin/sh",
-        "set -eu",
-        "",
+def _git_gate_init_repo_fn(repo_root: str, creds_dir: str) -> list[str]:
+    """The `init_repo` shell function, parameterized by the bare-repo root
+    and the per-bottle creds dir. Single source of the credential-wiring
+    logic, shared by the single-tenant daemon entrypoint (`/git`,
+    `/git-gate/creds`) and the consolidated per-bottle provisioning
+    (`/git/<bottle_id>`, `/git-gate/creds/<bottle_id>`)."""
+    return [
         "init_repo() {",
         "  name=$1",
         "  upstream_url=$2",
-        "  keyfile=/git-gate/creds/${name}-key",
-        "  hostsfile=/git-gate/creds/${name}-known_hosts",
+        f"  keyfile={creds_dir}/${{name}}-key",
+        f"  hostsfile={creds_dir}/${{name}}-known_hosts",
         "",
         # `|| true`: PRD 0018 chunk 3+ bind-mounts these RO from the
         # host, so chmod-syscalls fail with EROFS. The files already
@@ -153,14 +150,14 @@ def git_gate_render_entrypoint(upstreams: tuple[GitGateUpstream, ...]) -> str:
         "    chmod 600 \"$hostsfile\" 2>/dev/null || true",
         "  fi",
         "",
-        "  repo=/git/${name}.git",
+        f"  repo={repo_root}/${{name}}.git",
         "  if [ ! -d \"$repo\" ]; then",
         "    git init --bare \"$repo\" >/dev/null",
-        # --mirror=fetch sets remote.origin.fetch = +refs/*:refs/* so",
-        # a later `git fetch origin` mirrors the upstream's full ref",
-        # graph (heads, tags, notes) into the bare repo at canonical",
-        # paths. It does NOT set remote.origin.mirror=true, so an",
-        # explicit `git push origin <ref>:<ref>` still pushes one ref.",
+        # --mirror=fetch sets remote.origin.fetch = +refs/*:refs/* so a later
+        # `git fetch origin` mirrors the upstream's full ref graph (heads,
+        # tags, notes) into the bare repo at canonical paths. It does NOT set
+        # remote.origin.mirror=true, so an explicit `git push origin
+        # <ref>:<ref>` still pushes one ref.
         "    git -C \"$repo\" remote add --mirror=fetch origin \"$upstream_url\"",
         "  fi",
         "  git -C \"$repo\" config git-gate.identityFile \"$keyfile\"",
@@ -170,9 +167,19 @@ def git_gate_render_entrypoint(upstreams: tuple[GitGateUpstream, ...]) -> str:
         "  git -C \"$repo\" config http.receivepack true",
         "  install -m 755 /etc/git-gate/pre-receive \"$repo/hooks/pre-receive\"",
         "}",
-        "",
-        "mkdir -p /git",
     ]
+
+
+def git_gate_render_entrypoint(upstreams: tuple[GitGateUpstream, ...]) -> str:
+    """Posix-sh entrypoint. One `init_repo` call per upstream, then
+    `exec git daemon`. The function reads
+    `/git-gate/creds/<name>-{key,known_hosts}` (bind-mounted into
+    the bundle by the renderer) and wires them into each bare repo's
+    config; the access-hook + pre-receive hook pick those paths up
+    at fetch / push time."""
+    lines = ["#!/bin/sh", "set -eu", ""]
+    lines += _git_gate_init_repo_fn("/git", "/git-gate/creds")
+    lines += ["", "mkdir -p /git"]
     for u in upstreams:
         lines.append(f"init_repo {shlex.quote(u.name)} {shlex.quote(u.upstream_url)}")
     lines.extend([
@@ -190,6 +197,35 @@ def git_gate_render_entrypoint(upstreams: tuple[GitGateUpstream, ...]) -> str:
     return "\n".join(lines) + "\n"
 
 
+# A bottle id namespaces the consolidated gateway's repo + creds dirs; it is
+# embedded unquoted in the provisioning script, so restrict it to a shell- and
+# path-safe alphabet (registry ids are token_hex — this is defense in depth).
+_SAFE_BOTTLE_ID = re.compile(r"[A-Za-z0-9_-]+")
+
+
+def git_gate_render_provision(
+    bottle_id: str, upstreams: tuple[GitGateUpstream, ...],
+) -> str:
+    """Posix-sh script that provisions ONE bottle's bare repos into the
+    consolidated gateway (PRD 0070), under `/git/<bottle_id>/` with creds
+    read from `/git-gate/creds/<bottle_id>/`. Init-only — no `git daemon`,
+    since the shared gateway already serves every bottle; run inside the
+    running gateway when the bottle is registered.
+
+    Isolating each bottle's repo root and creds dir by id is what keeps one
+    bottle's push credentials out of another's repos on the shared gateway."""
+    if not _SAFE_BOTTLE_ID.fullmatch(bottle_id):
+        raise ValueError(f"git-gate: unsafe bottle id {bottle_id!r}")
+    repo_root = f"/git/{bottle_id}"
+    creds_dir = f"/git-gate/creds/{bottle_id}"
+    lines = ["#!/bin/sh", "set -eu", ""]
+    lines += _git_gate_init_repo_fn(repo_root, creds_dir)
+    lines += ["", f"mkdir -p {shlex.quote(repo_root)}"]
+    for u in upstreams:
+        lines.append(f"init_repo {shlex.quote(u.name)} {shlex.quote(u.upstream_url)}")
+    return "\n".join(lines) + "\n"
+
+
 def git_gate_render_hook() -> str:
     """The shared pre-receive hook: gitleaks-scan all incoming refs,
     then forward each accepted ref to the real upstream (`origin`)

bot_bottle/git_http_backend.py

@@ -188,6 +188,14 @@ class GitHttpHandler(BaseHTTPRequestHandler):
             "SERVER_PORT": str(self.server.server_port),  # type: ignore
             "SERVER_PROTOCOL": self.request_version,
         })
+        # Consolidated mode: attribute the gitleaks-allow supervise proposal
+        # (written by receive-pack's pre-receive hook, a child of the CGI we
+        # spawn below) to the calling bottle. The namespaced root is
+        # `<base>/<bottle_id>`, so its final component is the bottle id — the
+        # same per-bottle key egress uses. Single-tenant leaves the hook's
+        # container-stamped SUPERVISE_BOTTLE_SLUG untouched.
+        if getattr(self.server, "policy_resolver", None) is not None:
+            env["SUPERVISE_BOTTLE_SLUG"] = sandbox_root.name
         for header, variable in (
             ("accept", "HTTP_ACCEPT"),
             ("content-encoding", "HTTP_CONTENT_ENCODING"),

bot_bottle/orchestrator/client.py

@@ -0,0 +1,140 @@
+"""Host-side control-plane client (PRD 0070).
+
+The launch path talks to the orchestrator over its HTTP control plane to
+register, re-policy, and tear down bottles — the counterpart to the
+gateway-side `PolicyResolver` (which only reads `/resolve`). Where
+`PolicyResolver` is fail-closed and lives in the untrusted data plane, this
+is the trusted control-plane caller: a non-success response is an error the
+launch path must surface, not silently swallow.
+
+Stdlib-only, so the CLI can drive the orchestrator without any dependency.
+"""
+
+from __future__ import annotations
+
+import json
+import urllib.error
+import urllib.request
+from dataclasses import dataclass
+
+DEFAULT_TIMEOUT_SECONDS = 5.0
+
+
+class OrchestratorClientError(RuntimeError):
+    """A control-plane call failed (unreachable, or an unexpected status)."""
+
+
+@dataclass(frozen=True)
+class RegisteredBottle:
+    """What `POST /bottles` returns: the minted bottle id and the per-bottle
+    identity token the agent presents for app-layer attribution."""
+
+    bottle_id: str
+    identity_token: str
+
+
+class OrchestratorClient:
+    """Trusted host-side client for the orchestrator control plane."""
+
+    def __init__(self, base_url: str, *, timeout: float = DEFAULT_TIMEOUT_SECONDS) -> None:
+        self._base = base_url.rstrip("/")
+        self._timeout = timeout
+
+    def _request(
+        self, method: str, path: str, body: dict[str, object] | None = None,
+    ) -> tuple[int, dict[str, object]]:
+        """Send one request; return `(status, payload)`. Raises
+        `OrchestratorClientError` only when the orchestrator can't be reached
+        or returns malformed data — HTTP *status* codes are returned so
+        callers can treat 404 as a meaningful "no such bottle"."""
+        data = json.dumps(body).encode() if body is not None else None
+        headers = {"Content-Type": "application/json"} if data is not None else {}
+        req = urllib.request.Request(
+            f"{self._base}{path}", data=data, method=method, headers=headers,
+        )
+        try:
+            with urllib.request.urlopen(req, timeout=self._timeout) as resp:
+                raw = resp.read()
+                payload = json.loads(raw) if raw else {}
+                return resp.status, payload if isinstance(payload, dict) else {}
+        except urllib.error.HTTPError as e:
+            # A structured error response still carries a usable status.
+            try:
+                payload = json.loads(e.read() or b"{}")
+            except (ValueError, OSError):
+                payload = {}
+            return e.code, payload if isinstance(payload, dict) else {}
+        except (urllib.error.URLError, TimeoutError, OSError, ValueError) as e:
+            raise OrchestratorClientError(f"{method} {path}: {e}") from e
+
+    def _ok(self, method: str, path: str, body: dict[str, object] | None = None) -> dict[str, object]:
+        """`_request` that requires a 2xx, raising otherwise."""
+        status, payload = self._request(method, path, body)
+        if not 200 <= status < 300:
+            detail = payload.get("error", "")
+            raise OrchestratorClientError(f"{method} {path}: HTTP {status} {detail}".rstrip())
+        return payload
+
+    def health(self) -> bool:
+        """True iff the control plane answers `GET /health` with 200."""
+        try:
+            status, _ = self._request("GET", "/health")
+        except OrchestratorClientError:
+            return False
+        return status == 200
+
+    def register_bottle(
+        self,
+        source_ip: str,
+        *,
+        image_ref: str = "",
+        metadata: str = "",
+        policy: str = "",
+    ) -> RegisteredBottle:
+        """Register a bottle and broker its launch (`POST /bottles`). Returns
+        its minted id + identity token."""
+        payload = self._ok("POST", "/bottles", {
+            "source_ip": source_ip,
+            "image_ref": image_ref,
+            "metadata": metadata,
+            "policy": policy,
+        })
+        bottle_id = payload.get("bottle_id")
+        token = payload.get("identity_token")
+        if not isinstance(bottle_id, str) or not isinstance(token, str):
+            raise OrchestratorClientError("register: response missing bottle_id/identity_token")
+        return RegisteredBottle(bottle_id=bottle_id, identity_token=token)
+
+    def teardown_bottle(self, bottle_id: str) -> bool:
+        """Tear a bottle down (`DELETE /bottles/<id>`). False if the
+        orchestrator didn't know it (404) — idempotent for cleanup paths."""
+        status, _ = self._request("DELETE", f"/bottles/{bottle_id}")
+        if status == 404:
+            return False
+        if not 200 <= status < 300:
+            raise OrchestratorClientError(f"teardown {bottle_id}: HTTP {status}")
+        return True
+
+    def set_policy(self, bottle_id: str, policy: str) -> bool:
+        """Live-reload a bottle's policy (`PUT /bottles/<id>/policy`). False on
+        404 (unknown bottle)."""
+        status, _ = self._request("PUT", f"/bottles/{bottle_id}/policy", {"policy": policy})
+        if status == 404:
+            return False
+        if not 200 <= status < 300:
+            raise OrchestratorClientError(f"set_policy {bottle_id}: HTTP {status}")
+        return True
+
+    def list_bottles(self) -> list[dict[str, object]]:
+        """Every registered bottle's redacted record (`GET /bottles`)."""
+        payload = self._ok("GET", "/bottles")
+        bottles = payload.get("bottles")
+        return bottles if isinstance(bottles, list) else []
+
+
+__all__ = [
+    "OrchestratorClient",
+    "OrchestratorClientError",
+    "RegisteredBottle",
+    "DEFAULT_TIMEOUT_SECONDS",
+]

bot_bottle/orchestrator/gateway.py

@@ -25,6 +25,11 @@ from ..docker_cmd import run_docker
 
 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.
+# Agents attach here with a pinned IP and reach the gateway's egress /
+# git-http / supervise ports by its address — no host port publishing, and
+# the source IP the gateway attributes by is the address on this network.
+GATEWAY_NETWORK = "bot-bottle-gateway"
 
 # The real sidecar-bundle image + its Dockerfile. Kept as a local constant
 # rather than imported from backend.docker.sidecar_bundle, which would drag
@@ -77,11 +82,13 @@ class DockerGateway(Gateway):
         image_ref: str = GATEWAY_IMAGE,
         *,
         name: str = GATEWAY_NAME,
+        network: str = GATEWAY_NETWORK,
         build_context: Path | None = None,
         dockerfile: str | None = GATEWAY_DOCKERFILE,
     ) -> None:
         self.image_ref = image_ref
         self.name = name
+        self.network = network
         self._build_context = build_context or _REPO_ROOT
         self._dockerfile = dockerfile
 
@@ -112,9 +119,22 @@ class DockerGateway(Gateway):
         ])
         return self.name in proc.stdout.split()
 
+    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').
+        Docker picks the subnet; the launcher reads it back to allocate IPs."""
+        if run_docker(["docker", "network", "inspect", self.network]).returncode == 0:
+            return
+        proc = run_docker(["docker", "network", "create", self.network])
+        if proc.returncode != 0 and "already exists" not in proc.stderr:
+            raise GatewayError(
+                f"gateway network {self.network} failed to create: {proc.stderr.strip()}"
+            )
+
     def ensure_running(self) -> None:
         if self.is_running():
             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.
         run_docker(["docker", "rm", "--force", self.name])
@@ -122,6 +142,7 @@ class DockerGateway(Gateway):
             "docker", "run", "--detach",
             "--name", self.name,
             "--label", GATEWAY_LABEL,
+            "--network", self.network,
             self.image_ref,
         ])
         if proc.returncode != 0:
@@ -135,5 +156,5 @@ class DockerGateway(Gateway):
 
 __all__ = [
     "Gateway", "DockerGateway", "GatewayError",
-    "GATEWAY_NAME", "GATEWAY_LABEL", "GATEWAY_IMAGE",
+    "GATEWAY_NAME", "GATEWAY_LABEL", "GATEWAY_IMAGE", "GATEWAY_NETWORK",
 ]

bot_bottle/orchestrator/lifecycle.py

@@ -0,0 +1,141 @@
+"""Orchestrator process 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.
+
+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.
+"""
+
+from __future__ import annotations
+
+import subprocess
+import sys
+import time
+import urllib.error
+import urllib.request
+
+from .. import log
+from ..paths import 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."""
+
+
+class OrchestratorProcess:
+    """Manages the local orchestrator control-plane process for the docker
+    backend. Backend-neutral callers only need `ensure_running()` + `url`."""
+
+    def __init__(
+        self,
+        host: str = DEFAULT_HOST,
+        port: int = DEFAULT_PORT,
+        *,
+        broker: str = "docker",
+        gateway: bool = True,
+    ) -> None:
+        self.host = host
+        self.port = port
+        self._broker = broker
+        self._gateway = gateway
+
+    @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}"
+
+    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 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()
+        deadline = time.monotonic() + startup_timeout
+        while time.monotonic() < deadline:
+            if self.is_healthy():
+                log.info("orchestrator healthy", context={"url": self.url})
+                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"
+        )
+
+    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()
+
+
+__all__ = [
+    "OrchestratorProcess",
+    "OrchestratorStartError",
+    "DEFAULT_HOST",
+    "DEFAULT_PORT",
+    "DEFAULT_STARTUP_TIMEOUT_SECONDS",
+]

bot_bottle/orchestrator/registration.py

@@ -0,0 +1,57 @@
+"""Consolidated registration inputs (PRD 0070, docker slice).
+
+Bridges the existing per-bottle `prepare` output to the consolidated
+registry: turns a prepared bottle's egress plan into the backend-neutral
+inputs `Orchestrator.launch_bottle` takes — the egress **policy** blob and
+launch **metadata**.
+
+The policy blob is the exact routes YAML the per-bottle egress sidecar used
+to read from a file; in the consolidated model the multi-tenant gateway's
+`PolicyResolver` fetches it from the registry per request (keyed by source
+IP) instead. Same render, so consolidated and single-tenant egress apply
+byte-identical policy — a bottle's allow-list doesn't change when it moves
+onto the shared gateway.
+
+Host-side glue (imports `bot_bottle.egress`), used by the launch path — not
+by the lean orchestrator control-plane process itself.
+"""
+
+from __future__ import annotations
+
+import json
+from dataclasses import dataclass
+
+from ..egress import EgressPlan, egress_render_routes
+
+
+@dataclass(frozen=True)
+class RegistrationInputs:
+    """What `Orchestrator.launch_bottle` needs to register a bottle, derived
+    from its prepared plan. `policy` is served verbatim by the gateway's
+    `/resolve`; `metadata` is opaque forward-compat state — it carries the
+    human slug so the console / supervise can show a name, not just the
+    minted bottle id."""
+
+    policy: str
+    metadata: str
+
+
+def egress_policy(plan: EgressPlan) -> str:
+    """The bottle's egress policy blob: the routes YAML the gateway serves
+    and the addon parses with `load_config`. Identical to the per-bottle
+    `routes.yaml` render, so the consolidated path applies the same
+    allow-list."""
+    return egress_render_routes(plan.routes, log=plan.log)
+
+
+def registration_inputs(plan: EgressPlan) -> RegistrationInputs:
+    """Assemble the orchestrator registration inputs from a prepared egress
+    plan. `metadata` records the slug so the shared registry can map a minted
+    bottle id back to its human name."""
+    return RegistrationInputs(
+        policy=egress_policy(plan),
+        metadata=json.dumps({"slug": plan.slug}),
+    )
+
+
+__all__ = ["RegistrationInputs", "egress_policy", "registration_inputs"]

bot_bottle/supervise_server.py

@@ -15,6 +15,12 @@ The bottle slug arrives via SUPERVISE_BOTTLE_SLUG env (stamped at
 container creation by the backend's start step). SUPERVISE_DB_PATH
 points at the bind-mounted host database.
 
+Consolidated (PRD 0070): when BOT_BOTTLE_ORCHESTRATOR_URL is set, one
+shared server fronts every bottle and attributes each proposal to the
+calling bottle by source IP (resolved from the orchestrator) instead of a
+fixed slug — an unattributed source fails closed. Unset → the legacy
+per-bottle single-tenant server, unchanged.
+
 Speaks MCP over HTTP+JSON-RPC. Methods handled:
 
   * `initialize`           — handshake; returns server info + caps.
@@ -40,16 +46,18 @@ import time
 import typing
 import urllib.error
 import urllib.request
-from dataclasses import dataclass
+from dataclasses import dataclass, replace
 
 try:
     # Same-directory imports inside the bundle container; these files are
     # COPYed flat under /app by Dockerfile.sidecars.
     from egress_addon_core import LOG_OFF, load_config
+    from policy_resolver import PolicyResolveError, PolicyResolver
     import supervise as _sv
 except ModuleNotFoundError:
     # Package imports for host-side tests and tooling.
     from .egress_addon_core import LOG_OFF, load_config
+    from .policy_resolver import PolicyResolveError, PolicyResolver
     from . import supervise as _sv
 
 
@@ -73,6 +81,12 @@ DEFAULT_RESPONSE_TIMEOUT_SECONDS = 30.0
 MIN_RESPONSE_POLL_INTERVAL_SECONDS = 0.05
 EGRESS_LIST_TIMEOUT_SECONDS = 5.0
 
+# Consolidated (multi-tenant) mode: when set, one shared supervise server
+# fronts every bottle and attributes each proposal to the calling bottle by
+# source IP (resolved from the orchestrator), instead of a single
+# SUPERVISE_BOTTLE_SLUG env. Unset → legacy per-bottle single-tenant.
+ORCHESTRATOR_URL_ENV = "BOT_BOTTLE_ORCHESTRATOR_URL"
+
 
 @dataclass(frozen=True)
 class JsonRpcRequest:
@@ -511,9 +525,30 @@ class MCPHandler(http.server.BaseHTTPRequestHandler):
         if method == "tools/list":
             return handle_tools_list(req.params)
         if method == "tools/call":
-            return handle_tools_call(req.params, config)
+            # Attribute the proposal to the calling bottle. Single-tenant → the
+            # env slug on `config`; consolidated → the source-IP-resolved
+            # bottle id, so one shared server queues each bottle's proposal
+            # under its own slug.
+            return handle_tools_call(req.params, self._attributed_config(config))
         raise _RpcClientError(ERR_METHOD_NOT_FOUND, f"method not found: {method}")
 
+    def _attributed_config(self, config: ServerConfig) -> ServerConfig:
+        """The ServerConfig with `bottle_slug` bound to *this request's* bottle.
+        Single-tenant (no resolver): unchanged. Consolidated: the bottle id
+        attributed from the source IP — **fail-closed**, an unattributed or
+        unreachable source raises so no proposal is queued under the wrong (or
+        empty) slug."""
+        resolver = getattr(self.server, "policy_resolver", None)
+        if resolver is None:
+            return config
+        try:
+            bottle_id = resolver.resolve_bottle_id(self.client_address[0])
+        except PolicyResolveError as e:
+            raise _RpcInternalError(f"orchestrator unreachable, cannot attribute: {e}") from e
+        if not bottle_id:
+            raise _RpcInternalError("request source is not attributed to a bottle")
+        return replace(config, bottle_slug=bottle_id)
+
     def _write_jsonrpc(self, body: bytes) -> None:
         self.send_response(200)
         self.send_header("Content-Type", "application/json")
@@ -537,6 +572,9 @@ class MCPServer(socketserver.ThreadingMixIn, http.server.HTTPServer):
     allow_reuse_address = True
     daemon_threads = True
     config: ServerConfig = ServerConfig(bottle_slug="")
+    # None → single-tenant (proposals use config.bottle_slug); set → consolidated
+    # (each proposal attributed to the source-IP-resolved bottle).
+    policy_resolver: "PolicyResolver | None" = None
 
 
 # --- Entry point -----------------------------------------------------------
@@ -548,15 +586,17 @@ def serve(
     port: int = _sv.SUPERVISE_PORT,
     bind: str = "0.0.0.0",
     response_timeout_seconds: float = DEFAULT_RESPONSE_TIMEOUT_SECONDS,
+    resolver: "PolicyResolver | None" = None,
 ) -> typing.NoReturn:
     server = MCPServer((bind, port), MCPHandler)
     server.config = ServerConfig(
         bottle_slug=bottle_slug,
         response_timeout_seconds=response_timeout_seconds,
     )
+    server.policy_resolver = resolver
+    mode = "multi-tenant" if resolver else f"slug={bottle_slug!r}"
     sys.stderr.write(
-        f"supervise listening on {bind}:{port}; "
-        f"slug={bottle_slug!r}; "
+        f"supervise listening on {bind}:{port}; {mode}; "
         f"tools: {', '.join(t['name'] for t in TOOL_DEFINITIONS)}\n"  # type: ignore[arg-type]
     )
     sys.stderr.flush()
@@ -571,8 +611,12 @@ def serve(
 
 def main(argv: list[str]) -> int:
     del argv  # config is env-only, no CLI flags
+    orch_url = os.environ.get(ORCHESTRATOR_URL_ENV, "").strip()
+    resolver = PolicyResolver(orch_url) if orch_url else None
     bottle_slug = os.environ.get("SUPERVISE_BOTTLE_SLUG", "")
-    if not bottle_slug:
+    # Consolidated mode resolves the slug per request, so the env slug is
+    # optional there; single-tenant still requires it.
+    if not bottle_slug and resolver is None:
         sys.stderr.write("supervise: SUPERVISE_BOTTLE_SLUG env is unset\n")
         return 2
     port = int(os.environ.get("SUPERVISE_PORT", str(_sv.SUPERVISE_PORT)))
@@ -587,6 +631,7 @@ def main(argv: list[str]) -> int:
         port=port,
         bind=bind,
         response_timeout_seconds=response_timeout_seconds,
+        resolver=resolver,
     )
     return 0  # serve() does not return
 

tests/unit/test_gateway_net.py

@@ -0,0 +1,42 @@
+"""Unit: shared-gateway source-IP allocation (PRD 0070)."""
+
+from __future__ import annotations
+
+import unittest
+
+from bot_bottle.backend.docker.gateway_net import NoFreeAddressError, next_free_ip
+
+
+class TestNextFreeIp(unittest.TestCase):
+    def test_skips_router_and_returns_first_host(self) -> None:
+        # .1 is docker's router; the first assignable address is .2.
+        self.assertEqual("172.18.0.2", next_free_ip("172.18.0.0/16", []))
+
+    def test_skips_taken_addresses(self) -> None:
+        # Gateway container holds .2, a live bottle holds .3 -> next is .4.
+        self.assertEqual(
+            "172.18.0.4", next_free_ip("172.18.0.0/16", ["172.18.0.2", "172.18.0.3"]),
+        )
+
+    def test_taken_order_does_not_matter(self) -> None:
+        self.assertEqual(
+            "172.18.0.2", next_free_ip("172.18.0.0/16", ["172.18.0.3", "172.18.0.5"]),
+        )
+
+    def test_allocation_is_deterministic(self) -> None:
+        taken = ["172.18.0.2"]
+        self.assertEqual(next_free_ip("172.18.0.0/16", taken),
+                         next_free_ip("172.18.0.0/16", taken))
+
+    def test_raises_when_subnet_exhausted(self) -> None:
+        # /30: hosts are .1 (router, reserved) and .2; taking .2 leaves none.
+        with self.assertRaises(NoFreeAddressError):
+            next_free_ip("10.9.9.0/30", ["10.9.9.2"])
+
+    def test_accepts_host_bits_set_cidr(self) -> None:
+        # A container's inspected address arrives as e.g. 172.18.0.2/16.
+        self.assertEqual("172.18.0.2", next_free_ip("172.18.0.5/16", []))
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/unit/test_git_gate_provision_render.py

@@ -0,0 +1,67 @@
+"""Unit: consolidated per-bottle git-gate provisioning render (PRD 0070)."""
+
+from __future__ import annotations
+
+import unittest
+
+from bot_bottle.git_gate_render import (
+    GitGateUpstream,
+    git_gate_render_entrypoint,
+    git_gate_render_provision,
+)
+
+
+def _ups(*names: str) -> tuple[GitGateUpstream, ...]:
+    return tuple(
+        GitGateUpstream(
+            name=n,
+            upstream_url=f"ssh://git@github.com/x/{n}.git",
+            upstream_host="github.com",
+            upstream_port="22",
+            identity_file="",
+            known_host_key="",
+        )
+        for n in names
+    )
+
+
+class TestProvisionRender(unittest.TestCase):
+    def test_namespaces_repos_and_creds_by_bottle_id(self) -> None:
+        script = git_gate_render_provision("bottleab12", _ups("foo"))
+        self.assertIn("repo=/git/bottleab12/${name}.git", script)
+        self.assertIn("keyfile=/git-gate/creds/bottleab12/${name}-key", script)
+        self.assertIn("mkdir -p /git/bottleab12", script)
+
+    def test_one_init_repo_call_per_upstream(self) -> None:
+        script = git_gate_render_provision("b1", _ups("foo", "bar"))
+        calls = [l for l in script.splitlines() if l.startswith("init_repo ")]
+        self.assertEqual(2, len(calls))
+
+    def test_provision_does_not_start_the_daemon(self) -> None:
+        # The shared gateway already serves; provisioning is init-only.
+        self.assertNotIn("git daemon", git_gate_render_provision("b1", _ups("foo")))
+
+    def test_installs_pre_receive_hook(self) -> None:
+        script = git_gate_render_provision("b1", _ups("foo"))
+        self.assertIn("install -m 755 /etc/git-gate/pre-receive", script)
+
+    def test_rejects_unsafe_bottle_id(self) -> None:
+        for bad in ("../etc", "a/b", "a b", "a;rm", ""):
+            with self.assertRaises(ValueError):
+                git_gate_render_provision(bad, _ups("foo"))
+
+
+class TestEntrypointUnchanged(unittest.TestCase):
+    """The shared `_git_gate_init_repo_fn` refactor must not alter the
+    single-tenant daemon entrypoint's output."""
+
+    def test_entrypoint_still_single_tenant_flat(self) -> None:
+        script = git_gate_render_entrypoint(_ups("foo"))
+        self.assertIn("repo=/git/${name}.git", script)      # flat, not namespaced
+        self.assertIn("keyfile=/git-gate/creds/${name}-key", script)
+        self.assertIn("--base-path=/git", script)
+        self.assertIn("exec git daemon", script)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/unit/test_git_http_backend.py

@@ -13,6 +13,17 @@ from bot_bottle.git_gate import GIT_GATE_TIMEOUT_SECS
 from bot_bottle.git_http_backend import GitHttpHandler, MAX_BODY_BYTES
 
 
+class _FixedResolver:
+    """Maps every source IP to one bottle id (consolidated-mode stub)."""
+
+    def __init__(self, bottle_id: str) -> None:
+        self._bottle_id = bottle_id
+
+    def resolve_bottle_id(self, source_ip: str, identity_token: str = "") -> str:
+        del source_ip, identity_token
+        return self._bottle_id
+
+
 class TestGitHttpBackend(unittest.TestCase):
     def test_real_git_push_reaches_bare_repo(self):
         from http.server import ThreadingHTTPServer
@@ -94,6 +105,62 @@ class TestGitHttpBackend(unittest.TestCase):
             ).strip()
             self.assertEqual(head, cloned)
 
+    def test_consolidated_push_stamps_bottle_slug_for_the_hook(self):
+        # In consolidated mode the backend attributes the push by source IP and
+        # stamps SUPERVISE_BOTTLE_SLUG=<bottle_id> into the CGI env, so the
+        # gitleaks-allow pre-receive hook queues its proposal under the right
+        # bottle. The hook here just records what it received.
+        from http.server import ThreadingHTTPServer
+
+        bottle_id = "bottleab12"
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            bare = root / bottle_id / "repo.git"  # namespaced per bottle (slice 10)
+            bare.parent.mkdir(parents=True)
+            subprocess.run(["git", "init", "--bare", str(bare)],
+                           check=True, capture_output=True, text=True)
+            subprocess.run(
+                ["git", "-C", str(bare), "config", "http.receivepack", "true"],
+                check=True,
+            )
+            capture = root / "slug-capture"
+            hook = bare / "hooks" / "pre-receive"
+            hook.write_text(
+                f"#!/bin/sh\nprintf '%s' \"${{SUPERVISE_BOTTLE_SLUG:-UNSET}}\" > "
+                f"{capture}\ncat >/dev/null\nexit 0\n"
+            )
+            hook.chmod(0o755)
+
+            old_root = os.environ.get("GIT_PROJECT_ROOT")
+            os.environ["GIT_PROJECT_ROOT"] = str(root)  # base; backend nests per bottle
+            self.addCleanup(self._restore_env, old_root)
+
+            server = ThreadingHTTPServer(("127.0.0.1", 0), GitHttpHandler)
+            server.policy_resolver = _FixedResolver(bottle_id)  # type: ignore[attr-defined]
+            thread = threading.Thread(target=server.serve_forever, daemon=True)
+            thread.start()
+            self.addCleanup(server.shutdown)
+            self.addCleanup(server.server_close)
+
+            work = root / "work"
+            work.mkdir()
+            subprocess.run(["git", "init"], cwd=work, check=True,
+                           capture_output=True, text=True)
+            subprocess.run(["git", "config", "user.name", "test"], cwd=work, check=True)
+            subprocess.run(["git", "config", "user.email", "t@example.invalid"],
+                           cwd=work, check=True)
+            (work / "README.md").write_text("test\n")
+            subprocess.run(["git", "add", "README.md"], cwd=work, check=True)
+            subprocess.run(["git", "commit", "-m", "init"], cwd=work,
+                           check=True, capture_output=True, text=True)
+
+            url = f"http://127.0.0.1:{server.server_port}/repo.git"
+            subprocess.run(
+                ["git", "push", url, "HEAD:refs/heads/main"],
+                cwd=work, check=True, capture_output=True, text=True, timeout=5,
+            )
+            self.assertEqual(bottle_id, capture.read_text())
+
     def test_post_forwards_git_cgi_headers(self):
         from http.server import ThreadingHTTPServer
 

tests/unit/test_orchestrator_client.py

@@ -0,0 +1,106 @@
+"""Unit: host-side orchestrator control-plane client (PRD 0070). HTTP mocked."""
+
+from __future__ import annotations
+
+import json
+import unittest
+import urllib.error
+from unittest.mock import MagicMock, patch
+
+from bot_bottle.orchestrator.client import (
+    OrchestratorClient,
+    OrchestratorClientError,
+    RegisteredBottle,
+)
+
+_URLOPEN = "bot_bottle.orchestrator.client.urllib.request.urlopen"
+
+
+def _resp(status: int, payload: object) -> MagicMock:
+    m = MagicMock()
+    inner = m.__enter__.return_value
+    inner.status = status
+    inner.read.return_value = json.dumps(payload).encode()
+    return m
+
+
+def _http_error(code: int, payload: object = None) -> urllib.error.HTTPError:
+    del payload  # the client tolerates an empty error body; keep the signature
+    return urllib.error.HTTPError("http://x", code, "err", {}, None)  # type: ignore[arg-type]
+
+
+class TestRegister(unittest.TestCase):
+    def setUp(self) -> None:
+        self.c = OrchestratorClient("http://orch:8080")
+
+    def test_register_returns_id_and_token(self) -> None:
+        with patch(_URLOPEN, return_value=_resp(201, {"bottle_id": "b1", "identity_token": "tok"})):
+            got = self.c.register_bottle("10.0.0.2", policy="routes: []\n", metadata="{}")
+        self.assertEqual(RegisteredBottle("b1", "tok"), got)
+
+    def test_register_posts_source_ip_and_policy(self) -> None:
+        with patch(_URLOPEN, return_value=_resp(201, {"bottle_id": "b", "identity_token": "t"})) as m:
+            self.c.register_bottle("10.0.0.9", policy="P", metadata="M", image_ref="img")
+        sent = json.loads(m.call_args.args[0].data)
+        self.assertEqual("10.0.0.9", sent["source_ip"])
+        self.assertEqual("P", sent["policy"])
+        self.assertEqual("img", sent["image_ref"])
+
+    def test_register_missing_fields_raises(self) -> None:
+        with patch(_URLOPEN, return_value=_resp(201, {"bottle_id": "b1"})):
+            with self.assertRaises(OrchestratorClientError):
+                self.c.register_bottle("10.0.0.2")
+
+    def test_register_non_2xx_raises(self) -> None:
+        with patch(_URLOPEN, side_effect=_http_error(400, {"error": "bad"})):
+            with self.assertRaises(OrchestratorClientError):
+                self.c.register_bottle("")
+
+
+class TestTeardown(unittest.TestCase):
+    def setUp(self) -> None:
+        self.c = OrchestratorClient("http://orch:8080")
+
+    def test_teardown_true_on_success(self) -> None:
+        with patch(_URLOPEN, return_value=_resp(200, {"torn_down": True})):
+            self.assertTrue(self.c.teardown_bottle("b1"))
+
+    def test_teardown_false_on_404(self) -> None:
+        # Idempotent: an already-gone bottle is a clean no-op.
+        with patch(_URLOPEN, side_effect=_http_error(404)):
+            self.assertFalse(self.c.teardown_bottle("gone"))
+
+    def test_teardown_uses_delete(self) -> None:
+        with patch(_URLOPEN, return_value=_resp(200, {"torn_down": True})) as m:
+            self.c.teardown_bottle("b1")
+        self.assertEqual("DELETE", m.call_args.args[0].get_method())
+
+
+class TestHealthAndPolicy(unittest.TestCase):
+    def setUp(self) -> None:
+        self.c = OrchestratorClient("http://orch:8080")
+
+    def test_health_true_on_200(self) -> None:
+        with patch(_URLOPEN, return_value=_resp(200, {"status": "ok"})):
+            self.assertTrue(self.c.health())
+
+    def test_health_false_when_unreachable(self) -> None:
+        with patch(_URLOPEN, side_effect=urllib.error.URLError("refused")):
+            self.assertFalse(self.c.health())
+
+    def test_set_policy_false_on_404(self) -> None:
+        with patch(_URLOPEN, side_effect=_http_error(404)):
+            self.assertFalse(self.c.set_policy("gone", "P"))
+
+    def test_set_policy_true_on_success(self) -> None:
+        with patch(_URLOPEN, return_value=_resp(200, {"updated": True})):
+            self.assertTrue(self.c.set_policy("b1", "P"))
+
+    def test_unreachable_raises(self) -> None:
+        with patch(_URLOPEN, side_effect=urllib.error.URLError("refused")):
+            with self.assertRaises(OrchestratorClientError):
+                self.c.register_bottle("10.0.0.2")
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/unit/test_orchestrator_gateway.py

@@ -51,6 +51,33 @@ class TestDockerGateway(unittest.TestCase):
         self.assertEqual(1, len(runs))
         self.assertIn(self.sc.name, runs[0])
         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])
+
+    def test_ensure_running_creates_network_when_missing(self) -> None:
+        calls: list[list[str]] = []
+
+        def fake(argv: list[str]) -> Mock:
+            calls.append(argv)
+            if argv[:3] == ["docker", "network", "inspect"]:
+                return _proc(returncode=1, stderr="No such network")
+            return _proc(stdout="") if argv[:2] == ["docker", "ps"] else _proc()
+
+        with patch(_RUN_DOCKER, side_effect=fake):
+            self.sc.ensure_running()
+        creates = [c for c in calls if c[:3] == ["docker", "network", "create"]]
+        self.assertEqual([["docker", "network", "create", self.sc.network]], creates)
+
+    def test_ensure_running_reuses_existing_network(self) -> None:
+        calls: list[list[str]] = []
+
+        def fake(argv: list[str]) -> Mock:
+            calls.append(argv)
+            return _proc(stdout="") if argv[:2] == ["docker", "ps"] else _proc()
+
+        with patch(_RUN_DOCKER, side_effect=fake):
+            self.sc.ensure_running()  # network inspect returns 0 → exists
+        self.assertEqual([], [c for c in calls if c[:3] == ["docker", "network", "create"]])
 
     def test_ensure_running_raises_on_docker_failure(self) -> None:
         def fake(argv: list[str]) -> Mock:

tests/unit/test_orchestrator_lifecycle.py

@@ -0,0 +1,87 @@
+"""Unit: orchestrator process lifecycle — idempotent singleton (PRD 0070)."""
+
+from __future__ import annotations
+
+import tempfile
+import unittest
+import urllib.error
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+from bot_bottle.orchestrator.lifecycle import (
+    OrchestratorProcess,
+    OrchestratorStartError,
+)
+from tests.unit import use_bottle_root
+
+_URLOPEN = "bot_bottle.orchestrator.lifecycle.urllib.request.urlopen"
+_POPEN = "bot_bottle.orchestrator.lifecycle.subprocess.Popen"
+_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):
+    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)
+
+    def test_url(self) -> None:
+        self.assertEqual("http://127.0.0.1:8099", self.p.url)
+
+    def test_is_healthy_true_on_200(self) -> None:
+        with patch(_URLOPEN, return_value=_health(200)):
+            self.assertTrue(self.p.is_healthy())
+
+    def test_is_healthy_false_on_error(self) -> None:
+        with patch(_URLOPEN, side_effect=urllib.error.URLError("refused")):
+            self.assertFalse(self.p.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_spawns_then_waits_for_health(self) -> None:
+        # First check (before spawn) fails; after spawn the poll succeeds.
+        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)
+        self.assertIn("bot_bottle.orchestrator", argv)
+        self.assertEqual("docker", argv[argv.index("--broker") + 1])
+
+    def test_argv_omits_gateway_when_disabled(self) -> None:
+        self.assertNotIn("--gateway", OrchestratorProcess(gateway=False)._argv())
+
+    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())
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/unit/test_orchestrator_registration.py

@@ -0,0 +1,56 @@
+"""Unit: consolidated registration inputs — egress policy round-trip (PRD 0070)."""
+
+from __future__ import annotations
+
+import json
+import unittest
+from pathlib import Path
+
+from bot_bottle.egress import EgressPlan, EgressRoute
+from bot_bottle.egress_addon_core import LOG_BLOCKS, load_config
+from bot_bottle.orchestrator.registration import (
+    RegistrationInputs,
+    egress_policy,
+    registration_inputs,
+)
+
+
+def _plan(routes: tuple[EgressRoute, ...], *, slug: str = "demo", log: int = 0) -> EgressPlan:
+    return EgressPlan(
+        slug=slug,
+        routes_path=Path("/unused/routes.yaml"),
+        routes=routes,
+        token_env_map={},
+        log=log,
+    )
+
+
+class TestEgressPolicy(unittest.TestCase):
+    def test_policy_round_trips_through_load_config(self) -> None:
+        # The policy the gateway serves must parse back to the same allow-list
+        # the per-bottle sidecar applied — moving onto the shared gateway must
+        # not change a bottle's egress.
+        routes = (EgressRoute(host="api.example.com"), EgressRoute(host="pypi.org"))
+        cfg = load_config(egress_policy(_plan(routes)))
+        self.assertEqual(("api.example.com", "pypi.org"), tuple(r.host for r in cfg.routes))
+
+    def test_policy_preserves_log_level(self) -> None:
+        plan = _plan((EgressRoute(host="x.example.com"),), log=LOG_BLOCKS)
+        self.assertEqual(LOG_BLOCKS, load_config(egress_policy(plan)).log)
+
+    def test_empty_routes_yield_deny_all(self) -> None:
+        cfg = load_config(egress_policy(_plan(())))
+        self.assertEqual((), cfg.routes)  # no routes → default-deny
+
+
+class TestRegistrationInputs(unittest.TestCase):
+    def test_bundles_policy_and_slug_metadata(self) -> None:
+        plan = _plan((EgressRoute(host="api.example.com"),), slug="my-bot")
+        inputs = registration_inputs(plan)
+        self.assertIsInstance(inputs, RegistrationInputs)
+        self.assertEqual("my-bot", json.loads(inputs.metadata)["slug"])
+        self.assertEqual(egress_policy(plan), inputs.policy)  # same blob egress_policy renders
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/unit/test_supervise_server.py

@@ -6,6 +6,7 @@ import sys
 import tempfile
 import threading
 import time
+import types
 import unittest
 from pathlib import Path
 from unittest.mock import patch
@@ -629,5 +630,57 @@ class TestHttpEndToEnd(unittest.TestCase):
             conn.close()
 
 
+class _FakeResolver:
+    def __init__(self, bottle_id: str | None = None, raises: bool = False) -> None:
+        self._bottle_id = bottle_id
+        self._raises = raises
+        self.calls: list[str] = []
+
+    def resolve_bottle_id(self, source_ip: str, identity_token: str = "") -> str | None:
+        del identity_token
+        self.calls.append(source_ip)
+        if self._raises:
+            # Raise the exact class supervise_server catches (it imports
+            # policy_resolver flat inside the bundle, package-side in tests).
+            raise supervise_server.PolicyResolveError("orchestrator down")
+        return self._bottle_id
+
+
+def _handler(resolver: object) -> MCPHandler:
+    """A bare MCPHandler wired with a server (carrying the resolver) and a
+    client address, enough to exercise `_attributed_config` off-socket."""
+    h: MCPHandler = MCPHandler.__new__(MCPHandler)
+    h.server = types.SimpleNamespace(policy_resolver=resolver)  # type: ignore[assignment]
+    h.client_address = ("10.0.0.7", 4321)
+    return h
+
+
+class TestAttributedConfig(unittest.TestCase):
+    """Consolidated supervise: each proposal is attributed to the calling
+    bottle by source IP; single-tenant keeps the env slug (PRD 0070)."""
+
+    def test_single_tenant_keeps_env_slug(self) -> None:
+        cfg = _handler(None)._attributed_config(ServerConfig(bottle_slug="dev"))
+        self.assertEqual("dev", cfg.bottle_slug)
+
+    def test_consolidated_binds_source_ip_bottle(self) -> None:
+        r = _FakeResolver(bottle_id="bottle-x")
+        cfg = _handler(r)._attributed_config(ServerConfig(bottle_slug="ignored"))
+        self.assertEqual("bottle-x", cfg.bottle_slug)  # resolved slug wins
+        self.assertEqual(["10.0.0.7"], r.calls)
+
+    def test_unattributed_source_fails_closed(self) -> None:
+        with self.assertRaises(_RpcInternalError):
+            _handler(_FakeResolver(bottle_id=None))._attributed_config(
+                ServerConfig(bottle_slug="x")
+            )
+
+    def test_resolver_error_fails_closed(self) -> None:
+        with self.assertRaises(_RpcInternalError):
+            _handler(_FakeResolver(raises=True))._attributed_config(
+                ServerConfig(bottle_slug="x")
+            )
+
+
 if __name__ == "__main__":
     unittest.main()