refactor(verdict): unify read path on authoritative case_outcomes

Read paths recomputed per-case pass/connectivity independently — report
generation, the logs endpoint, and the frontend each derived it, and the
frontend's every(passed) recompute ignored the engine's authoritative
verdict. Extract resolve_case_verdicts: a single pure seam that prefers
stored case_outcomes verbatim and approximates only for legacy runs. The
logs endpoint now surfaces case_verdicts so the frontend reads instead of
recomputing.

backend/agenteval/evaluation/case_verdict.py

@@ -0,0 +1,67 @@
+"""Case-verdict read seam — the single place the read path derives pass/connectivity.
+
+The engine is the authority: it runs ``judgement.combine_case_outcome`` once and
+writes each case's verdict into ``summary.case_outcomes``. Every read surface
+(report generation, the run-logs endpoint) must present *that* verdict, never
+recompute it — otherwise WEIGHTED/ANY logic and connectivity cases diverge from
+what was judged.
+
+This module is that single seam. It reads the authoritative ``case_outcomes``
+when present, and only for older runs that predate it falls back to a documented
+approximation from persisted turns/results. Pure — no I/O; callers build the
+per-case ``CaseEvidence`` from whatever they already have in hand.
+"""
+
+from dataclasses import dataclass
+
+from agenteval.models import CaseOutcomeSummary
+
+
+@dataclass(frozen=True)
+class CaseEvidence:
+    """What the legacy approximation needs about one case's persisted record.
+
+    ``result_passes`` is the per-rule pass flags (empty means no judged rule
+    result exists for the case — the connectivity-vs-fault fork).
+    """
+
+    has_turns: bool
+    all_replied: bool
+    result_passes: tuple[bool, ...] = ()
+
+
+def resolve_case_verdicts(
+    *,
+    case_outcomes: dict[str, CaseOutcomeSummary],
+    evidence: dict[str, CaseEvidence],
+    errored_case_ids: set[str],
+) -> dict[str, CaseOutcomeSummary]:
+    """Resolve every case in ``evidence`` to its authoritative-or-approximated verdict.
+
+    Authoritative ``case_outcomes`` win verbatim. For a case missing from it (an
+    older run), approximate per CONTEXT.md / ADR-0002: a case with no judged
+    results but turns that all replied and no case-level error is a *connectivity*
+    case (counts as passed); a case with results passes iff every rule passed;
+    anything else (a fault) fails.
+    """
+    verdicts: dict[str, CaseOutcomeSummary] = {}
+    for case_id, ev in evidence.items():
+        authoritative = case_outcomes.get(case_id)
+        if authoritative is not None:
+            verdicts[case_id] = authoritative
+            continue
+
+        connectivity = (
+            not ev.result_passes
+            and ev.has_turns
+            and ev.all_replied
+            and case_id not in errored_case_ids
+        )
+        if connectivity:
+            passed = True
+        elif not ev.result_passes:
+            passed = False
+        else:
+            passed = all(ev.result_passes)
+        verdicts[case_id] = CaseOutcomeSummary(passed=passed, connectivity=connectivity)
+    return verdicts

backend/agenteval/evaluation/report.py

@@ -9,6 +9,7 @@ from datetime import datetime, timezone
 from pathlib import Path
 from typing import Any, Optional
 
+from agenteval.evaluation.case_verdict import CaseEvidence, resolve_case_verdicts
 from agenteval.evaluation.metrics import aggregate_runs
 from agenteval.evaluation.report_render import render_html, render_json, render_markdown
 from agenteval.models import Campaign, EvalRun, RunStatus, RunSummary
@@ -68,37 +69,31 @@ def generate_report(run_id: str, session=None) -> dict[str, Any]:
 
     summary = run.summary or RunSummary()
     errored_case_ids = {e.get("case_id") for e in summary.case_errors}
-    # 权威判定：引擎经 judgement.combine_case_outcome 算一次写入 summary；
+    # 权威判定：引擎经 combine_case_outcome 算一次写入 summary，读路径只读不重算。
-    # 老 run 没有该字段时退回从持久化结果反推（WEIGHTED/ANY 只能近似）。
+    # resolve_case_verdicts 统一处理「权威优先、老 run 近似回退」（唯一落点）。
-    authoritative = summary.case_outcomes
+    evidence = {
+        case_id: CaseEvidence(
+            has_turns=bool(item["turns"]),
+            all_replied=item["all_replied"],
+            result_passes=tuple(r["passed"] for r in item["results"]),
+        )
+        for case_id, item in case_map.items()
+    }
+    verdicts = resolve_case_verdicts(
+        case_outcomes=summary.case_outcomes,
+        evidence=evidence,
+        errored_case_ids=errored_case_ids,
+    )
 
     cases = []
     for case_id in sorted(case_map.keys()):
         item = case_map[case_id]
