refactor(orchestrator): drop the PolicyResolver cache (#352)

Review: the resolver is called rarely enough that a round-trip doesn't
matter, and correctness beats speed — always fetching means a revocation,
policy change, or teardown the orchestrator knows about is honored
immediately instead of lingering for a cache TTL. Remove the TTL cache
(and `invalidate`); `resolve` now hits the orchestrator every call. Noted
in the docstring that any future caching should use orchestrator-driven
invalidation, not a blind TTL.

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

bot_bottle/orchestrator/control_plane.py

@@ -4,14 +4,18 @@ The backend-agnostic control-plane RPC (CLI / console -> orchestrator) over
 **HTTP** — the universal transport chosen in 0070 (works on every host; no
 vsock / unix-socket portability caveats):
 
-    GET    /health              -> 200 {"status": "ok"}
-    GET    /sidecar             -> 200 {"configured", ["name", "running"]}
-    GET    /bottles             -> 200 {"bottles": [ <redacted record>, ... ]}
-    POST   /bottles             -> 201 {"bottle_id", "identity_token"}    (launch)
-                                   body: {"source_ip", ["image_ref"], ["metadata"]}
-    DELETE /bottles/<bottle_id> -> 200 {"torn_down": true} | 404          (teardown)
-    POST   /attribute           -> 200 {"bottle_id"} | 403
-                                   body: {"source_ip", "identity_token"}
+    GET    /health                    -> 200 {"status": "ok"}
+    GET    /sidecar                   -> 200 {"configured", ["name","running"]}
+    GET    /bottles                   -> 200 {"bottles": [ <redacted record>, ...]}
+    POST   /bottles                   -> 201 {"bottle_id","identity_token"}  (launch)
+                                         body: {"source_ip", ["image_ref"],
+                                                ["metadata"], ["policy"]}
+    PUT    /bottles/<bottle_id>/policy -> 200 {"updated": true} | 404  (live reload)
+                                         body: {"policy"}
+    DELETE /bottles/<bottle_id>        -> 200 {"torn_down": true} | 404  (teardown)
+    POST   /attribute                 -> 200 {"bottle_id"} | 403
+    POST   /resolve                   -> 200 {"bottle_id","policy"} | 403
+                                         body: {"source_ip","identity_token"}
 
 `POST /bottles` / `DELETE` drive the full launch lifecycle: they mint (or
 tear down) the bottle in the registry AND broker the backend-native launch
@@ -49,7 +53,7 @@ def _parse_json_object(body: bytes) -> Json:
     return obj
 
 
-def dispatch(  # pylint: disable=too-many-return-statements
+def dispatch(  # pylint: disable=too-many-return-statements,too-many-branches
     orch: Orchestrator, method: str, path: str, body: bytes
 ) -> tuple[int, Json]:
     """Route one control-plane request to a (status, payload) pair. Pure —
@@ -75,20 +79,35 @@ def dispatch(  # pylint: disable=too-many-return-statements
             return 400, {"error": "source_ip (string) is required"}
         image_ref = data.get("image_ref")
         metadata = data.get("metadata")
+        policy = data.get("policy")
         rec = orch.launch_bottle(
             source_ip,
             image_ref=image_ref if isinstance(image_ref, str) else "",
             metadata=metadata if isinstance(metadata, str) else "",
+            policy=policy if isinstance(policy, str) else "",
         )
         return 201, {"bottle_id": rec.bottle_id, "identity_token": rec.identity_token}
 
+    if method == "PUT" and route.startswith("/bottles/") and route.endswith("/policy"):
+        bottle_id = route[len("/bottles/"):-len("/policy")]
+        try:
+            data = _parse_json_object(body)
+        except ValueError as e:
+            return 400, {"error": f"invalid JSON: {e}"}
+        policy = data.get("policy")
+        if not isinstance(policy, str):
+            return 400, {"error": "policy (string) is required"}
+        if orch.set_policy(bottle_id, policy):
+            return 200, {"updated": True}
+        return 404, {"error": "no such bottle"}
+
     if method == "DELETE" and route.startswith("/bottles/"):
         bottle_id = route[len("/bottles/"):]
         if orch.teardown_bottle(bottle_id):
             return 200, {"torn_down": True}
         return 404, {"error": "no such bottle"}
 
-    if method == "POST" and route == "/attribute":
+    if method == "POST" and route in ("/attribute", "/resolve"):
         try:
             data = _parse_json_object(body)
         except ValueError as e:
@@ -100,6 +119,11 @@ def dispatch(  # pylint: disable=too-many-return-statements
         rec = orch.attribute(source_ip, token)
         if rec is None:
             return 403, {"error": "unattributed"}
+        # /attribute is the lightweight identity check; /resolve additionally
+        # returns the bottle's policy — the source-IP-keyed lookup the
+        # multi-tenant sidecar makes per request.
+        if route == "/resolve":
+            return 200, {"bottle_id": rec.bottle_id, "policy": rec.policy}
         return 200, {"bottle_id": rec.bottle_id}
 
     return 404, {"error": "not found"}
@@ -134,6 +158,9 @@ class Handler(http.server.BaseHTTPRequestHandler):
     def do_POST(self) -> None:
         self._serve("POST")
 
+    def do_PUT(self) -> None:
+        self._serve("PUT")
+
     def do_DELETE(self) -> None:
         self._serve("DELETE")
 

bot_bottle/orchestrator/registry.py

@@ -64,9 +64,13 @@ class BottleRecord:
     identity_token: str
     state: str = "active"
     created_at: float = 0.0
-    # Opaque JSON blob for forward-compat (policy refs, pool slot, ...) —
-    # the registry doesn't interpret it, so new fields need no migration.
+    # Opaque JSON blob for forward-compat (pool slot, ...) — the registry
+    # doesn't interpret it, so new fields need no migration.
     metadata: str = ""
+    # The bottle's sidecar policy (opaque JSON — egress allowlist / routes /
+    # git config). The registry stores and serves it verbatim, keyed by
+    # source IP; the multi-tenant sidecar interprets it.
+    policy: str = ""
 
     def redacted(self) -> dict[str, object]:
         """Public view for listing — never exposes the identity token."""
@@ -76,6 +80,7 @@ class BottleRecord:
             "state": self.state,
             "created_at": self.created_at,
             "metadata": self.metadata,
+            "policy": self.policy,
         }
 
 
@@ -97,6 +102,10 @@ _MIGRATIONS = TableMigrations(
         # the "one active bottle per source IP" check cheap.
         "CREATE INDEX IF NOT EXISTS idx_orchestrator_bottles_source_ip "
         "ON orchestrator_bottles (source_ip)",
+        # v3 — per-bottle policy (opaque JSON the sidecar interprets): the
+        # egress allowlist / routes / git config selected by source IP. The
+        # multi-tenant sidecar resolves it per request via `attribute`.
+        "ALTER TABLE orchestrator_bottles ADD COLUMN policy TEXT NOT NULL DEFAULT ''",
     ],
 )
 
@@ -110,6 +119,7 @@ def _row_to_record(row: sqlite3.Row) -> BottleRecord:
         state=row["state"],
         created_at=row["created_at"],
         metadata=row["metadata"],
+        policy=row["policy"],
     )
 
 
@@ -136,6 +146,7 @@ class RegistryStore(DbStore):
         bottle_id: str | None = None,
         identity_token: str | None = None,
         metadata: str = "",
+        policy: str = "",
     ) -> BottleRecord:
         """Register (or replace) a bottle. Mints a bottle_id / identity token
         when not supplied. Live — this is the control-plane reload path."""
@@ -146,12 +157,13 @@ class RegistryStore(DbStore):
             state="active",
             created_at=time.time(),
             metadata=metadata,
+            policy=policy,
         )
         with self._connect() as conn:
             conn.execute(
                 "INSERT OR REPLACE INTO orchestrator_bottles "
-                "(bottle_id, source_ip, identity_token, state, created_at, metadata) "
-                "VALUES (?, ?, ?, ?, ?, ?)",
+                "(bottle_id, source_ip, identity_token, state, created_at, metadata, policy) "
+                "VALUES (?, ?, ?, ?, ?, ?, ?)",
                 (
                     rec.bottle_id,
                     rec.source_ip,
@@ -159,11 +171,23 @@ class RegistryStore(DbStore):
                     rec.state,
                     rec.created_at,
                     rec.metadata,
+                    rec.policy,
                 ),
             )
         self._chmod()
         return rec
 
+    def set_policy(self, bottle_id: str, policy: str) -> bool:
+        """Update a bottle's policy in place (live reload). Returns True if
+        the bottle exists."""
+        with self._connect() as conn:
+            cur = conn.execute(
+                "UPDATE orchestrator_bottles SET policy = ? WHERE bottle_id = ?",
+                (policy, bottle_id),
+            )
+        self._chmod()
+        return cur.rowcount > 0
+
     def deregister(self, bottle_id: str) -> bool:
         """Remove a bottle. Returns True if a row was deleted."""
         with self._connect() as conn:

bot_bottle/orchestrator/service.py

@@ -46,10 +46,12 @@ class Orchestrator:
         image_ref: str = "",
         slot: int | None = None,
         metadata: str = "",
+        policy: str = "",
     ) -> BottleRecord:
-        """Register a bottle and broker its launch. Rolls the registry entry
-        back if the launch doesn't take, so a failure leaves no orphan."""
-        rec = self.registry.register(source_ip, metadata=metadata)
+        """Register a bottle (with its sidecar policy) and broker its launch.
+        Rolls the registry entry back if the launch doesn't take, so a
+        failure leaves no orphan."""
+        rec = self.registry.register(source_ip, metadata=metadata, policy=policy)
         req = LaunchRequest(
             op="launch",
             bottle_id=rec.bottle_id,
@@ -77,9 +79,16 @@ class Orchestrator:
         return True
 
     def attribute(self, source_ip: str, identity_token: str) -> BottleRecord | None:
-        """Fail-closed attribution (delegates to the registry)."""
+        """Fail-closed attribution (delegates to the registry). The returned
+        record carries the bottle's `policy` — this is the source-IP-keyed
+        resolution the multi-tenant sidecar makes per request."""
         return self.registry.attribute(source_ip, identity_token)
 
+    def set_policy(self, bottle_id: str, policy: str) -> bool:
+        """Update a bottle's sidecar policy in place (live reload). False if
+        the bottle is unknown."""
+        return self.registry.set_policy(bottle_id, policy)
+
     # --- consolidated sidecar ----------------------------------------------
 
     def ensure_sidecar(self) -> None:

bot_bottle/policy_resolver.py

@@ -0,0 +1,75 @@
+"""Sidecar-side per-client policy resolver (PRD 0070).
+
+The consolidated sidecar serves every bottle from one process, so for each
+request it must apply the *calling* bottle's policy, selected by the source
+IP the attribution invariant makes unspoofable. This resolves that policy
+from the orchestrator's control plane (`POST /resolve`), keyed on the
+`(source_ip, identity_token)` pair.
+
+**Always fresh — no cache.** The resolver is called rarely enough that a
+round-trip doesn't matter, and correctness matters more than speed: every
+resolve reflects the orchestrator's *current* view, so a revocation, a
+policy change, or a bottle teardown the orchestrator knows about is honored
+immediately rather than lingering for a cache TTL. (If this ever becomes a
+hot path, add caching with orchestrator-driven invalidation — not a blind
+TTL.)
+
+**Fail-closed:** an unattributed client (the orchestrator answers `403`)
+resolves to `None`, and the caller (the egress addon, git-gate) must then
+deny — exactly as an unknown bottle should be treated. Orchestrator
+*errors* (unreachable / unexpected status) raise, so the caller can fail
+closed too rather than silently serving stale or empty policy.
+
+The resolved value is the policy blob the orchestrator stores verbatim; the
+consumer parses it (e.g. the egress addon's `load_config`). This module is
+stdlib-only and free of bot-bottle imports so it can be COPYed flat into
+the sidecar bundle.
+"""
+
+from __future__ import annotations
+
+import json
+import urllib.error
+import urllib.request
+
+DEFAULT_TIMEOUT_SECONDS = 2.0
+
+
+class PolicyResolveError(RuntimeError):
+    """The orchestrator was unreachable or returned an unexpected status —
+    distinct from a clean `403` (unattributed), which returns None."""
+
+
+class PolicyResolver:
+    """Resolves each client's policy from the orchestrator, fresh per call."""
+
+    def __init__(self, base_url: str, *, timeout: float = DEFAULT_TIMEOUT_SECONDS) -> None:
+        self._base = base_url.rstrip("/")
+        self._timeout = timeout
+
+    def resolve(self, source_ip: str, identity_token: str) -> str | None:
+        """The calling bottle's policy blob, or None if unattributed. Always
+        fetches from the orchestrator so revocations / changes / teardowns
+        are honored immediately. Raises `PolicyResolveError` if the
+        orchestrator can't be reached / errors."""
+        body = json.dumps(
+            {"source_ip": source_ip, "identity_token": identity_token}
+        ).encode()
+        req = urllib.request.Request(
+            f"{self._base}/resolve", data=body, method="POST",
+            headers={"Content-Type": "application/json"},
+        )
+        try:
+            with urllib.request.urlopen(req, timeout=self._timeout) as resp:
+                payload = json.loads(resp.read())
+        except urllib.error.HTTPError as e:
+            if e.code == 403:
+                return None  # unattributed → fail closed (caller denies)
+            raise PolicyResolveError(f"/resolve returned HTTP {e.code}") from e
+        except (urllib.error.URLError, TimeoutError, OSError, ValueError) as e:
+            raise PolicyResolveError(f"/resolve unreachable or malformed: {e}") from e
+        policy = payload.get("policy") if isinstance(payload, dict) else None
+        return policy if isinstance(policy, str) else ""
+
+
+__all__ = ["PolicyResolver", "PolicyResolveError"]

tests/unit/test_orchestrator_control_plane.py

@@ -123,6 +123,54 @@ class TestDispatch(unittest.TestCase):
         self.assertEqual(200, status)
         self.assertEqual(False, payload["configured"])
 
+    def test_launch_stores_policy_and_resolve_returns_it(self) -> None:
+        _, reg = dispatch(
+            self.orch, "POST", "/bottles",
+            _body({"source_ip": "10.243.0.5", "policy": '{"a":1}'}),
+        )
+        status, payload = dispatch(
+            self.orch, "POST", "/resolve",
+            _body({"source_ip": "10.243.0.5", "identity_token": reg["identity_token"]}),
+        )
+        self.assertEqual(200, status)
+        self.assertEqual(reg["bottle_id"], payload["bottle_id"])
+        self.assertEqual('{"a":1}', payload["policy"])
+
+    def test_resolve_forbidden_on_bad_token(self) -> None:
+        dispatch(self.orch, "POST", "/bottles", _body({"source_ip": "10.243.0.5"}))
+        status, _ = dispatch(
+            self.orch, "POST", "/resolve",
+            _body({"source_ip": "10.243.0.5", "identity_token": "nope"}),
+        )
+        self.assertEqual(403, status)
+
+    def test_put_policy_updates_live(self) -> None:
+        _, reg = dispatch(self.orch, "POST", "/bottles", _body({"source_ip": "10.243.0.1"}))
+        bid = reg["bottle_id"]
+        status, payload = dispatch(
+            self.orch, "PUT", f"/bottles/{bid}/policy", _body({"policy": '{"v":9}'})
+        )
+        self.assertEqual(200, status)
+        self.assertEqual(True, payload["updated"])
+        _, resolved = dispatch(
+            self.orch, "POST", "/resolve",
+            _body({"source_ip": "10.243.0.1", "identity_token": reg["identity_token"]}),
+        )
+        self.assertEqual('{"v":9}', resolved["policy"])
+
+    def test_put_policy_missing_bottle_404(self) -> None:
+        status, _ = dispatch(
+            self.orch, "PUT", "/bottles/ghost/policy", _body({"policy": "{}"})
+        )
+        self.assertEqual(404, status)
+
+    def test_put_policy_requires_string(self) -> None:
+        _, reg = dispatch(self.orch, "POST", "/bottles", _body({"source_ip": "10.243.0.1"}))
+        status, _ = dispatch(
+            self.orch, "PUT", f"/bottles/{reg['bottle_id']}/policy", _body({})
+        )
+        self.assertEqual(400, status)
+
 
 class TestServerRoundTrip(unittest.TestCase):
     def test_http_register_health_attribute(self) -> None:

tests/unit/test_orchestrator_registry.py

@@ -97,6 +97,30 @@ class TestRegistryStore(unittest.TestCase):
         self.assertNotIn("identity_token", rec.redacted())
         self.assertEqual("10.243.0.1", rec.redacted()["source_ip"])
 
+    def test_register_with_policy_resolves_via_attribution(self) -> None:
+        rec = self.store.register("10.243.0.1", policy='{"allow":["x"]}')
+        self.assertEqual('{"allow":["x"]}', rec.policy)
+        got = self.store.attribute("10.243.0.1", rec.identity_token)
+        assert got is not None
+        self.assertEqual('{"allow":["x"]}', got.policy)
+
+    def test_set_policy_updates_live(self) -> None:
+        rec = self.store.register("10.243.0.1")
+        self.assertEqual("", rec.policy)
+        self.assertTrue(self.store.set_policy(rec.bottle_id, '{"v":2}'))
+        got = self.store.get(rec.bottle_id)
+        assert got is not None
+        self.assertEqual('{"v":2}', got.policy)
+
+    def test_set_policy_missing_is_false(self) -> None:
+        self.assertFalse(self.store.set_policy("ghost", "{}"))
+
+    def test_policy_defaults_empty_and_persists(self) -> None:
+        rec = self.store.register("10.243.0.1")
+        got = RegistryStore(self.db).get(rec.bottle_id)
+        assert got is not None
+        self.assertEqual("", got.policy)
+
 
 if __name__ == "__main__":
     unittest.main()

tests/unit/test_orchestrator_service.py

@@ -86,6 +86,22 @@ class TestOrchestrator(unittest.TestCase):
     def test_teardown_unknown_is_false(self) -> None:
         self.assertFalse(self.orch.teardown_bottle("ghost"))
 
+    def test_launch_with_policy_is_resolvable(self) -> None:
+        rec = self.orch.launch_bottle("10.243.0.1", policy='{"routes":[]}')
+        got = self.orch.attribute("10.243.0.1", rec.identity_token)
+        assert got is not None
+        self.assertEqual('{"routes":[]}', got.policy)
+
+    def test_set_policy_live_reload(self) -> None:
+        rec = self.orch.launch_bottle("10.243.0.3")
+        self.assertTrue(self.orch.set_policy(rec.bottle_id, '{"x":1}'))
+        got = self.orch.attribute("10.243.0.3", rec.identity_token)
+        assert got is not None
+        self.assertEqual('{"x":1}', got.policy)
+
+    def test_set_policy_unknown_is_false(self) -> None:
+        self.assertFalse(self.orch.set_policy("ghost", "{}"))
+
     def test_launch_rolls_back_registry_on_broker_failure(self) -> None:
         orch = Orchestrator(self.store, _FailingBroker(self.secret), self.secret)
         with self.assertRaises(RuntimeError):

tests/unit/test_policy_resolver.py

@@ -0,0 +1,71 @@
+"""Unit tests for the sidecar-side PolicyResolver (PRD 0070). HTTP mocked."""
+
+from __future__ import annotations
+
+import json
+import unittest
+import urllib.error
+from unittest.mock import MagicMock, patch
+
+from bot_bottle.policy_resolver import PolicyResolveError, PolicyResolver
+
+_URLOPEN = "bot_bottle.policy_resolver.urllib.request.urlopen"
+
+
+def _resp(payload: object) -> MagicMock:
+    """A urlopen() return value: a context manager whose read() yields JSON."""
+    m = MagicMock()
+    m.__enter__.return_value.read.return_value = json.dumps(payload).encode()
+    return m
+
+
+def _http_error(code: int) -> urllib.error.HTTPError:
+    return urllib.error.HTTPError("http://x/resolve", code, "err", {}, None)  # type: ignore[arg-type]
+
+
+class TestPolicyResolver(unittest.TestCase):
+    def setUp(self) -> None:
+        self.r = PolicyResolver("http://orch:8080")
+
+    def test_resolve_returns_policy(self) -> None:
+        with patch(_URLOPEN, return_value=_resp({"bottle_id": "b1", "policy": "P"})):
+            self.assertEqual("P", self.r.resolve("10.243.0.1", "tok"))
+
+    def test_resolve_always_fetches_fresh(self) -> None:
+        # No cache — every resolve hits the orchestrator so revocations /
+        # policy changes are honored immediately.
+        with patch(_URLOPEN, return_value=_resp({"policy": "P"})) as m:
+            self.r.resolve("10.243.0.1", "tok")
+            self.r.resolve("10.243.0.1", "tok")
+        self.assertEqual(2, m.call_count)
+
+    def test_unattributed_403_is_none_fail_closed(self) -> None:
+        with patch(_URLOPEN, side_effect=_http_error(403)):
+            self.assertIsNone(self.r.resolve("10.243.0.9", "tok"))
+
+    def test_other_http_error_raises(self) -> None:
+        with patch(_URLOPEN, side_effect=_http_error(500)):
+            with self.assertRaises(PolicyResolveError):
+                self.r.resolve("10.243.0.1", "tok")
+
+    def test_unreachable_raises(self) -> None:
+        with patch(_URLOPEN, side_effect=urllib.error.URLError("refused")):
+            with self.assertRaises(PolicyResolveError):
+                self.r.resolve("10.243.0.1", "tok")
+
+    def test_missing_policy_field_is_empty(self) -> None:
+        with patch(_URLOPEN, return_value=_resp({"bottle_id": "b1"})):
+            self.assertEqual("", self.r.resolve("10.243.0.1", "tok"))
+
+    def test_posts_source_ip_and_token(self) -> None:
+        with patch(_URLOPEN, return_value=_resp({"policy": "P"})) as m:
+            self.r.resolve("10.243.0.7", "the-token")
+        req = m.call_args.args[0]
+        self.assertTrue(req.full_url.endswith("/resolve"))
+        sent = json.loads(req.data)
+        self.assertEqual("10.243.0.7", sent["source_ip"])
+        self.assertEqual("the-token", sent["identity_token"])
+
+
+if __name__ == "__main__":
+    unittest.main()