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

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()