From c494ee2e42595892d4a213adaae586670736eb6f Mon Sep 17 00:00:00 2001 From: Vardhman Gupta Date: Sun, 1 Nov 2026 14:09:10 +0530 Subject: [PATCH 1/3] FIX: Reject concurrent SDK scenario runs and make attempt accounting atomic (#3748) --- pyrit/memory/memory_interface.py | 71 +++++ pyrit/scenario/core/scenario.py | 118 ++++--- .../test_interface_scenario_results.py | 75 ++++- tests/unit/scenario/core/test_scenario.py | 295 ++++++++++++++++++ 4 files changed, 507 insertions(+), 52 deletions(-) diff --git a/pyrit/memory/memory_interface.py b/pyrit/memory/memory_interface.py index e230165ae0..5db76a7830 100644 --- a/pyrit/memory/memory_interface.py +++ b/pyrit/memory/memory_interface.py @@ -7158,6 +7158,32 @@ def _execute_update_scenario_metadata_fields( flag_modified(entry, "scenario_metadata") session.commit() + def _execute_increment_scenario_number_tries( + self, + *, + scenario_result_id: str, + ) -> int: + """ + Atomically increment and return the number_tries counter for a scenario result. + + Args: + scenario_result_id (str): The ID of the scenario result. + + Returns: + int: The new number_tries value after incrementing. + + Raises: + ValueError: If the scenario result is not found in memory. + """ + with closing(self._get_session()) as session: + entry = session.query(ScenarioResultEntry).filter_by(id=scenario_result_id).first() + if not entry: + raise ValueError(f"Scenario result with ID {scenario_result_id} not found in memory") + entry.number_tries += 1 + new_tries = entry.number_tries + session.commit() + return new_tries + def _execute_get_scenario_result_header(self, *, scenario_result_id: str) -> ScenarioResult | None: """Return one ScenarioResult header without hydrating linked attack results.""" with closing(self._get_session()) as session: @@ -11747,6 +11773,51 @@ async def update_scenario_metadata_fields_async( self._execute_update_scenario_metadata_fields, scenario_result_id=scenario_result_id, fields=fields ) + def increment_scenario_number_tries( + self, + *, + scenario_result_id: str, + ) -> int: + """ + Use ``increment_scenario_number_tries_async``. + + This synchronous API is deprecated and can block the caller. + + Returns: + int: The new number_tries value after incrementing. + """ + print_deprecation_message( + old_item="MemoryInterface.increment_scenario_number_tries", + new_item="MemoryInterface.increment_scenario_number_tries_async", + removed_in="1.4.0", + ) + return self._execute_increment_scenario_number_tries( + scenario_result_id=scenario_result_id, + ) + + @legacy_sync_override(lambda: MemoryInterface.increment_scenario_number_tries) + async def increment_scenario_number_tries_async( + self, + *, + scenario_result_id: str, + ) -> int: + """ + Atomically increment and return the number_tries counter for a scenario result. + + Args: + scenario_result_id (str): The ID of the scenario result to update. + + Returns: + int: The new number_tries value after incrementing. + + Raises: + ValueError: If the scenario result is not found. + """ + return await self._run_database_operation_async( + self._execute_increment_scenario_number_tries, + scenario_result_id=scenario_result_id, + ) + def get_scenario_result_header(self, *, scenario_result_id: str) -> ScenarioResult | None: """ Use ``get_scenario_result_header_async``. diff --git a/pyrit/scenario/core/scenario.py b/pyrit/scenario/core/scenario.py index 3396be5b94..cb6cb2ddfd 100644 --- a/pyrit/scenario/core/scenario.py +++ b/pyrit/scenario/core/scenario.py @@ -11,6 +11,7 @@ import ast import asyncio import logging +import threading import uuid from abc import ABC, abstractmethod from collections.abc import Mapping, Sequence @@ -163,6 +164,16 @@ class Scenario(ABC): #: Maximum number of resolved concrete techniques; None leaves the selection unrestricted. MAX_CONCRETE_TECHNIQUES: ClassVar[int | None] = None + #: Active scenario result IDs currently executing in this process. + _active_runs: ClassVar[set[str]] = set() + _active_runs_lock: ClassVar[threading.Lock] = threading.Lock() + + @classmethod + def _clear_active_runs(cls) -> None: + """Clear active scenario runs. Intended for test isolation.""" + with cls._active_runs_lock: + cls._active_runs.clear() + def __init_subclass__(cls, **kwargs: Any) -> None: """ Enforce the keyword-only constructor contract on subclasses. @@ -1705,57 +1716,68 @@ async def run_async(self) -> ScenarioResult: # Type narrowing: create local variable that type checker knows is non-None scenario_result_id: str = self._scenario_result_id - # Implement retry logic - last_exception = None - for retry_attempt in range(self._max_retries + 1): # +1 for initial attempt - try: - return await self._execute_scenario_async() - except asyncio.CancelledError: + with Scenario._active_runs_lock: + if scenario_result_id in Scenario._active_runs: + raise RuntimeError( + f"Scenario '{self._name}' execution is already in progress for run '{scenario_result_id}'." + ) + Scenario._active_runs.add(scenario_result_id) + + try: + # Implement retry logic + last_exception = None + for retry_attempt in range(self._max_retries + 1): # +1 for initial attempt try: - ( - await self._memory.update_scenario_run_state_async( - scenario_result_id=scenario_result_id, - scenario_run_state=ScenarioRunState.CANCELLED, - error_message="Scenario run was cancelled", - error_type="CancelledError", + return await self._execute_scenario_async() + except asyncio.CancelledError: + try: + ( + await self._memory.update_scenario_run_state_async( + scenario_result_id=scenario_result_id, + scenario_run_state=ScenarioRunState.CANCELLED, + error_message="Scenario run was cancelled", + error_type="CancelledError", + ) ) + except Exception: + logger.exception(f"Failed to persist cancellation state for scenario '{self._name}'") + raise + except Exception as e: + last_exception = e + + # Get current scenario to check number of tries + scenario_results = await self._memory.get_scenario_results_async( + scenario_result_ids=[scenario_result_id] ) - except Exception: - logger.exception(f"Failed to persist cancellation state for scenario '{self._name}'") - raise - except Exception as e: - last_exception = e - - # Get current scenario to check number of tries - scenario_results = await self._memory.get_scenario_results_async( - scenario_result_ids=[scenario_result_id] - ) - current_tries = scenario_results[0].number_tries if scenario_results else retry_attempt + 1 + current_tries = scenario_results[0].number_tries if scenario_results else retry_attempt + 1 - # Check if we have more retries available - remaining_retries = self._max_retries - retry_attempt + # Check if we have more retries available + remaining_retries = self._max_retries - retry_attempt - if remaining_retries > 0: + if remaining_retries > 0: + logger.error( + f"Scenario '{self._name}' failed on attempt {current_tries} with error: {str(e)}. " + f"Retrying... ({remaining_retries} retries remaining)", + exc_info=True, + ) + # Continue to next iteration for retry + continue + # No more retries, log final failure logger.error( - f"Scenario '{self._name}' failed on attempt {current_tries} with error: {str(e)}. " - f"Retrying... ({remaining_retries} retries remaining)", + f"Scenario '{self._name}' failed after {current_tries} attempts " + f"(initial + {self._max_retries} retries) with error: {str(e)}. Giving up.", exc_info=True, ) - # Continue to next iteration for retry - continue - # No more retries, log final failure - logger.error( - f"Scenario '{self._name}' failed after {current_tries} attempts " - f"(initial + {self._max_retries} retries) with error: {str(e)}. Giving up.", - exc_info=True, - ) - (await self._mark_scenario_failed_async(scenario_result_id=scenario_result_id, error=e)) - raise + (await self._mark_scenario_failed_async(scenario_result_id=scenario_result_id, error=e)) + raise - # This should never be reached, but just in case - if last_exception: - raise last_exception - raise RuntimeError(f"Scenario '{self._name}' completed unexpectedly without result") + # This should never be reached, but just in case + if last_exception: + raise last_exception + raise RuntimeError(f"Scenario '{self._name}' completed unexpectedly without result") + finally: + with Scenario._active_runs_lock: + Scenario._active_runs.discard(scenario_result_id) async def _execute_scenario_async(self) -> ScenarioResult: """ @@ -1781,15 +1803,9 @@ async def _execute_scenario_async(self) -> ScenarioResult: raise ValueError("self._scenario_result_id is not initialized") scenario_result_id: str = self._scenario_result_id - # Increment number_tries at the start of each run - scenario_results = await self._memory.get_scenario_results_async(scenario_result_ids=[scenario_result_id]) - if scenario_results: - current_scenario = scenario_results[0] - current_scenario.number_tries += 1 - await self._memory.update_scenario_result_async(scenario_result=current_scenario) - logger.info(f"Scenario '{self._name}' attempt #{current_scenario.number_tries}") - else: - raise ValueError(f"Scenario result with ID {scenario_result_id} not found") + # Increment number_tries atomically at the start of each run + current_tries = await self._memory.increment_scenario_number_tries_async(scenario_result_id=scenario_result_id) + logger.info(f"Scenario '{self._name}' attempt #{current_tries}") # Mark scenario as in progress ( diff --git a/tests/unit/memory/memory_interface/test_interface_scenario_results.py b/tests/unit/memory/memory_interface/test_interface_scenario_results.py index a96b97dfdd..2d5ea0da50 100644 --- a/tests/unit/memory/memory_interface/test_interface_scenario_results.py +++ b/tests/unit/memory/memory_interface/test_interface_scenario_results.py @@ -1,11 +1,17 @@ # Copyright (c) Microsoft Corporation. # Licensed under the MIT license. +import asyncio from datetime import UTC, datetime, timedelta from uuid import uuid4 import pytest -from unit.mocks import get_mock_scorer_identifier, make_scenario_result, run_memory_session_async +from unit.mocks import ( + get_mock_scorer_identifier, + get_mock_target_identifier, + make_scenario_result, + run_memory_session_async, +) from pyrit.memory import MemoryInterface from pyrit.memory.memory_models import ( @@ -1172,3 +1178,70 @@ async def test_add_scenario_result_without_objective_target_raises(sqlite_instan (await sqlite_instance.add_scenario_results_to_memory_async(scenario_results=[scenario])) assert (await sqlite_instance.get_scenario_results_async(scenario_name="Targetless Scenario")) == [] + + +async def test_increment_scenario_number_tries_async(sqlite_instance: MemoryInterface): + """increment_scenario_number_tries_async atomically increments and returns the new try count.""" + scenario = make_scenario_result( + scenario_name="Tries Scenario", + scenario_version=1, + objective_target_identifier=get_mock_target_identifier(), + objective_scorer_identifier=get_mock_scorer_identifier(), + attack_results={}, + ) + await sqlite_instance.add_scenario_results_to_memory_async(scenario_results=[scenario]) + scenario_id = str(scenario.id) + + first = await sqlite_instance.increment_scenario_number_tries_async(scenario_result_id=scenario_id) + assert first == 1 + + second = await sqlite_instance.increment_scenario_number_tries_async(scenario_result_id=scenario_id) + assert second == 2 + + stored = (await sqlite_instance.get_scenario_results_async(scenario_result_ids=[scenario_id]))[0] + assert stored.number_tries == 2 + + +async def test_increment_scenario_number_tries_sync(sqlite_instance: MemoryInterface): + """Synchronous increment_scenario_number_tries works and updates try count.""" + scenario = make_scenario_result( + scenario_name="Sync Tries Scenario", + scenario_version=1, + objective_target_identifier=get_mock_target_identifier(), + objective_scorer_identifier=get_mock_scorer_identifier(), + attack_results={}, + ) + await sqlite_instance.add_scenario_results_to_memory_async(scenario_results=[scenario]) + scenario_id = str(scenario.id) + + with pytest.deprecated_call(): + first = sqlite_instance.increment_scenario_number_tries(scenario_result_id=scenario_id) + assert first == 1 + + +async def test_increment_scenario_number_tries_missing_id_raises(sqlite_instance: MemoryInterface): + """increment_scenario_number_tries_async raises ValueError when scenario_result_id does not exist.""" + missing_id = str(uuid4()) + with pytest.raises(ValueError, match=f"Scenario result with ID {missing_id} not found in memory"): + await sqlite_instance.increment_scenario_number_tries_async(scenario_result_id=missing_id) + + +async def test_increment_scenario_number_tries_concurrent(sqlite_instance: MemoryInterface): + """Concurrent increments are serialized by the transaction and never lose counts.""" + scenario = make_scenario_result( + scenario_name="Concurrent Tries Scenario", + scenario_version=1, + objective_target_identifier=get_mock_target_identifier(), + objective_scorer_identifier=get_mock_scorer_identifier(), + attack_results={}, + ) + await sqlite_instance.add_scenario_results_to_memory_async(scenario_results=[scenario]) + scenario_id = str(scenario.id) + + results = await asyncio.gather( + *[sqlite_instance.increment_scenario_number_tries_async(scenario_result_id=scenario_id) for _ in range(10)] + ) + assert sorted(results) == list(range(1, 11)) + + stored = (await sqlite_instance.get_scenario_results_async(scenario_result_ids=[scenario_id]))[0] + assert stored.number_tries == 10 diff --git a/tests/unit/scenario/core/test_scenario.py b/tests/unit/scenario/core/test_scenario.py index c87ddd3685..f0b8a6f9ac 100644 --- a/tests/unit/scenario/core/test_scenario.py +++ b/tests/unit/scenario/core/test_scenario.py @@ -115,6 +115,35 @@ async def mock_run_async(*args, **kwargs): return AsyncMock(side_effect=mock_run_async) +def _make_mock_atomic_attacks(count: int = 6): + """Create a list of mock AtomicAttack instances and matching AttackResults for testing.""" + attacks = [] + results = [] + mock_attack = MagicMock() + mock_attack.get_objective_target.return_value = MagicMock() + mock_attack.get_attack_scoring_config.return_value = MagicMock() + for i in range(count): + run = MagicMock(spec=AtomicAttack) + run.group_kind = ScenarioRunPlanGroupKind.ATTACK + run.seed_group_adaptations = {} + run.atomic_attack_name = f"attack_run_{i}" + run.display_group = f"attack_run_{i}" + run._attack = mock_attack + run._scenario_result_id = None + run.set_scenario_result_id = MagicMock(side_effect=lambda sid, r=run: setattr(r, "_scenario_result_id", sid)) + type(run).objectives = PropertyMock(return_value=[f"objective{i}"]) + attacks.append(run) + results.append( + AttackResult( + conversation_id=f"conv-{i}", + objective=f"objective{i}", + outcome=AttackOutcome.SUCCESS, + executed_turns=1, + ) + ) + return attacks, results + + @pytest.fixture def mock_atomic_attacks(): """Create mock AtomicAttack instances for testing.""" @@ -2601,3 +2630,269 @@ async def test_max_concurrency_one_serializes_via_single_worker( for run in mock_atomic_attacks: run.run_async.assert_called_once_with(executor=ANY, return_partial_on_failure=True, operation_id=None) + + async def test_concurrent_run_async_rejects_second_execution_before_target_work( + self, sqlite_instance: MemoryInterface, mock_objective_target + ): + """Two concurrent run_async calls on the same initialized scenario reject the second before target work.""" + Scenario._clear_active_runs() + attacks, sample_results = _make_mock_atomic_attacks(count=6) + + target_calls = 0 + in_flight = 0 + peak_concurrency = 0 + target_lock = asyncio.Lock() + + def make_run_async(idx): + async def run_async(*, executor, **kwargs): + nonlocal target_calls, in_flight, peak_concurrency + semaphore = executor._get_semaphore() + async with semaphore: + async with target_lock: + target_calls += 1 + in_flight += 1 + peak_concurrency = max(peak_concurrency, in_flight) + await asyncio.sleep(0.01) + async with target_lock: + in_flight -= 1 + + _stamp_scenario_linkage( + attack_results=[sample_results[idx]], + atomic_attack=attacks[idx], + ) + await save_attack_results_to_memory_async([sample_results[idx]]) + return AttackExecutorResult( + completed_results=[sample_results[idx]], + incomplete_objectives=[], + ) + + return AsyncMock(side_effect=run_async) + + for i, attack in enumerate(attacks): + attack.run_async = make_run_async(i) + + scenario = ConcreteScenario( + name="Test Concurrent Scenario", + version=1, + atomic_attacks_to_return=attacks, + ) + scenario.set_params_from_args( + args={ + "objective_target": mock_objective_target, + "max_concurrency": 2, + } + ) + await scenario.initialize_async() + run_id = scenario._scenario_result_id + + # Invoke two concurrent run_async calls on the same initialized scenario instance + results = await asyncio.gather(scenario.run_async(), scenario.run_async(), return_exceptions=True) + + succeeded = [r for r in results if isinstance(r, ScenarioResult)] + rejected = [r for r in results if isinstance(r, RuntimeError)] + + assert len(succeeded) == 1, f"Expected 1 succeeded run, got {len(succeeded)}" + assert len(rejected) == 1, f"Expected 1 rejected run, got {len(rejected)}" + assert f"execution is already in progress for run '{run_id}'" in str(rejected[0]) + + # Exact expected metrics from Issue #3748: + assert peak_concurrency <= 2, f"Peak concurrency {peak_concurrency} exceeded max_concurrency 2" + assert target_calls == 6, f"Expected exactly 6 target calls, got {target_calls}" + assert len(succeeded[0].attack_results) == 6 + + # Check SQLite persistence + db_results = await sqlite_instance.get_attack_results_async(result_selection=AttackResultSelection.ALL_RESULTS) + assert len(db_results) == 6, f"Expected 6 persisted attack results, got {len(db_results)}" + + db_scenarios = await sqlite_instance.get_scenario_results_async(scenario_result_ids=[run_id]) + assert len(db_scenarios) == 1 + assert db_scenarios[0].number_tries == 1 + assert db_scenarios[0].scenario_run_state == ScenarioRunState.COMPLETED + + async def test_concurrent_run_async_on_different_instances_same_run_id_rejects_second( + self, sqlite_instance: MemoryInterface, mock_objective_target + ): + """Two concurrent run_async calls on different Scenario instances sharing a run_id reject the second.""" + Scenario._clear_active_runs() + attacks, sample_results = _make_mock_atomic_attacks(count=3) + + for i, attack in enumerate(attacks): + attack.run_async = create_mock_run_async([sample_results[i]], atomic_attack=attack) + + scenario1 = ConcreteScenario( + name="Test Scenario", + version=1, + atomic_attacks_to_return=attacks, + ) + scenario1.set_params_from_args(args={"objective_target": mock_objective_target, "max_concurrency": 2}) + await scenario1.initialize_async() + run_id = scenario1._scenario_result_id + + scenario2 = ConcreteScenario( + name="Test Scenario", + version=1, + atomic_attacks_to_return=attacks, + scenario_result_id=run_id, + ) + scenario2.set_params_from_args(args={"objective_target": mock_objective_target, "max_concurrency": 2}) + await scenario2.initialize_async() + + results = await asyncio.gather(scenario1.run_async(), scenario2.run_async(), return_exceptions=True) + succeeded = [r for r in results if isinstance(r, ScenarioResult)] + rejected = [r for r in results if isinstance(r, RuntimeError)] + + assert len(succeeded) == 1 + assert len(rejected) == 1 + assert f"execution is already in progress for run '{run_id}'" in str(rejected[0]) + + async def test_cancellation_cleans_up_guard_and_allows_legitimate_resume( + self, sqlite_instance: MemoryInterface, mock_objective_target + ): + """Cancellation releases the same-run guard and marks CANCELLED; subsequent run_async resumes cleanly.""" + Scenario._clear_active_runs() + attacks, sample_results = _make_mock_atomic_attacks(count=6) + + two_completed = asyncio.Event() + completed_count = 0 + pause_event = asyncio.Event() + + def make_cancellable_run_async(idx): + async def run_async(*, executor, **kwargs): + nonlocal completed_count + if idx < 2: + _stamp_scenario_linkage(attack_results=[sample_results[idx]], atomic_attack=attacks[idx]) + await save_attack_results_to_memory_async([sample_results[idx]]) + completed_count += 1 + if completed_count == 2: + two_completed.set() + return AttackExecutorResult(completed_results=[sample_results[idx]], incomplete_objectives=[]) + await pause_event.wait() + _stamp_scenario_linkage(attack_results=[sample_results[idx]], atomic_attack=attacks[idx]) + await save_attack_results_to_memory_async([sample_results[idx]]) + return AttackExecutorResult(completed_results=[sample_results[idx]], incomplete_objectives=[]) + + return AsyncMock(side_effect=run_async) + + for i, attack in enumerate(attacks): + attack.run_async = make_cancellable_run_async(i) + + scenario = ConcreteScenario( + name="Cancellable Scenario", + version=1, + atomic_attacks_to_return=attacks, + ) + scenario.set_params_from_args(args={"objective_target": mock_objective_target, "max_concurrency": 2}) + await scenario.initialize_async() + run_id = scenario._scenario_result_id + + task = asyncio.create_task(scenario.run_async()) + await two_completed.wait() + + # Cancel the active run while in progress + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + + # Verify guard was released after worker cleanup + assert run_id not in Scenario._active_runs + + # Verify state in SQLite is CANCELLED and 2 results saved + stored = (await sqlite_instance.get_scenario_results_async(scenario_result_ids=[run_id]))[0] + assert stored.scenario_run_state == ScenarioRunState.CANCELLED + assert stored.number_tries == 1 + + partial_attack_results = await sqlite_instance.get_attack_results_async( + result_selection=AttackResultSelection.ALL_RESULTS + ) + assert len(partial_attack_results) == 2 + + # Legitimate later resume on the same scenario instance + pause_event.set() + resume_result = await scenario.run_async() + + assert resume_result.scenario_run_state == ScenarioRunState.COMPLETED + assert run_id not in Scenario._active_runs + + # Stored in SQLite: all 6 attack results, number_tries is 2 + final_stored = (await sqlite_instance.get_scenario_results_async(scenario_result_ids=[run_id]))[0] + assert final_stored.scenario_run_state == ScenarioRunState.COMPLETED + assert final_stored.number_tries == 2 + + final_attack_results = await sqlite_instance.get_attack_results_async( + result_selection=AttackResultSelection.ALL_RESULTS + ) + assert len(final_attack_results) == 6 + + async def test_concurrent_run_async_distinct_scenarios_run_in_parallel( + self, sqlite_instance: MemoryInterface, mock_objective_target + ): + """Two different scenarios with distinct run_ids run concurrently without blocking each other.""" + Scenario._clear_active_runs() + attacks1, sample_results1 = _make_mock_atomic_attacks(count=2) + attacks2, sample_results2 = _make_mock_atomic_attacks(count=2) + + for i, a in enumerate(attacks1): + a.run_async = create_mock_run_async([sample_results1[i]], atomic_attack=a) + for i, a in enumerate(attacks2): + a.run_async = create_mock_run_async([sample_results2[i]], atomic_attack=a) + + scenario1 = ConcreteScenario(name="Distinct Scenario 1", version=1, atomic_attacks_to_return=attacks1) + scenario1.set_params_from_args(args={"objective_target": mock_objective_target, "max_concurrency": 2}) + await scenario1.initialize_async() + + scenario2 = ConcreteScenario(name="Distinct Scenario 2", version=1, atomic_attacks_to_return=attacks2) + scenario2.set_params_from_args(args={"objective_target": mock_objective_target, "max_concurrency": 2}) + await scenario2.initialize_async() + + res1, res2 = await asyncio.gather(scenario1.run_async(), scenario2.run_async()) + assert isinstance(res1, ScenarioResult) + assert isinstance(res2, ScenarioResult) + assert res1.id != res2.id + + async def test_cancellation_state_persistence_failure_logs_and_propagates( + self, sqlite_instance: MemoryInterface, mock_objective_target + ): + """When persisting CANCELLED state in memory raises, the error is logged and CancelledError propagates.""" + Scenario._clear_active_runs() + attacks, sample_results = _make_mock_atomic_attacks(count=1) + + async def cancel_run(*args, **kwargs): + raise asyncio.CancelledError + + attacks[0].run_async = AsyncMock(side_effect=cancel_run) + + scenario = ConcreteScenario(name="Cancel Fail Scenario", version=1, atomic_attacks_to_return=attacks) + scenario.set_params_from_args(args={"objective_target": mock_objective_target}) + await scenario.initialize_async() + + orig_update = sqlite_instance.update_scenario_run_state_async + + async def mock_update(*, scenario_run_state, **kwargs): + if scenario_run_state == ScenarioRunState.CANCELLED: + raise RuntimeError("DB disconnected") + return await orig_update(scenario_run_state=scenario_run_state, **kwargs) + + with patch.object(sqlite_instance, "update_scenario_run_state_async", side_effect=mock_update): + with pytest.raises(asyncio.CancelledError): + await scenario.run_async() + + assert scenario._scenario_result_id not in Scenario._active_runs + + async def test_run_async_raises_runtime_error_when_loop_completes_without_result( + self, sqlite_instance: MemoryInterface, mock_objective_target + ): + """When retry loop does not produce a result, raises RuntimeError.""" + Scenario._clear_active_runs() + attacks, _ = _make_mock_atomic_attacks(count=1) + + scenario = ConcreteScenario(name="No Result Scenario", version=1, atomic_attacks_to_return=attacks) + scenario.set_params_from_args(args={"objective_target": mock_objective_target}) + await scenario.initialize_async() + + # Force empty retry range + scenario._max_retries = -1 + + with pytest.raises(RuntimeError, match="completed unexpectedly without result"): + await scenario.run_async() + + assert scenario._scenario_result_id not in Scenario._active_runs From c10e5479fff39ec28c88d26d300de528d547a028 Mon Sep 17 00:00:00 2001 From: Vardhman Gupta Date: Sun, 1 Nov 2026 14:43:32 +0530 Subject: [PATCH 2/3] MAINT: Trigger CI runner re-execution From 17066d46649560c31dab3eafc3885d3da18f9997 Mon Sep 17 00:00:00 2001 From: Vardhman Gupta Date: Mon, 2 Nov 2026 11:03:16 +0530 Subject: [PATCH 3/3] TEST: Add event rendezvous and watchdog to parallel scenario test (#3748) --- tests/unit/scenario/core/test_scenario.py | 52 ++++++++++++++++++++--- 1 file changed, 46 insertions(+), 6 deletions(-) diff --git a/tests/unit/scenario/core/test_scenario.py b/tests/unit/scenario/core/test_scenario.py index f0b8a6f9ac..6cf19c11e7 100644 --- a/tests/unit/scenario/core/test_scenario.py +++ b/tests/unit/scenario/core/test_scenario.py @@ -2826,15 +2826,45 @@ async def run_async(*, executor, **kwargs): async def test_concurrent_run_async_distinct_scenarios_run_in_parallel( self, sqlite_instance: MemoryInterface, mock_objective_target ): - """Two different scenarios with distinct run_ids run concurrently without blocking each other.""" + """Two different scenarios with distinct run_ids run concurrently without blocking each other. + + Uses an event-based rendezvous barrier so both scenarios must actively enter + execution in parallel before either can complete, strictly verifying that + unrelated scenarios are never serialized. + """ Scenario._clear_active_runs() attacks1, sample_results1 = _make_mock_atomic_attacks(count=2) attacks2, sample_results2 = _make_mock_atomic_attacks(count=2) - for i, a in enumerate(attacks1): - a.run_async = create_mock_run_async([sample_results1[i]], atomic_attack=a) - for i, a in enumerate(attacks2): - a.run_async = create_mock_run_async([sample_results2[i]], atomic_attack=a) + entered1 = asyncio.Event() + entered2 = asyncio.Event() + + base_run_1 = create_mock_run_async([sample_results1[0]], atomic_attack=attacks1[0]) + base_run_2 = create_mock_run_async([sample_results2[0]], atomic_attack=attacks2[0]) + + async def run_with_rendezvous( + entered_this: asyncio.Event, + entered_other: asyncio.Event, + base_run: Any, + *args: Any, + **kwargs: Any, + ) -> Any: + entered_this.set() + # Both scenarios must enter their attack execution before either can proceed + await asyncio.wait_for(entered_other.wait(), timeout=5.0) + return await base_run(*args, **kwargs) + + async def attack1_run(*args: Any, **kwargs: Any) -> Any: + return await run_with_rendezvous(entered1, entered2, base_run_1, *args, **kwargs) + + async def attack2_run(*args: Any, **kwargs: Any) -> Any: + return await run_with_rendezvous(entered2, entered1, base_run_2, *args, **kwargs) + + attacks1[0].run_async = AsyncMock(side_effect=attack1_run) + attacks1[1].run_async = create_mock_run_async([sample_results1[1]], atomic_attack=attacks1[1]) + + attacks2[0].run_async = AsyncMock(side_effect=attack2_run) + attacks2[1].run_async = create_mock_run_async([sample_results2[1]], atomic_attack=attacks2[1]) scenario1 = ConcreteScenario(name="Distinct Scenario 1", version=1, atomic_attacks_to_return=attacks1) scenario1.set_params_from_args(args={"objective_target": mock_objective_target, "max_concurrency": 2}) @@ -2844,7 +2874,17 @@ async def test_concurrent_run_async_distinct_scenarios_run_in_parallel( scenario2.set_params_from_args(args={"objective_target": mock_objective_target, "max_concurrency": 2}) await scenario2.initialize_async() - res1, res2 = await asyncio.gather(scenario1.run_async(), scenario2.run_async()) + task1 = asyncio.create_task(scenario1.run_async()) + task2 = asyncio.create_task(scenario2.run_async()) + + try: + res1, res2 = await asyncio.wait_for(asyncio.gather(task1, task2), timeout=10.0) + except BaseException: + task1.cancel() + task2.cancel() + await asyncio.gather(task1, task2, return_exceptions=True) + raise + assert isinstance(res1, ScenarioResult) assert isinstance(res2, ScenarioResult) assert res1.id != res2.id