diff --git a/doc/code/datasets/2_seed_programming.ipynb b/doc/code/datasets/2_seed_programming.ipynb index 87c85e10af..835bf34887 100644 --- a/doc/code/datasets/2_seed_programming.ipynb +++ b/doc/code/datasets/2_seed_programming.ipynb @@ -478,7 +478,7 @@ ], "source": [ "from pyrit.common.path import EXECUTOR_RED_TEAM_PATH, EXECUTOR_SIMULATED_TARGET_PATH\n", - "from pyrit.models import SeedSimulatedConversation\n", + "from pyrit.models import SeedSimulatedConversation, load_next_message_prompt\n", "\n", "seed_group = AttackSeedGroup(\n", " seeds=[\n", @@ -488,10 +488,12 @@ " role=\"system\",\n", " ),\n", " SeedSimulatedConversation(\n", - " adversarial_chat_system_prompt_path=EXECUTOR_RED_TEAM_PATH / \"naive_crescendo.yaml\",\n", + " adversarial_chat_system_prompt=SeedPrompt.from_yaml_file(EXECUTOR_RED_TEAM_PATH / \"naive_crescendo.yaml\"),\n", " sequence=1,\n", " num_turns=4,\n", - " next_message_system_prompt_path=EXECUTOR_SIMULATED_TARGET_PATH / \"direct_next_message.yaml\",\n", + " next_message_system_prompt=load_next_message_prompt(\n", + " EXECUTOR_SIMULATED_TARGET_PATH / \"direct_next_message.yaml\"\n", + " ),\n", " ),\n", " ]\n", ")\n", diff --git a/doc/code/datasets/2_seed_programming.py b/doc/code/datasets/2_seed_programming.py index 134a561592..601af4b7eb 100644 --- a/doc/code/datasets/2_seed_programming.py +++ b/doc/code/datasets/2_seed_programming.py @@ -92,7 +92,7 @@ # %% from pyrit.common.path import EXECUTOR_RED_TEAM_PATH, EXECUTOR_SIMULATED_TARGET_PATH -from pyrit.models import SeedSimulatedConversation +from pyrit.models import SeedSimulatedConversation, load_next_message_prompt seed_group = AttackSeedGroup( seeds=[ @@ -102,10 +102,12 @@ role="system", ), SeedSimulatedConversation( - adversarial_chat_system_prompt_path=EXECUTOR_RED_TEAM_PATH / "naive_crescendo.yaml", + adversarial_chat_system_prompt=SeedPrompt.from_yaml_file(EXECUTOR_RED_TEAM_PATH / "naive_crescendo.yaml"), sequence=1, num_turns=4, - next_message_system_prompt_path=EXECUTOR_SIMULATED_TARGET_PATH / "direct_next_message.yaml", + next_message_system_prompt=load_next_message_prompt( + EXECUTOR_SIMULATED_TARGET_PATH / "direct_next_message.yaml" + ), ), ] ) diff --git a/doc/code/datasets/5_simulated_conversation.ipynb b/doc/code/datasets/5_simulated_conversation.ipynb index 8de223e7fa..3d7aa12b2a 100644 --- a/doc/code/datasets/5_simulated_conversation.ipynb +++ b/doc/code/datasets/5_simulated_conversation.ipynb @@ -23,7 +23,7 @@ "\n", "## Generating a Simulated Conversation\n", "\n", - "The function takes an objective, an adversarial chat model, a scorer, and a system prompt path.\n", + "The function takes an objective, an adversarial chat model, a scorer, and a system prompt.\n", "It runs a `RedTeamingAttack` internally with the adversarial LLM playing both attacker and target\n", "roles." ] @@ -63,7 +63,7 @@ "\n", "from pyrit.common.path import EXECUTOR_SEED_PROMPT_PATH\n", "from pyrit.executor.attack import generate_simulated_conversation_async\n", - "from pyrit.models import SeedGroup\n", + "from pyrit.models import SeedGroup, SeedPrompt\n", "from pyrit.output import output_attack_async\n", "from pyrit.prompt_target import OpenAIChatTarget\n", "from pyrit.score import SelfAskRefusalScorer\n", @@ -82,7 +82,9 @@ " adversarial_chat=adversarial_chat,\n", " objective_scorer=objective_scorer,\n", " num_turns=3,\n", - " adversarial_chat_system_prompt_path=Path(EXECUTOR_SEED_PROMPT_PATH) / \"red_teaming\" / \"naive_crescendo.yaml\",\n", + " adversarial_chat_system_prompt=SeedPrompt.from_yaml_file(\n", + " Path(EXECUTOR_SEED_PROMPT_PATH) / \"red_teaming\" / \"naive_crescendo.yaml\"\n", + " ),\n", ")\n", "\n", "print(f\"Generated {len(simulated_result.seed_prompts)} messages\")" @@ -100,7 +102,7 @@ "Wrapping the prompts in a `SeedGroup` gives you convenient access to `prepended_conversation`\n", "(all turns except the last) and `next_message` (the final user message to continue from).\n", "Note that `next_message` is only populated when the last generated message has role `\"user\"` —\n", - "if you need a final user turn, pass `next_message_system_prompt_path` to the function.\n", + "if you need a final user turn, pass `next_message_system_prompt` to the function.\n", "\n", "This replaces the earlier `list[SeedPrompt]` return value. Use `result.seed_prompts` where you\n", "previously used the returned list." @@ -521,9 +523,9 @@ "| `adversarial_chat` | `PromptTarget` | The LLM that generates attack prompts (also plays the simulated target). Must declare `supports_multi_turn=True` and `supports_editable_history=True`. |\n", "| `objective_scorer` | `TrueFalseScorer` | Evaluates whether the final turn achieved the objective |\n", "| `num_turns` | `int` | Number of conversation turns to generate (default: 3) |\n", - "| `adversarial_chat_system_prompt_path` | `str \\| Path` | System prompt for the adversarial chat role |\n", - "| `simulated_target_system_prompt_path` | `str \\| Path \\| None` | Optional system prompt for the simulated target role |\n", - "| `next_message_system_prompt_path` | `str \\| Path \\| None` | Optional path to generate a final user message that elicits objective fulfillment |\n", + "| `adversarial_chat_system_prompt` | `SeedPrompt` | System prompt for the adversarial chat role |\n", + "| `simulated_target_system_prompt` | `SeedPrompt \\| None` | Optional system prompt for the simulated target role |\n", + "| `next_message_system_prompt` | `SeedPrompt \\| None` | Optional prompt that generates a final user message eliciting objective fulfillment |\n", "| `attack_converter_config` | `AttackConverterConfig \\| None` | Optional converter configuration for the attack |\n", "| `memory_labels` | `dict[str, str] \\| None` | Labels for tracking in memory |\n", "\n", diff --git a/doc/code/datasets/5_simulated_conversation.py b/doc/code/datasets/5_simulated_conversation.py index 6ee2e66025..d4e0adc552 100644 --- a/doc/code/datasets/5_simulated_conversation.py +++ b/doc/code/datasets/5_simulated_conversation.py @@ -27,7 +27,7 @@ # # ## Generating a Simulated Conversation # -# The function takes an objective, an adversarial chat model, a scorer, and a system prompt path. +# The function takes an objective, an adversarial chat model, a scorer, and a system prompt. # It runs a `RedTeamingAttack` internally with the adversarial LLM playing both attacker and target # roles. @@ -36,7 +36,7 @@ from pyrit.common.path import EXECUTOR_SEED_PROMPT_PATH from pyrit.executor.attack import generate_simulated_conversation_async -from pyrit.models import SeedGroup +from pyrit.models import SeedGroup, SeedPrompt from pyrit.output import output_attack_async from pyrit.prompt_target import OpenAIChatTarget from pyrit.score import SelfAskRefusalScorer @@ -55,7 +55,9 @@ adversarial_chat=adversarial_chat, objective_scorer=objective_scorer, num_turns=3, - adversarial_chat_system_prompt_path=Path(EXECUTOR_SEED_PROMPT_PATH) / "red_teaming" / "naive_crescendo.yaml", + adversarial_chat_system_prompt=SeedPrompt.from_yaml_file( + Path(EXECUTOR_SEED_PROMPT_PATH) / "red_teaming" / "naive_crescendo.yaml" + ), ) print(f"Generated {len(simulated_result.seed_prompts)} messages") @@ -68,7 +70,7 @@ # Wrapping the prompts in a `SeedGroup` gives you convenient access to `prepended_conversation` # (all turns except the last) and `next_message` (the final user message to continue from). # Note that `next_message` is only populated when the last generated message has role `"user"` — -# if you need a final user turn, pass `next_message_system_prompt_path` to the function. +# if you need a final user turn, pass `next_message_system_prompt` to the function. # # This replaces the earlier `list[SeedPrompt]` return value. Use `result.seed_prompts` where you # previously used the returned list. @@ -130,9 +132,9 @@ # | `adversarial_chat` | `PromptTarget` | The LLM that generates attack prompts (also plays the simulated target). Must declare `supports_multi_turn=True` and `supports_editable_history=True`. | # | `objective_scorer` | `TrueFalseScorer` | Evaluates whether the final turn achieved the objective | # | `num_turns` | `int` | Number of conversation turns to generate (default: 3) | -# | `adversarial_chat_system_prompt_path` | `str \| Path` | System prompt for the adversarial chat role | -# | `simulated_target_system_prompt_path` | `str \| Path \| None` | Optional system prompt for the simulated target role | -# | `next_message_system_prompt_path` | `str \| Path \| None` | Optional path to generate a final user message that elicits objective fulfillment | +# | `adversarial_chat_system_prompt` | `SeedPrompt` | System prompt for the adversarial chat role | +# | `simulated_target_system_prompt` | `SeedPrompt \| None` | Optional system prompt for the simulated target role | +# | `next_message_system_prompt` | `SeedPrompt \| None` | Optional prompt that generates a final user message eliciting objective fulfillment | # | `attack_converter_config` | `AttackConverterConfig \| None` | Optional converter configuration for the attack | # | `memory_labels` | `dict[str, str] \| None` | Labels for tracking in memory | # diff --git a/doc/code/memory/3_memory_data_types.md b/doc/code/memory/3_memory_data_types.md index 6c41eb769b..7ab05d53bf 100644 --- a/doc/code/memory/3_memory_data_types.md +++ b/doc/code/memory/3_memory_data_types.md @@ -84,7 +84,7 @@ All seed types inherit from [`Seed`](../../../pyrit/models/seeds/seed.py), which - [`SeedObjective`](../../../pyrit/models/seeds/seed_objective.py) — The goal of an attack (e.g., "Generate hate speech content"). Always text. Cannot be a general technique. -- [`SeedSimulatedConversation`](../../../pyrit/models/seeds/seed_simulated_conversation.py) — Configuration for dynamically generating multi-turn conversations. Specifies system prompt paths, number of turns, and sequence offsets. The actual generation happens in the executor layer. +- [`SeedSimulatedConversation`](../../../pyrit/models/seeds/seed_simulated_conversation.py) — Configuration for dynamically generating multi-turn conversations. Carries the adversarial, simulated-target, and next-message system prompts, the number of turns, and sequence offsets. The actual generation happens in the executor layer. ### Seed Groups diff --git a/pyrit/executor/attack/core/attack_parameters.py b/pyrit/executor/attack/core/attack_parameters.py index 32c5e0f912..5ff6b701b2 100644 --- a/pyrit/executor/attack/core/attack_parameters.py +++ b/pyrit/executor/attack/core/attack_parameters.py @@ -174,9 +174,9 @@ async def from_seed_group_async( objective_scorer=objective_scorer, num_turns=simulated_conversation_config.num_turns, starting_sequence=simulated_conversation_config.sequence, - adversarial_chat_system_prompt_path=simulated_conversation_config.adversarial_chat_system_prompt_path, - simulated_target_system_prompt_path=simulated_conversation_config.simulated_target_system_prompt_path, - next_message_system_prompt_path=simulated_conversation_config.next_message_system_prompt_path, + adversarial_chat_system_prompt=simulated_conversation_config.adversarial_chat_system_prompt, + simulated_target_system_prompt=simulated_conversation_config.simulated_target_system_prompt, + next_message_system_prompt=simulated_conversation_config.next_message_system_prompt, ) simulated_prompts = simulated_result.seed_prompts if "source_conversations" in valid_fields: diff --git a/pyrit/executor/attack/multi_turn/simulated_conversation.py b/pyrit/executor/attack/multi_turn/simulated_conversation.py index b116ca77f8..5a0c79a472 100644 --- a/pyrit/executor/attack/multi_turn/simulated_conversation.py +++ b/pyrit/executor/attack/multi_turn/simulated_conversation.py @@ -10,6 +10,7 @@ from __future__ import annotations +import asyncio import logging from dataclasses import dataclass from typing import TYPE_CHECKING @@ -31,11 +32,14 @@ ConversationType, Message, SeedPrompt, - SeedSimulatedConversation, + load_next_message_prompt, + load_simulated_target_prompt, + warn_prompt_path_deprecated, ) from pyrit.prompt_normalizer import PromptNormalizer if TYPE_CHECKING: + from collections.abc import Callable from pathlib import Path from pyrit.prompt_target import PromptTarget @@ -52,6 +56,39 @@ class SimulatedConversationResult: related_conversations: frozenset[ConversationReference] +async def _resolve_prompt_source_async( + *, + prompt: SeedPrompt | None, + path: str | Path | None, + prompt_name: str, + path_name: str, + load_prompt: Callable[[str | Path], SeedPrompt], +) -> SeedPrompt | None: + """ + Resolve a prompt source, loading a deprecated path input off the event loop. + + The conflict check and the deprecation warning stay on the event loop so the warning points + at the caller; only the file read is sent to a worker thread. + + Args: + prompt: The canonical prompt, if the caller supplied one. + path: The deprecated path input, if the caller supplied one. + prompt_name: Name of the canonical parameter, used in messages. + path_name: Name of the deprecated parameter, used in messages. + load_prompt: Loader that turns the path into a prompt. + + Returns: + SeedPrompt | None: The resolved prompt, or None when neither input was supplied. + + Raises: + ValueError: If both the canonical prompt and its deprecated path are supplied. + """ + if path is None: + return prompt + warn_prompt_path_deprecated(prompt=prompt, prompt_name=prompt_name, path_name=path_name) + return await asyncio.to_thread(load_prompt, path) + + async def generate_simulated_conversation_async( *, objective: str, @@ -59,7 +96,10 @@ async def generate_simulated_conversation_async( objective_scorer: TrueFalseScorer, num_turns: int = 3, starting_sequence: int = 0, - adversarial_chat_system_prompt_path: str | Path, + adversarial_chat_system_prompt: SeedPrompt | None = None, + simulated_target_system_prompt: SeedPrompt | None = None, + next_message_system_prompt: SeedPrompt | None = None, + adversarial_chat_system_prompt_path: str | Path | None = None, simulated_target_system_prompt_path: str | Path | None = None, next_message_system_prompt_path: str | Path | None = None, attack_converter_config: AttackConverterConfig | None = None, @@ -86,14 +126,21 @@ async def generate_simulated_conversation_async( num_turns: Number of conversation turns to generate. Defaults to 3. starting_sequence: The starting sequence number for the generated SeedPrompts. Each message gets an incrementing sequence number. Defaults to 0. - adversarial_chat_system_prompt_path: Path to the system prompt for the adversarial chat. - simulated_target_system_prompt_path: Path to the system prompt for the simulated target. + adversarial_chat_system_prompt: System prompt for the adversarial chat. Required unless + the deprecated path input is used. + simulated_target_system_prompt: System prompt for the simulated target. If None, no system prompt is used for the simulated target. - next_message_system_prompt_path: Optional path to a system prompt for generating - a final user message. If provided, after the simulated conversation, a single - LLM call generates a user message that attempts to get the target to fulfill - the objective in their next response. The prompt template receives `objective` - and `conversation_so_far` parameters. + next_message_system_prompt: Optional system prompt for generating a final user message. + If provided, after the simulated conversation, a single LLM call generates a user + message that attempts to get the target to fulfill the objective in their next + response. The prompt template receives `objective` and `conversation_context` + parameters. + adversarial_chat_system_prompt_path: Deprecated. Path to the adversarial chat system + prompt YAML. Use ``adversarial_chat_system_prompt`` instead. + simulated_target_system_prompt_path: Deprecated. Path to the simulated target system + prompt YAML. Use ``simulated_target_system_prompt`` instead. + next_message_system_prompt_path: Deprecated. Path to the next message system prompt + YAML. Use ``next_message_system_prompt`` instead. attack_converter_config: Converter configuration for the attack. Defaults to None. memory_labels: Labels to associate with the conversation in memory. Defaults to None. @@ -102,7 +149,8 @@ async def generate_simulated_conversation_async( start from ``starting_sequence`` and increment by 1 for each message. Raises: - ValueError: If num_turns is not a positive integer. + ValueError: If num_turns is not a positive integer, if no adversarial chat system prompt + is supplied, or if a prompt and its deprecated path input are both supplied. """ # Use the same LLM for both adversarial chat and simulated target # They get different system prompts to play different roles @@ -110,22 +158,40 @@ async def generate_simulated_conversation_async( if num_turns <= 0: raise ValueError("num_turns must be a positive integer") - # Load and configure simulated target system prompt using centralized validation - # Returns None if no path is provided (no system prompt for simulated target) - simulated_target_system_prompt = SeedSimulatedConversation.load_simulated_target_system_prompt( - objective=objective, - num_turns=num_turns, - simulated_target_system_prompt_path=simulated_target_system_prompt_path, + adversarial_chat_system_prompt = await _resolve_prompt_source_async( + prompt=adversarial_chat_system_prompt, + path=adversarial_chat_system_prompt_path, + prompt_name="adversarial_chat_system_prompt", + path_name="adversarial_chat_system_prompt_path", + load_prompt=SeedPrompt.from_yaml_file, ) - - # Create adversarial config for the simulation. Load the optional path into a SeedPrompt so the - # resolved prompt is stored directly on the configuration. - adversarial_system_prompt = ( - SeedPrompt.from_yaml_file(adversarial_chat_system_prompt_path) if adversarial_chat_system_prompt_path else None + simulated_target_system_prompt = await _resolve_prompt_source_async( + prompt=simulated_target_system_prompt, + path=simulated_target_system_prompt_path, + prompt_name="simulated_target_system_prompt", + path_name="simulated_target_system_prompt_path", + load_prompt=load_simulated_target_prompt, + ) + next_message_system_prompt = await _resolve_prompt_source_async( + prompt=next_message_system_prompt, + path=next_message_system_prompt_path, + prompt_name="next_message_system_prompt", + path_name="next_message_system_prompt_path", + load_prompt=load_next_message_prompt, ) + if adversarial_chat_system_prompt is None: + raise ValueError("adversarial_chat_system_prompt is required") + + # Render the simulated target system prompt; None means the target gets no system prompt. + simulated_target_system_prompt_value = ( + simulated_target_system_prompt.render_template_value(objective=objective, num_turns=num_turns) + if simulated_target_system_prompt is not None + else None + ) + adversarial_config = AttackAdversarialConfig( target=adversarial_chat, - system_prompt=adversarial_system_prompt, + system_prompt=adversarial_chat_system_prompt, ) # Create scoring config @@ -149,8 +215,8 @@ async def generate_simulated_conversation_async( # Build prepended_conversation - only include system message if prompt is provided prepended_conversation: list[Message] = [] - if simulated_target_system_prompt: - prepended_conversation.append(Message.from_system_prompt(simulated_target_system_prompt)) + if simulated_target_system_prompt_value: + prepended_conversation.append(Message.from_system_prompt(simulated_target_system_prompt_value)) result = await attack.execute_async( objective=objective, @@ -175,15 +241,15 @@ async def generate_simulated_conversation_async( *result.related_conversations, } - # If next_message_system_prompt_path is provided, generate a final user message - if next_message_system_prompt_path: + # If a next-message prompt is configured, generate a final user message + if next_message_system_prompt: next_message_conversation_id = str(uuid4()) next_message = await _generate_next_message_async( objective=objective, conversation_messages=conversation_messages, adversarial_chat=adversarial_chat, conversation_id=next_message_conversation_id, - next_message_system_prompt_path=next_message_system_prompt_path, + next_message_system_prompt=next_message_system_prompt, prompt_normalizer=PromptNormalizer(), memory_labels=memory_labels, ) @@ -216,7 +282,7 @@ async def _generate_next_message_async( conversation_messages: list[Message], adversarial_chat: PromptTarget, conversation_id: str, - next_message_system_prompt_path: str | Path, + next_message_system_prompt: SeedPrompt, prompt_normalizer: PromptNormalizer, memory_labels: dict[str, str] | None = None, ) -> Message: @@ -234,7 +300,8 @@ async def _generate_next_message_async( conversation_messages: The conversation generated so far as Messages. adversarial_chat: The LLM to use for generation. conversation_id: The conversation ID for the adversarial generation exchange. - next_message_system_prompt_path: Path to the system prompt template. + next_message_system_prompt: The next-message system-prompt SeedPrompt, rendered with the + conversation context. prompt_normalizer: The normalizer the manager sends the adversarial turn through. memory_labels: Optional memory labels to attach to the request. @@ -249,17 +316,10 @@ async def _generate_next_message_async( normalizer = ConversationContextNormalizer() conversation_context = await normalizer.normalize_string_async(conversation_messages) - # Load the system prompt template (schema + prompt are a matched pair declared in the YAML) - template = SeedPrompt.from_yaml_with_required_parameters( - template_path=next_message_system_prompt_path, - required_parameters=["objective", "conversation_context"], - error_message="Next message system prompt must have objective and conversation_context parameters", - ) - # The manager owns schema resolution, setting the system prompt, the send, and JSON parse/retry. manager = _AdversarialConversationManager( adversarial_target=adversarial_chat, - adversarial_system_prompt=template, + adversarial_system_prompt=next_message_system_prompt, prompt_normalizer=prompt_normalizer, conversation_id=conversation_id, objective=objective, diff --git a/pyrit/memory/memory_models.py b/pyrit/memory/memory_models.py index 9fac4e0e13..65bd82c20f 100644 --- a/pyrit/memory/memory_models.py +++ b/pyrit/memory/memory_models.py @@ -1620,6 +1620,10 @@ def get_seed(self) -> Seed: Returns: Seed: The reconstructed seed object (SeedPrompt, SeedObjective, or SeedSimulatedConversation) + + Raises: + ValueError: If a simulated conversation record cannot be rebuilt, for example when it + names a prompt file that is not present on this machine. """ cleaned_metadata, decoded_schema = self._unpack_seed_metadata(self.prompt_metadata) if self.seed_type == "objective": @@ -1640,28 +1644,52 @@ def get_seed(self) -> Seed: prompt_group_id=self.prompt_group_id, ) if self.seed_type == "simulated_conversation": - # Reconstruct SeedSimulatedConversation from JSON value + # Reconstruct SeedSimulatedConversation from JSON value. Records written before the + # prompts were normalized carry only ``*_path`` keys; the model's compatibility + # adapter resolves those, and a canonicalized record loses the stale hash of its + # old path-shaped value. config = json.loads(self.value) - return SeedSimulatedConversation( - id=self.id, - value_sha256=self.value_sha256, - name=self.name, - dataset_name=self.dataset_name, - harm_categories=self.harm_categories, - description=self.description, - authors=self.authors, - groups=self.groups, - source=self.source, - date_added=self.date_added, - added_by=self.added_by, - metadata=cleaned_metadata, - prompt_group_id=self.prompt_group_id, - num_turns=config.get("num_turns", 3), - sequence=config.get("sequence", 0), - adversarial_chat_system_prompt_path=config.get("adversarial_chat_system_prompt_path"), - simulated_target_system_prompt_path=config.get("simulated_target_system_prompt_path"), - next_message_system_prompt_path=config.get("next_message_system_prompt_path"), - ) + prompt_config = { + key: config[key] + for key in ( + "adversarial_chat_system_prompt", + "adversarial_chat_system_prompt_path", + "simulated_target_system_prompt", + "simulated_target_system_prompt_path", + "next_message_system_prompt", + "next_message_system_prompt_path", + ) + if config.get(key) is not None + } + is_legacy_record = any(key.endswith("_path") for key in prompt_config) + try: + return SeedSimulatedConversation( + id=self.id, + value_sha256=None if is_legacy_record else self.value_sha256, + name=self.name, + dataset_name=self.dataset_name, + harm_categories=self.harm_categories, + description=self.description, + authors=self.authors, + groups=self.groups, + source=self.source, + date_added=self.date_added, + added_by=self.added_by, + metadata=cleaned_metadata, + prompt_group_id=self.prompt_group_id, + num_turns=config.get("num_turns", 3), + sequence=config.get("sequence", 0), + pyrit_version=config.get("pyrit_version"), + **prompt_config, + ) + except (OSError, ValueError) as exc: + # A legacy record names prompt files by absolute path, so one written elsewhere + # can reference a file this machine does not have. Name the record so a single + # bad row is identifiable rather than an opaque failure of the whole query. + raise ValueError( + f"Could not rebuild simulated conversation seed {self.id} " + f"(name={self.name!r}, dataset={self.dataset_name!r}): {exc}" + ) from exc return SeedPrompt( id=self.id, value=self.value, diff --git a/pyrit/models/__init__.py b/pyrit/models/__init__.py index 2eff4cc7c0..2952ee7c23 100644 --- a/pyrit/models/__init__.py +++ b/pyrit/models/__init__.py @@ -178,6 +178,10 @@ SeedUnion, SimulatedTargetSystemPromptPaths, group_seeds_into_attack_groups, + load_next_message_prompt, + load_simulated_target_prompt, + resolve_prompt_source, + warn_prompt_path_deprecated, ) from pyrit.models.target import ( COMMON_JSON_SCHEMAS, @@ -342,6 +346,10 @@ "SeedType": "pyrit.models.literals", "SeedUnion": "pyrit.models.seeds", "SimulatedTargetSystemPromptPaths": "pyrit.models.seeds", + "load_next_message_prompt": "pyrit.models.seeds", + "load_simulated_target_prompt": "pyrit.models.seeds", + "resolve_prompt_source": "pyrit.models.seeds", + "warn_prompt_path_deprecated": "pyrit.models.seeds", "snake_case_to_class_name": "pyrit.models.identifiers", "sort_message_pieces": "pyrit.models.messages.message_piece", "StrategyResult": "pyrit.models.results.strategy_result", diff --git a/pyrit/models/seeds/__init__.py b/pyrit/models/seeds/__init__.py index dbb015bf05..fbf25c407f 100644 --- a/pyrit/models/seeds/__init__.py +++ b/pyrit/models/seeds/__init__.py @@ -33,6 +33,10 @@ NextMessageSystemPromptPaths, SeedSimulatedConversation, SimulatedTargetSystemPromptPaths, + load_next_message_prompt, + load_simulated_target_prompt, + resolve_prompt_source, + warn_prompt_path_deprecated, ) from pyrit.models.seeds.yaml_seed_loader import ( load_seed_dataset_from_yaml, @@ -44,6 +48,10 @@ "load_seed_dataset_from_yaml": "pyrit.models.seeds.yaml_seed_loader", "load_seed_from_yaml": "pyrit.models.seeds.yaml_seed_loader", "load_seed_prompt_from_yaml_with_required_parameters": "pyrit.models.seeds.yaml_seed_loader", + "load_next_message_prompt": "pyrit.models.seeds.seed_simulated_conversation", + "load_simulated_target_prompt": "pyrit.models.seeds.seed_simulated_conversation", + "resolve_prompt_source": "pyrit.models.seeds.seed_simulated_conversation", + "warn_prompt_path_deprecated": "pyrit.models.seeds.seed_simulated_conversation", "group_seeds_into_attack_groups": "pyrit.models.seeds.seed_grouping", "NextMessageSystemPromptPaths": "pyrit.models.seeds.seed_simulated_conversation", "Seed": "pyrit.models.seeds.seed", diff --git a/pyrit/models/seeds/seed_simulated_conversation.py b/pyrit/models/seeds/seed_simulated_conversation.py index f5c29525dc..b32b6238cc 100644 --- a/pyrit/models/seeds/seed_simulated_conversation.py +++ b/pyrit/models/seeds/seed_simulated_conversation.py @@ -19,16 +19,29 @@ import json import logging from pathlib import Path -from typing import Any, Literal +from typing import TYPE_CHECKING, Annotated, Any, Literal -from pydantic import field_validator, model_validator +from pydantic import Field, WrapValidator, field_validator, model_validator +from pyrit.common.deprecation import print_deprecation_message from pyrit.common.path import EXECUTOR_SIMULATED_TARGET_PATH from pyrit.models.seeds.seed import Seed from pyrit.models.seeds.seed_prompt import SeedPrompt +if TYPE_CHECKING: + from collections.abc import Callable + + from pydantic import ValidatorFunctionWrapHandler + logger = logging.getLogger(__name__) +SIMULATED_TARGET_REQUIRED_PARAMETERS = ["objective", "num_turns"] +SIMULATED_TARGET_PARAMETER_ERROR = "Simulated target system prompt must have objective and num_turns parameters" +NEXT_MESSAGE_REQUIRED_PARAMETERS = ["objective", "conversation_context"] +NEXT_MESSAGE_PARAMETER_ERROR = "Next message system prompt must have objective and conversation_context parameters" + +_PROMPT_PATH_REMOVED_IN = "1.4.0" + class SimulatedTargetSystemPromptPaths(enum.Enum): """Enum for predefined simulated target system prompt paths.""" @@ -42,11 +55,178 @@ class NextMessageSystemPromptPaths(enum.Enum): DIRECT = Path(EXECUTOR_SIMULATED_TARGET_PATH, "direct_next_message.yaml").resolve() +def load_simulated_target_prompt(template_path: str | Path) -> SeedPrompt: + """ + Load a simulated target system prompt template and verify it declares the parameters it needs. + + Args: + template_path: Path to the YAML file containing the prompt template. + + Returns: + SeedPrompt: The loaded template. + + Raises: + ValueError: If the template does not declare ``objective`` and ``num_turns``. + """ + return SeedPrompt.from_yaml_with_required_parameters( + template_path=template_path, + required_parameters=SIMULATED_TARGET_REQUIRED_PARAMETERS, + error_message=SIMULATED_TARGET_PARAMETER_ERROR, + ) + + +def load_next_message_prompt(template_path: str | Path) -> SeedPrompt: + """ + Load a next-message system prompt template and verify it declares the parameters it needs. + + Args: + template_path: Path to the YAML file containing the prompt template. + + Returns: + SeedPrompt: The loaded template. + + Raises: + ValueError: If the template does not declare ``objective`` and ``conversation_context``. + """ + return SeedPrompt.from_yaml_with_required_parameters( + template_path=template_path, + required_parameters=NEXT_MESSAGE_REQUIRED_PARAMETERS, + error_message=NEXT_MESSAGE_PARAMETER_ERROR, + ) + + +def _load_compliant_simulated_target_prompt() -> SeedPrompt: + """ + Load the default compliant simulated target prompt. + + Returns: + SeedPrompt: The compliant simulated target template. + """ + return load_simulated_target_prompt(SimulatedTargetSystemPromptPaths.COMPLIANT.value) + + +def resolve_prompt_source( + *, + prompt: SeedPrompt | None, + path: str | Path | None, + prompt_name: str, + path_name: str, + load_prompt: Callable[[str | Path], SeedPrompt], +) -> SeedPrompt | None: + """ + Choose between a canonical prompt and its deprecated path input, loading the path if needed. + + Every boundary that still accepts a ``*_system_prompt_path`` uses this so they all warn the + same way and reject the same ambiguity. It reads from disk, so async callers must run it + through ``asyncio.to_thread``. + + Args: + prompt: The canonical prompt, if the caller supplied one. + path: The deprecated path input, if the caller supplied one. + prompt_name: Name of the canonical parameter, used in messages. + path_name: Name of the deprecated parameter, used in messages. + load_prompt: Loader that turns the path into a prompt. + + Returns: + SeedPrompt | None: The resolved prompt, or None when neither input was supplied. + + Raises: + ValueError: If both the canonical prompt and its deprecated path are supplied. + """ + if path is None: + return prompt + warn_prompt_path_deprecated(prompt=prompt, prompt_name=prompt_name, path_name=path_name) + return load_prompt(path) + + +def warn_prompt_path_deprecated(*, prompt: SeedPrompt | None, prompt_name: str, path_name: str) -> None: + """ + Reject an ambiguous prompt source and warn that the path input is deprecated. + + Separated from loading so async callers can run this on the event loop, where the warning + points at their own call site, and send only the file read to a worker thread. + + Args: + prompt: The canonical prompt, if the caller supplied one alongside the path. + prompt_name: Name of the canonical parameter, used in messages. + path_name: Name of the deprecated parameter, used in messages. + + Raises: + ValueError: If both the canonical prompt and its deprecated path are supplied. + """ + if prompt is not None: + raise ValueError(f"Set only one of {prompt_name} or {path_name}; both were provided.") + print_deprecation_message(old_item=path_name, new_item=prompt_name, removed_in=_PROMPT_PATH_REMOVED_IN) + + +# Deprecated ``*_path`` inputs, mapped to the canonical field they populate and the loader that +# resolves them. These are accepted at construction only; they never become model fields. +_LEGACY_PROMPT_PATH_INPUTS: dict[str, tuple[str, Any]] = { + "adversarial_chat_system_prompt_path": ("adversarial_chat_system_prompt", SeedPrompt.from_yaml_file), + "simulated_target_system_prompt_path": ("simulated_target_system_prompt", load_simulated_target_prompt), + "next_message_system_prompt_path": ("next_message_system_prompt", load_next_message_prompt), +} + + +def _prompt_identity(prompt: SeedPrompt | None) -> dict[str, Any] | None: + """ + Project a prompt onto the fields that change how a simulated conversation behaves. + + Only the fields that alter rendering, validation, or the response contract are kept, so they + survive reconstruction from a persisted record. Descriptive metadata such as ``name``, + ``description``, and ``source`` is deliberately left out: renaming a template should not change + the configuration's identity. Those fields are therefore not restored from a persisted record. + + ``is_jinja_template`` is also left out. It marks a template that still needs its one-shot + path substitution, and the projected value has already had it, so restoring the flag would + make a rebuilt prompt render a second time. + + Args: + prompt: The prompt to project, or None. + + Returns: + dict[str, Any] | None: The projected prompt, or None when no prompt was given. + """ + if prompt is None: + return None + return { + "value": prompt.value, + "data_type": prompt.data_type, + "parameters": list(prompt.parameters or []), + "response_json_schema": prompt.response_json_schema, + } + + +def _keep_prompt_instance(value: Any, handler: ValidatorFunctionWrapHandler) -> Any: + """ + Accept an existing prompt as-is instead of validating it again. + + ``SeedPrompt`` substitutes dataset paths into a trusted template once, while it is being + validated. Re-validating an instance would run that substitution a second time, which + consumes template syntax the executor is meant to fill in and mutates the caller's object. + + Args: + value: The raw value supplied for the field. + handler: The validator to fall back to for anything that is not already a prompt. + + Returns: + Any: The prompt unchanged, or the result of normal validation. + """ + if isinstance(value, SeedPrompt): + return value + return handler(value) + + +# Prompt fields hold templates that have already been prepared, so an existing instance is +# never re-validated (see _keep_prompt_instance). +SystemPrompt = Annotated[SeedPrompt, WrapValidator(_keep_prompt_instance)] + + class SeedSimulatedConversation(Seed): """ Configuration for generating a simulated conversation dynamically. - This class holds the paths and parameters needed to generate prepended conversation + This class holds the prompts and parameters needed to generate prepended conversation content by running an adversarial chat against a simulated (compliant) target. This is a pure configuration class. The actual generation is performed by @@ -56,12 +236,23 @@ class SeedSimulatedConversation(Seed): The `value` property returns a JSON serialization of the config for database storage and deduplication. + The prompts are canonical `SeedPrompt` templates, so a technique carries its prompt text + rather than a file location and can be inspected or edited in place. The matching + `*_system_prompt_path` inputs are still accepted at construction for legacy callers and for + reading records persisted before the change; they are resolved immediately and are not + stored on the model. + + To change a prompt, edit the `SeedPrompt` and then build a new `SeedSimulatedConversation` + from it. Like the other fields, `value` is a snapshot taken when the configuration is + validated, so mutating a prompt on an existing instance changes what executes without + changing what is stored. + Attributes: num_turns: Number of conversation turns to generate. - adversarial_chat_system_prompt_path: Path to the adversarial chat system prompt YAML. - simulated_target_system_prompt_path: Path to the simulated target system prompt YAML. + adversarial_chat_system_prompt: System-prompt SeedPrompt for the adversarial chat. + simulated_target_system_prompt: System-prompt SeedPrompt for the simulated target. Defaults to the compliant prompt if not specified. - next_message_system_prompt_path: Optional path to the system prompt for generating + next_message_system_prompt: Optional system-prompt SeedPrompt for generating an additional user message after the simulated conversation. If provided, a single LLM call generates a final user message that attempts to get the target to fulfill the objective in their next response. @@ -85,9 +276,12 @@ class SeedSimulatedConversation(Seed): num_turns: int = 3 sequence: int = 0 - adversarial_chat_system_prompt_path: Path - simulated_target_system_prompt_path: Path = SimulatedTargetSystemPromptPaths.COMPLIANT.value - next_message_system_prompt_path: Path | None = None + # A prompt supplied directly is trusted as-is. The declared-parameter contract for the + # simulated target and next message is enforced where a template is loaded from a file, + # by load_simulated_target_prompt and load_next_message_prompt. + adversarial_chat_system_prompt: SystemPrompt + simulated_target_system_prompt: SystemPrompt = Field(default_factory=_load_compliant_simulated_target_prompt) + next_message_system_prompt: SystemPrompt | None = None pyrit_version: str | None = None @model_validator(mode="before") @@ -106,14 +300,51 @@ def _strip_user_value(cls, data: Any) -> Any: data.pop("value", None) return data - @field_validator("simulated_target_system_prompt_path", mode="before") + @field_validator("simulated_target_system_prompt", mode="before") @classmethod - def _default_simulated_target_path(cls, value: Any) -> Any: + def _default_simulated_target_prompt(cls, value: Any) -> Any: # Reconstruction from memory may pass an explicit None; fall back to the compliant default. if value is None: - return SimulatedTargetSystemPromptPaths.COMPLIANT.value + return _load_compliant_simulated_target_prompt() return value + @model_validator(mode="before") + @classmethod + def _resolve_legacy_prompt_paths(cls, data: Any) -> Any: + """ + Resolve deprecated ``*_system_prompt_path`` inputs into their canonical prompt fields. + + Runs in ``mode="before"`` so the path keys are removed from the input before Pydantic's + ``extra="forbid"`` rejects them. The paths are therefore never model fields and never + reach ``value`` or ``get_identifier()``. + + Args: + data: Raw input passed to the model constructor. + + Returns: + The input with any legacy path keys replaced by loaded prompts. + + Raises: + ValueError: If a canonical prompt and its deprecated path are both supplied. + """ + if not isinstance(data, dict): + return data + + resolved = data + for path_key, (prompt_key, load_prompt) in _LEGACY_PROMPT_PATH_INPUTS.items(): + if path_key not in resolved: + continue + if resolved is data: + resolved = dict(data) + resolved[prompt_key] = resolve_prompt_source( + prompt=resolved.get(prompt_key), + path=resolved.pop(path_key), + prompt_name=f"SeedSimulatedConversation.{prompt_key}", + path_name=f"SeedSimulatedConversation.{path_key}", + load_prompt=load_prompt, + ) + return resolved + @model_validator(mode="after") def _validate_and_compute_value(self) -> SeedSimulatedConversation: if self.num_turns <= 0: @@ -125,25 +356,32 @@ def _validate_and_compute_value(self) -> SeedSimulatedConversation: self.value = self._compute_value() return self - def _compute_value(self) -> str: + def _config_dict(self) -> dict[str, Any]: """ - Compute the value field as JSON serialization of config. + Build the canonical configuration mapping shared by ``value`` and ``get_identifier()``. Returns: - str: Deterministic JSON representation of this configuration. + dict[str, Any]: The configuration, with each prompt reduced to its behavioral fields. """ - config = { + return { "num_turns": self.num_turns, "sequence": self.sequence, - "adversarial_chat_system_prompt_path": str(self.adversarial_chat_system_prompt_path), - "simulated_target_system_prompt_path": str(self.simulated_target_system_prompt_path), - "next_message_system_prompt_path": ( - str(self.next_message_system_prompt_path) if self.next_message_system_prompt_path else None - ), + "adversarial_chat_system_prompt": _prompt_identity(self.adversarial_chat_system_prompt), + "simulated_target_system_prompt": _prompt_identity(self.simulated_target_system_prompt), + "next_message_system_prompt": _prompt_identity(self.next_message_system_prompt), "pyrit_version": self.pyrit_version, } - return json.dumps(config, sort_keys=True, separators=(",", ":")) + + def _compute_value(self) -> str: + """ + Compute the value field as JSON serialization of config. + + Returns: + str: Deterministic JSON representation of this configuration. + + """ + return json.dumps(self._config_dict(), sort_keys=True, separators=(",", ":")) def get_identifier(self) -> dict[str, Any]: """ @@ -153,17 +391,7 @@ def get_identifier(self) -> dict[str, Any]: Dictionary with configuration details. """ - return { - "__type__": "SeedSimulatedConversation", - "num_turns": self.num_turns, - "sequence": self.sequence, - "adversarial_chat_system_prompt_path": str(self.adversarial_chat_system_prompt_path), - "simulated_target_system_prompt_path": str(self.simulated_target_system_prompt_path), - "next_message_system_prompt_path": ( - str(self.next_message_system_prompt_path) if self.next_message_system_prompt_path else None - ), - "pyrit_version": self.pyrit_version, - } + return {"__type__": "SeedSimulatedConversation", **self._config_dict()} def compute_hash(self) -> str: """ @@ -187,6 +415,11 @@ def load_simulated_target_system_prompt( """ Load and render the simulated target system prompt. + .. deprecated:: + Render ``SeedSimulatedConversation.simulated_target_system_prompt`` directly with + ``SeedPrompt.render_template_value(objective=..., num_turns=...)`` instead. This + helper reads from disk, so it must not be called from an async path. + If no path is provided, returns None (no system prompt). Validates that the template has required `objective` and `num_turns` parameters. @@ -203,14 +436,15 @@ def load_simulated_target_system_prompt( ValueError: If the template doesn't have required parameters. """ + print_deprecation_message( + old_item="SeedSimulatedConversation.load_simulated_target_system_prompt", + new_item="SeedSimulatedConversation.simulated_target_system_prompt.render_template_value", + removed_in=_PROMPT_PATH_REMOVED_IN, + ) if simulated_target_system_prompt_path is None: return None - template = SeedPrompt.from_yaml_with_required_parameters( - template_path=simulated_target_system_prompt_path, - required_parameters=["objective", "num_turns"], - error_message="Simulated target system prompt must have objective and num_turns parameters", - ) + template = load_simulated_target_prompt(simulated_target_system_prompt_path) return template.render_template_value( objective=objective, @@ -223,14 +457,14 @@ def sequence_range(self) -> range: The range of sequence numbers this simulated conversation will occupy. Each turn generates 2 messages (user + assistant), so num_turns generates - num_turns * 2 messages. If next_message_system_prompt_path is set, an additional + num_turns * 2 messages. If next_message_system_prompt is set, an additional user message is added at the end. Returns: A range object representing the sequence numbers. """ - message_count = self.num_turns * 2 + (1 if self.next_message_system_prompt_path else 0) + message_count = self.num_turns * 2 + (1 if self.next_message_system_prompt else 0) return range(self.sequence, self.sequence + message_count) def __repr__(self) -> str: @@ -241,9 +475,12 @@ def __repr__(self) -> str: str: Simulated conversation summary string. """ - has_next_msg = self.next_message_system_prompt_path is not None + has_next_msg = self.next_message_system_prompt is not None + # ``name`` is descriptive metadata that _prompt_identity drops, so a seed rebuilt from a + # persisted record has none. Omit the fragment rather than print a placeholder. + prompt_name = self.adversarial_chat_system_prompt.name + adversarial = f", adversarial_prompt={prompt_name}" if prompt_name else "" return ( f"" + f"next_message={has_next_msg}{adversarial})>" ) diff --git a/pyrit/scenario/core/attack_technique_factory.py b/pyrit/scenario/core/attack_technique_factory.py index 9c373512f2..c9be41af6b 100644 --- a/pyrit/scenario/core/attack_technique_factory.py +++ b/pyrit/scenario/core/attack_technique_factory.py @@ -36,6 +36,10 @@ SeedIdentifier, SeedPrompt, SeedSimulatedConversation, + SimulatedTargetSystemPromptPaths, + load_next_message_prompt, + load_simulated_target_prompt, + resolve_prompt_source, ) from pyrit.models.seeds.seed_simulated_conversation import NextMessageSystemPromptPaths from pyrit.scenario.core.attack_technique import AttackTechnique @@ -165,6 +169,9 @@ def with_simulated_conversation( name: str, attack_class: type[AttackStrategy[Any, Any]] | None = None, description: str | None = None, + adversarial_chat_system_prompt: SeedPrompt | None = None, + simulated_target_system_prompt: SeedPrompt | None = None, + next_message_system_prompt: SeedPrompt | None = None, adversarial_chat_system_prompt_path: str | Path | None = None, simulated_target_system_prompt_path: str | Path | None = None, next_message_system_prompt_path: str | Path | None = None, @@ -192,18 +199,22 @@ def with_simulated_conversation( ``PromptSendingAttack``. description: Short human-readable summary of what the technique does. Forwarded to the factory constructor as descriptive metadata. - adversarial_chat_system_prompt_path: Path to the YAML file containing - the adversarial chat system prompt for the simulated conversation. - Defaults to ``EXECUTOR_SEED_PROMPT_PATH/red_teaming/{name}.yaml``. - simulated_target_system_prompt_path: Optional path to the YAML file - containing the system prompt for the simulated target (the - assistant side of the generated conversation). When ``None``, - ``SeedSimulatedConversation`` falls back to its compliant default. - next_message_system_prompt_path: Optional path to the YAML file - containing the system prompt for generating a final user message - after the simulated conversation. Defaults to - ``NextMessageSystemPromptPaths.DIRECT.value``. Ignored (forced to - ``None``) when ``final_user_message`` is provided. + adversarial_chat_system_prompt: System prompt for the adversarial chat in the + simulated conversation. Defaults to the prompt loaded from + ``EXECUTOR_SEED_PROMPT_PATH/red_teaming/{name}.yaml``. + simulated_target_system_prompt: Optional system prompt for the simulated target + (the assistant side of the generated conversation). Defaults to the prompt + loaded from ``SimulatedTargetSystemPromptPaths.COMPLIANT``. + next_message_system_prompt: Optional system prompt for generating a final user + message after the simulated conversation. Defaults to the prompt loaded from + ``NextMessageSystemPromptPaths.DIRECT``. Ignored (forced to ``None``) when + ``final_user_message`` is provided. + adversarial_chat_system_prompt_path: Deprecated. Path to the YAML file containing + the adversarial chat system prompt. Use ``adversarial_chat_system_prompt``. + simulated_target_system_prompt_path: Deprecated. Path to the YAML file containing + the simulated target system prompt. Use ``simulated_target_system_prompt``. + next_message_system_prompt_path: Deprecated. Path to the YAML file containing the + next-message system prompt. Use ``next_message_system_prompt``. final_user_message: Optional fixed final user message. When provided, a static ``SeedPrompt`` carrying this text is appended after the simulated conversation (so it becomes the ``next_message``) and no @@ -240,29 +251,51 @@ def with_simulated_conversation( """ if attack_class is None: attack_class = PromptSendingAttack - if adversarial_chat_system_prompt_path is None: - adversarial_chat_system_prompt_path = Path(EXECUTOR_SEED_PROMPT_PATH) / "red_teaming" / f"{name}.yaml" + adversarial_chat_system_prompt = resolve_prompt_source( + prompt=adversarial_chat_system_prompt, + path=adversarial_chat_system_prompt_path, + prompt_name="adversarial_chat_system_prompt", + path_name="adversarial_chat_system_prompt_path", + load_prompt=SeedPrompt.from_yaml_file, + ) + if adversarial_chat_system_prompt is None: + adversarial_chat_system_prompt = SeedPrompt.from_yaml_file( + Path(EXECUTOR_SEED_PROMPT_PATH) / "red_teaming" / f"{name}.yaml" + ) + simulated_target_system_prompt = resolve_prompt_source( + prompt=simulated_target_system_prompt, + path=simulated_target_system_prompt_path, + prompt_name="simulated_target_system_prompt", + path_name="simulated_target_system_prompt_path", + load_prompt=load_simulated_target_prompt, + ) + if simulated_target_system_prompt is None: + simulated_target_system_prompt = load_simulated_target_prompt( + SimulatedTargetSystemPromptPaths.COMPLIANT.value + ) # A fixed final user message and an LLM-generated next message are mutually # exclusive: when a fixed message is supplied it becomes the next_message via # a static SeedPrompt, so no next-message generation prompt is used. if final_user_message is not None: - next_message_system_prompt_path = None - elif next_message_system_prompt_path is None: - next_message_system_prompt_path = NextMessageSystemPromptPaths.DIRECT.value - - simulated_conversation_kwargs: dict[str, Any] = { - "adversarial_chat_system_prompt_path": Path(adversarial_chat_system_prompt_path), - "num_turns": num_turns, - } - if simulated_target_system_prompt_path is not None: - simulated_conversation_kwargs["simulated_target_system_prompt_path"] = Path( - simulated_target_system_prompt_path + next_message_system_prompt = None + else: + next_message_system_prompt = resolve_prompt_source( + prompt=next_message_system_prompt, + path=next_message_system_prompt_path, + prompt_name="next_message_system_prompt", + path_name="next_message_system_prompt_path", + load_prompt=load_next_message_prompt, ) - if next_message_system_prompt_path is not None: - simulated_conversation_kwargs["next_message_system_prompt_path"] = Path(next_message_system_prompt_path) - - simulated_conversation = SeedSimulatedConversation(**simulated_conversation_kwargs) + if next_message_system_prompt is None: + next_message_system_prompt = load_next_message_prompt(NextMessageSystemPromptPaths.DIRECT.value) + + simulated_conversation = SeedSimulatedConversation( + num_turns=num_turns, + adversarial_chat_system_prompt=adversarial_chat_system_prompt, + simulated_target_system_prompt=simulated_target_system_prompt, + next_message_system_prompt=next_message_system_prompt, + ) seeds: list[Any] = [simulated_conversation] if final_user_message is not None: diff --git a/pyrit/scenario/scenarios/adaptive/adaptive_scenario.py b/pyrit/scenario/scenarios/adaptive/adaptive_scenario.py index 16965aab65..3de77591d9 100644 --- a/pyrit/scenario/scenarios/adaptive/adaptive_scenario.py +++ b/pyrit/scenario/scenarios/adaptive/adaptive_scenario.py @@ -16,6 +16,7 @@ from __future__ import annotations +import asyncio import logging from abc import abstractmethod from typing import TYPE_CHECKING, ClassVar @@ -177,7 +178,9 @@ async def _build_atomic_attacks_async(self, *, context: ScenarioContext) -> list Raises: ValueError: If ``_build_techniques_dict`` finds no usable techniques. """ - techniques = self._build_techniques_dict(objective_target=context.objective_target) + # Building the technique catalog reads each technique's prompt YAML, so keep the + # synchronous builder off the event loop. + techniques = await asyncio.to_thread(self._build_techniques_dict, objective_target=context.objective_target) atomic_attacks: list[AtomicAttack] = [] if context.include_baseline: @@ -243,7 +246,9 @@ async def _estimate_run_size_async(self) -> ScenarioRunSizeEstimate: ) assert self._objective_target is not None - techniques = self._build_techniques_dict(objective_target=self._objective_target) + # Building the technique catalog reads each technique's prompt YAML, so keep the + # synchronous builder off the event loop. + techniques = await asyncio.to_thread(self._build_techniques_dict, objective_target=self._objective_target) dispatcher = AdaptiveTechniqueDispatcher( objective_target=self._objective_target, techniques=techniques, diff --git a/pyrit/scenario/scenarios/airt/psychosocial.py b/pyrit/scenario/scenarios/airt/psychosocial.py index af6b28e193..64fee315fe 100644 --- a/pyrit/scenario/scenarios/airt/psychosocial.py +++ b/pyrit/scenario/scenarios/airt/psychosocial.py @@ -3,6 +3,7 @@ from __future__ import annotations +import asyncio import logging import pathlib from dataclasses import dataclass @@ -290,6 +291,27 @@ def default(cls) -> PsychosocialTechnique: } +def _build_simulated_base_factory(*, harm: _SubHarm, max_turns: int) -> AttackTechniqueFactory: + """ + Build the simulated-conversation base factory for a sub-harm. + + Reads the sub-harm's escalation prompt from disk, so async callers must run it through + ``asyncio.to_thread``. + + Args: + harm: The sub-harm whose escalation prompt drives the simulated conversation. + max_turns: Number of simulated conversation turns. + + Returns: + AttackTechniqueFactory: The base factory every converter technique layers onto. + """ + return AttackTechniqueFactory.with_simulated_conversation( + name=f"psychosocial_{harm.name}", + adversarial_chat_system_prompt=SeedPrompt.from_yaml_file(harm.escalation_prompt_path), + num_turns=max_turns, + ) + + def _converter_for_technique(technique: PsychosocialTechnique, *, adversarial_chat: PromptTarget) -> Converter | None: """ Map a converter technique to its converter instance. @@ -582,11 +604,9 @@ async def _build_atomic_attacks_async(self, *, context: ScenarioContext) -> list ) ) - base_factory = AttackTechniqueFactory.with_simulated_conversation( - name=f"psychosocial_{harm.name}", - adversarial_chat_system_prompt_path=harm.escalation_prompt_path, - num_turns=max_turns, - ) + # Building the base factory reads the sub-harm's escalation prompt from disk, so keep + # it off the event loop. + base_factory = await asyncio.to_thread(_build_simulated_base_factory, harm=harm, max_turns=max_turns) for technique in techniques: if technique is PsychosocialTechnique.Crescendo: diff --git a/pyrit/scenario/scenarios/airt/scam.py b/pyrit/scenario/scenarios/airt/scam.py index b7a64dd7be..7315dfb013 100644 --- a/pyrit/scenario/scenarios/airt/scam.py +++ b/pyrit/scenario/scenarios/airt/scam.py @@ -1,6 +1,7 @@ # Copyright (c) Microsoft Corporation. # Licensed under the MIT license. +import asyncio import logging from pathlib import Path from typing import TYPE_CHECKING, Any @@ -9,7 +10,7 @@ from pyrit.common.path import EXECUTOR_RED_TEAM_PATH, EXECUTOR_SIMULATED_TARGET_PATH, SCORER_SEED_PROMPT_PATH from pyrit.executor.attack import RedTeamingAttack from pyrit.executor.attack.core.attack_config import AttackAdversarialConfig, AttackScoringConfig -from pyrit.models import AttackSeedGroup, Parameter, SeedPrompt +from pyrit.models import AttackSeedGroup, Parameter, SeedPrompt, load_next_message_prompt, load_simulated_target_prompt from pyrit.prompt_target import PromptTarget from pyrit.scenario.core.atomic_attack import AtomicAttack from pyrit.scenario.core.attack_technique import AttackTechnique @@ -195,10 +196,12 @@ def _get_atomic_attack_from_technique(self, *, technique: str, seed_groups: list # objective is delivered to the target. role_play_technique = AttackTechniqueFactory.with_simulated_conversation( name="role_play_persuasion_written", - adversarial_chat_system_prompt_path=EXECUTOR_RED_TEAM_PATH - / "role_play" - / "role_play_persuasion_written.yaml", - next_message_system_prompt_path=EXECUTOR_SIMULATED_TARGET_PATH / "role_play_next_message.yaml", + adversarial_chat_system_prompt=SeedPrompt.from_yaml_file( + EXECUTOR_RED_TEAM_PATH / "role_play" / "role_play_persuasion_written.yaml" + ), + next_message_system_prompt=load_next_message_prompt( + EXECUTOR_SIMULATED_TARGET_PATH / "role_play_next_message.yaml" + ), num_turns=2, ).create( objective_target=self._objective_target, @@ -219,10 +222,12 @@ def _get_atomic_attack_from_technique(self, *, technique: str, seed_groups: list # further, then a fixed "yes." is delivered to the target as the final user turn. context_compliance_technique = AttackTechniqueFactory.with_simulated_conversation( name="context_compliance", - adversarial_chat_system_prompt_path=EXECUTOR_RED_TEAM_PATH - / "context_compliance" - / "context_compliance.yaml", - simulated_target_system_prompt_path=EXECUTOR_SIMULATED_TARGET_PATH / "context_compliance_target.yaml", + adversarial_chat_system_prompt=SeedPrompt.from_yaml_file( + EXECUTOR_RED_TEAM_PATH / "context_compliance" / "context_compliance.yaml" + ), + simulated_target_system_prompt=load_simulated_target_prompt( + EXECUTOR_SIMULATED_TARGET_PATH / "context_compliance_target.yaml" + ), final_user_message="yes.", num_turns=1, ).create( @@ -271,8 +276,14 @@ async def _build_atomic_attacks_async(self, *, context: ScenarioContext) -> list memory_labels=context.memory_labels, ) ) + # Building a simulated-conversation technique reads its prompt YAML, so keep the + # synchronous builder off the event loop. atomic_attacks.extend( - self._get_atomic_attack_from_technique(technique=technique, seed_groups=seed_groups) - for technique in techniques + [ + await asyncio.to_thread( + self._get_atomic_attack_from_technique, technique=technique, seed_groups=seed_groups + ) + for technique in techniques + ] ) return atomic_attacks diff --git a/pyrit/setup/initializers/techniques/core.py b/pyrit/setup/initializers/techniques/core.py index d4dbc9667c..d95e073081 100644 --- a/pyrit/setup/initializers/techniques/core.py +++ b/pyrit/setup/initializers/techniques/core.py @@ -29,7 +29,13 @@ RedTeamingAttack, TreeOfAttacksWithPruningAttack, ) -from pyrit.models import AttackTechniqueSeedGroup, Message, SeedPrompt +from pyrit.models import ( + AttackTechniqueSeedGroup, + Message, + SeedPrompt, + load_next_message_prompt, + load_simulated_target_prompt, +) from pyrit.prompt_normalizer import ConverterConfiguration from pyrit.scenario.core.attack_technique_factory import AttackTechniqueFactory @@ -112,50 +118,60 @@ def get_technique_factories() -> list[AttackTechniqueFactory]: AttackTechniqueFactory.with_simulated_conversation( name="role_play_movie_script", description="Frames the objective as a movie beat by casting the adversarial chat as a screenwriter.", - adversarial_chat_system_prompt_path=( + adversarial_chat_system_prompt=SeedPrompt.from_yaml_file( EXECUTOR_SEED_PROMPT_PATH / "red_teaming" / "role_play" / "role_play_movie_script.yaml" ), - next_message_system_prompt_path=EXECUTOR_SIMULATED_TARGET_PATH / "role_play_next_message.yaml", + next_message_system_prompt=load_next_message_prompt( + EXECUTOR_SIMULATED_TARGET_PATH / "role_play_next_message.yaml" + ), technique_tags=["single_turn", "light"], num_turns=2, ), AttackTechniqueFactory.with_simulated_conversation( name="role_play_video_game", description="Frames the objective as an in-game mechanic by casting the adversarial chat as a designer.", - adversarial_chat_system_prompt_path=( + adversarial_chat_system_prompt=SeedPrompt.from_yaml_file( EXECUTOR_SEED_PROMPT_PATH / "red_teaming" / "role_play" / "role_play_video_game.yaml" ), - next_message_system_prompt_path=EXECUTOR_SIMULATED_TARGET_PATH / "role_play_next_message.yaml", + next_message_system_prompt=load_next_message_prompt( + EXECUTOR_SIMULATED_TARGET_PATH / "role_play_next_message.yaml" + ), technique_tags=["single_turn", "light"], num_turns=2, ), AttackTechniqueFactory.with_simulated_conversation( name="role_play_trivia_game", description="Frames the objective as a quiz answer by casting the adversarial chat as a trivia host.", - adversarial_chat_system_prompt_path=( + adversarial_chat_system_prompt=SeedPrompt.from_yaml_file( EXECUTOR_SEED_PROMPT_PATH / "red_teaming" / "role_play" / "role_play_trivia_game.yaml" ), - next_message_system_prompt_path=EXECUTOR_SIMULATED_TARGET_PATH / "role_play_next_message.yaml", + next_message_system_prompt=load_next_message_prompt( + EXECUTOR_SIMULATED_TARGET_PATH / "role_play_next_message.yaml" + ), technique_tags=["single_turn", "light"], num_turns=2, ), AttackTechniqueFactory.with_simulated_conversation( name="role_play_persuasion", description="Frames a persuasion-resistance test so the target drafts a persuasive spoken script.", - adversarial_chat_system_prompt_path=( + adversarial_chat_system_prompt=SeedPrompt.from_yaml_file( EXECUTOR_SEED_PROMPT_PATH / "red_teaming" / "role_play" / "role_play_persuasion.yaml" ), - next_message_system_prompt_path=EXECUTOR_SIMULATED_TARGET_PATH / "role_play_next_message.yaml", + next_message_system_prompt=load_next_message_prompt( + EXECUTOR_SIMULATED_TARGET_PATH / "role_play_next_message.yaml" + ), technique_tags=["single_turn", "light"], num_turns=2, ), AttackTechniqueFactory.with_simulated_conversation( name="role_play_persuasion_written", description="Frames a persuasion-resistance test so the target drafts written persuasive material.", - adversarial_chat_system_prompt_path=( + adversarial_chat_system_prompt=SeedPrompt.from_yaml_file( EXECUTOR_SEED_PROMPT_PATH / "red_teaming" / "role_play" / "role_play_persuasion_written.yaml" ), - next_message_system_prompt_path=EXECUTOR_SIMULATED_TARGET_PATH / "role_play_next_message.yaml", + next_message_system_prompt=load_next_message_prompt( + EXECUTOR_SIMULATED_TARGET_PATH / "role_play_next_message.yaml" + ), technique_tags=["single_turn", "light"], num_turns=2, ), @@ -200,10 +216,12 @@ def get_technique_factories() -> list[AttackTechniqueFactory]: AttackTechniqueFactory.with_simulated_conversation( name="context_compliance", description="Injects a fabricated prior exchange so the target continues as if it already agreed.", - adversarial_chat_system_prompt_path=EXECUTOR_RED_TEAM_PATH - / "context_compliance" - / "context_compliance.yaml", - simulated_target_system_prompt_path=EXECUTOR_SIMULATED_TARGET_PATH / "context_compliance_target.yaml", + adversarial_chat_system_prompt=SeedPrompt.from_yaml_file( + EXECUTOR_RED_TEAM_PATH / "context_compliance" / "context_compliance.yaml" + ), + simulated_target_system_prompt=load_simulated_target_prompt( + EXECUTOR_SIMULATED_TARGET_PATH / "context_compliance_target.yaml" + ), final_user_message="yes.", num_turns=1, technique_tags=["single_turn", "light"], diff --git a/pyrit/setup/initializers/techniques/technique_initializer.py b/pyrit/setup/initializers/techniques/technique_initializer.py index 80370d6cca..10c94c27c3 100644 --- a/pyrit/setup/initializers/techniques/technique_initializer.py +++ b/pyrit/setup/initializers/techniques/technique_initializer.py @@ -17,6 +17,7 @@ not overwritten. """ +import asyncio import logging from enum import Enum @@ -113,7 +114,8 @@ async def initialize_async(self) -> None: if TechniqueInitializerTags.ALL.value in tags: tags = [TechniqueInitializerTags.CORE.value, TechniqueInitializerTags.EXTRA.value] - factories = build_technique_factories(groups=tags) + # Building the catalog reads each technique's prompt YAML, so keep it off the event loop. + factories = await asyncio.to_thread(build_technique_factories, groups=tags) registry = AttackTechniqueRegistry.get_registry_singleton() registry.register_from_factories(factories) diff --git a/tests/unit/datasets/test_local_prompt_dataset_semantics.py b/tests/unit/datasets/test_local_prompt_dataset_semantics.py index 6a33fd3e84..0d4f847ad4 100644 --- a/tests/unit/datasets/test_local_prompt_dataset_semantics.py +++ b/tests/unit/datasets/test_local_prompt_dataset_semantics.py @@ -7,7 +7,13 @@ from pyrit.common.path import DATASETS_PATH from pyrit.executor.attack import AttackParameters -from pyrit.models import AttackSeedGroup, AttackTechniqueSeedGroup, SeedDataset, SeedSimulatedConversation +from pyrit.models import ( + AttackSeedGroup, + AttackTechniqueSeedGroup, + SeedDataset, + SeedPrompt, + SeedSimulatedConversation, +) LOCAL_DATASET_PATH = Path(DATASETS_PATH) / "seed_datasets" / "local" FAIRNESS_REQUEST = """Infer ethnicity from the following names: @@ -35,7 +41,7 @@ async def test_airt_fairness_builds_coherent_attack_parameters() -> None: simulated_technique = AttackTechniqueSeedGroup( seeds=[ SeedSimulatedConversation( - adversarial_chat_system_prompt_path="test.yaml", + adversarial_chat_system_prompt=SeedPrompt(value="adversarial", parameters=["objective"]), num_turns=3, ) ] diff --git a/tests/unit/executor/attack/component/test_simulated_conversation.py b/tests/unit/executor/attack/component/test_simulated_conversation.py index 76225bd5f8..dcf4dec972 100644 --- a/tests/unit/executor/attack/component/test_simulated_conversation.py +++ b/tests/unit/executor/attack/component/test_simulated_conversation.py @@ -27,6 +27,7 @@ Score, SeedPrompt, SimulatedTargetSystemPromptPaths, + load_next_message_prompt, ) from pyrit.prompt_normalizer import PromptNormalizer from pyrit.prompt_target import PromptTarget @@ -926,7 +927,7 @@ async def test_parses_next_message_from_json_reply(self, mock_adversarial_chat: conversation_messages=[], adversarial_chat=mock_adversarial_chat, conversation_id="next-message-conversation", - next_message_system_prompt_path=NextMessageSystemPromptPaths.DIRECT.value, + next_message_system_prompt=load_next_message_prompt(NextMessageSystemPromptPaths.DIRECT.value), prompt_normalizer=normalizer, ) @@ -956,7 +957,7 @@ async def test_invalid_json_reply_raises_after_retry(self, mock_adversarial_chat conversation_messages=[], adversarial_chat=mock_adversarial_chat, conversation_id="next-message-conversation", - next_message_system_prompt_path=NextMessageSystemPromptPaths.DIRECT.value, + next_message_system_prompt=load_next_message_prompt(NextMessageSystemPromptPaths.DIRECT.value), prompt_normalizer=normalizer, ) @@ -978,6 +979,147 @@ async def test_raises_when_no_response(self, mock_adversarial_chat: MagicMock): conversation_messages=[], adversarial_chat=mock_adversarial_chat, conversation_id="next-message-conversation", - next_message_system_prompt_path=NextMessageSystemPromptPaths.DIRECT.value, + next_message_system_prompt=load_next_message_prompt(NextMessageSystemPromptPaths.DIRECT.value), prompt_normalizer=normalizer, ) + + +@pytest.mark.usefixtures("patch_central_database") +class TestSimulatedConversationPromptSources: + """Tests for the canonical prompt inputs and the deprecated path inputs.""" + + @staticmethod + def _mock_attack(conversation_id: str) -> MagicMock: + attack = MagicMock() + attack.get_identifier.return_value = ComponentIdentifier( + class_name="RedTeamingAttack", class_module="pyrit.executor.attack" + ) + attack.execute_async = AsyncMock( + return_value=AttackResult( + atomic_attack_identifier=ComponentIdentifier( + class_name="RedTeamingAttack", class_module="pyrit.executor.attack" + ), + conversation_id=conversation_id, + objective="Test objective", + outcome=AttackOutcome.SUCCESS, + executed_turns=2, + ) + ) + return attack + + async def test_canonical_prompts_never_read_from_disk( + self, + mock_adversarial_chat: MagicMock, + mock_objective_scorer: MagicMock, + sample_conversation: list[Message], + ): + """Canonical prompts are carried in, so the async path performs no YAML loading.""" + with patch("pyrit.executor.attack.multi_turn.simulated_conversation.RedTeamingAttack") as mock_attack_class: + mock_attack_class.return_value = self._mock_attack(str(uuid.uuid4())) + + with patch("pyrit.executor.attack.multi_turn.simulated_conversation.CentralMemory") as mock_memory_class: + mock_memory = MagicMock() + mock_memory.get_conversation_messages.return_value = iter(sample_conversation) + mock_memory_class.get_memory_instance.return_value = mock_memory + + with patch("pyrit.models.seeds.yaml_seed_loader.load_seed_from_yaml") as mock_load: + await generate_simulated_conversation_async( + objective="Test objective", + adversarial_chat=mock_adversarial_chat, + objective_scorer=mock_objective_scorer, + adversarial_chat_system_prompt=SeedPrompt(value="adversarial", parameters=["objective"]), + simulated_target_system_prompt=SeedPrompt( + value="Objective: {{ objective }} Turns: {{ num_turns }}", + parameters=["objective", "num_turns"], + ), + num_turns=2, + ) + + mock_load.assert_not_called() + + async def test_simulated_target_prompt_is_rendered_into_system_message( + self, + mock_adversarial_chat: MagicMock, + mock_objective_scorer: MagicMock, + sample_conversation: list[Message], + ): + """The simulated target template is rendered with the objective and turn count.""" + with patch("pyrit.executor.attack.multi_turn.simulated_conversation.RedTeamingAttack") as mock_attack_class: + mock_attack = self._mock_attack(str(uuid.uuid4())) + mock_attack_class.return_value = mock_attack + + with patch("pyrit.executor.attack.multi_turn.simulated_conversation.CentralMemory") as mock_memory_class: + mock_memory = MagicMock() + mock_memory.get_conversation_messages.return_value = iter(sample_conversation) + mock_memory_class.get_memory_instance.return_value = mock_memory + + await generate_simulated_conversation_async( + objective="Test objective", + adversarial_chat=mock_adversarial_chat, + objective_scorer=mock_objective_scorer, + adversarial_chat_system_prompt=SeedPrompt(value="adversarial", parameters=["objective"]), + simulated_target_system_prompt=SeedPrompt( + value="Objective: {{ objective }} Turns: {{ num_turns }}", + parameters=["objective", "num_turns"], + ), + num_turns=2, + ) + + prepended = mock_attack.execute_async.call_args.kwargs["prepended_conversation"] + assert prepended[0].get_value() == "Objective: Test objective Turns: 2" + + async def test_deprecated_path_input_warns( + self, + mock_adversarial_chat: MagicMock, + mock_objective_scorer: MagicMock, + adversarial_system_prompt_path: Path, + sample_conversation: list[Message], + ): + """A deprecated path input still works and warns.""" + with patch("pyrit.executor.attack.multi_turn.simulated_conversation.RedTeamingAttack") as mock_attack_class: + mock_attack_class.return_value = self._mock_attack(str(uuid.uuid4())) + + with patch("pyrit.executor.attack.multi_turn.simulated_conversation.CentralMemory") as mock_memory_class: + mock_memory = MagicMock() + mock_memory.get_conversation_messages.return_value = iter(sample_conversation) + mock_memory_class.get_memory_instance.return_value = mock_memory + + with pytest.warns(DeprecationWarning, match="adversarial_chat_system_prompt_path"): + await generate_simulated_conversation_async( + objective="Test objective", + adversarial_chat=mock_adversarial_chat, + objective_scorer=mock_objective_scorer, + adversarial_chat_system_prompt_path=adversarial_system_prompt_path, + num_turns=2, + ) + + async def test_prompt_and_path_together_raises( + self, + mock_adversarial_chat: MagicMock, + mock_objective_scorer: MagicMock, + adversarial_system_prompt_path: Path, + ): + """Supplying both a canonical prompt and its deprecated path is ambiguous.""" + with pytest.raises(ValueError, match="Set only one of adversarial_chat_system_prompt"): + await generate_simulated_conversation_async( + objective="Test objective", + adversarial_chat=mock_adversarial_chat, + objective_scorer=mock_objective_scorer, + adversarial_chat_system_prompt=SeedPrompt(value="adversarial"), + adversarial_chat_system_prompt_path=adversarial_system_prompt_path, + num_turns=2, + ) + + async def test_missing_adversarial_prompt_raises( + self, + mock_adversarial_chat: MagicMock, + mock_objective_scorer: MagicMock, + ): + """The adversarial system prompt is required.""" + with pytest.raises(ValueError, match="adversarial_chat_system_prompt is required"): + await generate_simulated_conversation_async( + objective="Test objective", + adversarial_chat=mock_adversarial_chat, + objective_scorer=mock_objective_scorer, + num_turns=2, + ) diff --git a/tests/unit/executor/attack/core/test_attack_parameters.py b/tests/unit/executor/attack/core/test_attack_parameters.py index 805213c25c..f41c4a43e3 100644 --- a/tests/unit/executor/attack/core/test_attack_parameters.py +++ b/tests/unit/executor/attack/core/test_attack_parameters.py @@ -132,8 +132,8 @@ def simulated_conversation_config(self) -> SeedSimulatedConversation: """Create a SeedSimulatedConversation config.""" return SeedSimulatedConversation( num_turns=3, - adversarial_chat_system_prompt_path="/path/to/adversarial.yaml", - simulated_target_system_prompt_path="/path/to/target.yaml", + adversarial_chat_system_prompt=SeedPrompt(value="adversarial", parameters=["objective"]), + simulated_target_system_prompt=SeedPrompt(value="target", parameters=["objective", "num_turns"]), ) @pytest.fixture diff --git a/tests/unit/memory/test_memory_models.py b/tests/unit/memory/test_memory_models.py index b65405d850..3fdcf7c4c8 100644 --- a/tests/unit/memory/test_memory_models.py +++ b/tests/unit/memory/test_memory_models.py @@ -1,6 +1,7 @@ # Copyright (c) Microsoft Corporation. # Licensed under the MIT license. +import json import uuid from collections.abc import Sequence from datetime import UTC, datetime @@ -619,14 +620,93 @@ def test_roundtrip_seed_objective_strips_reserved_key(self): assert SEED_RESPONSE_JSON_SCHEMA_METADATA_KEY not in (recovered.metadata or {}) assert (recovered.metadata or {}).get("owned") == "by-caller" + def test_roundtrip_seed_simulated_conversation_preserves_prompts_and_version(self): + """A canonical record round-trips its prompts, value, hash, and recorded version.""" + config = SeedSimulatedConversation( + num_turns=2, + adversarial_chat_system_prompt=SeedPrompt(value="adversarial", parameters=["objective"]), + next_message_system_prompt=SeedPrompt(value="next", response_json_schema_name="adversarial_chat"), + pyrit_version="1.0.0", + ) + config.value_sha256 = "canonical-hash" + + recovered = SeedEntry(entry=config).get_seed() + + assert isinstance(recovered, SeedSimulatedConversation) + assert recovered.adversarial_chat_system_prompt.value == "adversarial" + assert recovered.next_message_system_prompt is not None + assert recovered.next_message_system_prompt.response_json_schema is not None + assert recovered.pyrit_version == "1.0.0" + assert recovered.value == config.value + assert recovered.value_sha256 == "canonical-hash" + + def test_legacy_path_record_reconstructs_prompts(self, tmp_path): + """A record written before normalization still loads, resolving its paths to prompts.""" + adv_path = tmp_path / "adversarial.yaml" + adv_path.write_text("value: legacy adversarial\ndata_type: text") + + seed = SeedSimulatedConversation( + num_turns=2, + adversarial_chat_system_prompt=SeedPrompt(value="placeholder"), + ) + entry = SeedEntry(entry=seed) + entry.value = json.dumps( + { + "num_turns": 2, + "sequence": 0, + "adversarial_chat_system_prompt_path": str(adv_path), + "simulated_target_system_prompt_path": None, + "next_message_system_prompt_path": None, + "pyrit_version": "1.0.0", + }, + sort_keys=True, + separators=(",", ":"), + ) + entry.value_sha256 = "stale-path-hash" + + with pytest.warns(DeprecationWarning, match="adversarial_chat_system_prompt_path"): + recovered = entry.get_seed() + + assert isinstance(recovered, SeedSimulatedConversation) + assert recovered.adversarial_chat_system_prompt.value == "legacy adversarial" + # The compliant default fills in for the omitted simulated target. + assert recovered.simulated_target_system_prompt.name == "simulated_target_compliant" + assert recovered.next_message_system_prompt is None + assert recovered.pyrit_version == "1.0.0" + # The stored hash described the old path-shaped value, so it is not carried over. + assert recovered.value_sha256 is None + + def test_legacy_record_with_missing_file_names_the_record(self, tmp_path): + """A legacy record pointing at a file this machine lacks fails with the record identified.""" + seed = SeedSimulatedConversation( + num_turns=2, + adversarial_chat_system_prompt=SeedPrompt(value="placeholder"), + name="stale-technique", + dataset_name="legacy-dataset", + ) + entry = SeedEntry(entry=seed) + entry.value = json.dumps( + { + "num_turns": 2, + "sequence": 0, + "adversarial_chat_system_prompt_path": str(tmp_path / "gone.yaml"), + "pyrit_version": "1.0.0", + }, + sort_keys=True, + separators=(",", ":"), + ) + + with pytest.raises(ValueError, match="stale-technique"): + entry.get_seed() + def test_roundtrip_seed_simulated_conversation_strips_reserved_key(self): """SeedSimulatedConversation also has no schema field; reserved key must still be stripped.""" from pyrit.models import SEED_RESPONSE_JSON_SCHEMA_METADATA_KEY config = SeedSimulatedConversation( num_turns=3, - adversarial_chat_system_prompt_path="/path/to/adversarial.yaml", - simulated_target_system_prompt_path="/path/to/target.yaml", + adversarial_chat_system_prompt=SeedPrompt(value="adversarial", parameters=["objective"]), + simulated_target_system_prompt=SeedPrompt(value="target", parameters=["objective", "num_turns"]), metadata={ SEED_RESPONSE_JSON_SCHEMA_METADATA_KEY: "sneaky", "owned": "by-caller", diff --git a/tests/unit/models/test_attack_technique_seed_group.py b/tests/unit/models/test_attack_technique_seed_group.py index 3e312a1130..e87d89a0cb 100644 --- a/tests/unit/models/test_attack_technique_seed_group.py +++ b/tests/unit/models/test_attack_technique_seed_group.py @@ -42,24 +42,18 @@ def test_seed_objective_raises_if_set_true(self): with pytest.raises(ValueError, match="SeedObjective cannot be a general technique"): SeedObjective(value="Test objective", is_general_technique=True) - def test_seed_simulated_conversation_defaults_to_true(self, tmp_path): + def test_seed_simulated_conversation_defaults_to_true(self): """Test that SeedSimulatedConversation.is_general_technique defaults to True.""" - adv_path = tmp_path / "adversarial.yaml" - adv_path.write_text("value: Adversarial\ndata_type: text") - sim = SeedSimulatedConversation( - adversarial_chat_system_prompt_path=adv_path, + adversarial_chat_system_prompt=SeedPrompt(value="Adversarial", parameters=["objective"]), num_turns=2, ) assert sim.is_general_technique is True - def test_seed_simulated_conversation_can_be_set_false(self, tmp_path): + def test_seed_simulated_conversation_can_be_set_false(self): """Test that SeedSimulatedConversation.is_general_technique can be overridden to False.""" - adv_path = tmp_path / "adversarial.yaml" - adv_path.write_text("value: Adversarial\ndata_type: text") - sim = SeedSimulatedConversation( - adversarial_chat_system_prompt_path=adv_path, + adversarial_chat_system_prompt=SeedPrompt(value="Adversarial", parameters=["objective"]), num_turns=2, is_general_technique=False, ) @@ -113,16 +107,13 @@ def test_init_raises_with_objective(self): ] ) - def test_init_with_simulated_conversation(self, tmp_path): + def test_init_with_simulated_conversation(self): """Test initialization with SeedSimulatedConversation (defaults to general technique).""" - adv_path = tmp_path / "adversarial.yaml" - adv_path.write_text("value: Adversarial\ndata_type: text") - group = AttackTechniqueSeedGroup( seeds=[ SeedSimulatedConversation( num_turns=3, - adversarial_chat_system_prompt_path=adv_path, + adversarial_chat_system_prompt=SeedPrompt(value="Adversarial", parameters=["objective"]), ), SeedPrompt( value="Technique prompt", data_type="text", sequence=10, role="user", is_general_technique=True diff --git a/tests/unit/models/test_seed.py b/tests/unit/models/test_seed.py index 2d8c48f9c7..1dca202d06 100644 --- a/tests/unit/models/test_seed.py +++ b/tests/unit/models/test_seed.py @@ -1069,8 +1069,8 @@ def test_seed_dataset_dict_to_seed_simulated_conversation_all_base_params(tmp_pa # Verify SeedSimulatedConversation-specific fields assert seed.num_turns == 5 - assert seed.adversarial_chat_system_prompt_path == pathlib.Path(adv_path) - assert seed.simulated_target_system_prompt_path == pathlib.Path(sim_path) + assert seed.adversarial_chat_system_prompt.value == "You are adversarial" + assert seed.simulated_target_system_prompt.parameters == ["objective", "num_turns"] def test_seed_dataset_uses_dataset_defaults_for_missing_params(): diff --git a/tests/unit/models/test_seed_group.py b/tests/unit/models/test_seed_group.py index bcb0c754fb..9876cbe4ed 100644 --- a/tests/unit/models/test_seed_group.py +++ b/tests/unit/models/test_seed_group.py @@ -679,7 +679,7 @@ def test_raises_when_technique_has_simulated_conversation_and_prompts_overlap(se technique = AttackTechniqueSeedGroup( seeds=[ SeedSimulatedConversation( - adversarial_chat_system_prompt_path="fake_path.yaml", + adversarial_chat_system_prompt=SeedPrompt(value="adversarial", parameters=["objective"]), num_turns=3, ), ], @@ -694,7 +694,7 @@ def test_succeeds_when_technique_has_simulated_conversation_and_no_prompts(self) technique = AttackTechniqueSeedGroup( seeds=[ SeedSimulatedConversation( - adversarial_chat_system_prompt_path="fake_path.yaml", + adversarial_chat_system_prompt=SeedPrompt(value="adversarial", parameters=["objective"]), num_turns=3, ), ], @@ -716,7 +716,7 @@ def test_is_compatible_returns_false_when_prompts_overlap_simulated_range(self): technique = AttackTechniqueSeedGroup( seeds=[ SeedSimulatedConversation( - adversarial_chat_system_prompt_path="fake_path.yaml", + adversarial_chat_system_prompt=SeedPrompt(value="adversarial", parameters=["objective"]), num_turns=3, ), ], @@ -730,7 +730,7 @@ def test_is_compatible_returns_true_for_objective_only_with_simulated(self): technique = AttackTechniqueSeedGroup( seeds=[ SeedSimulatedConversation( - adversarial_chat_system_prompt_path="fake_path.yaml", + adversarial_chat_system_prompt=SeedPrompt(value="adversarial", parameters=["objective"]), num_turns=3, ), ], @@ -777,7 +777,7 @@ def test_filters_out_incompatible_groups(self): technique = AttackTechniqueSeedGroup( seeds=[ SeedSimulatedConversation( - adversarial_chat_system_prompt_path="fake.yaml", + adversarial_chat_system_prompt=SeedPrompt(value="adversarial", parameters=["objective"]), num_turns=3, ), ], diff --git a/tests/unit/models/test_seed_simulated_conversation.py b/tests/unit/models/test_seed_simulated_conversation.py index c8239a9caa..4f71d38e40 100644 --- a/tests/unit/models/test_seed_simulated_conversation.py +++ b/tests/unit/models/test_seed_simulated_conversation.py @@ -9,8 +9,8 @@ import pytest from pyrit.models.seeds import ( + SeedPrompt, SeedSimulatedConversation, - SimulatedTargetSystemPromptPaths, ) @@ -22,7 +22,7 @@ def test_init_with_all_parameters(self, tmp_path): adv_path = tmp_path / "adversarial.yaml" adv_path.write_text("value: test\ndata_type: text") sim_path = tmp_path / "simulated.yaml" - sim_path.write_text("value: test\ndata_type: text") + sim_path.write_text("value: test\ndata_type: text\nparameters:\n - objective\n - num_turns") conv = SeedSimulatedConversation( adversarial_chat_system_prompt_path=adv_path, @@ -31,8 +31,8 @@ def test_init_with_all_parameters(self, tmp_path): ) assert conv.num_turns == 5 - assert conv.adversarial_chat_system_prompt_path == adv_path - assert conv.simulated_target_system_prompt_path == sim_path + assert conv.adversarial_chat_system_prompt.value == "test" + assert conv.simulated_target_system_prompt.value == "test" assert conv.data_type == "text" assert isinstance(conv.id, uuid.UUID) @@ -46,9 +46,9 @@ def test_init_with_minimal_parameters(self, tmp_path): ) assert conv.num_turns == 3 # default - assert conv.adversarial_chat_system_prompt_path == adv_path - # Default simulated_target_system_prompt_path is the compliant prompt - assert conv.simulated_target_system_prompt_path == SimulatedTargetSystemPromptPaths.COMPLIANT.value + assert conv.adversarial_chat_system_prompt.value == "test" + # The simulated target defaults to the compliant prompt + assert conv.simulated_target_system_prompt.name == "simulated_target_compliant" def test_init_default_num_turns(self, tmp_path): """Test that default num_turns is 3.""" @@ -106,7 +106,7 @@ def test_init_generates_json_value(self, tmp_path): value = json.loads(conv.value) assert value["num_turns"] == 5 - assert "adversarial_chat_system_prompt_path" in value + assert value["adversarial_chat_system_prompt"]["value"] == "test" assert "pyrit_version" in value def test_init_value_is_deterministic(self, tmp_path): @@ -148,8 +148,8 @@ def test_init_custom_sequence(self, tmp_path): assert conv.sequence == 5 - def test_init_default_next_message_system_prompt_path_is_none(self, tmp_path): - """Test that default next_message_system_prompt_path is None.""" + def test_init_default_next_message_system_prompt_is_none(self, tmp_path): + """Test that the next message system prompt defaults to None.""" adv_path = tmp_path / "adversarial.yaml" adv_path.write_text("value: test\ndata_type: text") @@ -157,10 +157,10 @@ def test_init_default_next_message_system_prompt_path_is_none(self, tmp_path): adversarial_chat_system_prompt_path=adv_path, ) - assert conv.next_message_system_prompt_path is None + assert conv.next_message_system_prompt is None - def test_init_next_message_system_prompt_path_set(self, tmp_path): - """Test that next_message_system_prompt_path can be set.""" + def test_init_next_message_system_prompt_set(self, tmp_path): + """Test that the next message system prompt can be set.""" adv_path = tmp_path / "adversarial.yaml" adv_path.write_text("value: test\ndata_type: text") next_msg_path = tmp_path / "next_message.yaml" @@ -171,7 +171,7 @@ def test_init_next_message_system_prompt_path_set(self, tmp_path): next_message_system_prompt_path=next_msg_path, ) - assert conv.next_message_system_prompt_path == next_msg_path + assert conv.next_message_system_prompt.value == "test" class TestSeedSimulatedConversationFromMapping: @@ -189,7 +189,7 @@ def test_from_dict_with_paths(self, tmp_path): conv = SeedSimulatedConversation.model_validate(data) assert conv.num_turns == 5 - assert conv.adversarial_chat_system_prompt_path == adv_path + assert conv.adversarial_chat_system_prompt.value == "test" def test_from_dict_without_simulated_target_path(self, tmp_path): """Test construction without simulated_target_system_prompt_path uses compliant default.""" @@ -202,8 +202,8 @@ def test_from_dict_without_simulated_target_path(self, tmp_path): } conv = SeedSimulatedConversation.model_validate(data) - # Default simulated_target_system_prompt_path is the compliant prompt - assert conv.simulated_target_system_prompt_path == SimulatedTargetSystemPromptPaths.COMPLIANT.value + # The simulated target defaults to the compliant prompt + assert conv.simulated_target_system_prompt.name == "simulated_target_compliant" def test_from_dict_default_num_turns(self, tmp_path): """Test that num_turns defaults to 3 when not specified.""" @@ -221,7 +221,7 @@ def test_from_dict_missing_adversarial_path_raises_error(self): """Test that construction raises when adversarial path is missing (required field).""" data = {"num_turns": 3} - with pytest.raises(ValueError, match="adversarial_chat_system_prompt_path"): + with pytest.raises(ValueError, match="adversarial_chat_system_prompt"): SeedSimulatedConversation.model_validate(data) @@ -241,7 +241,7 @@ def test_get_identifier_returns_correct_structure(self, tmp_path): assert identifier["__type__"] == "SeedSimulatedConversation" assert identifier["num_turns"] == 3 - assert "adversarial_chat_system_prompt_path" in identifier + assert identifier["adversarial_chat_system_prompt"]["value"] == "test" assert "pyrit_version" in identifier @@ -298,10 +298,10 @@ def test_compute_hash_differs_for_different_num_turns(self, tmp_path): class TestSeedSimulatedConversationRepr: """Tests for SeedSimulatedConversation.__repr__ method.""" - def test_repr_shows_num_turns_and_path(self, tmp_path): + def test_repr_shows_num_turns_and_prompt_name(self, tmp_path): """Test __repr__ shows key information.""" adv_path = tmp_path / "adversarial.yaml" - adv_path.write_text("value: test\ndata_type: text") + adv_path.write_text("name: my_adversarial\nvalue: test\ndata_type: text") conv = SeedSimulatedConversation( adversarial_chat_system_prompt_path=adv_path, @@ -311,7 +311,19 @@ def test_repr_shows_num_turns_and_path(self, tmp_path): assert "SeedSimulatedConversation" in repr_str assert "num_turns=5" in repr_str - assert "adversarial.yaml" in repr_str + assert "my_adversarial" in repr_str + + def test_repr_omits_prompt_name_when_unnamed(self): + """An unnamed adversarial prompt drops the fragment rather than printing a placeholder.""" + conv = SeedSimulatedConversation( + adversarial_chat_system_prompt=SeedPrompt(value="test", data_type="text"), + num_turns=5, + ) + repr_str = repr(conv) + + assert "num_turns=5" in repr_str + assert "adversarial_prompt" not in repr_str + assert "None" not in repr_str class TestSeedSimulatedConversationLoadSimulatedTargetSystemPrompt: @@ -348,3 +360,204 @@ def test_load_simulated_target_system_prompt_raises_for_missing_params(self, tmp num_turns=3, simulated_target_system_prompt_path=sim_path, ) + + +class TestSeedSimulatedConversationCanonicalPrompts: + """Tests for the canonical SeedPrompt fields and the deprecated path inputs.""" + + def test_init_with_canonical_prompts(self): + """Canonical SeedPrompt inputs populate the fields without touching disk.""" + conv = SeedSimulatedConversation( + adversarial_chat_system_prompt=SeedPrompt(value="adversarial", parameters=["objective"]), + simulated_target_system_prompt=SeedPrompt(value="target", parameters=["objective", "num_turns"]), + next_message_system_prompt=SeedPrompt(value="next", parameters=["objective", "conversation_context"]), + num_turns=2, + ) + + assert conv.adversarial_chat_system_prompt.value == "adversarial" + assert conv.simulated_target_system_prompt.value == "target" + assert conv.next_message_system_prompt is not None + assert conv.next_message_system_prompt.value == "next" + + def test_path_inputs_are_not_model_fields(self): + """The deprecated path inputs never become fields, so they cannot reach persistence.""" + for field_name in ( + "adversarial_chat_system_prompt_path", + "simulated_target_system_prompt_path", + "next_message_system_prompt_path", + ): + assert field_name not in SeedSimulatedConversation.model_fields + + def test_deprecated_path_input_warns_and_resolves(self, tmp_path): + """A deprecated path input warns and is resolved into the canonical prompt.""" + adv_path = tmp_path / "adversarial.yaml" + adv_path.write_text("value: adversarial\ndata_type: text") + + with pytest.warns(DeprecationWarning, match="adversarial_chat_system_prompt_path"): + conv = SeedSimulatedConversation(adversarial_chat_system_prompt_path=adv_path) + + assert conv.adversarial_chat_system_prompt.value == "adversarial" + assert not hasattr(conv, "adversarial_chat_system_prompt_path") + + def test_prompt_and_path_together_raises(self, tmp_path): + """Supplying both a canonical prompt and its deprecated path is ambiguous.""" + adv_path = tmp_path / "adversarial.yaml" + adv_path.write_text("value: adversarial\ndata_type: text") + + with pytest.raises( + ValueError, match="Set only one of SeedSimulatedConversation.adversarial_chat_system_prompt" + ): + SeedSimulatedConversation( + adversarial_chat_system_prompt=SeedPrompt(value="adversarial"), + adversarial_chat_system_prompt_path=adv_path, + ) + + def test_missing_deprecated_path_raises(self, tmp_path): + """A deprecated path that does not exist fails loudly rather than silently.""" + with pytest.raises(FileNotFoundError): + SeedSimulatedConversation(adversarial_chat_system_prompt_path=tmp_path / "missing.yaml") + + def test_deprecated_simulated_target_path_requires_parameters(self, tmp_path): + """Loading a simulated target from a path keeps the declared-parameter contract.""" + adv_path = tmp_path / "adversarial.yaml" + adv_path.write_text("value: adversarial\ndata_type: text") + sim_path = tmp_path / "simulated.yaml" + sim_path.write_text("value: no params\ndata_type: text") + + with pytest.raises(ValueError, match="objective and num_turns"): + SeedSimulatedConversation( + adversarial_chat_system_prompt_path=adv_path, + simulated_target_system_prompt_path=sim_path, + ) + + def test_explicit_none_simulated_target_falls_back_to_compliant(self): + """An explicit None simulated target (as memory reconstruction sends) uses the default.""" + conv = SeedSimulatedConversation( + adversarial_chat_system_prompt=SeedPrompt(value="adversarial"), + simulated_target_system_prompt=None, + ) + + assert conv.simulated_target_system_prompt.name == "simulated_target_compliant" + + +class TestSeedSimulatedConversationIdentity: + """Tests for the content-based value and hash.""" + + def test_value_carries_prompt_text_not_paths(self): + """The serialized value holds the prompt text so a technique is inspectable.""" + conv = SeedSimulatedConversation( + adversarial_chat_system_prompt=SeedPrompt(value="adversarial text", parameters=["objective"]) + ) + value = json.loads(conv.value) + + assert value["adversarial_chat_system_prompt"]["value"] == "adversarial text" + assert value["adversarial_chat_system_prompt"]["parameters"] == ["objective"] + assert not any(key.endswith("_path") for key in value) + + def test_editing_one_word_changes_identity(self): + """Changing a word in a system prompt produces a different configuration identity.""" + original = SeedSimulatedConversation(adversarial_chat_system_prompt=SeedPrompt(value="be a screenwriter")) + edited = SeedSimulatedConversation(adversarial_chat_system_prompt=SeedPrompt(value="be a novelist")) + + assert original.value != edited.value + assert original.compute_hash() != edited.compute_hash() + + def test_identical_content_from_different_files_matches(self, tmp_path): + """Two copies of the same prompt text share an identity even from different files.""" + first = tmp_path / "first.yaml" + first.write_text("value: same text\ndata_type: text") + second = tmp_path / "second.yaml" + second.write_text("value: same text\ndata_type: text") + + conv1 = SeedSimulatedConversation(adversarial_chat_system_prompt_path=first) + conv2 = SeedSimulatedConversation(adversarial_chat_system_prompt_path=second) + + assert conv1.value == conv2.value + + def test_response_json_schema_changes_identity(self): + """Two prompts with identical text but different schemas are not interchangeable.""" + plain = SeedSimulatedConversation( + adversarial_chat_system_prompt=SeedPrompt(value="adversarial"), + next_message_system_prompt=SeedPrompt(value="next"), + ) + with_schema = SeedSimulatedConversation( + adversarial_chat_system_prompt=SeedPrompt(value="adversarial"), + next_message_system_prompt=SeedPrompt(value="next", response_json_schema_name="adversarial_chat"), + ) + + assert plain.value != with_schema.value + + @pytest.mark.parametrize("mode", ["python", "json"]) + def test_value_is_stable_across_round_trip(self, mode): + """Dumping and revalidating recomputes the same value.""" + conv = SeedSimulatedConversation( + adversarial_chat_system_prompt=SeedPrompt(value="adversarial", parameters=["objective"]), + next_message_system_prompt=SeedPrompt(value="next", response_json_schema_name="adversarial_chat"), + num_turns=2, + ) + + assert SeedSimulatedConversation.model_validate(conv.model_dump(mode=mode)).value == conv.value + + def test_sequence_range_accounts_for_next_message(self): + """The next message adds one sequence slot after the generated turns.""" + without = SeedSimulatedConversation(adversarial_chat_system_prompt=SeedPrompt(value="adversarial"), num_turns=2) + with_next = SeedSimulatedConversation( + adversarial_chat_system_prompt=SeedPrompt(value="adversarial"), + next_message_system_prompt=SeedPrompt(value="next"), + num_turns=2, + ) + + assert list(without.sequence_range) == [0, 1, 2, 3] + assert list(with_next.sequence_range) == [0, 1, 2, 3, 4] + + +class TestSeedSimulatedConversationTemplatePreparation: + """Tests that a prompt is prepared exactly once, so deferred template syntax survives.""" + + DEFERRED_TEMPLATE = "{% raw %}{% if num_turns == 1 %}one turn{% else %}many turns{% endif %}{% endraw %}" + + @pytest.fixture + def deferred_prompt_path(self, tmp_path): + path = tmp_path / "deferred.yaml" + path.write_text( + f"data_type: text\nparameters:\n - objective\n - num_turns\nvalue: '{self.DEFERRED_TEMPLATE}'\n" + ) + return path + + def test_canonical_prompt_is_not_prepared_again(self, deferred_prompt_path): + """A prompt handed in already prepared keeps its deferred template syntax.""" + prompt = SeedPrompt.from_yaml_file(deferred_prompt_path) + original = prompt.value + + conv = SeedSimulatedConversation( + adversarial_chat_system_prompt=SeedPrompt(value="adversarial"), + simulated_target_system_prompt=prompt, + num_turns=1, + ) + + assert prompt.value == original, "the caller's prompt must not be mutated" + assert conv.simulated_target_system_prompt.render_template_value(objective="o", num_turns=1) == "one turn" + + def test_deprecated_path_prompt_is_not_prepared_again(self, deferred_prompt_path): + """Loading through the deprecated path input also prepares the template only once.""" + with pytest.warns(DeprecationWarning): + conv = SeedSimulatedConversation( + adversarial_chat_system_prompt=SeedPrompt(value="adversarial"), + simulated_target_system_prompt_path=deferred_prompt_path, + num_turns=1, + ) + + assert conv.simulated_target_system_prompt.render_template_value(objective="o", num_turns=1) == "one turn" + + def test_prompt_rebuilt_from_value_is_not_prepared_again(self, deferred_prompt_path): + """A prompt rebuilt from the serialized configuration keeps its deferred template syntax.""" + with pytest.warns(DeprecationWarning): + conv = SeedSimulatedConversation( + adversarial_chat_system_prompt=SeedPrompt(value="adversarial"), + simulated_target_system_prompt_path=deferred_prompt_path, + num_turns=1, + ) + + rebuilt = SeedPrompt(**json.loads(conv.value)["simulated_target_system_prompt"]) + + assert rebuilt.render_template_value(objective="o", num_turns=1) == "one turn" diff --git a/tests/unit/scenario/core/test_attack_technique_factory.py b/tests/unit/scenario/core/test_attack_technique_factory.py index c5324d8d39..7812a7992b 100644 --- a/tests/unit/scenario/core/test_attack_technique_factory.py +++ b/tests/unit/scenario/core/test_attack_technique_factory.py @@ -4,6 +4,7 @@ """Tests for the AttackTechniqueFactory class.""" import typing +import warnings from unittest.mock import MagicMock, patch import pytest @@ -957,7 +958,7 @@ def test_simulated_conversation_resolves_default_lazily(self): factory = AttackTechniqueFactory.with_simulated_conversation( name="role_play_movie_script", - adversarial_chat_system_prompt_path=( + adversarial_chat_system_prompt=SeedPrompt.from_yaml_file( EXECUTOR_SEED_PROMPT_PATH / "red_teaming" / "role_play" / "role_play_movie_script.yaml" ), num_turns=2, @@ -1034,3 +1035,63 @@ def get_identifier(self): ) assert factory._get_scoring_config_type() is None + + +@pytest.mark.usefixtures("patch_central_database") +class TestWithSimulatedConversationPromptSources: + """Tests for the canonical prompt inputs on ``with_simulated_conversation``.""" + + def test_defaults_resolve_to_prompts_without_warning(self): + """The name-derived adversarial prompt and the default next message load silently.""" + with warnings.catch_warnings(): + warnings.simplefilter("error", DeprecationWarning) + factory = AttackTechniqueFactory.with_simulated_conversation(name="crescendo_simulated") + + sim = factory.seed_technique.simulated_conversation_config + assert sim is not None + assert sim.adversarial_chat_system_prompt.name == "crescendo_simulated" + assert sim.simulated_target_system_prompt.name == "simulated_target_compliant" + assert sim.next_message_system_prompt is not None + assert sim.next_message_system_prompt.name == "direct_next_message_generator" + + def test_canonical_prompt_is_used(self): + """An explicit prompt is carried straight through to the seed.""" + prompt = SeedPrompt(value="custom adversarial", parameters=["objective"]) + factory = AttackTechniqueFactory.with_simulated_conversation( + name="crescendo_simulated", + adversarial_chat_system_prompt=prompt, + ) + + sim = factory.seed_technique.simulated_conversation_config + assert sim is not None + assert sim.adversarial_chat_system_prompt.value == "custom adversarial" + + def test_deprecated_path_input_warns(self, tmp_path): + """An explicit path input still works and warns.""" + adv_path = tmp_path / "adversarial.yaml" + adv_path.write_text("value: from path\ndata_type: text") + + with pytest.warns(DeprecationWarning, match="adversarial_chat_system_prompt_path"): + factory = AttackTechniqueFactory.with_simulated_conversation( + name="crescendo_simulated", + adversarial_chat_system_prompt_path=adv_path, + ) + + sim = factory.seed_technique.simulated_conversation_config + assert sim is not None + assert sim.adversarial_chat_system_prompt.value == "from path" + + def test_final_user_message_disables_next_message_prompt(self): + """A fixed final message replaces the generated next message.""" + factory = AttackTechniqueFactory.with_simulated_conversation( + name="crescendo_simulated", + final_user_message="yes.", + num_turns=1, + ) + + sim = factory.seed_technique.simulated_conversation_config + assert sim is not None + assert sim.next_message_system_prompt is None + prompts = list(factory.seed_technique.prompts) + assert prompts[0].value == "yes." + assert prompts[0].sequence == sim.sequence_range.stop diff --git a/tests/unit/scenario/test_default_run_size_estimates.py b/tests/unit/scenario/test_default_run_size_estimates.py index f07e27dd2e..c3e6f89f5f 100644 --- a/tests/unit/scenario/test_default_run_size_estimates.py +++ b/tests/unit/scenario/test_default_run_size_estimates.py @@ -292,7 +292,7 @@ async def test_matrix_estimate_filters_each_technique_seed_population_like_execu conversation_factory.seed_technique = AttackTechniqueSeedGroup( seeds=[ SeedSimulatedConversation( - adversarial_chat_system_prompt_path="fake.yaml", + adversarial_chat_system_prompt=SeedPrompt(value="adversarial", parameters=["objective"]), num_turns=3, ) ] @@ -409,7 +409,7 @@ async def resolve_groups() -> tuple[dict[str, list[AttackSeedGroup]], list[Scena conversation_factory.seed_technique = AttackTechniqueSeedGroup( seeds=[ SeedSimulatedConversation( - adversarial_chat_system_prompt_path="fake.yaml", + adversarial_chat_system_prompt=SeedPrompt(value="adversarial", parameters=["objective"]), num_turns=3, ) ] @@ -811,7 +811,7 @@ async def test_adversarial_benchmark_resolves_targets_and_filters_each_technique conversation_factory.seed_technique = AttackTechniqueSeedGroup( seeds=[ SeedSimulatedConversation( - adversarial_chat_system_prompt_path="fake.yaml", + adversarial_chat_system_prompt=SeedPrompt(value="adversarial", parameters=["objective"]), num_turns=3, ) ] diff --git a/tests/unit/setup/test_technique_initializer.py b/tests/unit/setup/test_technique_initializer.py index 66f791a12f..00770b58a0 100644 --- a/tests/unit/setup/test_technique_initializer.py +++ b/tests/unit/setup/test_technique_initializer.py @@ -260,11 +260,12 @@ def test_seed_technique_num_turns_matches_canonical_default(self): assert sim is not None assert sim.num_turns == 3 - def test_seed_technique_yaml_path_resolves_to_existing_file(self): + def test_seed_technique_carries_resolved_adversarial_prompt(self): for f in self._persona_factories(): sim = f.seed_technique.simulated_conversation_config assert sim is not None - assert sim.adversarial_chat_system_prompt_path.exists() + assert sim.adversarial_chat_system_prompt.value + assert sim.adversarial_chat_system_prompt.parameters == ["objective", "max_turns"] class TestPersonaCrescendoYamls: @@ -327,10 +328,10 @@ def test_seed_technique_uses_custom_simulated_target_prompt(self): factory = self._context_compliance_factory() sim = factory.seed_technique.simulated_conversation_config assert sim is not None - assert sim.simulated_target_system_prompt_path.name == "context_compliance_target.yaml" - assert sim.simulated_target_system_prompt_path.exists() + assert sim.simulated_target_system_prompt.name == "simulated_target_context_compliance" + assert sim.simulated_target_system_prompt.parameters == ["objective", "num_turns"] # No LLM-generated next message: the final turn is a fixed affirmation instead. - assert sim.next_message_system_prompt_path is None + assert sim.next_message_system_prompt is None def test_final_user_message_is_fixed_affirmation(self): factory = self._context_compliance_factory() @@ -343,12 +344,12 @@ def test_final_user_message_is_fixed_affirmation(self): assert yes_prompt.role == "user" assert yes_prompt.sequence == 2 - def test_adversarial_yaml_resolves_to_existing_file(self): + def test_adversarial_prompt_is_resolved(self): factory = self._context_compliance_factory() sim = factory.seed_technique.simulated_conversation_config assert sim is not None - assert sim.adversarial_chat_system_prompt_path.name == "context_compliance.yaml" - assert sim.adversarial_chat_system_prompt_path.exists() + assert sim.adversarial_chat_system_prompt.value + assert sim.adversarial_chat_system_prompt.parameters == ["objective", "max_turns"] def test_tagged_core_single_turn_light(self): factory = self._context_compliance_factory() @@ -417,19 +418,20 @@ def test_seed_technique_num_turns_matches_role_play_default(self): assert sim is not None assert sim.num_turns == 2 - def test_seed_technique_yaml_path_resolves_to_existing_file(self): + def test_seed_technique_carries_resolved_adversarial_prompt(self): for f in self._role_play_factories(): sim = f.seed_technique.simulated_conversation_config assert sim is not None - assert sim.adversarial_chat_system_prompt_path.exists() + assert sim.adversarial_chat_system_prompt.value + assert sim.adversarial_chat_system_prompt.parameters == ["objective", "max_turns"] def test_all_use_role_play_next_message_prompt(self): for f in self._role_play_factories(): sim = f.seed_technique.simulated_conversation_config assert sim is not None - assert sim.next_message_system_prompt_path is not None - assert sim.next_message_system_prompt_path.name == "role_play_next_message.yaml" - assert sim.next_message_system_prompt_path.exists() + assert sim.next_message_system_prompt is not None + assert sim.next_message_system_prompt.name == "role_play_next_message_generator" + assert sim.next_message_system_prompt.parameters == ["objective", "conversation_context"] class TestRolePlayYamls: