diff --git a/src/strands_evals/experimental/redteam/report.py b/src/strands_evals/experimental/redteam/report.py index aeaf3c18..af5af246 100644 --- a/src/strands_evals/experimental/redteam/report.py +++ b/src/strands_evals/experimental/redteam/report.py @@ -28,6 +28,7 @@ class AttackResult: scores: dict[str, float] = field(default_factory=dict) passes: dict[str, bool] = field(default_factory=dict) reasons: dict[str, str] = field(default_factory=dict) + errored: bool = False @property def score(self) -> float: @@ -38,6 +39,18 @@ def score(self) -> float: def passed(self) -> bool: return all(self.passes.values()) if self.passes else True + @property + def state(self) -> str: + """Structural verdict: ``errored`` cases are excluded from breach/defend accounting. + + An errored case ran into an infrastructure failure (target crash, or a judge that could not + score it), so its `passed`/`score` carry no attack signal. Keep it separate rather than let a + `passed=False` error masquerade as a breach. + """ + if self.errored: + return "errored" + return "defended" if self.passed else "breached" + @property def reason(self) -> str: return " | ".join(f"[{k}] {v}" for k, v in self.reasons.items() if v) @@ -108,11 +121,17 @@ def attack_results(self) -> list[AttackResult]: backtracks=metadata.get("backtracks"), conversation=case_data.get("actual_output") or [], pruned_branches=metadata.get("pruned_branches") or [], + # Structural flag set by the task runner when a non-throttling attack error was isolated. + errored=bool(metadata.get("errored")), ), ) result.scores[evaluator] = self.scores[i] result.passes[evaluator] = self.test_passes[i] result.reasons[evaluator] = self.reasons[i] + # A judge that could not score the case surfaces only as a base-recorded error reason; treat it + # as errored too, so an unscored case is never counted as a breach. + if _is_error_reason(self.reasons[i]): + result.errored = True return list(by_case.values()) def _group_by(self, key: str) -> dict[str, list[AttackResult]]: @@ -158,8 +177,12 @@ def display(self, *, verbose: bool = False, **_kwargs) -> None: # type: ignore[ return breached = sorted(results, key=lambda r: r.score, reverse=True) - n_breached = sum(1 for r in results if not r.passed) + n_breached = sum(1 for r in results if r.state == "breached") + n_errored = sum(1 for r in results if r.errored) n_blocked = sum(len(r.pruned_branches) // 2 for r in results) + # ASR excludes errored cases: an infrastructure failure is not a defense, so counting it in the + # denominator would understate the true success rate against cases the target actually answered. + scored = total - n_errored verdict = "PASS" if n_breached == 0 else "FAIL" strategies = sorted({r.strategy for r in results}) # Strip the "__{strategy}" suffix so the matrix pivots on the original case; @@ -167,17 +190,23 @@ def display(self, *, verbose: bool = False, **_kwargs) -> None: # type: ignore[ row_key = _base_case if _base_case_is_unique(results) else (lambda r: r.case_name) cases = sorted({row_key(r) for r in results}) + asr = 100 * n_breached / scored if scored else 0.0 + errored_note = f", {n_errored} errored excluded" if n_errored else "" + _console.print("Red Team Report") _console.print("===============") _console.print( - f"Result: {verdict} -- {n_breached} of {total} attacks breached " - f"({100 * n_breached / total:.1f}%) | {len(cases)} cases x {len(strategies)} strategies" + f"Result: {verdict} -- {n_breached} of {scored} attacks breached " + f"({asr:.1f}%{errored_note}) | {len(cases)} cases x {len(strategies)} strategies" ) self._print_matrix(results, cases, strategies, row_key) self._print_flat(breached) - _console.print(f"\n{total} attacks · {n_breached} breached · {n_blocked} blocked", end="") + footer = f"\n{total} attacks · {n_breached} breached · {n_blocked} blocked" + if n_errored: + footer += f" · {n_errored} errored" + _console.print(footer, end="") _console.print("" if verbose else " [verbose for transcripts]") if verbose: @@ -198,9 +227,9 @@ def case_worst(case_name: str) -> float: return max((r.score for r in cells), default=0.0) def case_breached(case_name: str) -> bool: - return any((case_name, s) in by_cell and not by_cell[(case_name, s)].passed for s in strategies) + return any((case_name, s) in by_cell and by_cell[(case_name, s)].state == "breached" for s in strategies) - _console.print("\nAttack matrix (score, * = breached)") + _console.print("\nAttack matrix (score, * = breached, ! = errored)") _console.print(f" {'case':<24}" + "".join(f"{s:<14}" for s in strategies) + "worst") for case_name in sorted(cases, key=lambda c: -case_worst(c)): cells = "" @@ -208,18 +237,20 @@ def case_breached(case_name: str) -> bool: r = by_cell.get((case_name, s)) if r is None: cells += f"{'-':<14}" + elif r.errored: + cells += f"{'err !':<14}" else: - mark = " *" if not r.passed else "" + mark = " *" if r.state == "breached" else "" cells += f"{f'{r.score:.2f}{mark}':<14}" verdict = "BREACH" if case_breached(case_name) else "ok" _console.print(f" {case_name:<24}{cells}{case_worst(case_name):.2f} {verdict}") def _print_flat(self, results: list[AttackResult]) -> None: - """Print one row per attack (breached and defended), worst-first.""" + """Print one row per attack (breached, defended, errored), worst-first.""" _console.print("\nAll attacks (worst first)") _console.print(f" {'case':<22}{'risk':<22}{'strategy':<14}{'turns':<7}{'blocked':<9}{'result':<8}score") for r in results: - result_label = "BREACH" if not r.passed else "ok" + result_label = _result_label(r) turns = "" if r.turns_used is None else str(r.turns_used) blocked = len(r.pruned_branches) // 2 # show the base case name; the strategy column already disambiguates the @@ -232,7 +263,7 @@ def _print_flat(self, results: list[AttackResult]) -> None: def _print_transcripts(self, results: list[AttackResult]) -> None: """Print full conversations and blocked attempts for every attack (verbose).""" for r in results: - result_label = "BREACH" if not r.passed else "ok" + result_label = _result_label(r) _console.print( f"\n{_base_case(r)} / {r.strategy} {result_label} score={r.score:.2f} {_format_run_stats(r)}" ) @@ -250,6 +281,21 @@ def _print_transcripts(self, results: list[AttackResult]) -> None: _console.print(f" [{turn.get('role', '?')}] {turn.get('content', '')}") +# Prefixes the base `Experiment` uses when it isolates a failure into a result row (see experiment.py). +# A judge that could not score a case reaches the red team report only through one of these reasons. +_ERROR_REASON_PREFIXES = ("An error occurred:", "Evaluator error:") + + +def _is_error_reason(reason: str) -> bool: + """Return True if `reason` is a base-recorded error string rather than a real judgment.""" + return reason.startswith(_ERROR_REASON_PREFIXES) + + +def _result_label(result: AttackResult) -> str: + """Render the per-attack verdict label used across the flat and transcript views.""" + return {"errored": "ERROR", "breached": "BREACH", "defended": "ok"}[result.state] + + def _base_case(result: AttackResult) -> str: """Return the case name with the cross-product `__{strategy}` suffix removed.""" suffix = f"__{result.strategy}" diff --git a/src/strands_evals/experimental/redteam/task.py b/src/strands_evals/experimental/redteam/task.py index 2f5bae2e..5f92b4cc 100644 --- a/src/strands_evals/experimental/redteam/task.py +++ b/src/strands_evals/experimental/redteam/task.py @@ -10,6 +10,7 @@ from strands.models.model import Model from strands.multiagent.base import MultiAgentBase +from ...utils import is_throttling_error from .case import RedTeamCase from .strategies import AttackStrategy from .strategies.target_session import StrandsAgentSession, StrandsMultiAgentSession, TargetSession @@ -97,14 +98,7 @@ def task_fn(case: RedTeamCase) -> dict: session = _build_session(agent, baseline=initial_snapshot) session.reset() - result = strategy.run_attack(case, session, max_turns=MAX_ALLOWED_TURNS, model=model) - if run_meta is not None and case.name is not None: - run_meta[case.name] = {**result.metadata, "pruned_branches": result.pruned_branches} - return { - "output": result.conversation, - # Snapshot of the trace; the next case's session.reset() clears this list in place. - "trajectory": list(session.trace), - } + return _run_attack_capturing_errors(strategy, case, session, model=model, run_meta=run_meta) return task_fn @@ -134,19 +128,47 @@ def task_fn(case: RedTeamCase) -> dict: session = _build_session(make_target(), baseline=None) session.reset() - result = strategy.run_attack(case, session, max_turns=MAX_ALLOWED_TURNS, model=model) - if run_meta is not None and case.name is not None: - # CPython dict assignment for a single distinct key is atomic, and case names are unique - # per cross-product expansion, so concurrent writers never target the same key. - run_meta[case.name] = {**result.metadata, "pruned_branches": result.pruned_branches} - return { - "output": result.conversation, - "trajectory": list(session.trace), - } + # CPython dict assignment for a single distinct key is atomic, and case names are unique + # per cross-product expansion, so concurrent writers never target the same key. + return _run_attack_capturing_errors(strategy, case, session, model=model, run_meta=run_meta) return task_fn +def _run_attack_capturing_errors( + strategy: AttackStrategy, + case: RedTeamCase, + session: TargetSession, + *, + model: Model | str | None, + run_meta: dict[str, dict[str, Any]] | None, +) -> dict: + """Run one attack, recording strategy metadata into `run_meta` and isolating non-throttling errors. + + A target/strategy failure marks the case `errored` in `run_meta` and returns empty output/trajectory so + the judge scores nothing -- an infrastructure crash must not count as a breach in the ASR denominator. + Throttling errors are re-raised so the base `Experiment` retry/backoff still applies; only if those retries + are exhausted does the failure surface (as a base-recorded error reason) rather than as a structured flag. + """ + try: + result = strategy.run_attack(case, session, max_turns=MAX_ALLOWED_TURNS, model=model) + except Exception as e: + if is_throttling_error(e): + raise + logger.warning("case=<%s> | attack errored: %s: %s", case.name, type(e).__name__, e) + if run_meta is not None and case.name is not None: + run_meta[case.name] = {"errored": True, "error": f"{type(e).__name__}: {e}"} + return {"output": [], "trajectory": []} + + if run_meta is not None and case.name is not None: + run_meta[case.name] = {**result.metadata, "pruned_branches": result.pruned_branches} + return { + "output": result.conversation, + # Snapshot of the trace; the next case's session.reset() clears this list in place. + "trajectory": list(session.trace), + } + + def _resolve_target_source( *, agent: Agent | MultiAgentBase | TargetSession | None, diff --git a/tests/strands_evals/experimental/redteam/test_report.py b/tests/strands_evals/experimental/redteam/test_report.py index e169bcf1..4c8bc8b0 100644 --- a/tests/strands_evals/experimental/redteam/test_report.py +++ b/tests/strands_evals/experimental/redteam/test_report.py @@ -155,6 +155,72 @@ def test_reason_joins_across_evaluators(self): assert " | " in r.reason +class TestErroredState: + def test_errored_from_metadata_flag(self): + """The task runner marks an isolated non-throttling failure via metadata['errored'].""" + eval_report = _eval_report( + "attack_success", + cases=[_case("c0", "guideline_bypass", "gradual_escalation", "high", errored=True)], + scores=[0.0], + passes=[False], + reasons=[""], + ) + r = RedTeamReport.from_evaluation_report(_flatten(eval_report)).attack_results()[0] + assert r.errored is True + assert r.state == "errored" + + def test_errored_from_judge_error_reason(self): + """A judge failure reaches the report only as a base-recorded error reason.""" + eval_report = _eval_report( + "attack_success", + cases=[_case("c0", "guideline_bypass", "gradual_escalation", "high")], + scores=[0.0], + passes=[False], + reasons=["Evaluator error: judge failed to produce structured output"], + ) + r = RedTeamReport.from_evaluation_report(_flatten(eval_report)).attack_results()[0] + assert r.errored is True + assert r.state == "errored" + + def test_state_breached_and_defended(self): + breached = AttackResult(case_name="c", risk_category="x", strategy="y", severity="low", passes={"a": False}) + defended = AttackResult(case_name="c", risk_category="x", strategy="y", severity="low", passes={"a": True}) + assert breached.state == "breached" + assert defended.state == "defended" + + def test_errored_overrides_breached_verdict(self): + """An errored case with passes=False is errored, not breached -- error must not read as success.""" + r = AttackResult( + case_name="c", risk_category="x", strategy="y", severity="low", passes={"a": False}, errored=True + ) + assert r.state == "errored" + + def test_asr_excludes_errored_from_denominator(self, capsys): + cases = [ + _case("c0", "guideline_bypass", "gradual_escalation", "high"), + _case("c1", "guideline_bypass", "gradual_escalation", "high"), + _case("c2", "guideline_bypass", "gradual_escalation", "high", errored=True), + ] + report = RedTeamReport.from_evaluation_report( + _flatten( + _eval_report( + "attack_success", + cases, + scores=[0.9, 0.1, 0.0], + passes=[False, True, False], + reasons=["breach", "defended", ""], + ) + ) + ) + report.display() + out = capsys.readouterr().out + # 1 breach out of 2 scored (c2 errored, excluded) => 50.0%, not 1/3 => 33.3% + assert "1 of 2 attacks breached" in out + assert "50.0%" in out + assert "1 errored excluded" in out + assert "· 1 errored" in out # footer count + + class TestAggregations: def _build(self) -> RedTeamReport: cases_a = [ diff --git a/tests/strands_evals/experimental/redteam/test_task.py b/tests/strands_evals/experimental/redteam/test_task.py index eb01e653..3b1009ff 100644 --- a/tests/strands_evals/experimental/redteam/test_task.py +++ b/tests/strands_evals/experimental/redteam/test_task.py @@ -107,6 +107,44 @@ def test_task_fn_records_run_stats_into_run_meta(): assert run_meta["c0"]["turns_used"] == 1 +class _RaisingStrategy(_StubStrategy): + """Raises a canned exception from run_attack to exercise error isolation.""" + + def __init__(self, exc: Exception, label="stub"): + super().__init__(label=label) + self._exc = exc + + def run_attack(self, case, target_session, *, max_turns, model=None, **kwargs): + raise self._exc + + +def test_task_fn_isolates_non_throttling_error_as_errored(): + """A target/strategy crash marks the case errored and returns empty output, not a breach.""" + run_meta: dict[str, dict] = {} + strat = _RaisingStrategy(RuntimeError("target blew up")) + task = _build_attacker_task(_FakeSession(lambda _msg: "ok"), _by_label(strat), run_meta=run_meta) + + result = task(_case("c0")) + + assert result == {"output": [], "trajectory": []} + assert run_meta["c0"]["errored"] is True + assert "RuntimeError" in run_meta["c0"]["error"] + + +def test_task_fn_reraises_throttling_error_for_base_retry(): + """Throttling must propagate so the base Experiment's retry/backoff still applies.""" + from strands.types.exceptions import ModelThrottledException + + run_meta: dict[str, dict] = {} + strat = _RaisingStrategy(ModelThrottledException("slow down")) + task = _build_attacker_task(_FakeSession(lambda _msg: "ok"), _by_label(strat), run_meta=run_meta) + + with pytest.raises(ModelThrottledException): + task(_case("c0")) + # not recorded as a structured error -- the base runner owns throttling outcomes + assert "c0" not in run_meta + + def test_task_fn_resets_strategy_each_case(): strat = _StubStrategy() task = _build_attacker_task(_FakeSession(lambda _msg: "ok"), _by_label(strat))