-        if case_id in authoritative:
+        verdict = verdicts[case_id]
-            outcome = authoritative[case_id]
-            connectivity = outcome.connectivity
-            passed = outcome.passed
-        else:
-            # 连通用例：无任何判定结果，且每轮都收到回复、无用例级错误（CONTEXT.md）
-            connectivity = (
-                not item["results"]
-                and bool(item["turns"])
-                and item["all_replied"]
-                and case_id not in errored_case_ids
-            )
-            if connectivity:
-                passed = True
-            elif not item["results"]:
-                # 故障用例（无结果且非连通）＝不通过（ADR-0002）
-                passed = False
-            else:
-                passed = all(r["passed"] for r in item["results"])
         cases.append(
             {
                 "case_id": case_id,
-                "passed": passed,
+                "passed": verdict.passed,
-                "connectivity": connectivity,
+                "connectivity": verdict.connectivity,
                 "turns": sorted(item["turns"], key=lambda x: x["round"]),
                 "results": item["results"],
             }

backend/agenteval/web/routers/runs.py

@@ -6,8 +6,9 @@ from fastapi import APIRouter, Depends, HTTPException
 from pydantic import BaseModel
 from sqlmodel import Session
 
+from agenteval.evaluation.case_verdict import CaseEvidence, resolve_case_verdicts
 from agenteval.evaluation.engine import EvalEngine
-from agenteval.models import EvalRun, RunStatus, RunTrigger
+from agenteval.models import EvalRun, RunStatus, RunSummary, RunTrigger
 from agenteval.storage.db import get_session, iso_utc
 from agenteval.storage.repository import RunRepository, ScenarioRepository, TargetRepository
 from agenteval.task_registry import TaskRegistry
@@ -169,6 +170,37 @@ async def get_run_logs(run_id: str, session: Session = Depends(get_db)) -> dict:
         for r in results
     ]
 
+    # Authoritative per-case verdicts: resolve_case_verdicts prefers the engine's
+    # stored case_outcomes and approximates only for legacy runs (single seam).
+    evidence: dict[str, dict] = {}
+    for t in turns:
+        ev = evidence.get(t.case_id)
+        if ev is None:
+            ev = {"has_turns": True, "all_replied": True, "passes": []}
+            evidence[t.case_id] = ev
+        else:
+            ev["has_turns"] = True
+        if t.get_reply() is None:
+            ev["all_replied"] = False
+    for r in results:
+        ev = evidence.setdefault(r.case_id, {"has_turns": False, "all_replied": True, "passes": []})
+        ev["passes"].append(r.passed)
+    summary = run.summary or RunSummary()
+    errored_case_ids = {e.get("case_id") for e in summary.case_errors}
+    verdicts = resolve_case_verdicts(
+        case_outcomes=summary.case_outcomes,
+        evidence={
+            cid: CaseEvidence(
+                has_turns=ev["has_turns"],
+                all_replied=ev["all_replied"],
+                result_passes=tuple(ev["passes"]),
+            )
+            for cid, ev in evidence.items()
+        },
+        errored_case_ids=errored_case_ids,
+    )
+    case_verdicts = {cid: {"passed": v.passed, "connectivity": v.connectivity} for cid, v in verdicts.items()}
+
     scenario_snapshot: dict = {}
     scenario = ScenarioRepository(session).get(run.scenario_id)
     if scenario:
@@ -191,4 +223,9 @@ async def get_run_logs(run_id: str, session: Session = Depends(get_db)) -> dict:
                 "rule_pass_threshold": case.rule_pass_threshold,
             }
 
-    return {"turns": turns_data, "results": results_data, "scenario_snapshot": scenario_snapshot}
+    return {
+        "turns": turns_data,
+        "results": results_data,
+        "case_verdicts": case_verdicts,
+        "scenario_snapshot": scenario_snapshot,
+    }

frontend/web/src/api.ts

@@ -278,6 +278,7 @@ export interface CaseSnapshot {
 export interface RunLogsResponse {
   turns: RunLogsTurn[]
   results: RunLogsResult[]
+  case_verdicts: Record<string, { passed: boolean; connectivity: boolean }>
   scenario_snapshot: Record<string, CaseSnapshot>
 }
 

frontend/web/src/hooks/useRunSession.ts

@@ -160,9 +160,10 @@ export function useRunSession(): RunSession {
           reason: r.reason,
         })
       }
+      const verdicts = data.case_verdicts ?? {}
       for (const cs of caseMap.values()) {
         cs.turns.sort((a, b) => a.roundIndex - b.roundIndex)
-        cs.passed = cs.ruleResults.length > 0 && cs.ruleResults.every((r) => r.passed)
+        cs.passed = verdicts[cs.caseId]?.passed ?? false
       }
       const cases = Array.from(caseMap.values())
       const total = cases.length

