TEST: Stabilize scenario estimate readiness coordination

doc/contributing/4_running_tests.md

@@ -87,6 +87,13 @@ rounding cannot turn a valid query budget into a failed assertion.
 For scenario correctness tests, wait for the owned task to finish, including
 persistence and scheduler handoff. Use events to control concurrent steps and a
 finite test watchdog to detect deadlocks, not short per-step latency limits.
+Catalog queue-admission and configured-estimate independence tests use the existing
+estimate watchdog fixture. Observe the queued semaphore acquisition before checking
+that its execution deadline has not started, and verify configured completion while
+catalog capacity is still occupied. Check each deadline against completed admissions
+and spy on catalog acquisition during configured work to assert independence directly.
+Release gates and drain owned requests and service workers in `finally`, including
+when a readiness assertion fails.
 For frontend modal tests, use fake timers and wait for an accessible control on
 the destination page. A route change does not mean that modal cleanup has restored
 background accessibility.

tests/unit/backend/test_scenario_service.py

@@ -953,7 +953,9 @@ async def estimate_async(
             await service.close_async()
             await asyncio.gather(catalog_task, return_exceptions=True)
 
-    async def test_catalog_queue_wait_does_not_start_execution_timeout(self) -> None:
+    async def test_catalog_queue_wait_does_not_start_execution_timeout_async(
+        self, *, estimate_watchdog_seconds: float
+    ) -> None:
         """A queued catalog estimate starts its timeout only after acquiring capacity."""
         metadata = [_make_scenario_metadata(registry_name=f"test.scenario_{index}") for index in range(2)]
         estimate = ScenarioRunSizeEstimate(
@@ -962,8 +964,10 @@ async def test_catalog_queue_wait_does_not_start_execution_timeout(self) -> None
         )
         first_estimate_started = asyncio.Event()
         release_first_estimate = asyncio.Event()
+        second_estimate_queued = asyncio.Event()
         second_timeout_started = asyncio.Event()
         estimate_count = 0
+        admitted_count = 0
         timeout_count = 0
 
         async def estimate_async(
@@ -990,31 +994,57 @@ async def wait_async(
             timeout: float,
         ) -> tuple[set[asyncio.Task[ScenarioRunSizeEstimate]], set[asyncio.Task[ScenarioRunSizeEstimate]]]:
             nonlocal timeout_count
-            assert timeout > 0
+            assert timeout == estimate_watchdog_seconds
+            assert admitted_count > timeout_count
             timeout_count += 1
             if timeout_count == 2:
                 second_timeout_started.set()
             return await original_wait(tasks, timeout=timeout)
 
         service = ScenarioService()
-        service._registry = MagicMock()
+        service._registry = MagicMock(spec=ScenarioRegistry)
         service._registry.get_all_registered_class_metadata.return_value = metadata
         service._estimate_semaphore = asyncio.Semaphore(1)
         service._run_default_estimate_async = AsyncMock(side_effect=estimate_async)
+        acquire = service._estimate_semaphore.acquire
 
-        with patch("pyrit.backend.services.scenario_service.asyncio.wait", side_effect=wait_async):
+        async def acquire_async() -> bool:
+            nonlocal admitted_count
+            if service._estimate_semaphore.locked():
+                second_estimate_queued.set()
+            acquired = await acquire()
+            admitted_count += 1
+            return acquired
+
+        with (
+            patch.object(service._estimate_semaphore, "acquire", side_effect=acquire_async),
+            patch("pyrit.backend.services.scenario_service.asyncio.wait", side_effect=wait_async),
+        ):
             catalog_task = asyncio.create_task(service.list_scenarios_async())
-            async with asyncio.timeout(1):
-                await first_estimate_started.wait()
-            await asyncio.sleep(0)
-            assert not second_timeout_started.is_set()
+            try:
+                async with asyncio.timeout(estimate_watchdog_seconds):
+                    await first_estimate_started.wait()
+                    await second_estimate_queued.wait()
+                    assert service._estimate_semaphore.locked()
+                    assert service._run_default_estimate_async.await_count == 1
+                    assert timeout_count == 1
+                    assert not second_timeout_started.is_set()
 
-            release_first_estimate.set()
-            result = await catalog_task
+                    release_first_estimate.set()
+                    result = await catalog_task
+            finally:
+                release_first_estimate.set()
+                async with asyncio.timeout(estimate_watchdog_seconds):
+                    await asyncio.gather(catalog_task, return_exceptions=True)
+                    await service.close_async()
 
         assert second_timeout_started.is_set()
+        assert admitted_count == 2
         assert timeout_count == 2
+        assert service._run_default_estimate_async.await_count == 2
         assert all(item.default_run_size == estimate for item in result.items)
+        assert service.outstanding_estimates() == 0
+        assert service._estimate_tasks == {}
 
     async def test_catalog_execution_timeout_is_unavailable_and_cached(self) -> None:
         """A genuine estimate execution timeout is unavailable and reused from cache."""
@@ -1190,7 +1220,9 @@ async def acquire_second_async() -> bool:
         assert second_result == estimate
         assert maximum_active == 1
 
-    async def test_configured_estimate_does_not_wait_for_catalog_estimate(self) -> None:
+    async def test_configured_estimate_does_not_wait_for_catalog_estimate_async(
+        self, *, estimate_watchdog_seconds: float
+    ) -> None:
         """Configured estimates use separate capacity from default catalog estimates."""
         metadata = _make_scenario_metadata()
         default_estimate = ScenarioRunSizeEstimate(
@@ -1218,25 +1250,48 @@ async def default_estimate_async(
             return default_estimate
 
         service = ScenarioService()
-        service._registry = MagicMock()
+        service._registry = MagicMock(spec=ScenarioRegistry)
         service._registry.get_registered_class_metadata.return_value = metadata
         service._estimate_semaphore = asyncio.Semaphore(1)
         service._run_default_estimate_async = AsyncMock(side_effect=default_estimate_async)
         service._estimate_configured_run_size_async = AsyncMock(return_value=configured_estimate)
 
         catalog_task = asyncio.create_task(service._get_default_run_size_estimate_async(metadata=metadata))
-        await asyncio.wait_for(started.wait(), timeout=1)
-        interactive = await asyncio.wait_for(
-            service.estimate_scenario_run_size_async(
-                scenario_name=metadata.registry_name,
-                request=ScenarioRunSizeEstimateRequest(),
-            ),
-            timeout=1,
-        )
-        release.set()
+        configured_task = None
+        try:
+            async with asyncio.timeout(estimate_watchdog_seconds):
+                await started.wait()
+                with patch.object(
+                    service._estimate_semaphore,
+                    "acquire",
+                    new_callable=AsyncMock,
+                    wraps=service._estimate_semaphore.acquire,
+                ) as catalog_acquire:
+                    configured_task = asyncio.create_task(
+                        service.estimate_scenario_run_size_async(
+                            scenario_name=metadata.registry_name,
+                            request=ScenarioRunSizeEstimateRequest(),
+                        )
+                    )
+                    interactive = await configured_task
+                    catalog_acquire.assert_not_awaited()
+
+                assert interactive == configured_estimate
+                assert not release.is_set()
+                assert not catalog_task.done()
+                assert service._estimate_semaphore.locked()
+                service._estimate_configured_run_size_async.assert_awaited_once()
+                release.set()
+                assert await catalog_task == default_estimate
+        finally:
+            release.set()
+            async with asyncio.timeout(estimate_watchdog_seconds):
+                await asyncio.gather(*get_defined_tasks(catalog_task, configured_task), return_exceptions=True)
+                await service.close_async()
 
-        assert interactive == configured_estimate
-        assert await catalog_task == default_estimate
+        assert service.outstanding_estimates() == 0
+        assert service._estimate_tasks == {}
+        assert service._configured_estimate_tasks == {}
 
     async def test_concurrent_configured_estimates_share_one_task(self) -> None:
         """Equivalent configured requests share one cancellation-safe execution task."""