Module `scenario.report`

Post-hoc report generation for red-team scenario runs.

Public API: scenario.save_redteam_report(result, red_team=…, test_name=…)

Each call writes a JSON file into a timestamped batch directory under ./redteam-reports/ (one directory per Python process by default). The scenario.report.app module is a Streamlit dashboard that reads from such a directory.

Expand source code

"""Post-hoc report generation for red-team scenario runs.

Public API:
    scenario.save_redteam_report(result, red_team=..., test_name=...)

Each call writes a JSON file into a timestamped batch directory under
``./redteam-reports/`` (one directory per Python process by default).
The ``scenario.report.app`` module is a Streamlit dashboard that reads
from such a directory.
"""

from ._save import save_redteam_report, set_batch_dir, current_batch_dir
from ._aggregate import aggregate_fixes, load_or_generate as load_or_generate_fixes

# Skip the Streamlit dashboard entry point during pdoc API doc generation.
# `app` is launched via `streamlit run`, not imported, and `streamlit` is an
# optional extra (`pip install langwatch-scenario[report]`) so the docs build
# environment can't import it.
__pdoc__ = {"app": False}

__all__ = [
    "save_redteam_report",
    "set_batch_dir",
    "current_batch_dir",
    "aggregate_fixes",
    "load_or_generate_fixes",
]

Functions

def aggregate_fixes(reports: list[dict], *, model: str = 'openai/gpt-4.1', temperature: float = 0.2) ‑> dict

Return {'clusters': […]} from a single LLM call.

Expand source code

def aggregate_fixes(
    reports: list[dict],
    *,
    model: str = "openai/gpt-4.1",
    temperature: float = 0.2,
) -> dict:
    """Return {'clusters': [...]} from a single LLM call."""
    findings_block = _build_findings_block(reports)
    prompt = _AGGREGATE_PROMPT.format(findings_block=findings_block)
    resp = cast(
        ModelResponse,
        litellm.completion(
            model=model,
            messages=[{"role": "user", "content": prompt}],
            response_format={"type": "json_object"},
            temperature=temperature,
        ),
    )
    raw = cast(Choices, resp.choices[0]).message.content or "{}"
    try:
        data = json.loads(raw)
    except json.JSONDecodeError:
        m = re.search(r"\{.*\}", raw, re.DOTALL)
        data = json.loads(m.group(0)) if m else {}
    clusters = data.get("clusters") if isinstance(data, dict) else None
    if not isinstance(clusters, list):
        clusters = []
    return {"clusters": clusters}

def current_batch_dir() ‑> pathlib.Path | None

Return the current batch directory, or None if no report has been saved yet.

Expand source code

def current_batch_dir() -> Optional[Path]:
    """Return the current batch directory, or None if no report has been saved yet."""
    return _BATCH_DIR

def load_or_generate_fixes(batch_dir: Path, reports: list[dict], *, model: str = 'openai/gpt-4.1', force: bool = False) ‑> dict

Return aggregated fixes, cached to disk per batch directory.

Expand source code

def load_or_generate(
    batch_dir: Path,
    reports: list[dict],
    *,
    model: str = "openai/gpt-4.1",
    force: bool = False,
) -> dict:
    """Return aggregated fixes, cached to disk per batch directory."""
    cache = batch_dir / _CACHE_NAME
    if cache.exists() and not force:
        try:
            return json.loads(cache.read_text())
        except (OSError, json.JSONDecodeError):
            # Cache is missing/unreadable/corrupt; fall back to fresh aggregation.
            pass
    result = aggregate_fixes(reports, model=model)
    try:
        cache.write_text(json.dumps(result, indent=2))
    except OSError:
        # Best-effort cache write failure should not block returning results.
        pass
    return result

def save_redteam_report(result: Optional[ScenarioResult] = None, *, red_team: RedTeamAgent, test_name: str, criteria: Optional[Iterable[str]] = None, description: Optional[str] = None, analyzer_model: Optional[str] = None, analyze: bool = True, out_dir: Optional[Union[str, Path]] = None, error: Optional[str] = None, elapsed_seconds: Optional[float] = None, messages_override: Optional[Iterable[Any]] = None, failing_turn_hint: Optional[int] = None) ‑> pathlib.Path

Persist a red-team scenario run to JSON for dashboard consumption.

Pass result=None, error="..." to record a run that raised before completion (e.g. a per-turn check function raised an exception). The dashboard will show the card as "errored" with no transcript.

Args

result: Object returned by await run()(…), or None if the run raised an exception.
red_team: The RedTeamAgent instance used in the run (for metadata).
test_name: Short identifier for the test.
criteria: Judge criteria used, for the report display.
description: Scenario description, for the report display.
analyzer_model: Model for the suggestion pass. Defaults to the red team's metaprompt_model.
analyze: If False, skip the LLM analyzer call.
out_dir: Directory to write into. Defaults to the current batch dir (one per process, under ./redteam-reports/<timestamp>/).
error: Exception message if the run raised. Saved in place of reasoning.
elapsed_seconds: Wall-clock seconds for the run, if known.

Returns

Path to the written JSON file.

Expand source code

