TEST Use shared watchdog for backend correctness helpers

doc/contributing/4_running_tests.md

@@ -94,6 +94,14 @@ timeout coverage expires the real cancellation path only after the request start
 Event-controlled analytics expiry tests freeze both deadline clocks so floating-point
 rounding cannot turn a valid query budget into a failed assertion.
 
+Backend manual-send completion and first-request analytics correctness tests also
+use the shared watchdog. Manual sends are polled until a terminal snapshot, including
+finalization. Expiring the test's wait does not cancel the send; its lifecycle owner
+still drains pending work before mocks and memory close. Explicit zero-timeout tests
+check this ownership without requiring the send to finish quickly.
+The async and repeated manual-send test classes use 600-second whole-test backstops
+so the 300-second completion watchdog can expire and allow cleanup before pytest stops the test.
+
 The target-type route test starts target metadata discovery, waits for a signal
 that it has started, and keeps it paused while checking that `/api/health`
 responds within 10 seconds. Waiting for that signal does not use an extra worker

tests/unit/backend/conftest.py

@@ -19,12 +19,18 @@
 from pyrit.backend.services.message_send_service import get_message_send_service
 from pyrit.memory import SQLiteMemory
 from test_support.fixtures import authentication_policy, compatibility_headers, compatibility_id  # noqa: F401
+from unit.async_utils import TEST_WATCHDOG_SECONDS
 
 
 @pytest.fixture
 async def analytics_service(sqlite_instance: SQLiteMemory) -> AsyncGenerator[AttackResultAnalytics, None]:
     """Use real isolated memory with finite correctness budgets, not production latency assertions."""
-    execution = partial(AnalyticsExecution, queue_timeout=300.0, report_timeout=300.0, quick_timeout=300.0)
+    execution = partial(
+        AnalyticsExecution,
+        queue_timeout=TEST_WATCHDOG_SECONDS,
+        report_timeout=TEST_WATCHDOG_SECONDS,
+        quick_timeout=TEST_WATCHDOG_SECONDS,
+    )
     assert get_analytics_service.cache_info().currsize == 0
     with patch("pyrit.analytics.attack_result_analytics.AnalyticsExecution", side_effect=execution):
         try:

tests/unit/backend/mocks.py

@@ -14,12 +14,13 @@
 from pyrit.memory import MemoryInterface
 from pyrit.models import AtomicAttackIdentifier, AttackOutcome, AttackResult, ComponentIdentifier
 from pyrit.prompt_target import PromptTarget
+from unit.async_utils import TEST_WATCHDOG_SECONDS
 
 
 async def _settle_send_async(
-    *, service: MessageSendService, status: MessageSendStatus, timeout_seconds: float = 10
+    *, service: MessageSendService, status: MessageSendStatus, timeout_seconds: float = TEST_WATCHDOG_SECONDS
 ) -> MessageSendStatus:
-    """Wait for a terminal snapshot without cancelling the send on timeout."""
+    """Wait for a terminal snapshot with a deadlock backstop, without cancelling the send."""
     async with asyncio.timeout(timeout_seconds):
         while status.state not in (MessageSendState.COMPLETED, MessageSendState.FAILED, MessageSendState.INTERRUPTED):
             status = await service.get_status_async(

tests/unit/backend/test_analytics_routes.py

@@ -43,7 +43,7 @@
     AttackAnalyticsResultsQuery,
     AttackOutcome,
 )
-from unit.async_utils import wait_for_completion_async, wait_for_event_async
+from unit.async_utils import TEST_WATCHDOG_SECONDS, wait_for_completion_async, wait_for_event_async
 from unit.memory.test_attack_analytics import make_result
 
 
@@ -124,7 +124,12 @@ def create() -> AttackResultAnalytics:
 
     assert get_analytics_service.cache_info().currsize == 0
     user = AuthenticatedUser(oid="test", name="Test", email="test@example.invalid", groups=["test-group"])
