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_request_converter_composition=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/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,
+    ScenarioDefaultRunSizeEstimate,
+    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=ScenarioDefaultRunSizeEstimate,
+    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,
+) -> ScenarioDefaultRunSizeEstimate:
+    """
+    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:
+        ScenarioDefaultRunSizeEstimate: 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
 # ============================================================================
@@ -154,7 +197,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 +211,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,