def save_redteam_report(
    result: Optional[ScenarioResult] = None,
    *,
    red_team: RedTeamAgent,
    test_name: str,
    criteria: Optional[Iterable[str]] = None,
    description: Optional[str] = None,
    analyzer_model: Optional[str] = None,
    analyze: bool = True,
    out_dir: Optional[Union[str, Path]] = None,
    error: Optional[str] = None,
    elapsed_seconds: Optional[float] = None,
    messages_override: Optional[Iterable[Any]] = None,
    failing_turn_hint: Optional[int] = None,
) -> Path:
    """Persist a red-team scenario run to JSON for dashboard consumption.

    Pass ``result=None, error="..."`` to record a run that raised before
    completion (e.g. a per-turn check function raised an exception). The
    dashboard will show the card as "errored" with no transcript.

    Args:
        result: Object returned by ``await scenario.run(...)``, or None if
            the run raised an exception.
        red_team: The RedTeamAgent instance used in the run (for metadata).
        test_name: Short identifier for the test.
        criteria: Judge criteria used, for the report display.
        description: Scenario description, for the report display.
        analyzer_model: Model for the suggestion pass. Defaults to the
            red team's ``metaprompt_model``.
        analyze: If False, skip the LLM analyzer call.
        out_dir: Directory to write into. Defaults to the current batch dir
            (one per process, under ``./redteam-reports/<timestamp>/``).
        error: Exception message if the run raised. Saved in place of reasoning.
        elapsed_seconds: Wall-clock seconds for the run, if known.

    Returns:
        Path to the written JSON file.
    """
    dest_dir = Path(out_dir).resolve() if out_dir else _default_batch_dir()
    dest_dir.mkdir(parents=True, exist_ok=True)

    strategy = _strategy_name(red_team)
    if messages_override is not None:
        messages = [_serialize_message(m) for m in messages_override]
    elif result is not None:
        messages = [_serialize_message(m) for m in (result.messages or [])]
    else:
        messages = []

    analysis: dict = {
        "failing_turn_index": failing_turn_hint,
        "failure_summary": "",
        "suggestions": [],
        "severity": "medium",
        "severity_rationale": "",
        "break_severity": "none",
        "break_rationale": "",
    }
    if analyze and messages:
        _success = bool(result.success) if result else False
        _reasoning = (result.reasoning if result else None) or (error or "")
        _failed = list((result.failed_criteria if result else []) or [])
        try:
            analysis = _analyze(
                target=red_team.target,
                strategy=strategy,
                turns=red_team.total_turns,
                criteria=list(criteria or []),
                success=_success,
                reasoning=_reasoning,
                failed_criteria=_failed,
                transcript=_transcript_for_prompt(messages),
                model=analyzer_model or red_team.metaprompt_model,
            )
            if analysis.get("failing_turn_index") is None and failing_turn_hint is not None:
                analysis["failing_turn_index"] = failing_turn_hint
        except Exception as e:
            analysis["failure_summary"] = f"(analyzer failed: {e})"

    if error is not None:
        success = False
        reasoning = f"EXCEPTION: {error}"
        passed_criteria: list[str] = []
        failed_criteria: list[str] = list(criteria or [])
        total_time = elapsed_seconds
        agent_time = None
        # If we have a transcript, the exception came from a per-turn check
        # catching the agent break red-handed — that's a defeat, not an infra
        # error. Classify as "broke".
        status = "broke" if messages else "errored"
        if not analysis["failure_summary"]:
            analysis["failure_summary"] = reasoning
    else:
        success = bool(result.success) if result else False
        reasoning = (result.reasoning if result else "") or ""
        passed_criteria = list((result.passed_criteria if result else []) or [])
        failed_criteria = list((result.failed_criteria if result else []) or [])
        total_time = (result.total_time if result else None) or elapsed_seconds
        agent_time = result.agent_time if result else None
        status = "held" if success else "broke"

    report = {
        "test_name": test_name,
        "description": description or "",
        "strategy": strategy,
        "target": red_team.target,
        "total_turns": red_team.total_turns,
        "attacker_model": red_team._model,
        "metaprompt_model": red_team.metaprompt_model,
        "criteria": list(criteria or []),
        "status": status,
        "success": success,
        "reasoning": reasoning,
        "passed_criteria": passed_criteria,
        "failed_criteria": failed_criteria,
        "total_time": total_time,
        "agent_time": agent_time,
        "messages": messages,
        "failing_turn_index": analysis["failing_turn_index"],
        "failure_summary": analysis["failure_summary"],
        "suggestions": analysis["suggestions"],
        "severity": analysis["severity"],
        "severity_rationale": analysis.get("severity_rationale", ""),
        "break_severity": analysis.get("break_severity", "none"),
        "break_rationale": analysis.get("break_rationale", ""),
        "timestamp": datetime.now().isoformat(timespec="seconds"),
    }

    stamp = f"{int(time.time() * 1000)}"
    safe_name = re.sub(r"[^A-Za-z0-9_-]+", "_", test_name).strip("_") or "run"
    filename = f"{stamp}_{safe_name}_{strategy}.json"
    path = dest_dir / filename
    path.write_text(json.dumps(report, indent=2, default=str))
    return path

def set_batch_dir(path: Union[str, Path]) ‑> pathlib.Path

Override the batch directory. Subsequent save_redteam_report() calls without an explicit out_dir will write here.

Expand source code

def set_batch_dir(path: Union[str, Path]) -> Path:
    """Override the batch directory. Subsequent ``save_redteam_report`` calls
    without an explicit ``out_dir`` will write here."""
    global _BATCH_DIR
    _BATCH_DIR = Path(path).resolve()
    _BATCH_DIR.mkdir(parents=True, exist_ok=True)
    return _BATCH_DIR