tests/unit/test_case_verdict.py

@@ -0,0 +1,105 @@
+"""Unit tests for resolve_case_verdicts — the single case-verdict read seam."""
+
+from agenteval.evaluation.case_verdict import CaseEvidence, resolve_case_verdicts
+from agenteval.models import CaseOutcomeSummary
+
+
+def _ev(has_turns=True, all_replied=True, result_passes=()):
+    return CaseEvidence(has_turns=has_turns, all_replied=all_replied, result_passes=tuple(result_passes))
+
+
+# ── authoritative wins ──────────────────────────────────────────────────────
+
+def test_authoritative_outcome_used_verbatim_ignoring_evidence():
+    # Evidence would approximate passed=True (all rules pass), but the authoritative
+    # verdict says failed (e.g. WEIGHTED/ANY logic) — authority must win.
+    verdicts = resolve_case_verdicts(
+        case_outcomes={"c1": CaseOutcomeSummary(passed=False, connectivity=False)},
+        evidence={"c1": _ev(result_passes=(True, True))},
+        errored_case_ids=set(),
+    )
+    assert verdicts["c1"] == CaseOutcomeSummary(passed=False, connectivity=False)
+
+
+def test_authoritative_connectivity_preserved():
+    verdicts = resolve_case_verdicts(
+        case_outcomes={"c1": CaseOutcomeSummary(passed=True, connectivity=True)},
+        evidence={"c1": _ev(result_passes=(False,))},
+        errored_case_ids=set(),
+    )
+    assert verdicts["c1"].connectivity is True
+    assert verdicts["c1"].passed is True
+
+
+# ── legacy approximation (case_outcomes missing) ────────────────────────────
+
+def test_legacy_connectivity_case_passes():
+    # No judged results, has turns, every turn replied, not errored → connectivity pass.
+    verdicts = resolve_case_verdicts(
+        case_outcomes={},
+        evidence={"c1": _ev(has_turns=True, all_replied=True, result_passes=())},
+        errored_case_ids=set(),
+    )
+    assert verdicts["c1"] == CaseOutcomeSummary(passed=True, connectivity=True)
+
+
+def test_legacy_all_rules_pass():
+    verdicts = resolve_case_verdicts(
+        case_outcomes={},
+        evidence={"c1": _ev(result_passes=(True, True))},
+        errored_case_ids=set(),
+    )
+    assert verdicts["c1"] == CaseOutcomeSummary(passed=True, connectivity=False)
+
+
+def test_legacy_some_rule_fails():
+    verdicts = resolve_case_verdicts(
+        case_outcomes={},
+        evidence={"c1": _ev(result_passes=(True, False))},
+        errored_case_ids=set(),
+    )
+    assert verdicts["c1"] == CaseOutcomeSummary(passed=False, connectivity=False)
+
+
+def test_legacy_fault_no_results_not_all_replied():
+    # No results and a turn missing its reply → fault, not connectivity → fail (ADR-0002).
+    verdicts = resolve_case_verdicts(
+        case_outcomes={},
+        evidence={"c1": _ev(has_turns=True, all_replied=False, result_passes=())},
+        errored_case_ids=set(),
+    )
+    assert verdicts["c1"] == CaseOutcomeSummary(passed=False, connectivity=False)
+
+
+def test_legacy_errored_case_not_connectivity():
+    # A case with a case-level error is not connectivity even if it replied.
+    verdicts = resolve_case_verdicts(
+        case_outcomes={},
+        evidence={"c1": _ev(has_turns=True, all_replied=True, result_passes=())},
+        errored_case_ids={"c1"},
+    )
+    assert verdicts["c1"] == CaseOutcomeSummary(passed=False, connectivity=False)
+
+
+def test_legacy_no_turns_no_results_is_fault():
+    verdicts = resolve_case_verdicts(
+        case_outcomes={},
+        evidence={"c1": _ev(has_turns=False, all_replied=True, result_passes=())},
+        errored_case_ids=set(),
+    )
+    assert verdicts["c1"] == CaseOutcomeSummary(passed=False, connectivity=False)
+
+
+# ── mixed authoritative + legacy in one call ────────────────────────────────
+
+def test_mixed_authoritative_and_legacy():
+    verdicts = resolve_case_verdicts(
+        case_outcomes={"auth": CaseOutcomeSummary(passed=True, connectivity=False)},
+        evidence={
+            "auth": _ev(result_passes=(False,)),          # authority overrides
+            "legacy": _ev(result_passes=(True,)),          # approximated
+        },
+        errored_case_ids=set(),
+    )
+    assert verdicts["auth"] == CaseOutcomeSummary(passed=True, connectivity=False)
+    assert verdicts["legacy"] == CaseOutcomeSummary(passed=True, connectivity=False)