fix(provisioner): keep k8s calls off event loop (#3941)

* fix(provisioner): keep k8s calls off event loop

* test(provisioner): scan blocking IO by module name

* test(provisioner): exercise create path so BlockBuster has real work

backend/AGENTS.md

@@ -125,6 +125,9 @@ deterministic scope step before routing each candidate to a fix and/or a
 Regression tests related to Docker/provisioner behavior:
 - `tests/test_docker_sandbox_mode_detection.py` (mode detection from `config.yaml`)
 - `tests/test_provisioner_kubeconfig.py` (kubeconfig file/directory handling)
+- `tests/test_provisioner_request_threading.py` (keeps provisioner sandbox CRUD
+  endpoints as sync FastAPI handlers so synchronous K8s client calls run in the
+  Starlette worker pool instead of on the ASGI event loop)
 
 Blocking-IO runtime gate (`tests/blocking_io/`):
 - Wraps every item under `tests/blocking_io/` with a strict Blockbuster

backend/tests/conftest.py

@@ -54,8 +54,16 @@ def provisioner_module():
     assert spec is not None
     assert spec.loader is not None
     module = importlib.util.module_from_spec(spec)
-    spec.loader.exec_module(module)
-    return module
+    previous_module = sys.modules.get(spec.name)
+    sys.modules[spec.name] = module
+    try:
+        spec.loader.exec_module(module)
+        yield module
+    finally:
+        if previous_module is None:
+            sys.modules.pop(spec.name, None)
+        else:
+            sys.modules[spec.name] = previous_module
 
 
 # ---------------------------------------------------------------------------

backend/tests/test_provisioner_request_threading.py

@@ -0,0 +1,159 @@
+"""Regression tests for provisioner request-path K8s IO threading."""
+
+from __future__ import annotations
+
+import asyncio
+import inspect
+import threading
+import time
+from contextlib import contextmanager
+from types import SimpleNamespace
+
+import httpx
+import pytest
+from blockbuster import BlockBuster
+
+
+class _RecordingCoreV1:
+    def __init__(
+        self,
+        *,
+        event_loop_thread_id: int,
+        ready_after_service_reads: dict[str, int] | None = None,
+    ) -> None:
+        self.event_loop_thread_id = event_loop_thread_id
+        self.thread_ids: list[int] = []
+        self.service_sandboxes: set[str] = {"sandbox-existing"}
+        self.ready_after_service_reads = ready_after_service_reads or {}
+        self.service_read_counts: dict[str, int] = {}
+        self.created_pods: list[str] = []
+        self.created_services: list[str] = []
+
+    def _record_k8s_call(self) -> None:
+        thread_id = threading.get_ident()
+        self.thread_ids.append(thread_id)
+        time.sleep(0)
+        if thread_id == self.event_loop_thread_id:
+            raise AssertionError("Kubernetes client call ran on the ASGI event-loop thread")
+        try:
+            asyncio.get_running_loop()
+        except RuntimeError:
+            return
+        raise AssertionError("Kubernetes client call ran inside an asyncio event loop")
+
+    def read_namespaced_service(self, _name: str, _namespace: str):
+        self._record_k8s_call()
+        sandbox_id = _sandbox_id_from_service_name(_name)
+        self.service_read_counts[sandbox_id] = self.service_read_counts.get(sandbox_id, 0) + 1
+        ready_after_reads = self.ready_after_service_reads.get(sandbox_id, 1)
+        if sandbox_id not in self.service_sandboxes or self.service_read_counts[sandbox_id] < ready_after_reads:
+            return _service_without_node_port(sandbox_id)
+        return _service(sandbox_id)
+
+    def read_namespaced_pod(self, _name: str, _namespace: str):
+        self._record_k8s_call()
+        return SimpleNamespace(status=SimpleNamespace(phase="Running"))
+
+    def create_namespaced_pod(self, _namespace: str, pod) -> None:
+        self._record_k8s_call()
+        sandbox_id = pod.metadata.labels["sandbox-id"]
+        self.created_pods.append(sandbox_id)
+
+    def create_namespaced_service(self, _namespace: str, service) -> None:
+        self._record_k8s_call()
+        sandbox_id = service.metadata.labels["sandbox-id"]
+        self.created_services.append(sandbox_id)
+        self.service_sandboxes.add(sandbox_id)
+
+    def delete_namespaced_service(self, _name: str, _namespace: str) -> None:
+        self._record_k8s_call()
+
+    def delete_namespaced_pod(self, _name: str, _namespace: str) -> None:
+        self._record_k8s_call()
+
+    def list_namespaced_service(self, _namespace: str, *, label_selector: str):
+        self._record_k8s_call()
+        assert label_selector == "app=deer-flow-sandbox"
+        return SimpleNamespace(items=[_service("sandbox-listed")])
+
+
+def _service(sandbox_id: str):
+    return SimpleNamespace(
+        metadata=SimpleNamespace(labels={"sandbox-id": sandbox_id}),
+        spec=SimpleNamespace(ports=[SimpleNamespace(name="http", node_port=32123)]),
+    )
+
+
+def _service_without_node_port(sandbox_id: str):
+    return SimpleNamespace(
+        metadata=SimpleNamespace(labels={"sandbox-id": sandbox_id}),
+        spec=SimpleNamespace(ports=[]),
+    )
+
+
+def _sandbox_id_from_service_name(name: str) -> str:
+    assert name.startswith("sandbox-")
+    assert name.endswith("-svc")
+    return name[len("sandbox-") : -len("-svc")]
+
+
+@contextmanager
+def _detect_provisioner_blocking_io(provisioner_module):
+    detector = BlockBuster(scanned_modules=[provisioner_module.__name__])
+    detector.activate()
+    try:
+        yield
+    finally:
+        detector.deactivate()
+
+
+def test_sandbox_business_route_handlers_are_sync(provisioner_module) -> None:
+    """FastAPI runs sync handlers in its worker pool, away from the event loop."""
+    for handler in (
+        provisioner_module.create_sandbox,
+        provisioner_module.destroy_sandbox,
+        provisioner_module.get_sandbox,
+        provisioner_module.list_sandboxes,
+    ):
+        assert not inspect.iscoroutinefunction(handler)
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    ("method", "path", "json_body", "expected_created_sandbox"),
+    [
+        ("POST", "/api/sandboxes", {"sandbox_id": "sandbox-existing", "thread_id": "thread-1", "user_id": "user-1"}, None),
+        ("POST", "/api/sandboxes", {"sandbox_id": "sandbox-new", "thread_id": "thread-1", "user_id": "user-1"}, "sandbox-new"),
+        ("DELETE", "/api/sandboxes/sandbox-existing", None, None),
+        ("GET", "/api/sandboxes/sandbox-existing", None, None),
+        ("GET", "/api/sandboxes", None, None),
+    ],
+    ids=["create-existing", "create-new", "destroy", "get", "list"],
+)
+async def test_sandbox_business_routes_run_k8s_client_off_event_loop_thread(
+    method: str,
+    path: str,
+    json_body: dict[str, str] | None,
+    expected_created_sandbox: str | None,
+    monkeypatch: pytest.MonkeyPatch,
+    provisioner_module,
+) -> None:
+    fake_core_v1 = _RecordingCoreV1(
+        event_loop_thread_id=threading.get_ident(),
+        ready_after_service_reads={"sandbox-new": 3},
+    )
+    monkeypatch.setattr(provisioner_module, "core_v1", fake_core_v1)
+
+    with _detect_provisioner_blocking_io(provisioner_module):
+        transport = httpx.ASGITransport(app=provisioner_module.app)
+        async with httpx.AsyncClient(transport=transport, base_url="http://testserver") as client:
+            if json_body is None:
+                response = await client.request(method, path)
+            else:
+                response = await client.request(method, path, json=json_body)
+
+    assert response.status_code == 200
+    assert fake_core_v1.thread_ids
+    if expected_created_sandbox is not None:
+        assert fake_core_v1.created_pods == [expected_created_sandbox]
+        assert fake_core_v1.created_services == [expected_created_sandbox]

