fix(runtime): add deterministic executor starvation regressions (#5335)

* test(runtime): add deterministic executor starvation regressions

* docs(tests): document executor starvation test contract

* fix(runtime): address executor starvation review feedback

* docs(tests): clarify executor cleanup guidance

* docs(tests): clarify executor cleanup guidance

* docs(tests): clarify executor cleanup guidance

* test(runtime): apply remaining ruff formatting fix

* test(runtime): register executor test guidance path

* test(runtime): bound dedicated executor regression wait

* test(ci): align guidance budget test with checker semantics

backend/tests/AGENTS.md

@@ -0,0 +1,15 @@
+# Backend Tests
+
+Backend tests must preserve the runtime invariants they exercise without changing production execution topology.
+
+## Executor starvation tests
+
+`test_executor_starvation.py` covers the deterministic starvation semantics from RFC #4560:
+
+- default-executor saturation and queueing;
+- cancellation of an awaiter while an already-started synchronous worker continues;
+- isolation between the asyncio default executor and DeerFlow's dedicated file-I/O executor.
+
+Use explicit synchronization such as `threading.Event` rather than sleep-based timing thresholds for worker lifecycle assertions. Every test must release blocked workers and restore any process-global monkeypatches so teardown cannot leak threads or state into later tests.
+
+Stress/soak testing, AnyIO worker instrumentation, Uvicorn multi-process behavior, and broad production executor redesign are separate concerns and should not be folded into these deterministic regressions.

backend/tests/test_agent_guidance_check.py

@@ -10,6 +10,7 @@ CHECKER_PATH = REPO_ROOT / "scripts" / "check_agent_guidance.py"
 EXPECTED_GUIDANCE_PATHS = {
     "AGENTS.md",
     "backend/AGENTS.md",
+    "backend/tests/AGENTS.md",
     "frontend/AGENTS.md",
     "backend/app/gateway/AGENTS.md",
     "backend/app/channels/AGENTS.md",
@@ -129,14 +130,14 @@ def test_repository_has_the_approved_scoped_guidance_shape() -> None:
     assert actual == EXPECTED_GUIDANCE_PATHS
 
 
-def test_repository_guidance_stays_below_soft_budgets_and_avoids_doc_indexes() -> None:
+def test_repository_guidance_stays_below_hard_budgets_and_avoids_doc_indexes() -> None:
     for relative_text in EXPECTED_GUIDANCE_PATHS:
         relative = PurePosixPath(relative_text)
         path = REPO_ROOT / relative_text
         assert path.is_file(), relative
-        soft, _ = checker.agent_budget(relative)
+        _, hard = checker.agent_budget(relative)
         text = path.read_text(encoding="utf-8")
-        assert checker.normalized_utf8_size(text) <= soft, relative
+        assert checker.normalized_utf8_size(text) <= hard, relative
         assert "Subsystem Index" not in text
 
 

backend/tests/test_executor_starvation.py

@@ -0,0 +1,117 @@
+from __future__ import annotations
+
+import asyncio
+import threading
+from concurrent.futures import ThreadPoolExecutor
+
+
+def test_default_executor_saturation_keeps_work_queued():
+    async def scenario():
+        loop = asyncio.get_running_loop()
+        executor = ThreadPoolExecutor(max_workers=1)
+        loop.set_default_executor(executor)
+        release = threading.Event()
+        first_started = asyncio.Event()
+        second_started = asyncio.Event()
+
+        def blocker():
+            loop.call_soon_threadsafe(first_started.set)
+            release.wait()
+
+        try:
+            first = asyncio.create_task(asyncio.to_thread(blocker))
+            await first_started.wait()
+
+            second = asyncio.create_task(asyncio.to_thread(loop.call_soon_threadsafe, second_started.set))
+            await asyncio.sleep(0)
+            assert not second_started.is_set()
+
+            release.set()
+            await asyncio.gather(first, second)
+            assert second_started.is_set()
+        finally:
+            release.set()
+            executor.shutdown(wait=True, cancel_futures=True)
+
+    asyncio.run(scenario())
+
+
+def test_waiter_timeout_does_not_stop_started_sync_work():
+    async def scenario():
+        loop = asyncio.get_running_loop()
+        executor = ThreadPoolExecutor(max_workers=1)
+        loop.set_default_executor(executor)
+        release = threading.Event()
+        started = asyncio.Event()
+        finished = threading.Event()
+        sentinel_started = asyncio.Event()
+
+        def blocker():
+            loop.call_soon_threadsafe(started.set)
+            release.wait()
+            finished.set()
+
+        try:
+            running = asyncio.create_task(asyncio.to_thread(blocker))
+            await started.wait()
+
+            try:
+                await asyncio.wait_for(running, timeout=0)
+            except TimeoutError:
+                pass
+
+            assert running.cancelled()
+            assert not finished.is_set()
+
+            sentinel = asyncio.create_task(asyncio.to_thread(loop.call_soon_threadsafe, sentinel_started.set))
+            await asyncio.sleep(0)
+            assert not sentinel_started.is_set()
+
+            release.set()
+            await sentinel
+            assert finished.is_set()
+            assert sentinel_started.is_set()
+        finally:
+            release.set()
+            executor.shutdown(wait=True, cancel_futures=True)
+
+    asyncio.run(scenario())
+
+
+def test_dedicated_file_io_pool_runs_while_default_executor_is_saturated(monkeypatch):
+    async def scenario():
+        from deerflow.utils import file_io
+
+        loop = asyncio.get_running_loop()
+        default_executor = ThreadPoolExecutor(max_workers=1)
+        dedicated_executor = ThreadPoolExecutor(max_workers=1)
+        monkeypatch.setattr(file_io, "_FILE_IO_EXECUTOR", dedicated_executor)
+        loop.set_default_executor(default_executor)
+        release = threading.Event()
+        default_started = asyncio.Event()
+        dedicated_started = asyncio.Event()
+
+        def default_blocker():
+            loop.call_soon_threadsafe(default_started.set)
+            release.wait()
+
+        def dedicated_work():
+            loop.call_soon_threadsafe(dedicated_started.set)
+            return "file-io"
+
+        try:
+            default_task = asyncio.create_task(asyncio.to_thread(default_blocker))
+            await default_started.wait()
+
+            file_task = asyncio.create_task(file_io.run_file_io(dedicated_work))
+            await asyncio.wait_for(dedicated_started.wait(), timeout=5)
+            assert await file_task == "file-io"
+
+            release.set()
+            await default_task
+        finally:
+            release.set()
+            dedicated_executor.shutdown(wait=True, cancel_futures=True)
+            default_executor.shutdown(wait=True, cancel_futures=True)
+
+    asyncio.run(scenario())