fix(egress): run supervise propose/poll RPCs off the mitmproxy event loop

`_supervise_token_block` and `_await_token_response` called the synchronous
`PolicyResolver` directly from mitmproxy's async request handling. Those RPCs
funnel through `PolicyResolver._post_json`, which blocks on
`urllib.request.urlopen`. Because the poll loop runs for the entire
operator-approval window, a slow or unreachable orchestrator would repeatedly
freeze the proxy event loop — stalling every other bottle's traffic — despite
`_await_token_response`'s contract of not blocking the loop (#471 review,
review #443).

Dispatch both the propose and poll RPCs via `asyncio.to_thread` so the
blocking urllib call runs in a worker thread and the event loop keeps serving
other flows. Fail-closed semantics are unchanged: `PolicyResolveError` still
propagates through the await and is caught exactly as before.

Regression test records the thread each RPC executes on and asserts it is not
the loop thread; it fails against the pre-fix inline calls and passes with the
dispatch. Full egress-addon + policy-resolver + supervise-server suites green
(155 tests).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

bot_bottle/gateway/egress/addon.py

@@ -612,7 +612,11 @@ class EgressAddon:
         source_ip = conn.peername[0] if conn is not None and conn.peername else ""
         token = self._conn_tokens.get(getattr(conn, "id", ""), "") if conn is not None else ""
         try:
-            proposal_id = self._resolver.propose_supervise(
+            # The resolver RPC is synchronous (blocking urllib); run it off the
+            # event loop so a slow/unreachable orchestrator can't freeze the
+            # proxy for every other bottle's traffic (#471 review).
+            proposal_id = await asyncio.to_thread(
+                self._resolver.propose_supervise,
                 source_ip, token,
                 tool=TOOL_EGRESS_TOKEN_ALLOW,
                 proposed_file=payload,
@@ -675,7 +679,13 @@ class EgressAddon:
         deadline = loop.time() + self._token_allow_timeout
         while True:
             try:
-                result = self._resolver.poll_supervise(source_ip, token, proposal_id)
+                # Poll off the event loop — the resolver RPC blocks on urllib,
+                # and this loop runs for the whole operator-approval window, so
+                # calling it inline would repeatedly freeze all proxy traffic
+                # until a decision or timeout (#471 review).
+                result = await asyncio.to_thread(
+                    self._resolver.poll_supervise, source_ip, token, proposal_id,
+                )
             except PolicyResolveError:
                 result = None
             if result is not None and result.get("status") in STATUSES:

tests/unit/test_egress_addon_request_flow.py

@@ -21,6 +21,7 @@ import asyncio
 import json
 import os
 import sys
+import threading
 import types
 import unittest
 from io import StringIO
@@ -599,6 +600,35 @@ class TestSuperviseBranch(unittest.TestCase):
         assert flow.response is not None
         self.assertEqual(403, flow.response.status_code)
 
+    def test_supervise_rpcs_run_off_the_event_loop_thread(self) -> None:
+        # The propose/poll RPCs block on urllib, and the poll loop runs for the
+        # whole operator-approval window; calling them inline would freeze the
+        # proxy event loop for every other bottle's traffic. They must run in a
+        # worker thread instead (#471 review). Prove it by recording the thread
+        # the RPCs execute on and asserting it isn't the loop's (main) thread.
+        loop_ident = threading.get_ident()
+        rpc_idents: list[int] = []
+
+        class _ThreadRecordingResolver(_StaticResolver):
+            def propose_supervise(self, *a: Any, **k: Any) -> str:
+                rpc_idents.append(threading.get_ident())
+                return super().propose_supervise(*a, **k)
+
+            def poll_supervise(self, *a: Any, **k: Any) -> "dict[str, object]":
+                rpc_idents.append(threading.get_ident())
+                return super().poll_supervise(*a, **k)
+
+        config = Config(routes=(Route(host="api.example.com"),))
+        addon = _addon(config, slug="test-bottle")
+        resolver = _ThreadRecordingResolver(config, bottle_id="test-bottle")
+        resolver.supervise_status = "approved"
+        addon._resolver = cast(Any, resolver)
+        addon._token_allow_timeout = 0.05
+        flow = _Flow(_Request(host="api.example.com", method="POST", body=f"k={_OPENAI_KEY}"))
+        _run_request(addon, flow)
+        self.assertTrue(rpc_idents, "propose/poll were never invoked")
+        self.assertNotIn(loop_ident, rpc_idents)  # every RPC ran off the loop thread
+
 
 # ---------------------------------------------------------------------------
 # Inbound DLP on responses