-    budgets = partial(AnalyticsExecution, report_timeout=30.0, quick_timeout=30.0, queue_timeout=30.0)
+    budgets = partial(
+        AnalyticsExecution,
+        report_timeout=TEST_WATCHDOG_SECONDS,
+        quick_timeout=TEST_WATCHDOG_SECONDS,
+        queue_timeout=TEST_WATCHDOG_SECONDS,
+    )
     with (
         patch("pyrit.backend.services.analytics_service.AttackResultAnalytics", side_effect=create),
         patch("pyrit.analytics.attack_result_analytics.AnalyticsExecution", side_effect=budgets),
@@ -142,6 +147,9 @@ def create() -> AttackResultAnalytics:
             assert all(response.status_code == 200 for response in responses)
             assert len(owners) == 1
             assert get_analytics_service() is owners[0]
+            execution = owners[0]._execution()
+            assert execution._queue_timeout == TEST_WATCHDOG_SECONDS
+            assert all(lane.timeout == TEST_WATCHDOG_SECONDS for lane in execution._lanes.values())
         finally:
             await close_analytics_service_async()
 

tests/unit/backend/test_message_send_service.py

@@ -4,6 +4,7 @@
 """Tests for the shared synchronous manual-message owner."""
 
 import asyncio
+import inspect
 import uuid
 from collections.abc import AsyncGenerator, AsyncIterator, Generator, Iterator, Sequence
 from contextlib import asynccontextmanager, contextmanager
@@ -2786,7 +2787,7 @@ def _submission(
     )
 
 
-@pytest.mark.timeout(20)
+@pytest.mark.timeout(2 * TEST_WATCHDOG_SECONDS)
 @pytest.mark.usefixtures("patch_central_database")
 class TestAsyncMessageSend:
     async def test_repetitions_cannot_bypass_original_capabilities_through_an_alias_async(
@@ -2883,6 +2884,48 @@ async def test_settle_send_waits_past_finalizing_async(self, terminal_state: Mes
             for call in service.get_status_async.await_args_list
         )
 
+    def test_settle_send_defaults_to_shared_watchdog(self, request: pytest.FixtureRequest) -> None:
+        parameter = inspect.signature(_settle_send_async).parameters["timeout_seconds"]
+        assert parameter.default == TEST_WATCHDOG_SECONDS
+        timeout = request.node.get_closest_marker("timeout")
+        assert timeout is not None
+        assert timeout.args[0] > TEST_WATCHDOG_SECONDS
+
+    async def test_settle_send_timeout_preserves_send_until_success_async(
+        self,
+        *,
+        message_send_service: MessageSendService,
+        send_dependencies: tuple[MagicMock, AsyncMock],
+    ) -> None:
+        started: asyncio.Future[None] = asyncio.get_running_loop().create_future()
+        release = asyncio.Event()
+
+        async def hold_send_async(**_: Any) -> None:
+            started.set_result(None)
+            await release.wait()
+
+        send_dependencies[1].side_effect = hold_send_async
+        async with message_send_lifecycle_async(message_send_service):
+            try:
+                status = await message_send_service.submit_async(attack_result_id="attack", request=_submission())
+                await wait_for_completion_async(future=started)
+                task = message_send_service._sends[status.send_id].task
+                assert task is not None
+
+                with pytest.raises(TimeoutError):
+                    await _settle_send_async(service=message_send_service, status=status, timeout_seconds=0)
+                assert not task.done()
+                assert task.cancelling() == 0
+
+                release.set()
+                final = await _settle_send_async(service=message_send_service, status=status)
+                assert final.state == MessageSendState.COMPLETED
+                assert task.done()
+                assert not task.cancelled()
+                assert not message_send_service._scheduler._conversations
+            finally:
+                release.set()
+
     @pytest.mark.parametrize("failure", ["assertion", "timeout"])
     async def test_send_lifecycle_drains_pending_work_on_test_failure_async(
         self,
@@ -2913,9 +2956,10 @@ async def hold_send_async(**_: Any) -> None:
                 if failure == "assertion":
                     raise AssertionError("controlled test failure")
                 try:
-                    await _settle_send_async(service=message_send_service, status=status, timeout_seconds=0.01)
+                    await _settle_send_async(service=message_send_service, status=status, timeout_seconds=0)
                 except TimeoutError:
                     assert not task.done()
+                    assert task.cancelling() == 0
                     assert not cancelled.is_set()
                     raise
 
@@ -3537,9 +3581,14 @@ async def hold_write_async(**_: Any) -> None:
         ).state == MessageSendState.INTERRUPTED
 
 
-@pytest.mark.timeout(20)
+@pytest.mark.timeout(2 * TEST_WATCHDOG_SECONDS)
 @pytest.mark.usefixtures("patch_central_database")
 class TestRepeatedMessageSend:
+    def test_whole_test_timeout_exceeds_send_watchdog(self, request: pytest.FixtureRequest) -> None:
+        timeout = request.node.get_closest_marker("timeout")
+        assert timeout is not None
+        assert timeout.args[0] > TEST_WATCHDOG_SECONDS
+
     async def test_missing_validated_target_fails_before_branch_registration_async(
         self,
         *,