feat(orchestrator): slice 13a — orchestrator process lifecycle (idempotent singleton)

First sub-slice of docker launch integration: the foundation the CLI needs
to ensure exactly one orchestrator control plane + shared gateway is up
before registering/launching bottles. Does NOT touch the real launch path
yet — it's the lifecycle primitive the cut-over slices build on.

- OrchestratorProcess.ensure_running(): idempotent singleton — returns the
  control-plane URL if a healthy one already answers /health, else spawns
  `python -m bot_bottle.orchestrator --broker docker --gateway` detached
  (start_new_session, output tee'd to <root>/orchestrator.log) and polls
  /health until healthy or timeout (OrchestratorStartError).
- 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) per the PRD's dev-harness sequencing — it
  already has the host user's docker access to broker launches; the
  data-plane gateway it manages is the container.

pyright 0 errors; pylint 9.83/10; unit suite green (1714 tests; 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/egress_addon.py

@@ -32,7 +32,7 @@ from egress_addon_core import (  # type: ignore[import-not-found]  # pylint: dis
     is_git_push_request,
     load_config,
     match_route,
-    resolve_client_config,
+    resolve_client_context,
     outbound_scan_headers,
     route_to_yaml_dict,
     scan_inbound,
@@ -99,17 +99,29 @@ class EgressAddon:
         # Absent → single-tenant (static routes file); behaviour unchanged.
         orch_url = os.environ.get(ORCHESTRATOR_URL_ENV, "").strip()
         self._resolver = PolicyResolver(orch_url) if orch_url else None
-        # Tokens the operator has approved this session (PRD 0062). In-memory
-        # only — a restart re-prompts. Mutated only from the asyncio loop that
-        # runs the addon hooks, so no lock is needed.
-        self.safe_tokens: set[str] = set()
+        # Tokens the operator has approved this session (PRD 0062), keyed by
+        # bottle so the shared gateway keeps each bottle's safelist separate —
+        # a global set would let bottle A's approved secret pass bottle B's DLP
+        # scan. In-memory only (a restart re-prompts); mutated only from the
+        # asyncio loop that runs the addon hooks, so no lock is needed.
+        self._safe_tokens: dict[str, set[str]] = {}
         self._supervise_slug = os.environ.get("SUPERVISE_BOTTLE_SLUG", "").strip()
         self._token_allow_timeout = _token_allow_timeout_from_env(os.environ)
         self._reload(initial=True)
         self._install_sighup()
 
-    def _supervise_available(self) -> bool:
-        return bool(self._supervise_slug)
+    @staticmethod
+    def _supervise_available(slug: str) -> bool:
+        """Supervise is reachable for this request iff we resolved a bottle to
+        attribute its proposals to (single-tenant env slug, or a source-IP
+        -attributed bottle id). Empty → fail closed (no queue to write to)."""
+        return bool(slug)
+
+    def _safe_tokens_for(self, slug: str) -> set[str]:
+        """This bottle's operator-approved DLP safelist (PRD 0062), created on
+        first use. Keyed by bottle so the shared gateway never leaks one
+        bottle's approved token into another's scan."""
+        return self._safe_tokens.setdefault(slug, set())
 
     def _reload(self, *, initial: bool = False) -> None:
         try:
@@ -218,19 +230,21 @@ class EgressAddon:
             + "\n"
         )
 
-    def _active_config(self, flow: http.HTTPFlow) -> Config:
-        """The Config to apply to this request. Single-tenant → the static
-        `self.config`. Consolidated → the calling bottle's Config, resolved
-        by source IP (fail-closed to deny-all if unattributed). The identity
-        token, if the agent injected one, is read then stripped so it never
-        leaks upstream."""
+    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."""
         if self._resolver is None:
-            return self.config
+            return self.config, self._supervise_slug
         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_config(self._resolver, client_ip, token)
+        return resolve_client_context(self._resolver, client_ip, token)
 
     async def request(self, flow: http.HTTPFlow) -> None:
         request_path, _, query = flow.request.path.partition("?")
@@ -239,14 +253,14 @@ class EgressAddon:
             self._serve_introspection(flow, request_path)
             return
 
-        config = self._active_config(flow)
+        config, slug = 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):
+            if not await self._handle_outbound_dlp(flow, route, slug):
                 return
             # The redact policy may have rewritten the request line; recompute
             # the path/query the git checks below rely on.
@@ -310,6 +324,7 @@ class EgressAddon:
         self,
         flow: http.HTTPFlow,
         route: Route,
+        slug: 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
@@ -331,7 +346,7 @@ class EgressAddon:
             )
             result = scan_outbound(
                 route, scan_text, os.environ,
-                safe_tokens=self.safe_tokens, crlf_text=crlf_text,
+                safe_tokens=self._safe_tokens_for(slug), crlf_text=crlf_text,
             )
             if result is None or result.severity != "block":
                 return True
@@ -366,10 +381,10 @@ class EgressAddon:
 
             # supervise (default): hold the request for operator approval.
             # Fall back to a hard 403 when supervise isn't wired for the bottle.
-            if not self._supervise_available():
+            if not self._supervise_available(slug):
                 self._block_dlp(flow, result)
                 return False
-            approved = await self._supervise_token_block(flow, request_path, result)
+            approved = await self._supervise_token_block(flow, request_path, result, slug)
             if not approved:
                 return False  # _supervise_token_block wrote the 403 response
             # loop: the approved value is now in safe_tokens; re-scan.
@@ -412,12 +427,16 @@ class EgressAddon:
         flow: http.HTTPFlow,
         request_path: str,
         result: ScanResult,
+        slug: str,
     ) -> bool:
         """Route a token DLP block to the operator's supervisor queue and wait.
 
-        Returns True if the operator approved (the matched value is added to
-        `self.safe_tokens` and the caller re-scans); False if the request must
-        be blocked (a 403 response has been written to `flow`)."""
+        `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`)."""
         host = flow.request.pretty_host
         payload = build_token_allow_payload(
             redact_tokens(host, env=os.environ),
@@ -426,7 +445,7 @@ class EgressAddon:
             result,
         )
         proposal = _sv.Proposal.new(
-            bottle_slug=self._supervise_slug,
+            bottle_slug=slug,
             tool=_sv.TOOL_EGRESS_TOKEN_ALLOW,
             proposed_file=payload,
             justification=_TOKEN_ALLOW_JUSTIFICATION,
@@ -449,13 +468,13 @@ class EgressAddon:
             **self._req_ctx(flow),
         }) + "\n")
 
-        response = await self._await_token_response(proposal.id)
-        _sv.archive_proposal(self._supervise_slug, proposal.id)
+        response = await self._await_token_response(proposal.id, slug)
+        _sv.archive_proposal(slug, proposal.id)
 
         if response is not None and response.status in (
             _sv.STATUS_APPROVED, _sv.STATUS_MODIFIED,
         ):
-            self.safe_tokens.add(result.matched)
+            self._safe_tokens_for(slug).add(result.matched)
             if self.config.log >= LOG_BLOCKS:
                 sys.stderr.write(json.dumps({
                     "event": "egress_token_allowed",
@@ -478,6 +497,7 @@ class EgressAddon:
     async def _await_token_response(
         self,
         proposal_id: str,
+        slug: str,
     ) -> "_sv.Response | None":
         """Poll the DB for the operator's response without blocking the
         proxy event loop. Returns the Response, or None on timeout."""
@@ -485,7 +505,7 @@ class EgressAddon:
         deadline = loop.time() + self._token_allow_timeout
         while True:
             try:
-                return _sv.read_response(self._supervise_slug, proposal_id)
+                return _sv.read_response(slug, proposal_id)
             except (OSError, ValueError, KeyError):
                 # Not written yet, or a partial/malformed write — retry until
                 # the deadline, then fail closed.
@@ -539,6 +559,9 @@ class EgressAddon:
         """
         if flow.websocket is None:  # type: ignore[union-attr]
             return
+        # WebSocket DLP runs against the static config only (single-tenant); in
+        # the consolidated gateway self.config has no routes, so this is inert
+        # until websocket routing is made source-IP-aware (a separate slice).
         route = match_route(self.config.routes, flow.request.pretty_host)
         if route is None:
             return
@@ -549,7 +572,7 @@ class EgressAddon:
             # not an injection vector here — scan only for credential leakage.
             result = scan_outbound(
                 route, content, os.environ,
-                safe_tokens=self.safe_tokens, crlf_text="",
+                safe_tokens=self._safe_tokens_for(self._supervise_slug), crlf_text="",
             )
             if result is not None and result.severity == "block":
                 sys.stderr.write(f"egress DLP: {result.reason}\n")

bot_bottle/egress_addon_core.py

@@ -421,6 +421,18 @@ class PolicyResolverLike(typing.Protocol):
         ...
 
 
+def _config_from_policy(policy: "str | None") -> "Config":
+    """Parse a resolved policy blob into a Config, fail-closed: None / empty /
+    unparseable all become a deny-all Config (no routes → every request
+    blocked)."""
+    if not policy:
+        return Config(routes=())  # unattributed or empty → deny-all
+    try:
+        return load_config(policy)
+    except ValueError:
+        return Config(routes=())  # unparseable policy → deny
+
+
 def resolve_client_config(
     resolver: PolicyResolverLike, client_ip: str, identity_token: str = ""
 ) -> "Config":
@@ -433,12 +445,33 @@ def resolve_client_config(
         policy = resolver.resolve(client_ip, identity_token)
     except Exception:  # noqa: BLE001  # pylint: disable=broad-exception-caught
         return Config(routes=())  # orchestrator unreachable/errored → deny
-    if not policy:
-        return Config(routes=())  # unattributed or empty → deny-all
+    return _config_from_policy(policy)
+
+
+class ContextResolverLike(typing.Protocol):
+    """The bit of `policy_resolver.PolicyResolver` `resolve_client_context`
+    needs — one round-trip returning both policy and bottle id."""
+
+    def resolve_policy_and_bottle_id(
+        self, source_ip: str, identity_token: str = ...,
+    ) -> "tuple[str | None, str | None]":
+        ...
+
+
+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 —
+    **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."""
     try:
-        return load_config(policy)
-    except ValueError:
-        return Config(routes=())  # unparseable policy → deny
+        policy, bottle_id = 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 "")
 
 
 # ---------------------------------------------------------------------------
@@ -833,7 +866,9 @@ __all__ = [
     "is_git_fetch_request",
     "load_config",
     "resolve_client_config",
+    "resolve_client_context",
     "PolicyResolverLike",
+    "ContextResolverLike",
     "match_route",
     "outbound_scan_headers",
     "parse_config",

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

@@ -100,5 +100,24 @@ class PolicyResolver:
         bottle_id = payload.get("bottle_id")
         return bottle_id if isinstance(bottle_id, str) and bottle_id else None
 
+    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."""
+        payload = self._post_resolve(source_ip, identity_token)
+        if payload is None:
+            return None, None
+        policy = payload.get("policy")
+        bottle_id = payload.get("bottle_id")
+        return (
+            policy if isinstance(policy, str) else "",
+            bottle_id if isinstance(bottle_id, str) and bottle_id else None,
+        )
+
 
 __all__ = ["PolicyResolver", "PolicyResolveError"]

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_egress_addon_log_redaction.py

@@ -46,7 +46,7 @@ def _addon() -> EgressAddon:
     """Return a bare EgressAddon with LOG_FULL config and no routes file."""
     a: EgressAddon = EgressAddon.__new__(EgressAddon)
     a.config = Config(routes=(), log=LOG_FULL)
-    a.safe_tokens = set()
+    a._safe_tokens = {}
     a._supervise_slug = ""
     a._token_allow_timeout = 300.0
     return a

tests/unit/test_egress_addon_request_flow.py

@@ -211,7 +211,7 @@ def _addon(config: Config) -> EgressAddon:
     """Bare EgressAddon with a supplied config and no supervise wiring."""
     a: EgressAddon = EgressAddon.__new__(EgressAddon)
     a.config = config
-    a.safe_tokens = set()
+    a._safe_tokens = {}
     a._supervise_slug = ""
     a._token_allow_timeout = 300.0
     a.routes_path = "/nonexistent/routes.yaml"
@@ -222,6 +222,29 @@ def _run_request(addon: EgressAddon, flow: _Flow) -> None:
     asyncio.run(addon.request(flow))  # type: ignore[arg-type]
 
 
+def _with_client_ip(flow: _Flow, ip: str) -> _Flow:
+    """Attach a mitmproxy-style client_conn so consolidated resolution can read
+    the source IP (`flow.client_conn.peername[0]`)."""
+    flow.client_conn = types.SimpleNamespace(peername=(ip, 54321))  # type: ignore[attr-defined]
+    return flow
+
+
+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:
+        self._map = ip_to_bottle
+
+    def resolve_policy_and_bottle_id(
+        self, source_ip: str, identity_token: str = "",
+    ) -> tuple[str | None, str | None]:
+        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
+
+
 # ---------------------------------------------------------------------------
 # Introspection endpoint
 # ---------------------------------------------------------------------------
@@ -418,7 +441,8 @@ class TestSuperviseBranch(unittest.TestCase):
         with patch.object(_ea_mod, "_sv", _fake_sv("approved")):
             _run_request(addon, flow)
         self.assertIsNone(flow.response)  # forwarded after approval
-        self.assertIn(_OPENAI_KEY, addon.safe_tokens)
+        # Approval lands in the calling bottle's safelist (keyed by slug).
+        self.assertIn(_OPENAI_KEY, addon._safe_tokens_for("test-bottle"))
 
     def test_operator_rejection_blocks(self) -> None:
         addon = self._supervised_addon()
@@ -735,5 +759,62 @@ class TestLogFullRequest(unittest.TestCase):
         self.assertTrue(any(e.get("event") == "egress_request" for e in logged))
 
 
+class TestSuperviseMultiTenant(unittest.TestCase):
+    """Consolidated gateway: supervise proposals + the DLP safelist are keyed
+    per bottle, resolved by source IP (PRD 0070)."""
+
+    def _consolidated_addon(self) -> EgressAddon:
+        # Static config is empty; the resolver supplies each bottle's config.
+        addon = _addon(Config(routes=()))
+        addon._resolver = cast(Any, _CtxResolver({"10.0.0.1": "bottle-a", "10.0.0.2": "bottle-b"}))
+        addon._token_allow_timeout = 0.05
+        return addon
+
+    def test_approval_is_scoped_to_the_calling_bottle(self) -> None:
+        addon = self._consolidated_addon()
+        # bottle-a (10.0.0.1) sends the token; the operator approves.
+        flow = _with_client_ip(
+            _Flow(_Request(host="api.example.com", method="POST", body=f"k={_OPENAI_KEY}")),
+            "10.0.0.1",
+        )
+        with patch.object(_ea_mod, "_sv", _fake_sv("approved")):
+            _run_request(addon, flow)
+        self.assertIsNone(flow.response)  # forwarded after approval
+        # The approval lands ONLY in bottle-a's safelist — never bottle-b's.
+        # A global set here would be the cross-tenant leak this slice closes.
+        self.assertIn(_OPENAI_KEY, addon._safe_tokens_for("bottle-a"))
+        self.assertNotIn(_OPENAI_KEY, addon._safe_tokens_for("bottle-b"))
+
+    def test_proposal_is_attributed_to_the_source_ip_bottle(self) -> None:
+        addon = self._consolidated_addon()
+        seen: list[str] = []
+        fake = _fake_sv("approved")
+
+        def _capture(**kw: Any) -> Any:
+            seen.append(kw["bottle_slug"])
+            return types.SimpleNamespace(id="p")
+
+        fake.Proposal = types.SimpleNamespace(new=_capture)
+        flow = _with_client_ip(
+            _Flow(_Request(host="api.example.com", method="POST", body=f"k={_OPENAI_KEY}")),
+            "10.0.0.2",
+        )
+        with patch.object(_ea_mod, "_sv", fake):
+            _run_request(addon, flow)
+        self.assertEqual(["bottle-b"], seen)  # proposal keyed by the resolved bottle
+
+    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.
+        flow = _with_client_ip(
+            _Flow(_Request(host="api.example.com", method="POST", body=f"k={_OPENAI_KEY}")),
+            "10.9.9.9",
+        )
+        with patch.object(_ea_mod, "_sv", _fake_sv("approved")):
+            _run_request(addon, flow)
+        self.assertIsNotNone(flow.response)  # blocked (no route, no supervise)
+        self.assertNotIn(_OPENAI_KEY, addon._safe_tokens_for(""))
+
+
 if __name__ == "__main__":
     unittest.main()

tests/unit/test_egress_multitenant.py

@@ -1,10 +1,10 @@
-"""Unit: resolve_client_config — fail-closed per-client egress config (PRD 0070)."""
+"""Unit: fail-closed per-client egress resolution — config + context (PRD 0070)."""
 
 from __future__ import annotations
 
 import unittest
 
-from bot_bottle.egress_addon_core import resolve_client_config
+from bot_bottle.egress_addon_core import resolve_client_config, resolve_client_context
 from bot_bottle.policy_resolver import PolicyResolveError
 
 
@@ -49,5 +49,52 @@ class TestResolveClientConfig(unittest.TestCase):
         self.assertEqual(("10.243.0.1", "tok"), r.calls[0])
 
 
+class _FakeContextResolver:
+    def __init__(
+        self, policy: str | None = None, bottle_id: str | None = None, raises: bool = False,
+    ) -> None:
+        self._policy = policy
+        self._bottle_id = bottle_id
+        self._raises = raises
+
+    def resolve_policy_and_bottle_id(
+        self, source_ip: str, identity_token: str = "",
+    ) -> tuple[str | None, str | None]:
+        if self._raises:
+            raise PolicyResolveError("orchestrator down")
+        return self._policy, self._bottle_id
+
+
+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"),
+            "10.243.0.1",
+        )
+        self.assertEqual(("example.com",), tuple(r.host for r in cfg.routes))
+        self.assertEqual("b1", slug)
+
+    def test_unattributed_denies_and_empty_slug(self) -> None:
+        cfg, slug = resolve_client_context(
+            _FakeContextResolver(policy=None, bottle_id=None), "10.243.0.9",
+        )
+        self.assertEqual((), cfg.routes)
+        self.assertEqual("", slug)  # no bottle → supervise unavailable
+
+    def test_resolver_error_denies_and_empty_slug(self) -> None:
+        cfg, slug = resolve_client_context(_FakeContextResolver(raises=True), "10.243.0.1")
+        self.assertEqual((), cfg.routes)
+        self.assertEqual("", slug)
+
+    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(
+            _FakeContextResolver(policy="routes: notalist\n", bottle_id="b2"), "10.243.0.1",
+        )
+        self.assertEqual((), cfg.routes)
+        self.assertEqual("b2", slug)
+
+
 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_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_policy_resolver.py

@@ -88,6 +88,20 @@ class TestPolicyResolver(unittest.TestCase):
             with self.assertRaises(PolicyResolveError):
                 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
+
+    def test_resolve_policy_and_bottle_id_403_is_none_none(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"))
+
+    def test_resolve_policy_and_bottle_id_error_raises(self) -> None:
+        with patch(_URLOPEN, side_effect=urllib.error.URLError("refused")):
+            with self.assertRaises(PolicyResolveError):
+                self.r.resolve_policy_and_bottle_id("10.243.0.1")
+
 
 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()