diff --git a/backend/tests/AGENTS.md b/backend/tests/AGENTS.md new file mode 100644 index 000000000..b6cea4f93 --- /dev/null +++ b/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. diff --git a/backend/tests/test_agent_guidance_check.py b/backend/tests/test_agent_guidance_check.py index b87fa3d10..07651b630 100644 --- a/backend/tests/test_agent_guidance_check.py +++ b/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 diff --git a/backend/tests/test_executor_starvation.py b/backend/tests/test_executor_starvation.py new file mode 100644 index 000000000..7645bb1ef --- /dev/null +++ b/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())