FEAT: Adding Scenario run to the REST API

.pyrit_conf_example

@@ -111,6 +111,12 @@ operation: op_trash_panda
 #   - /path/to/.env
 #   - /path/to/.env.local
 
+# Max Concurrent Scenario Runs
+# ----------------------------
+# Maximum number of scenario runs that can execute concurrently in the backend.
+# Applies only to the pyrit_backend server.
+max_concurrent_scenario_runs: 3
+
 # Silent Mode
 # -----------
 # If true, suppresses print statements during initialization.

pyrit/backend/models/__init__.py

@@ -48,8 +48,8 @@
     PreviewStep,
 )
 from pyrit.backend.models.scenarios import (
-    ScenarioListResponse,
-    ScenarioSummary,
+    ListRegisteredScenarioResponse,
+    RegisteredScenario,
 )
 from pyrit.backend.models.targets import (
     CreateTargetRequest,
@@ -96,8 +96,8 @@
     "CreateConverterResponse",
     "PreviewStep",
     # Scenarios
-    "ScenarioListResponse",
-    "ScenarioSummary",
+    "ListRegisteredScenarioResponse",
+    "RegisteredScenario",
     # Targets
     "CreateTargetRequest",
     "TargetInstance",

pyrit/backend/models/attacks.py

@@ -95,18 +95,24 @@ class AttackSummary(BaseModel):
 
     attack_result_id: str = Field(..., description="Database-assigned unique ID for this AttackResult")
     conversation_id: str = Field(..., description="Primary conversation of this attack result")
-    attack_type: str = Field(..., description="Attack class name (e.g., 'CrescendoAttack', 'ManualAttack')")
+    attack_type: str = Field("", description="Attack class name (e.g., 'CrescendoAttack', 'ManualAttack')")
     attack_specific_params: Optional[dict[str, Any]] = Field(None, description="Additional attack-specific parameters")
     target: Optional[TargetInfo] = Field(None, description="Target information from the stored identifier")
     converters: list[str] = Field(
         default_factory=list, description="Request converter class names applied in this attack"
     )
+    objective: str = Field("", description="Natural-language description of the attacker's objective")
     outcome: Optional[Literal["undetermined", "success", "failure"]] = Field(
         None, description="Attack outcome (null if not yet determined)"
     )
+    outcome_reason: str | None = Field(None, description="Reason for the outcome")
+    last_response: str | None = Field(None, description="Model response from the final turn")
     last_message_preview: Optional[str] = Field(
         None, description="Preview of the last message (truncated to ~100 chars)"
     )
+    score_value: str | None = Field(None, description="Score value from the objective scorer")
+    executed_turns: int = Field(0, ge=0, description="Number of turns executed")
+    execution_time_ms: int = Field(0, ge=0, description="Execution time in milliseconds")
     message_count: int = Field(0, description="Total number of messages in the attack")
     related_conversation_ids: list[str] = Field(
         default_factory=list, description="IDs of related conversations within this attack"

pyrit/backend/models/scenarios.py

@@ -5,17 +5,20 @@
 Scenario API response models.
 
 Scenarios are multi-attack security testing campaigns. These models represent
-the metadata about available scenarios (listing), not scenario execution results.
+the metadata about available scenarios (listing) and scenario execution (runs).
 """
 
-from typing import Optional
+from datetime import datetime
+from enum import Enum
+from typing import Any, Optional
 
 from pydantic import BaseModel, Field
 
+from pyrit.backend.models.attacks import AttackSummary
 from pyrit.backend.models.common import PaginationInfo
 
 
-class ScenarioSummary(BaseModel):
+class RegisteredScenario(BaseModel):
     """Summary of a registered scenario."""
 
     scenario_name: str = Field(..., description="Registry key (e.g., 'foundry.red_team_agent')")
@@ -30,8 +33,103 @@ class ScenarioSummary(BaseModel):
     max_dataset_size: Optional[int] = Field(None, description="Maximum items per dataset (None means unlimited)")
 
 
-class ScenarioListResponse(BaseModel):
+class ListRegisteredScenarioResponse(BaseModel):
     """Response for listing scenarios."""
 
-    items: list[ScenarioSummary] = Field(..., description="List of scenario summaries")
+    items: list[RegisteredScenario] = Field(..., description="List of scenario summaries")
     pagination: PaginationInfo = Field(..., description="Pagination metadata")
+
+
+# ============================================================================
+# Scenario Run Models
+# ============================================================================
+
+
+class ScenarioRunStatus(str, Enum):
+    """Status of a scenario run."""
+
+    PENDING = "pending"
+    INITIALIZING = "initializing"
+    RUNNING = "running"
+    COMPLETED = "completed"
+    FAILED = "failed"
+    CANCELLED = "cancelled"
+
+
+class RunScenarioRequest(BaseModel):
+    """Request body for starting a scenario run."""
+
+    scenario_name: str = Field(..., description="Registry key of the scenario to run")
+    target_name: str = Field(..., description="Name of a registered target from the TargetRegistry")
+    initializers: list[str] | None = Field(
+        None, description="Initializer names to run before scenario (e.g., ['target', 'load_default_datasets'])"
+    )
+    strategies: list[str] | None = Field(None, description="Strategy names to use (uses scenario default if omitted)")
+    dataset_names: list[str] | None = Field(None, description="Dataset names to use (uses scenario default if omitted)")
+    max_dataset_size: int | None = Field(None, ge=1, description="Maximum items per dataset")
+    max_concurrency: int = Field(10, ge=1, le=100, description="Maximum concurrent operations")
+    max_retries: int = Field(0, ge=0, le=20, description="Maximum retry attempts on failure")
+    labels: dict[str, str] | None = Field(None, description="Labels to attach to memory entries")
+    scenario_params: dict[str, Any] | None = Field(
+        None,
+        description="Custom parameters for the scenario (passed to scenario.set_params_from_args). "
+        "Keys are parameter names declared by the scenario's supported_parameters().",
+    )
+    initializer_args: dict[str, dict[str, Any]] | None = Field(
+        None,
+        description="Per-initializer arguments keyed by initializer name. "
+        "Each value is a dict of args passed to that initializer's set_params_from_args(). "
+        "Example: {'target': {'endpoint': 'https://...'}}.",
+    )
+    scenario_result_id: str | None = Field(
+        None,
+        description="Optional ID of an existing ScenarioResult to resume. "
+        "If provided, the scenario will resume from prior progress instead of starting fresh.",
+    )
+
+
+class ScenarioRunSummary(BaseModel):
+    """Response for a scenario run (status + result details)."""
+
+    scenario_result_id: str = Field(..., description="UUID of the ScenarioResult in memory")
+    scenario_name: str = Field(..., description="Registry key of the scenario being run")
+    scenario_version: int = Field(0, ge=0, description="Version of the scenario")
+    status: ScenarioRunStatus = Field(..., description="Current run status")
+    created_at: datetime = Field(..., description="When the run was created")
+    updated_at: datetime = Field(..., description="When the run status last changed")
+    error: str | None = Field(None, description="Error message if status is FAILED")
+    strategies_used: list[str] = Field(default_factory=list, description="Strategy names that were executed")
+    total_attacks: int = Field(0, ge=0, description="Total number of atomic attacks")
+    completed_attacks: int = Field(0, ge=0, description="Number of attacks that completed")
+    objective_achieved_rate: int = Field(0, ge=0, le=100, description="Success rate as percentage (0-100)")
+    labels: dict[str, str] = Field(default_factory=dict, description="Labels attached to this run")
+    completed_at: datetime | None = Field(None, description="When the scenario finished")
+
+
+class ScenarioRunListResponse(BaseModel):
+    """Response for listing scenario runs."""
+
+    items: list[ScenarioRunSummary] = Field(..., description="List of scenario runs")
+
+
+# ============================================================================
+# Scenario Results Detail Models
+# ============================================================================
+
+
+class AtomicAttackResults(BaseModel):
+    """Results grouped by atomic attack name."""
+
+    atomic_attack_name: str = Field(..., description="Name of the atomic attack (strategy)")
+    display_group: str | None = Field(None, description="Display group label for UI grouping")
+    results: list[AttackSummary] = Field(..., description="Individual attack results")
+    success_count: int = Field(0, ge=0, description="Number of successful attacks")
+    failure_count: int = Field(0, ge=0, description="Number of failed attacks")
+    total_count: int = Field(0, ge=0, description="Total number of attack results")
+
+
+class ScenarioRunDetail(BaseModel):
+    """Full detailed results of a scenario run."""
+
+    run: ScenarioRunSummary = Field(..., description="The scenario run summary")
+    attacks: list[AtomicAttackResults] = Field(..., description="Results grouped by atomic attack")