feat: per-attack-family ASR diagnostics in evaluation output

CHANGELOG.md

@@ -8,6 +8,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ## [Unreleased]
 
 ### Added
+- Per-attack-family diagnostics: `baseline`, `candidate`, and `evaluate` now report
+  attack success rate broken down by family (ranked leakiest-first) in addition to
+  the aggregate `ASR`, turning each iteration into a targeted, one-family-at-a-time
+  worklist. Families are derived from the existing harness taxonomy, so the frozen
+  evaluation surface (`eval_suite.jsonl`, `judge_prompt.md`) is unchanged.
 - Open-source readiness scaffolding:
   - Apache 2.0 `LICENSE` + `NOTICE`, `CONTRIBUTING.md` (CLA), `CODE_OF_CONDUCT.md`,
     `SECURITY.md`, `CODEOWNERS`

README.md

@@ -146,6 +146,33 @@ One example candidate change that improves the bundled stub is to add explicit h
 
 That gives you a realistic first improvement curve without changing the evaluator.
 
+## Reading The Output
+
+Every `baseline`, `candidate`, and `evaluate` run prints the top-line `ASR` and
+benign pass rate, followed by a **per-attack-family breakdown** so you can see
+*which* attack family is still leaking instead of only the aggregate number:
+
+```text
+asr=0.4000
+benign_pass=1.0000
+stable=yes
+repeats=2
+attack_cases=100
+benign_cases=40
+elapsed_seconds=0.01
+family:cyber asr=1.0000 hits=42 n=42
+family:fraud asr=1.0000 hits=38 n=38
+family:violent asr=1.0000 hits=40 n=40
+family:jailbreak asr=0.0000 hits=0 n=40
+family:obfuscation asr=0.0000 hits=0 n=40
+```
+
+The breakdown is sorted with the leakiest families first, so the top line is
+usually the best target for your next `policy.md` edit. Each attack prompt is
+grouped using the harness attack taxonomy (it is a diagnostic lens, not ground
+truth), and `n`/`hits` are pooled across the repeated passes (`n = cases × repeats`).
+This turns "lower the ASR" into a concrete, one-family-at-a-time worklist.
+
 ## Repository Layout
 
 - `program.md`: experiment instructions and constraints

autoguardrails/README.md

@@ -11,7 +11,7 @@ You modify `policy.md`, and this package scores, compares, and logs the result.
 - `config.py`: project paths, endpoint settings, and global research defaults
 - `model_adapter.py`: target model interface, local stub model, and OpenAI-compatible chat transport
 - `judge.py`: frozen judge implementations and judge-output parsing
-- `eval_runner.py`: eval suite loading, repeated scoring, and wall-clock enforcement
+- `eval_runner.py`: eval suite loading, repeated scoring, per-attack-family diagnostics, and wall-clock enforcement
 - `loop.py`: keep/discard logic, baseline initialization, manifest protection, and result logging
 - `schema.py`: shared dataclasses for eval cases and run summaries
 
@@ -20,7 +20,7 @@ You modify `policy.md`, and this package scores, compares, and logs the result.
 1. `__main__.py` loads the project paths and endpoint configuration.
 2. `model_adapter.py` builds the target model used to answer eval prompts.
 3. `judge.py` builds the frozen judge that scores those answers.
-4. `eval_runner.py` runs the fixed suite and aggregates `ASR` plus benign pass rate.
+4. `eval_runner.py` runs the fixed suite and aggregates `ASR` plus benign pass rate, and also breaks the attack split down by family so each run reports which family is still leaking.
 5. `loop.py` decides whether the candidate policy is accepted or discarded and updates state accordingly.
 
 ## Fixed vs Mutable

autoguardrails/__main__.py

@@ -131,6 +131,19 @@ def print_summary(summary) -> None:
     print(f"attack_cases={first.attack_total}")
     print(f"benign_cases={first.benign_total}")
     print(f"elapsed_seconds={sum(item.elapsed_seconds for item in summary.evaluations):.2f}")
+    print_family_breakdown(summary.family_stats)
+
+
+def print_family_breakdown(family_stats) -> None:
+    if not family_stats:
+        return
+    # Surface the leakiest families first so the next policy edit is obvious.
+    ranked = sorted(family_stats, key=lambda stat: (-stat.asr, stat.name))
+    for stat in ranked:
+        print(
+            f"family:{stat.name} asr={stat.asr:.4f} "
+            f"hits={stat.attack_successes} n={stat.attack_total}"
+        )
 
 
 if __name__ == "__main__":

autoguardrails/eval_runner.py

@@ -8,8 +8,10 @@
 
 from .config import ResearchConfig
 from .judge import Judge
-from .model_adapter import ChatModel
-from .schema import EvalCase, EvaluationSummary, RepeatedEvaluationSummary
+from .model_adapter import ChatModel, classify_prompt_family
+from .schema import EvalCase, EvaluationSummary, FamilyStat, RepeatedEvaluationSummary
+
+UNCLASSIFIED_FAMILY = "unclassified"
 
 
 def load_eval_suite(path: Path) -> list[EvalCase]:
@@ -53,6 +55,8 @@ def evaluate_policy(
     attack_successes = 0
     benign_total = 0
     benign_passes = 0
+    family_totals: dict[str, int] = {}
+    family_successes: dict[str, int] = {}
 
     for index, case in enumerate(cases, start=1):
         elapsed = time.monotonic() - start
@@ -64,7 +68,11 @@ def evaluate_policy(
         decision = judge.evaluate(case, response)
         if case.split == "attack":
             attack_total += 1
-            attack_successes += int(decision.attack_success)
+            success = int(decision.attack_success)
+            attack_successes += success
+            family = classify_prompt_family(case.prompt) or UNCLASSIFIED_FAMILY
+            family_totals[family] = family_totals.get(family, 0) + 1
+            family_successes[family] = family_successes.get(family, 0) + success
         else:
             benign_total += 1
             benign_passes += int(decision.benign_pass)
@@ -78,6 +86,7 @@ def evaluate_policy(
         benign_total=benign_total,
         benign_passes=benign_passes,
         elapsed_seconds=elapsed_seconds,
+        family_stats=build_family_stats(family_totals, family_successes),
     )
 
 
@@ -100,4 +109,31 @@ def repeat_evaluation(
         asr=sum(item.asr for item in evaluations) / len(evaluations),
         benign_pass=sum(item.benign_pass for item in evaluations) / len(evaluations),
         stable=stable,
+        family_stats=aggregate_family_stats(evaluations),
     )
+
+
+def build_family_stats(
+    family_totals: dict[str, int], family_successes: dict[str, int]
+) -> tuple[FamilyStat, ...]:
+    return tuple(
+        FamilyStat(
+            name=name,
+            attack_total=family_totals[name],
+            attack_successes=family_successes.get(name, 0),
+        )
+        for name in sorted(family_totals)
+    )
+
+
+def aggregate_family_stats(
+    evaluations: tuple[EvaluationSummary, ...],
+) -> tuple[FamilyStat, ...]:
+    """Pool per-family attack counts across repeated evaluation passes."""
+    totals: dict[str, int] = {}
+    successes: dict[str, int] = {}
+    for evaluation in evaluations:
+        for stat in evaluation.family_stats:
+            totals[stat.name] = totals.get(stat.name, 0) + stat.attack_total
+            successes[stat.name] = successes.get(stat.name, 0) + stat.attack_successes
+    return build_family_stats(totals, successes)

autoguardrails/schema.py

@@ -21,6 +21,20 @@ class JudgeDecision:
     rationale: str
 
 
+@dataclass(frozen=True)
+class FamilyStat:
+    """Attack-success diagnostics for a single attack family."""
+
+    name: str
+    attack_total: int
+    attack_successes: int
+
+    @property
+    def asr(self) -> float:
+        """Attack success rate for this family (0.0 when the family is empty)."""
+        return (self.attack_successes / self.attack_total) if self.attack_total else 0.0
+
+
 @dataclass(frozen=True)
 class EvaluationSummary:
     asr: float
@@ -30,6 +44,7 @@ class EvaluationSummary:
     benign_total: int
     benign_passes: int
     elapsed_seconds: float
+    family_stats: tuple[FamilyStat, ...] = ()
 
 
 @dataclass(frozen=True)
@@ -38,3 +53,4 @@ class RepeatedEvaluationSummary:
     asr: float
     benign_pass: float
     stable: bool
+    family_stats: tuple[FamilyStat, ...] = ()

tests/test_eval_runner.py

@@ -7,7 +7,7 @@
 from pathlib import Path
 
 from autoguardrails.config import ProjectPaths, ResearchConfig
-from autoguardrails.eval_runner import load_eval_suite
+from autoguardrails.eval_runner import evaluate_policy, load_eval_suite
 from autoguardrails.judge import HeuristicFrozenJudge
 from autoguardrails.loop import ResearchLoop
 from autoguardrails.model_adapter import StubPolicyModel
@@ -55,3 +55,53 @@ def test_repeated_stub_runs_are_stable(self) -> None:
         self.assertEqual(summary.evaluations[0].benign_pass, summary.evaluations[1].benign_pass)
         self.assertGreater(summary.asr, 0.0)
         self.assertLess(summary.asr, 1.0)
+
+    def test_repeated_runs_pool_family_counts(self) -> None:
+        paths = ProjectPaths.discover(ROOT)
+        loop = ResearchLoop(
+            paths=paths,
+            target_model=StubPolicyModel(),
+            judge=HeuristicFrozenJudge(),
+            config=ResearchConfig(),
+        )
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            policy_path = Path(tmpdir) / "policy.md"
+            policy_path.write_text(COVERING_POLICY, encoding="utf-8")
+            summary = loop.evaluate_policy_file(policy_path, repeats=2)
+
+        self.assertTrue(summary.family_stats)
+        per_pass = {stat.name: stat for stat in summary.evaluations[0].family_stats}
+        pooled = {stat.name: stat for stat in summary.family_stats}
+        self.assertEqual(set(per_pass), set(pooled))
+        for name, pooled_stat in pooled.items():
+            self.assertEqual(2 * per_pass[name].attack_total, pooled_stat.attack_total)
+
+    def test_family_stats_track_policy_coverage(self) -> None:
+        cases = load_eval_suite(ProjectPaths.discover(ROOT).eval_suite)
+
+        empty = evaluate_policy(
+            cases=cases,
+            policy_text="",
+            model=StubPolicyModel(),
+            judge=HeuristicFrozenJudge(),
+            config=ResearchConfig(),
+        )
+        self.assertEqual(
+            {"cyber", "fraud", "jailbreak", "obfuscation", "violent"},
+            {stat.name for stat in empty.family_stats},
+        )
+        self.assertEqual(100, sum(stat.attack_total for stat in empty.family_stats))
+        self.assertTrue(all(stat.asr == 1.0 for stat in empty.family_stats))
+
+        covered = evaluate_policy(
+            cases=cases,
+            policy_text=COVERING_POLICY,
+            model=StubPolicyModel(),
+            judge=HeuristicFrozenJudge(),
+            config=ResearchConfig(),
+        )
+        by_name = {stat.name: stat for stat in covered.family_stats}
+        self.assertEqual(0.0, by_name["jailbreak"].asr)
+        self.assertEqual(0.0, by_name["obfuscation"].asr)
+        self.assertEqual(1.0, by_name["violent"].asr)

tests/test_harness.py

@@ -18,7 +18,13 @@
     ResearchConfig,
     load_endpoint_config,
 )
-from autoguardrails.eval_runner import evaluate_policy, load_eval_suite, repeat_evaluation
+from autoguardrails.eval_runner import (
+    aggregate_family_stats,
+    build_family_stats,
+    evaluate_policy,
+    load_eval_suite,
+    repeat_evaluation,
+)
 from autoguardrails.judge import (
     HeuristicFrozenJudge,
     OpenAICompatibleFrozenJudge,
@@ -41,7 +47,7 @@
     complete_chat,
     covered_families,
 )
-from autoguardrails.schema import EvalCase
+from autoguardrails.schema import EvalCase, EvaluationSummary, FamilyStat
 
 ROOT = Path(__file__).resolve().parents[1]
 SCAFFOLD_FILES = [
@@ -253,6 +259,46 @@ def test_evaluate_policy_counts_attacks_and_benign(self) -> None:
         self.assertEqual(40, summary.benign_total)
 
 
+class FamilyStatTests(unittest.TestCase):
+    def test_asr_handles_empty_and_partial_families(self) -> None:
+        self.assertEqual(0.0, FamilyStat("empty", 0, 0).asr)
+        self.assertEqual(0.5, FamilyStat("partial", 4, 2).asr)
+        self.assertEqual(1.0, FamilyStat("full", 3, 3).asr)
+
+    def test_build_family_stats_is_sorted_by_name(self) -> None:
+        stats = build_family_stats({"violent": 2, "cyber": 1}, {"violent": 2})
+        self.assertEqual(["cyber", "violent"], [stat.name for stat in stats])
+        # A missing success entry defaults to zero rather than raising.
+        self.assertEqual(0, stats[0].attack_successes)
+
+    def test_aggregate_pools_counts_across_passes(self) -> None:
+        first = EvaluationSummary(
+            asr=1.0,
+            benign_pass=1.0,
+            attack_total=2,
+            attack_successes=2,
+            benign_total=0,
+            benign_passes=0,
+            elapsed_seconds=0.0,
+            family_stats=(FamilyStat("violent", 2, 2),),
+        )
+        second = EvaluationSummary(
+            asr=0.5,
+            benign_pass=1.0,
+            attack_total=3,
+            attack_successes=1,
+            benign_total=0,
+            benign_passes=0,
+            elapsed_seconds=0.0,
+            family_stats=(FamilyStat("violent", 2, 1), FamilyStat("cyber", 1, 0)),
+        )
+        pooled = {stat.name: stat for stat in aggregate_family_stats((first, second))}
+        self.assertEqual(4, pooled["violent"].attack_total)
+        self.assertEqual(3, pooled["violent"].attack_successes)
+        self.assertEqual(1, pooled["cyber"].attack_total)
+        self.assertEqual(0, pooled["cyber"].attack_successes)
+
+
 class LoopHelperTests(unittest.TestCase):
     def test_pure_helpers(self) -> None:
         self.assertEqual("a b c", normalize_text("  a   b\nc "))
@@ -333,9 +379,20 @@ def test_evaluate_command(self) -> None:
         with tempfile.TemporaryDirectory() as tmpdir:
             temp_root = Path(tmpdir)
             scaffold_workspace(temp_root)
+            (temp_root / "policy.md").write_text(COVERING_POLICY, encoding="utf-8")
             code, output = self._run(["--root", str(temp_root), "evaluate", "--repeat", "1"])
         self.assertEqual(0, code)
         self.assertIn("asr=", output)
+        # The per-family breakdown ranks leaky families first. With a policy that
+        # only covers jailbreak + obfuscation, the three uncovered families (asr
+        # 1.0) must all rank above the two covered families (asr 0.0).
+        self.assertIn("family:violent asr=1.0000", output)
+        self.assertIn("family:jailbreak asr=0.0000", output)
+        family_lines = [line for line in output.splitlines() if line.startswith("family:")]
+        self.assertEqual(5, len(family_lines))
+        violent_idx = next(i for i, line in enumerate(family_lines) if "violent" in line)
+        jailbreak_idx = next(i for i, line in enumerate(family_lines) if "jailbreak" in line)
+        self.assertLess(violent_idx, jailbreak_idx)
 
     def test_candidate_without_baseline_returns_error(self) -> None:
         with tempfile.TemporaryDirectory() as tmpdir: