ENH Add Scenario progress foundation

doc/code/framework.md

@@ -182,6 +182,7 @@ If you are contributing to PyRIT, that work will most likely land in one of the
 - `core` stays deliberately small so a default run doesn't print 200 techniques or take forever; the wider catalog lives in `extra` and is selected on demand. Users pick subsets by passing initializer tags (e.g. `core`, `extra`, `all`) or writing their own initializer, so different runs — including from the CLI — can register different technique sets without changing the catalog.
 - A technique tied to one scenario is fine; if it's pinned and non-reusable it can stay local to that scenario, but if another scenario could reuse it, promote it to a catalog module and tag it.
 - Tags describe a technique (behavioral tags like `single_turn`/`multi_turn`, owner tags like `airt`); they don't decide what a scenario runs. There is deliberately **no global `default` tag** — a default is scenario-relative, declared per scenario via `build_technique_class_from_factories` (the `factories` list is the pool, catalog tags become named aggregate presets, and `default_tags` / `default_names` set what runs when nothing is chosen).
+- Factories opt into additive request-converter composition with `supports_additional_request_converters=True`. This is a semantic capability, not just constructor-signature detection; the factory validates that opted-in attacks accept `attack_converter_config`.
 - **Does not own**: the conversation algorithm itself. Branching, turn management, and scoring decisions live in the executor it wraps — a technique only selects and configures existing components, and shouldn't implement new sending, scoring, or branching logic.
 
 **Framework Plans**:

pyrit/backend/models/scenarios.py

@@ -13,7 +13,7 @@
 from pydantic import BaseModel, Field
 
 from pyrit.backend.models.common import PaginationInfo
-from pyrit.models.catalog.scenario import RegisteredScenario, ScenarioRunSummary
+from pyrit.models.catalog.scenario import RegisteredScenario, ScenarioRunListItem
 
 __all__ = [
     "ListRegisteredScenariosResponse",
@@ -31,4 +31,4 @@ class ListRegisteredScenariosResponse(BaseModel):
 class ScenarioRunListResponse(BaseModel):
     """Response for listing scenario runs."""
 
-    items: list[ScenarioRunSummary] = Field(..., description="List of scenario runs")
+    items: list[ScenarioRunListItem] = Field(..., description="List of scenario runs")

pyrit/backend/routes/scenarios.py

@@ -13,6 +13,7 @@
 """
 
 from fastapi import APIRouter, HTTPException, Query, status
+from starlette.concurrency import run_in_threadpool
 
 from pyrit.backend.models.common import ProblemDetail
 from pyrit.backend.models.scenarios import (
@@ -25,8 +26,11 @@
 from pyrit.models.catalog.scenario import (
     RegisteredScenario,
     RunScenarioRequest,
+    ScenarioRunSizeEstimate,
+    ScenarioRunSizeEstimateRequest,
     ScenarioRunSummary,
 )
+from pyrit.models.scenario_progress import ScenarioRunProgress
 
 router = APIRouter(prefix="/scenarios", tags=["scenarios"])
 
@@ -86,6 +90,45 @@ async def get_scenario(scenario_name: str) -> RegisteredScenario:  # pyrit-async
     return scenario
 
 
+@router.post(
+    "/catalog/{scenario_name}/estimate",
+    response_model=ScenarioRunSizeEstimate,
+    responses={
+        400: {"model": ProblemDetail, "description": "Invalid estimate configuration"},
+        404: {"model": ProblemDetail, "description": "Scenario not found"},
+    },
+)
+async def estimate_scenario_run_size(  # pyrit-async-suffix-exempt
+    *,
+    scenario_name: str,
+    request: ScenarioRunSizeEstimateRequest,
+) -> ScenarioRunSizeEstimate:
+    """
+    Estimate a configured scenario without creating or persisting a run.
+
+    Args:
+        scenario_name: Registry name of the scenario.
+        request: Techniques, datasets, baseline choice, and scenario parameters to preview.
+
+    Returns:
+        ScenarioRunSizeEstimate: Structured request-specific planned-unit estimate.
+    """
+    service = get_scenario_service()
+    try:
+        estimate = await service.estimate_scenario_run_size_async(
+            scenario_name=scenario_name,
+            request=request,
+        )
+    except ValueError as exc:
+        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(exc)) from None
+    if estimate is None:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail=f"Scenario '{scenario_name}' not found",
+        )
+    return estimate
+
+
 # ============================================================================
 # Scenario Runs
 # ============================================================================
@@ -122,7 +165,9 @@ async def start_scenario_run(request: RunScenarioRequest) -> ScenarioRunSummary:
     "/runs",
     response_model=ScenarioRunListResponse,
 )
-async def list_scenario_runs(limit: int = Query(100, ge=1)) -> ScenarioRunListResponse:  # pyrit-async-suffix-exempt
+async def list_scenario_runs(
+    limit: int = Query(100, ge=1, le=100),
+) -> ScenarioRunListResponse:  # pyrit-async-suffix-exempt
     """
     List tracked scenario runs (most recent first).
 
@@ -133,7 +178,7 @@ async def list_scenario_runs(limit: int = Query(100, ge=1)) -> ScenarioRunListRe
         ScenarioRunListResponse: Runs, most recent first.
     """
     service = get_scenario_run_service()
-    return service.list_runs(limit=limit)
+    return await run_in_threadpool(service.list_runs, limit=limit)
 
 
 @router.get(
@@ -154,7 +199,12 @@ async def get_scenario_run(scenario_result_id: str) -> ScenarioRunSummary:  # py
         ScenarioRunSummary: Current run status (and result if completed).
     """
     service = get_scenario_run_service()
-    run = service.get_run(scenario_result_id=scenario_result_id)
+    active_snapshot = service.snapshot_active_run(scenario_result_id=scenario_result_id)
+    run = await run_in_threadpool(
+        service.get_run_from_storage,
+        scenario_result_id=scenario_result_id,
+        active_error=active_snapshot.error,
+    )
     if run is None:
         raise HTTPException(
             status_code=status.HTTP_404_NOT_FOUND,
@@ -163,6 +213,46 @@ async def get_scenario_run(scenario_result_id: str) -> ScenarioRunSummary:  # py
     return run
 
 
+@router.get(
+    "/runs/{scenario_result_id}/progress",
+    response_model=ScenarioRunProgress,
+    responses={
+        400: {"model": ProblemDetail, "description": "Invalid progress cursor"},
+        404: {"model": ProblemDetail, "description": "Run not found"},
+    },
+)
+async def get_scenario_run_progress(  # pyrit-async-suffix-exempt
+    *,
+    scenario_result_id: str,
+    since: str | None = Query(None, description="Opaque ascending progress cursor"),
+    limit: int = Query(100, ge=1, le=500),
+) -> ScenarioRunProgress:
+    """
+    Get a compact, refresh-safe page of scenario progress deltas.
+
+    Returns:
+        ScenarioRunProgress: The run plan and ascending result deltas.
+    """
+    service = get_scenario_run_service()
+    active_snapshot = service.snapshot_active_run(scenario_result_id=scenario_result_id)
+    try:
+        progress = await run_in_threadpool(
+            service.get_run_progress_from_storage,
+            scenario_result_id=scenario_result_id,
+            since=since,
+            limit=limit,
+            active_group_ids=active_snapshot.active_group_ids,
+        )
+    except ValueError as exc:
+        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(exc)) from None
+    if progress is None:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail=f"Scenario run '{scenario_result_id}' not found",
+        )
+    return progress
+
+
 @router.post(
     "/runs/{scenario_result_id}/cancel",
     response_model=ScenarioRunSummary,

pyrit/backend/services/attack_service.py

@@ -58,7 +58,7 @@
 from pyrit.backend.models.common import PaginationInfo
 from pyrit.backend.services.converter_service import get_converter_service
 from pyrit.backend.services.target_service import get_target_service
-from pyrit.memory import AttackResultsKeysetCursor, CentralMemory, data_serializer_factory
+from pyrit.memory import AttackResultKeysetCursor, CentralMemory, data_serializer_factory
 from pyrit.models import (
     AtomicAttackIdentifier,
     AttackIdentifier,
@@ -182,7 +182,7 @@ async def list_attacks_async(
         page_results = list(results[:limit])
         next_cursor = (
             self._encode_attack_cursor(
-                cursor=AttackResultsKeysetCursor.from_attack_result(page_results[-1]),
+                cursor=AttackResultKeysetCursor.from_attack_result(page_results[-1]),
                 fingerprint=filter_fingerprint,
             )
             if has_next_page and page_results
@@ -919,7 +919,7 @@ def _norm_labels(
         return hashlib.sha256(canonical.encode("utf-8")).hexdigest()[:16]
 
     @staticmethod
-    def _encode_attack_cursor(*, cursor: AttackResultsKeysetCursor, fingerprint: str) -> str:
+    def _encode_attack_cursor(*, cursor: AttackResultKeysetCursor, fingerprint: str) -> str:
         """
         Encode a keyset anchor and its filter fingerprint into an opaque pagination cursor.
 
@@ -940,7 +940,7 @@ def _encode_attack_cursor(*, cursor: AttackResultsKeysetCursor, fingerprint: str
         return base64.urlsafe_b64encode(raw).decode("ascii").rstrip("=")
 
     @staticmethod
-    def _decode_attack_cursor(*, cursor: str | None, fingerprint: str) -> AttackResultsKeysetCursor | None:
+    def _decode_attack_cursor(*, cursor: str | None, fingerprint: str) -> AttackResultKeysetCursor | None:
         """
         Decode the opaque list-attacks cursor into a keyset (seek) anchor.
 
@@ -952,7 +952,7 @@ def _decode_attack_cursor(*, cursor: str | None, fingerprint: str) -> AttackResu
         raising or seeking within the wrong result set.
 
         Returns:
-            The decoded ``AttackResultsKeysetCursor``, or ``None`` to start at the first page.
+            The decoded ``AttackResultKeysetCursor``, or ``None`` to start at the first page.
         """
         if not cursor:
             return None
@@ -985,7 +985,7 @@ def _decode_attack_cursor(*, cursor: str | None, fingerprint: str) -> AttackResu
             # A crafted cursor near datetime's min/max with a large UTC offset overflows the
             # representable range when shifted to UTC; treat it as malformed and restart at page one.
             return None
-        return AttackResultsKeysetCursor(timestamp=timestamp, attack_result_id=attack_result_id)
+        return AttackResultKeysetCursor(timestamp=timestamp, attack_result_id=attack_result_id)
 
     # ========================================================================
     # Private Helper Methods - Duplicate / Branch