[Vardhman Gupta (Kaap10)]

Description

Fixes #3748

When SDK callers execute concurrent tasks against the same Scenario instance or multiple Scenario instances sharing the same persisted scenario_result_id (via scenario_result_id resume), concurrent run_async() executions bypass the planned concurrency limit, duplicating target calls and attack results while racing on number_tries.

Changes:

• Scenario (pyrit/scenario/core/scenario.py):
  • Added class-level active run guard _active_runs: ClassVar[set[str]] protected by _active_runs_lock: ClassVar[threading.Lock].
  • In run_async(), acquires ownership of _scenario_result_id before starting the retry loop or scheduling target work. If already active in another task or thread, raises RuntimeError immediately with a clear diagnostic message.
  • Releases guard in the outermost finally: block of run_async(), guaranteeing cleanup on normal completion, unhandled exceptions, and cancellation.
  • Prior to releasing the guard upon cancellation, target worker tasks in _execute_atomic_attacks_parallel_async() are drained to ensure background calls have completed.
  • Replaced the vulnerable read-modify-write pattern on number_tries with atomic accounting via increment_scenario_number_tries_async().
• MemoryInterface (pyrit/memory/memory_interface.py):
  • Added increment_scenario_number_tries_async(scenario_result_id: str) to MemoryInterface and database implementations using @legacy_sync_override.
  • Executes atomic SQL update: UPDATE ScenarioResult SET number_tries = number_tries + 1 WHERE id = :id.
  • Added deprecated synchronous wrapper increment_scenario_number_tries(scenario_result_id: str) per PyRIT memory interface conventions.

cc Roman Lutz (@romanlutz)

Tests and Documentation

• Exact Issue Concurrent SDK run_async calls duplicate one persisted run and exceed its concurrency limit #3748 Reproduction Test: Added test_concurrent_run_async_rejects_second_execution_before_target_work in tests/unit/scenario/core/test_scenario.py. Verifies that two concurrent tasks with 6 objectives and max_concurrency=2 result in 1 task executing cleanly, the second task rejected with RuntimeError before target work, peak concurrency strictly <= 2, exactly 6 target calls, exactly 6 attack results, and number_tries == 1.
• Cross-Instance Same-Run Rejection: Added test_concurrent_run_async_on_different_instances_same_run_id_rejects_second to ensure multiple Scenario objects sharing the same run ID cannot run concurrently.
• Cancellation & Legitimate Resume: Added test_cancellation_cleans_up_guard_and_allows_legitimate_resume testing with real SQLite to verify cancellation safely releases the guard and enables subsequent legitimate resumption to complete the remaining objectives and reach number_tries = 2.
• Concurrency Isolation: Added test_concurrent_run_async_distinct_scenarios_run_in_parallel ensuring distinct scenarios run concurrently without mutual blocking.
• Atomic Accounting Tests: Added test_increment_scenario_number_tries_async, test_increment_scenario_number_tries_sync, test_increment_scenario_number_tries_missing_id_raises, and test_increment_scenario_number_tries_concurrent in tests/unit/memory/memory_interface/test_interface_scenario_results.py.
• Test Suite Verification:
  • uv run -m pytest tests/unit/scenario/core/test_scenario.py tests/unit/memory/memory_interface/test_interface_scenario_results.py: 167 passed in 25.97s.
  • Diff coverage on modified code: 98% (exceeds 90% requirement).
• Code Quality:
  • uv run -m ruff format: clean.
  • uv run -m ruff check: 0 errors.
• JupyText / Documentation: Documentation was not modified as internal execution concurrency guards and atomic attempt accounting do not change the public user-facing API surface.