From 9338b24cc1523a2b5742310d7eb39efa0f75f30e Mon Sep 17 00:00:00 2001 From: Copilot <223556219+Copilot@users.noreply.github.com> Date: Tue, 4 Aug 2026 01:36:19 -0700 Subject: [PATCH 1/2] FIX: Correlate multi-turn attack errors Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 0d1f97bd-49a5-4a6d-b814-fecef1e2a8cb --- pyrit/executor/attack/core/attack_strategy.py | 7 ++++-- .../attack/core/test_attack_strategy.py | 22 +++++++++++++++++++ 2 files changed, 27 insertions(+), 2 deletions(-) diff --git a/pyrit/executor/attack/core/attack_strategy.py b/pyrit/executor/attack/core/attack_strategy.py index 6bab4492b7..11c58b60ad 100644 --- a/pyrit/executor/attack/core/attack_strategy.py +++ b/pyrit/executor/attack/core/attack_strategy.py @@ -355,8 +355,11 @@ async def _on_error_async( collector = get_retry_collector() retry_events = collector.events if collector else [] - # Build a conversation_id — use context's if available, otherwise generate one - conversation_id = getattr(context, "conversation_id", None) or str(uuid.uuid4()) + # Multi-turn contexts keep the active ID on their conversation session. + conversation_id = getattr(context, "conversation_id", None) + if not conversation_id: + conversation_id = getattr(getattr(context, "session", None), "conversation_id", None) + conversation_id = conversation_id or str(uuid.uuid4()) error_result = AttackResult( conversation_id=conversation_id, diff --git a/tests/unit/executor/attack/core/test_attack_strategy.py b/tests/unit/executor/attack/core/test_attack_strategy.py index f4f8701245..433f6f3df9 100644 --- a/tests/unit/executor/attack/core/test_attack_strategy.py +++ b/tests/unit/executor/attack/core/test_attack_strategy.py @@ -15,6 +15,7 @@ AttackStrategy, _DefaultAttackStrategyEventHandler, ) +from pyrit.executor.attack.multi_turn.multi_turn_attack_strategy import ConversationSession, MultiTurnAttackContext from pyrit.executor.core import StrategyEvent, StrategyEventData from pyrit.memory.central_memory import CentralMemory from pyrit.models import ( @@ -631,6 +632,27 @@ async def test_on_error_persists_result_to_memory(self, sample_attack_context, m assert stored_result.error_type == "ValueError" assert stored_result.execution_time_ms == 500 + async def test_on_error_uses_multi_turn_session_conversation_id(self, mock_memory): + """Test that multi-turn failures remain correlated with their active conversation.""" + context = MultiTurnAttackContext( + params=AttackParameters(objective="Test harmful objective"), + session=ConversationSession(conversation_id="active-conversation-id"), + ) + + with patch("pyrit.memory.central_memory.CentralMemory.get_memory_instance", return_value=mock_memory): + handler = _DefaultAttackStrategyEventHandler() + event_data = StrategyEventData( + event=StrategyEvent.ON_ERROR, + strategy_name="TestStrategy", + strategy_id="test-id", + context=context, + error=TimeoutError("target timed out"), + ) + await handler.on_event_async(event_data) + + stored_result = mock_memory.add_attack_results_to_memory.call_args.kwargs["attack_results"][0] + assert stored_result.conversation_id == "active-conversation-id" + async def test_on_error_skips_when_no_error_or_context(self, mock_memory): """Test that error handler returns early when error or context is None""" with patch("pyrit.memory.central_memory.CentralMemory.get_memory_instance", return_value=mock_memory): From 9fd26b702dd4583a37f243528b986bb90bfe58bb Mon Sep 17 00:00:00 2001 From: Copilot <223556219+Copilot@users.noreply.github.com> Date: Thu, 6 Aug 2026 08:12:39 -0700 Subject: [PATCH 2/2] TEST Add Crescendo scenario resume coverage Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- .../core/test_scenario_crescendo_resume.py | 460 ++++++++++++++++++ 1 file changed, 460 insertions(+) create mode 100644 tests/unit/scenario/core/test_scenario_crescendo_resume.py diff --git a/tests/unit/scenario/core/test_scenario_crescendo_resume.py b/tests/unit/scenario/core/test_scenario_crescendo_resume.py new file mode 100644 index 0000000000..803ae3f09e --- /dev/null +++ b/tests/unit/scenario/core/test_scenario_crescendo_resume.py @@ -0,0 +1,460 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT license. + +import json +from collections.abc import Awaitable, Callable, Sequence +from typing import Any, ClassVar +from unittest.mock import AsyncMock, patch + +import httpx +import pytest +from openai import APITimeoutError, RateLimitError +from openai.types.chat import ChatCompletion, ChatCompletionMessage +from openai.types.chat.chat_completion import Choice + +from pyrit.exceptions import ScenarioPartialFailureException +from pyrit.executor.attack import AttackAdversarialConfig, AttackScoringConfig, CrescendoAttack +from pyrit.memory import CentralMemory +from pyrit.models import ( + AttackOutcome, + AttackSeedGroup, + ConversationRetryReason, + ConversationType, + ScenarioRunState, + SeedObjective, +) +from pyrit.prompt_target import OpenAIChatTarget, PromptTarget +from pyrit.scenario import DatasetConfiguration +from pyrit.scenario.core import AtomicAttack, AttackTechnique, BaselineAttackPolicy, Scenario, ScenarioTechnique +from pyrit.scenario.core.scenario_context import ScenarioContext +from pyrit.score import SelfAskRefusalScorer, SubStringScorer, TrueFalseScorer + +_ATOMIC_ATTACK_NAME = "crescendo" +_OBJECTIVE_A = "Recover objective A" +_OBJECTIVE_B = "Recover objective B" +_MALFORMED_REPLY = "not valid JSON" +_REFUSAL = "I refuse to help." + + +class _CrescendoScenarioTechnique(ScenarioTechnique): + CRESCENDO = ("crescendo", {"crescendo"}) + ALL = ("all", {"all"}) + + @classmethod + def get_aggregate_tags(cls) -> set[str]: + return {"all"} + + +class _CrescendoResumeScenario(Scenario): + BASELINE_ATTACK_POLICY: ClassVar[BaselineAttackPolicy] = BaselineAttackPolicy.Forbidden + + def __init__( + self, + *, + adversarial_chat: PromptTarget, + objective_scorer: TrueFalseScorer, + refusal_scorer: TrueFalseScorer, + scenario_result_id: str | None = None, + ) -> None: + super().__init__( + name="Crescendo Resume Scenario", + version=1, + technique_class=_CrescendoScenarioTechnique, + default_dataset_config=DatasetConfiguration(), + objective_scorer=objective_scorer, + scenario_result_id=scenario_result_id, + ) + self._adversarial_chat = adversarial_chat + self._refusal_scorer = refusal_scorer + + async def _resolve_seed_groups_by_dataset_async( + self, + *, + apply_sampling: bool = True, + ) -> dict[str, list[AttackSeedGroup]]: + return { + "resume": [ + AttackSeedGroup(seeds=[SeedObjective(value=_OBJECTIVE_A)]), + AttackSeedGroup(seeds=[SeedObjective(value=_OBJECTIVE_B)]), + ] + } + + async def _build_atomic_attacks_async(self, *, context: ScenarioContext) -> list[AtomicAttack]: + attack = CrescendoAttack( + objective_target=context.objective_target, + attack_adversarial_config=AttackAdversarialConfig(target=self._adversarial_chat), + attack_scoring_config=AttackScoringConfig( + objective_scorer=self._objective_scorer, + refusal_scorer=self._refusal_scorer, + ), + max_backtracks=1, + max_turns=1, + ) + return [ + AtomicAttack( + atomic_attack_name=_ATOMIC_ATTACK_NAME, + attack_technique=AttackTechnique(attack=attack), + seed_groups=list(context.seed_groups), + memory_labels=context.memory_labels, + ) + ] + + +def _build_chat_completion(*, text: str, completion_id: str) -> ChatCompletion: + return ChatCompletion( + id=completion_id, + choices=[ + Choice( + finish_reason="stop", + index=0, + logprobs=None, + message=ChatCompletionMessage(content=text, refusal=None, role="assistant"), + ) + ], + created=0, + model="gpt-test", + object="chat.completion", + ) + + +def _adversarial_reply(*, prompt: str) -> str: + return json.dumps( + { + "next_message": prompt, + "last_response_summary": "summary", + "rationale": "rationale", + } + ) + + +def _refusal_score_reply(*, refused: bool) -> str: + return json.dumps( + { + "score_value": str(refused).lower(), + "description": "refusal result", + "rationale": "controlled test score", + } + ) + + +def _build_target(*, model_name: str) -> OpenAIChatTarget: + return OpenAIChatTarget( + model_name=model_name, + endpoint="https://api.openai.com/v1", + api_key="test-api-key", + ) + + +def _build_rate_limit_error() -> RateLimitError: + request = httpx.Request("POST", "https://api.openai.com/v1/chat/completions") + response = httpx.Response(429, text="rate limited", request=request) + return RateLimitError("rate limited", response=response, body=None) + + +def _build_timeout_error() -> APITimeoutError: + request = httpx.Request("POST", "https://api.openai.com/v1/chat/completions") + return APITimeoutError(request=request) + + +def _build_boundary_side_effect( + *, + schedule: list[tuple[str, ChatCompletion | BaseException]], + events: list[str], + requests: list[dict[str, Any]], +) -> Callable[..., Awaitable[ChatCompletion]]: + async def _side_effect_async(**kwargs: Any) -> ChatCompletion: + requests.append(kwargs) + label, outcome = schedule.pop(0) + events.append(label) + if isinstance(outcome, BaseException): + raise outcome + return outcome + + return _side_effect_async + + +def _build_scenario( + *, + objective_target: OpenAIChatTarget, + adversarial_target: OpenAIChatTarget, + objective_scorer: SubStringScorer, + refusal_scorer: SelfAskRefusalScorer, + max_retries: int, + scenario_result_id: str | None = None, +) -> _CrescendoResumeScenario: + scenario = _CrescendoResumeScenario( + adversarial_chat=adversarial_target, + objective_scorer=objective_scorer, + refusal_scorer=refusal_scorer, + scenario_result_id=scenario_result_id, + ) + scenario.set_params_from_args( + args={ + "objective_target": objective_target, + "max_concurrency": 1, + "max_retries": max_retries, + "memory_labels": {"test": "crescendo-resume"}, + } + ) + return scenario + + +def _message_contents(requests: Sequence[dict[str, Any]]) -> list[list[dict[str, Any]]]: + return [request["messages"] for request in requests] + + +@pytest.mark.usefixtures("patch_central_database") +async def test_crescendo_scenario_retry_and_resume_use_real_components_without_stale_state() -> None: + objective_target = _build_target(model_name="objective-model") + adversarial_target = _build_target(model_name="adversarial-model") + scorer_target = _build_target(model_name="scorer-model") + objective_scorer = SubStringScorer(substring="RECOVERED") + refusal_scorer = SelfAskRefusalScorer(chat_target=scorer_target) + + events: list[str] = [] + adversarial_requests: list[dict[str, Any]] = [] + objective_requests: list[dict[str, Any]] = [] + scorer_requests: list[dict[str, Any]] = [] + adversarial_schedule = [ + ("adversarial:a:rate-limit", _build_rate_limit_error()), + ( + "adversarial:a:success", + _build_chat_completion(text=_adversarial_reply(prompt="A prompt"), completion_id="adv-a"), + ), + ( + "adversarial:b:timeout-attempt", + _build_chat_completion(text=_adversarial_reply(prompt="B timeout prompt"), completion_id="adv-b-timeout"), + ), + ( + "adversarial:b:malformed-1", + _build_chat_completion(text=_MALFORMED_REPLY, completion_id="adv-b-malformed-1"), + ), + ( + "adversarial:b:malformed-2", + _build_chat_completion(text=_MALFORMED_REPLY, completion_id="adv-b-malformed-2"), + ), + ( + "adversarial:b:refusal", + _build_chat_completion(text=_adversarial_reply(prompt="B refused prompt"), completion_id="adv-b-refusal"), + ), + ( + "adversarial:b:recovery", + _build_chat_completion(text=_adversarial_reply(prompt="B recovery prompt"), completion_id="adv-b-recovery"), + ), + ] + objective_schedule = [ + ("objective:a:success", _build_chat_completion(text="RECOVERED A", completion_id="objective-a")), + ("objective:b:timeout", _build_timeout_error()), + ("objective:b:refusal", _build_chat_completion(text=_REFUSAL, completion_id="objective-b-refusal")), + ("objective:b:recovery", _build_chat_completion(text="RECOVERED B", completion_id="objective-b-recovery")), + ] + scorer_schedule = [ + ( + "scorer:a:not-refusal", + _build_chat_completion(text=_refusal_score_reply(refused=False), completion_id="score-a"), + ), + ( + "scorer:b:refusal", + _build_chat_completion(text=_refusal_score_reply(refused=True), completion_id="score-b-1"), + ), + ( + "scorer:b:not-refusal", + _build_chat_completion(text=_refusal_score_reply(refused=False), completion_id="score-b-2"), + ), + ] + + first_scenario = _build_scenario( + objective_target=objective_target, + adversarial_target=adversarial_target, + objective_scorer=objective_scorer, + refusal_scorer=refusal_scorer, + max_retries=1, + ) + memory = CentralMemory.get_memory_instance() + + with ( + patch.object( + adversarial_target._async_client.chat.completions, + "create", + new=AsyncMock( + side_effect=_build_boundary_side_effect( + schedule=adversarial_schedule, + events=events, + requests=adversarial_requests, + ) + ), + ), + patch.object( + objective_target._async_client.chat.completions, + "create", + new=AsyncMock( + side_effect=_build_boundary_side_effect( + schedule=objective_schedule, + events=events, + requests=objective_requests, + ) + ), + ), + patch.object( + scorer_target._async_client.chat.completions, + "create", + new=AsyncMock( + side_effect=_build_boundary_side_effect( + schedule=scorer_schedule, + events=events, + requests=scorer_requests, + ) + ), + ), + patch("asyncio.sleep", new_callable=AsyncMock) as sleep_mock, + patch.object( + memory, + "update_scenario_run_state", + wraps=memory.update_scenario_run_state, + ) as update_state, + ): + await first_scenario.initialize_async() + with pytest.raises(ScenarioPartialFailureException, match="1 of 1 objectives incomplete"): + await first_scenario.run_async() + + scenario_result_id = first_scenario._scenario_result_id + assert scenario_result_id is not None + failed_result = memory.get_scenario_results(scenario_result_ids=[scenario_result_id])[0] + assert failed_result.scenario_run_state == ScenarioRunState.FAILED + assert failed_result.number_tries == 2 + assert [result.outcome for result in failed_result.attack_results[_ATOMIC_ATTACK_NAME]] == [ + AttackOutcome.SUCCESS, + AttackOutcome.ERROR, + AttackOutcome.ERROR, + ] + + resumed_scenario = _build_scenario( + objective_target=objective_target, + adversarial_target=adversarial_target, + objective_scorer=objective_scorer, + refusal_scorer=refusal_scorer, + max_retries=0, + scenario_result_id=scenario_result_id, + ) + await resumed_scenario.initialize_async() + assert resumed_scenario._atomic_attacks[0].objectives == [_OBJECTIVE_A, _OBJECTIVE_B] + + completed_result = await resumed_scenario.run_async() + + attack_results = completed_result.attack_results[_ATOMIC_ATTACK_NAME] + objective_a_results = [result for result in attack_results if result.objective == _OBJECTIVE_A] + objective_b_results = [result for result in attack_results if result.objective == _OBJECTIVE_B] + error_results = [result for result in attack_results if result.outcome == AttackOutcome.ERROR] + timeout_result = next(result for result in error_results if result.error_type == "Exception") + malformed_result = next(result for result in error_results if result.error_type == "InvalidJsonException") + recovered_result = next(result for result in objective_b_results if result.outcome == AttackOutcome.SUCCESS) + + assert completed_result.scenario_run_state == ScenarioRunState.COMPLETED + assert completed_result.number_tries == 3 + assert completed_result.error_message is None + assert completed_result.error_type is None + assert len(objective_a_results) == 1 + assert len(objective_b_results) == 3 + assert [result.outcome for result in objective_b_results] == [ + AttackOutcome.ERROR, + AttackOutcome.ERROR, + AttackOutcome.SUCCESS, + ] + assert len({result.attack_result_id for result in attack_results}) == len(attack_results) + assert len({result.conversation_id for result in attack_results}) == len(attack_results) + assert all(result.labels == {"test": "crescendo-resume"} for result in attack_results) + assert recovered_result.executed_turns == 1 + assert recovered_result.metadata["backtrack_count"] == 1 + assert recovered_result.last_response and recovered_result.last_response.converted_value == "RECOVERED B" + assert recovered_result.last_score and recovered_result.last_score.get_value() is True + assert recovered_result.last_score.objective == _OBJECTIVE_B + assert recovered_result.last_score.message_piece_id == recovered_result.last_response.id + + timeout_messages = memory.get_conversation_messages(conversation_id=timeout_result.conversation_id) + assert [message.api_role for message in timeout_messages] == ["user", "assistant"] + assert timeout_messages[0].get_value() == "B timeout prompt" + assert timeout_messages[1].get_piece().response_error == "processing" + assert "APITimeoutError" in timeout_messages[1].get_value() + + pruned_conversations = recovered_result.get_conversations_by_type(ConversationType.PRUNED) + adversarial_conversations = recovered_result.get_conversations_by_type(ConversationType.ADVERSARIAL) + assert len(pruned_conversations) == 1 + assert len(adversarial_conversations) == 1 + pruned_messages = memory.get_conversation_messages(conversation_id=pruned_conversations[0].conversation_id) + recovered_messages = memory.get_conversation_messages(conversation_id=recovered_result.conversation_id) + assert [message.get_value() for message in pruned_messages] == ["B refused prompt", _REFUSAL] + assert [message.get_value() for message in recovered_messages] == ["B recovery prompt", "RECOVERED B"] + + malformed_adversarial = malformed_result.get_conversations_by_type(ConversationType.ADVERSARIAL) + assert len(malformed_adversarial) == 1 + malformed_conversation = memory._get_conversation( + conversation_id=malformed_adversarial[0].conversation_id, + ) + assert malformed_conversation is not None + assert len(malformed_conversation.retries) == 1 + assert malformed_conversation.retries[0].reason == ConversationRetryReason.JSON_PARSING + malformed_messages = memory.get_conversation_messages( + conversation_id=malformed_adversarial[0].conversation_id, + ) + assert [message.api_role for message in malformed_messages] == ["system", "user", "assistant"] + assert malformed_messages[-1].get_value() == _MALFORMED_REPLY + + assert events == [ + "adversarial:a:rate-limit", + "adversarial:a:success", + "objective:a:success", + "scorer:a:not-refusal", + "adversarial:b:timeout-attempt", + "objective:b:timeout", + "adversarial:b:malformed-1", + "adversarial:b:malformed-2", + "adversarial:b:refusal", + "objective:b:refusal", + "scorer:b:refusal", + "adversarial:b:recovery", + "objective:b:recovery", + "scorer:b:not-refusal", + ] + assert not adversarial_schedule + assert not objective_schedule + assert not scorer_schedule + assert sleep_mock.await_count == 2 + assert all(call.args[0] >= 0 for call in sleep_mock.await_args_list) + observed_states = [call.kwargs["scenario_run_state"] for call in update_state.call_args_list] + assert observed_states == [ + ScenarioRunState.IN_PROGRESS, + ScenarioRunState.IN_PROGRESS, + ScenarioRunState.FAILED, + ScenarioRunState.IN_PROGRESS, + ScenarioRunState.COMPLETED, + ] + + adversarial_messages = _message_contents(adversarial_requests) + assert adversarial_messages[0] == adversarial_messages[1] + assert adversarial_messages[3] == adversarial_messages[4] + objective_messages = _message_contents(objective_requests) + assert [messages[-1]["content"] for messages in objective_messages] == [ + "A prompt", + "B timeout prompt", + "B refused prompt", + "B recovery prompt", + ] + assert all(_REFUSAL not in json.dumps(message) for message in objective_messages[-1]) + scorer_messages = _message_contents(scorer_requests) + scorer_inputs = [messages[-1]["content"] for messages in scorer_messages] + assert "conversation_objective: A prompt" in scorer_inputs[0] + assert "response_to_evaluate_input: RECOVERED A" in scorer_inputs[0] + assert "conversation_objective: B refused prompt" in scorer_inputs[1] + assert f"response_to_evaluate_input: {_REFUSAL}" in scorer_inputs[1] + assert "conversation_objective: B recovery prompt" in scorer_inputs[2] + assert "response_to_evaluate_input: RECOVERED B" in scorer_inputs[2] + + all_pieces = memory.get_message_pieces() + assert len({piece.id for piece in all_pieces}) == len(all_pieces) + assert len({piece.conversation_id for piece in recovered_messages}) == 1 + assert len({piece.conversation_id for piece in pruned_messages}) == 1 + assert timeout_result.total_retries == 0 + assert malformed_result.total_retries == 2 + assert [event.attempt_number for event in malformed_result.retry_events] == [1, 2] + assert all(event.exception_type == "InvalidJsonException" for event in malformed_result.retry_events) + assert all(event.component_role == "adversarial_chat" for event in malformed_result.retry_events)