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/scenarios.py

@@ -5,10 +5,12 @@
 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 StrEnum
+from typing import Any, Optional
 
 from pydantic import BaseModel, Field
 
@@ -35,3 +37,126 @@ class ScenarioListResponse(BaseModel):
 
     items: list[ScenarioSummary] = Field(..., description="List of scenario summaries")
     pagination: PaginationInfo = Field(..., description="Pagination metadata")
+
+
+# ============================================================================
+# Scenario Run Models
+# ============================================================================
+
+
+class ScenarioRunStatus(StrEnum):
+    """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")
+    memory_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 ScenarioRunResult(BaseModel):
+    """Summary of a completed scenario run's results."""
+
+    scenario_result_id: str = Field(..., description="UUID of the ScenarioResult in memory")
+    run_state: str = Field(..., description="Final scenario run state (COMPLETED, FAILED)")
+    strategies_used: list[str] = Field(..., description="Strategy names that were executed")
+    total_attacks: int = Field(..., ge=0, description="Total number of atomic attacks")
+    completed_attacks: int = Field(..., ge=0, description="Number of attacks that completed")
+    number_tries: int = Field(..., ge=0, description="Number of execution attempts")
+    completion_time: datetime | None = Field(None, description="When the scenario finished")
+
+
+class ScenarioRunResponse(BaseModel):
+    """Response for a scenario run (status + optional result)."""
+
+    run_id: str = Field(..., description="Unique identifier for this run")
+    scenario_name: str = Field(..., description="Registry key of the scenario being run")
+    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")
+    result: ScenarioRunResult | None = Field(None, description="Result details if status is COMPLETED")
+
+
+class ScenarioRunListResponse(BaseModel):
+    """Response for listing scenario runs."""
+
+    items: list[ScenarioRunResponse] = Field(..., description="List of scenario runs")
+
+
+# ============================================================================
+# Scenario Results Detail Models
+# ============================================================================
+
+
+class AttackResultDetail(BaseModel):
+    """Detailed result of a single attack within a scenario."""
+
+    attack_result_id: str = Field(..., description="Unique ID of this attack result")
+    conversation_id: str = Field(..., description="Conversation ID that produced this result")
+    objective: str = Field(..., description="Natural-language description of the attacker's objective")
+    outcome: str = Field(..., description="Attack outcome: success, failure, or undetermined")
+    outcome_reason: str | None = Field(None, description="Reason for the outcome")
+    last_response: str | None = Field(None, description="Model response from the final turn")
+    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")
+    timestamp: datetime | None = Field(None, description="When the result was created")
+
+
+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[AttackResultDetail] = 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 ScenarioResultDetailResponse(BaseModel):
+    """Full detailed results of a scenario run."""
+
+    scenario_result_id: str = Field(..., description="UUID of the ScenarioResult")
+    scenario_name: str = Field(..., description="Name of the scenario")
+    scenario_version: int = Field(..., description="Version of the scenario")
+    run_state: str = Field(..., description="Final run state (COMPLETED, FAILED, etc.)")
+    objective_achieved_rate: int = Field(..., ge=0, le=100, description="Success rate as percentage (0-100)")
+    number_tries: int = Field(..., ge=0, description="Number of execution attempts")
+    completion_time: datetime | None = Field(None, description="When the scenario finished")
+    labels: dict[str, str] = Field(default_factory=dict, description="Labels attached to this run")
+    attacks: list[AtomicAttackResults] = Field(..., description="Results grouped by atomic attack")

pyrit/backend/routes/scenarios.py

@@ -4,22 +4,40 @@
 """
 Scenario API routes.
 
-Provides endpoints for listing available scenarios and their metadata.
+Provides endpoints for listing available scenarios, their metadata,
+and managing scenario runs.
+
+Route structure:
+    /api/scenarios/catalog       — scenario catalog (list + detail)
+    /api/scenarios/runs          — scenario execution lifecycle
 """
 
 from typing import Optional
 
 from fastapi import APIRouter, HTTPException, Query, status
 
 from pyrit.backend.models.common import ProblemDetail
-from pyrit.backend.models.scenarios import ScenarioListResponse, ScenarioSummary
+from pyrit.backend.models.scenarios import (
+    RunScenarioRequest,
+    ScenarioListResponse,
+    ScenarioResultDetailResponse,
+    ScenarioRunListResponse,
+    ScenarioRunResponse,
+    ScenarioSummary,
+)
+from pyrit.backend.services.scenario_run_service import get_scenario_run_service
 from pyrit.backend.services.scenario_service import get_scenario_service
 
 router = APIRouter(prefix="/scenarios", tags=["scenarios"])
 
 
+# ============================================================================
+# Scenario Catalog
+# ============================================================================
+
+
 @router.get(
-    "",
+    "/catalog",
     response_model=ScenarioListResponse,
 )
 async def list_scenarios(
@@ -30,7 +48,7 @@ async def list_scenarios(
     List all available scenarios.
 
     Returns scenario metadata including strategies, datasets, and defaults.
-    Use GET /api/scenarios/{scenario_name} for full details on a specific scenario.
+    Use GET /api/scenarios/catalog/{scenario_name} for full details on a specific scenario.
 
     Returns:
         ScenarioListResponse: Paginated list of scenario summaries.
@@ -40,7 +58,7 @@ async def list_scenarios(
 
 
 @router.get(
-    "/{scenario_name:path}",
+    "/catalog/{scenario_name:path}",
     response_model=ScenarioSummary,
     responses={
         404: {"model": ProblemDetail, "description": "Scenario not found"},
@@ -66,3 +84,147 @@ async def get_scenario(scenario_name: str) -> ScenarioSummary:
         )
 
     return scenario
+
+
+# ============================================================================
+# Scenario Runs
+# ============================================================================
+
+
+@router.post(
+    "/runs",
+    response_model=ScenarioRunResponse,
+    status_code=status.HTTP_202_ACCEPTED,
+    responses={
+        400: {"model": ProblemDetail, "description": "Invalid request (bad scenario/target/strategy)"},
+    },
+)
+async def start_scenario_run(request: RunScenarioRequest) -> ScenarioRunResponse:
+    """
+    Start a new scenario run as a background task.
+
+    Returns immediately with a run_id that can be polled for status.
+
+    Args:
+        request: Scenario run configuration.
+
+    Returns:
+        ScenarioRunResponse: Run metadata with PENDING status.
+    """
+    service = get_scenario_run_service()
+    try:
+        return await service.start_run_async(request=request)
+    except ValueError as e:
+        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e)) from None
+
+
+@router.get(
+    "/runs",
+    response_model=ScenarioRunListResponse,
+)
+async def list_scenario_runs(limit: int = Query(100, ge=1)) -> ScenarioRunListResponse:
+    """
+    List tracked scenario runs (most recent first).
+
+    Args:
+        limit (int): Maximum number of runs to return. Defaults to 100.
+
+    Returns:
+        ScenarioRunListResponse: Runs, most recent first.
+    """
+    service = get_scenario_run_service()
+    return service.list_runs(limit=limit)
+
+
+@router.get(
+    "/runs/{run_id}",
+    response_model=ScenarioRunResponse,
+    responses={
+        404: {"model": ProblemDetail, "description": "Run not found"},
+    },
+)
+async def get_scenario_run(run_id: str) -> ScenarioRunResponse:
+    """
+    Get the current status and result of a scenario run.
+
+    Args:
+        run_id: The unique run identifier returned by POST /runs.
+
+    Returns:
+        ScenarioRunResponse: Current run status (and result if completed).
+    """
+    service = get_scenario_run_service()
+    run = service.get_run(run_id=run_id)
+    if run is None:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail=f"Scenario run '{run_id}' not found",
+        )
+    return run
+
+
+@router.delete(
+    "/runs/{run_id}",
+    response_model=ScenarioRunResponse,
+    responses={
+        404: {"model": ProblemDetail, "description": "Run not found"},
+        409: {"model": ProblemDetail, "description": "Run already in terminal state"},
+    },
+)
+async def cancel_scenario_run(run_id: str) -> ScenarioRunResponse:
+    """
+    Cancel a running scenario.
+
+    Args:
+        run_id: The unique run identifier to cancel.
+
+    Returns:
+        ScenarioRunResponse: Updated run with CANCELLED status.
+    """
+    service = get_scenario_run_service()
+    try:
+        result = await service.cancel_run_async(run_id=run_id)
+    except ValueError as e:
+        raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail=str(e)) from None
+
+    if result is None:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail=f"Scenario run '{run_id}' not found",
+        )
+    return result
+
+
+@router.get(
+    "/runs/{run_id}/results",
+    response_model=ScenarioResultDetailResponse,
+    responses={
+        404: {"model": ProblemDetail, "description": "Run not found"},
+        409: {"model": ProblemDetail, "description": "Run not yet completed"},
+    },
+)
+async def get_scenario_run_results(run_id: str) -> ScenarioResultDetailResponse:
+    """
+    Get detailed results for a completed scenario run.
+
+    Returns per-attack outcomes including objectives, responses, scores,
+    and success/failure counts.
+
+    Args:
+        run_id: The unique run identifier.
+
+    Returns:
+        ScenarioResultDetailResponse: Full attack-level results.
+    """
+    service = get_scenario_run_service()
+    try:
+        result = service.get_run_results(run_id=run_id)
+    except ValueError as e:
+        raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail=str(e)) from None
+
+    if result is None:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail=f"Scenario run '{run_id}' not found",
+        )
+    return result

pyrit/backend/services/__init__.py

@@ -15,6 +15,10 @@
     ConverterService,
     get_converter_service,
 )
+from pyrit.backend.services.scenario_run_service import (
+    ScenarioRunService,
+    get_scenario_run_service,
+)
 from pyrit.backend.services.scenario_service import (
     ScenarioService,
     get_scenario_service,
@@ -31,6 +35,8 @@
     "get_converter_service",
     "ScenarioService",
     "get_scenario_service",
+    "ScenarioRunService",
+    "get_scenario_run_service",
     "TargetService",
     "get_target_service",
 ]