feat(supervise+orchestrator): slice 12 — multi-tenant git-gate hook + supervise server

Finishes the supervise data-plane writers for the shared gateway. Both
still keyed off a single SUPERVISE_BOTTLE_SLUG env; now each proposal is
attributed to the calling bottle, keeping slices 10/11's per-bottle model.

- git_http_backend: in consolidated mode stamp SUPERVISE_BOTTLE_SLUG=
  <bottle_id> into the CGI env. The gitleaks-allow pre-receive hook runs as
  a child of the `git http-backend` we spawn, so it inherits the stamp and
  queues under the right bottle — no change to the (fragile) embedded hook.
  The bottle id is the namespaced root's final component (slice 10), the
  same key egress uses. Single-tenant leaves the container-stamped slug.
- supervise_server: resolve the proposal slug per request by source IP when
  BOT_BOTTLE_ORCHESTRATOR_URL is set (`_attributed_config`), fail-closed —
  an unattributed / unreachable source raises rather than queue under the
  wrong or empty slug. `serve`/`main` build + attach the resolver and make
  the env slug optional in consolidated mode. Single-tenant unchanged.

Tests: a real git push proving the slug reaches the hook's env; the
supervise attribution matrix (single-tenant slug kept, source-IP bottle
bound, unattributed + resolver-error fail closed).

Remaining supervise gap (noted): websocket DLP still self.config-only.

pyright 0 errors; pylint 9.83/10; unit suite green (1705 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/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/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"supervise listening on {bind}:{port}; {mode}; "
-        f"slug={bottle_slug!r}; "
         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_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_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()