docker/provisioner/README.md

@@ -36,6 +36,13 @@ The **Sandbox Provisioner** is a FastAPI service that dynamically manages sandbo
 
 5. **Cleanup**: When the session ends, `DELETE /api/sandboxes/{sandbox_id}` removes both the Pod and Service.
 
+The sandbox business endpoints are implemented as synchronous FastAPI handlers
+because the Kubernetes Python client used here is synchronous. Starlette runs
+sync handlers in its worker pool, keeping create/read/list/delete K8s API calls
+and the NodePort polling sleep off the ASGI event-loop thread. Keep `/health`
+lightweight; do not move the sandbox CRUD handlers back to `async def` unless
+the K8s client path is also made async or explicitly offloaded.
+
 ## Requirements
 
 Host machine with a running Kubernetes cluster (Docker Desktop K8s, OrbStack, minikube, kind, etc.)

docker/provisioner/app.py

@@ -289,7 +289,9 @@ def _build_volumes(thread_id: str) -> list[k8s_client.V1Volume]:
     return [skills_vol, userdata_vol]
 
 
-def _build_volume_mounts(thread_id: str, user_id: str = DEFAULT_USER_ID) -> list[k8s_client.V1VolumeMount]:
+def _build_volume_mounts(
+    thread_id: str, user_id: str = DEFAULT_USER_ID
+) -> list[k8s_client.V1VolumeMount]:
     """Build volume mount list, using subPath for PVC user-data."""
     userdata_mount = k8s_client.V1VolumeMount(
         name="user-data",
@@ -297,7 +299,9 @@ def _build_volume_mounts(thread_id: str, user_id: str = DEFAULT_USER_ID) -> list
         read_only=False,
     )
     if USERDATA_PVC_NAME:
-        userdata_mount.sub_path = f"deer-flow/users/{user_id}/threads/{thread_id}/user-data"
+        userdata_mount.sub_path = (
+            f"deer-flow/users/{user_id}/threads/{thread_id}/user-data"
+        )
 
     return [
         k8s_client.V1VolumeMount(
@@ -309,7 +313,9 @@ def _build_volume_mounts(thread_id: str, user_id: str = DEFAULT_USER_ID) -> list
     ]
 
 
-def _build_pod(sandbox_id: str, thread_id: str, user_id: str = DEFAULT_USER_ID) -> k8s_client.V1Pod:
+def _build_pod(
+    sandbox_id: str, thread_id: str, user_id: str = DEFAULT_USER_ID
+) -> k8s_client.V1Pod:
     """Construct a Pod manifest for a single sandbox."""
     return k8s_client.V1Pod(
         metadata=k8s_client.V1ObjectMeta(
@@ -442,7 +448,7 @@ async def health():
 
 
 @app.post("/api/sandboxes", response_model=SandboxResponse)
-async def create_sandbox(req: CreateSandboxRequest):
+def create_sandbox(req: CreateSandboxRequest):
     """Create a sandbox Pod + NodePort Service for *sandbox_id*.
 
     If the sandbox already exists, returns the existing information
@@ -470,7 +476,9 @@ async def create_sandbox(req: CreateSandboxRequest):
 
     # ── Create Pod ───────────────────────────────────────────────────
     try:
-        core_v1.create_namespaced_pod(K8S_NAMESPACE, _build_pod(sandbox_id, thread_id, user_id=user_id))
+        core_v1.create_namespaced_pod(
+            K8S_NAMESPACE, _build_pod(sandbox_id, thread_id, user_id=user_id)
+        )
         logger.info(f"Created Pod {_pod_name(sandbox_id)}")
     except ApiException as exc:
         if exc.status != 409:  # 409 = AlreadyExists
@@ -514,7 +522,7 @@ async def create_sandbox(req: CreateSandboxRequest):
 
 
 @app.delete("/api/sandboxes/{sandbox_id}")
-async def destroy_sandbox(sandbox_id: str):
+def destroy_sandbox(sandbox_id: str):
     """Destroy a sandbox Pod + Service."""
     errors: list[str] = []
 
@@ -543,7 +551,7 @@ async def destroy_sandbox(sandbox_id: str):
 
 
 @app.get("/api/sandboxes/{sandbox_id}", response_model=SandboxResponse)
-async def get_sandbox(sandbox_id: str):
+def get_sandbox(sandbox_id: str):
     """Return current status and URL for a sandbox."""
     node_port = _get_node_port(sandbox_id)
     if not node_port:
@@ -557,7 +565,7 @@ async def get_sandbox(sandbox_id: str):
 
 
 @app.get("/api/sandboxes")
-async def list_sandboxes():
+def list_sandboxes():
     """List every sandbox currently managed in the namespace."""
     try:
         services = core_v1.list_namespaced_service(