refactor(campaign): type Campaign.summary as CampaignSummary VO

Campaign.summary was a bare Optional[dict] while RunSummary is a typed VO —
scheduler state (spawned_indices/errors) flowed untyped through
set_/get_summary. Introduce CampaignSummary + SchedulerState (extra=allow,
validate_assignment), mirroring RunSummary; campaign_runner reads/writes the
VO. Also converge the ~10 repeated JSON column get/set pairs onto
_json_dumps/_json_loads helpers, unifying ensure_ascii=False and fixing the
set_modalities ensure_ascii=True trap.

backend/agenteval/evaluation/campaign_runner.py

@@ -29,7 +29,15 @@ from agenteval.evaluation.campaign_scheduler import (
     resolve_finalize,
 )
 from agenteval.evaluation.engine import EvalEngine
-from agenteval.models import Campaign, CampaignStatus, EvalRun, RunStatus, RunTrigger
+from agenteval.models import (
+    Campaign,
+    CampaignStatus,
+    CampaignSummary,
+    EvalRun,
+    RunStatus,
+    RunTrigger,
+    SchedulerState,
+)
 from agenteval.storage.db import get_session, utc_now
 from agenteval.storage.repository import (
     CampaignRepository,
@@ -39,9 +47,6 @@ from agenteval.storage.repository import (
 )
 from agenteval.task_registry import TaskRegistry
 
-_SPAWNED_KEY = "spawned_indices"
-_ERRORS_KEY = "errors"
-
 # Real wall-clock seconds between scheduler ticks. time_scale compresses the
 # *window*, not the tick cadence — production 24h campaigns still tick slowly.
 DEFAULT_TICK_SECONDS = 1.0
@@ -62,8 +67,9 @@ class AdvanceResult:
 
 
 def _spawned_indices(campaign: Campaign) -> set[int]:
-    scheduler = (campaign.summary or {}).get("scheduler", {})
+    if campaign.summary is None:
-    return set(scheduler.get(_SPAWNED_KEY, []))
+        return set()
+    return set(campaign.summary.scheduler.spawned_indices)
 
 
 def current_window_offset(campaign: Campaign) -> float:
@@ -152,7 +158,7 @@ async def advance_campaign(
     )
 
     result = AdvanceResult(finished=decision.finished)
-    errors: list[dict] = list((campaign.summary or {}).get("scheduler", {}).get(_ERRORS_KEY, []))
+    errors: list[dict] = list(campaign.summary.scheduler.errors) if campaign.summary else []
     for due in decision.due:
         # Cancellation stops further spawning; runs already in flight finish.
         if cancel_event is not None and cancel_event.is_set():
@@ -167,12 +173,10 @@ async def advance_campaign(
         spawned.add(due.index)
         # Persist progress per entry: a failure partway through a multi-entry
         # advance must never lose which entries already spawned, since restart
-        # recovery reads this back from the DB.
+        # recovery reads this back from the DB. Mutate the existing summary so
-        scheduler_state: dict = {_SPAWNED_KEY: sorted(spawned)}
+        # any unknown top-level keys survive the read-modify-write.
-        if errors:
+        summary = campaign.summary or CampaignSummary()
-            scheduler_state[_ERRORS_KEY] = errors
+        summary.scheduler = SchedulerState(spawned_indices=sorted(spawned), errors=errors)
-        summary = dict(campaign.summary or {})
-        summary["scheduler"] = scheduler_state
         campaign.summary = summary
         repo.update(campaign)
 

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/engine.py

@@ -8,13 +8,16 @@ via an ``asyncio.Event`` cancel token.
 import asyncio
 import uuid
 from dataclasses import dataclass
+from datetime import datetime
 from typing import Any, Callable, Optional
 
 from agenteval.channels.base import EvalChannel
 from agenteval.channels.factory import ChannelFactory
 from agenteval.config import get_settings
+from agenteval.evaluation.implicit_rules import derive_implicit_rules
 from agenteval.evaluation.judgement import CaseOutcome, RuleOutcome, combine_case_outcome
 from agenteval.evaluation.rules import RuleResult, get_rule
+from agenteval.evaluation.run_summary import build_run_summary
 from agenteval.model_gateway import ModelGateway
 from agenteval.models import (
     Case,
@@ -58,6 +61,14 @@ def _build_send_message(content: str) -> dict[str, Any]:
     }
 
 
+# 需要模型资源的规则类型 → 评测岗位（ModelPurpose）；其余规则无需模型
+RULE_PURPOSE = {
+    "llm_score": ModelPurpose.JUDGE,
+    "semantic_similarity": ModelPurpose.EMBEDDING,
+    "safety": ModelPurpose.MODERATION,
+}
+
+
 class EvalEngine:
     """Execute evaluation scenarios against targets.
 
@@ -130,9 +141,7 @@ class EvalEngine:
 
         try:
             total_cases = len(self.scenario.cases)
-            passed_cases = 0
+            case_outcomes: dict[str, CaseOutcome] = {}
-            failed_cases = 0
-            case_outcomes: dict[str, dict[str, bool]] = {}
 
             for idx, case in enumerate(self.scenario.cases, start=1):
                 self._check_cancel()
@@ -151,11 +160,7 @@ class EvalEngine:
                         case,
                         progress_callback,
                     )
-                case_outcomes[case.id] = {"passed": outcome.passed, "connectivity": outcome.connectivity}
+                case_outcomes[case.id] = outcome
-                if outcome.passed:
-                    passed_cases += 1
-                else:
-                    failed_cases += 1
                 await self._emit(
                     progress_callback,
                     "case_end",
@@ -170,34 +175,18 @@ class EvalEngine:
                 )
 
             results = self.run_repo.get_results(run.id)
-            total_rules = len(results)
-            passed_rules = sum(1 for r in results if r.passed)
-
             turns = self.run_repo.get_turns(run.id)
-            latencies = [t.latency_ms for t in turns if t.latency_ms is not None]
+            summary = build_run_summary(
-            avg_latency_ms = round(sum(latencies) / len(latencies), 1) if latencies else None
+                case_outcomes=case_outcomes,
-
+                latencies=[t.latency_ms for t in turns if t.latency_ms is not None],
-            summary = {
+                rule_passes=[r.passed for r in results],
-                "total_cases": total_cases,
+                case_errors=self._case_errors or None,
-                "passed_cases": passed_cases,
+                model_configs=(
-                "failed_cases": failed_cases,
+                    {purpose.value: config.snapshot() for purpose, config in self._resolved_models.items()}
-                "total_rules": total_rules,
+                    if self._resolved_models
-                "passed_rules": passed_rules,
+                    else None
-                # 通过率是用例级口径（CONTEXT.md）；规则级数字保留在 passed_rules/total_rules
+                ),
-                "pass_rate": round(passed_cases / total_cases, 4) if total_cases else 0.0,
+            )
-                # 平均时延（毫秒），供活动周期报告的时延轴聚合；无回复轮不计入
-                "avg_latency_ms": avg_latency_ms,
-                # 逐用例权威判定（judgement.py 算一次），报告/对比/渲染层只读不重算
-                "case_outcomes": case_outcomes,
-            }
-            # Surface fatal case-level errors (e.g. dynamic generation failures)
-            # so the report / DB record shows *why* a run produced no results.
-            if self._case_errors:
-                summary["case_errors"] = self._case_errors
-            if self._resolved_models:
-                summary["model_configs"] = {
-                    purpose.value: config.snapshot() for purpose, config in self._resolved_models.items()
-                }
             run.status = RunStatus.COMPLETED
             run.completed_at = utc_now()
             run.summary = summary
@@ -206,7 +195,7 @@ class EvalEngine:
                 "run_completed",
                 {
                     "status": "completed",
-                    "summary": summary,
+                    "summary": summary.model_dump(),
                 },
             )
         except CancelledError:
@@ -258,6 +247,36 @@ class EvalEngine:
 
     # ── case / turn execution ─────────────────────────────────────────
 
+    def _persist_turn(
+        self,
+        run: EvalRun,
+        case: Case,
+        round_index: int,
+        message: str,
+        sent_at: datetime,
+        *,
+        question_msg_id: Optional[str] = None,
+        reply: Optional[dict] = None,
+        received_at: Optional[datetime] = None,
+        latency_ms: Optional[int] = None,
+    ) -> Turn:
+        """Build, persist, and return a Turn. The three call sites (send-fail /
+        poll-except / happy path) differ only in which optional fields are set."""
+        turn = Turn(
+            id=str(uuid.uuid4()),
+            run_id=run.id,
+            case_id=case.id,
+            round_index=round_index,
+            sent_message=_build_send_message(message),
+            sent_at=sent_at,
+            question_msg_id=question_msg_id,
+            reply=reply,
+            received_at=received_at,
+            latency_ms=latency_ms,
+        )
+        self.result_repo.save_turn(turn)
+        return turn
+
     async def _run_case(
         self,
         run: EvalRun,
@@ -298,15 +317,7 @@ class EvalEngine:
             sent_at = utc_now()
             send_result = await self.channel.send(message)
             if not send_result.ok:
-                turn = Turn(
+                turn = self._persist_turn(run, case, round_index, message, sent_at)
-                    id=str(uuid.uuid4()),
-                    run_id=run.id,
-                    case_id=case.id,
-                    round_index=round_index,
-                    sent_message=_build_send_message(message),
-                    sent_at=sent_at,
-                )
-                self.result_repo.save_turn(turn)
                 await self._save_rule_results(run, case, turn, [], progress_callback)
                 await self._emit(
                     progress_callback,
@@ -326,17 +337,15 @@ class EvalEngine:
                 )
             except Exception as poll_exc:
                 received_at = utc_now()
-                turn = Turn(
+                turn = self._persist_turn(
-                    id=str(uuid.uuid4()),
+                    run,
-                    run_id=run.id,
+                    case,
-                    case_id=case.id,
+                    round_index,
-                    round_index=round_index,
+                    message,
-                    sent_message=_build_send_message(message),
+                    sent_at,
-                    sent_at=sent_at,
                     question_msg_id=send_result.question_msg_id,
                     received_at=received_at,
                 )
-                self.result_repo.save_turn(turn)
                 await self._emit(
                     progress_callback,
                     "turn_error",
@@ -353,19 +362,17 @@ class EvalEngine:
             if sent_at and received_at:
                 latency_ms = int((received_at - sent_at).total_seconds() * 1000)
 
-            turn = Turn(
+            turn = self._persist_turn(
-                id=str(uuid.uuid4()),
+                run,
-                run_id=run.id,
+                case,
-                case_id=case.id,
+                round_index,
-                round_index=round_index,
+                message,
-                sent_message=_build_send_message(message),
+                sent_at,
-                sent_at=sent_at,
                 question_msg_id=send_result.question_msg_id,
                 reply=reply.raw_message if reply else None,
                 received_at=received_at,
                 latency_ms=latency_ms,
             )
-            self.result_repo.save_turn(turn)
             dialog.append(turn)
 
             await self._emit(
@@ -401,27 +408,7 @@ class EvalEngine:
         from agenteval.models import EvalRuleConfig
 
         rules_config: list[EvalRuleConfig] = list(case.eval_rules)
-
+        implicit_config = derive_implicit_rules(case.expectations)
-        implicit_config: list[EvalRuleConfig] = []
-        if case.expectations.response_time_max_ms:
-            implicit_config.append(
-                EvalRuleConfig(
-                    type="response_time",
-                    params={
-                        "max_ms": case.expectations.response_time_max_ms,
-                    },
-                )
-            )
-        if case.expectations.keywords_include or case.expectations.keywords_exclude:
-            implicit_config.append(
-                EvalRuleConfig(
-                    type="keyword_match",
-                    params={
-                        "keywords": case.expectations.keywords_include,
-                        "exclude_keywords": case.expectations.keywords_exclude,
-                    },
-                )
-            )
 
         all_replied = bool(dialog) and all(t.reply is not None for t in dialog)
 
@@ -436,11 +423,7 @@ class EvalEngine:
 
         all_rules = [(cfg, False) for cfg in rules_config] + [(cfg, True) for cfg in implicit_config]
         for rule_config, is_implicit in all_rules:
-            purpose = {
+            purpose = RULE_PURPOSE.get(rule_config.type)
-                "llm_score": ModelPurpose.JUDGE,
-                "semantic_similarity": ModelPurpose.EMBEDDING,
-                "safety": ModelPurpose.MODERATION,
-            }.get(rule_config.type)
             try:
                 model_config = self._resolve_model(purpose) if purpose else None
                 rule = get_rule(

backend/agenteval/evaluation/implicit_rules.py

@@ -0,0 +1,30 @@
+"""Translate a case's期望 (Expectation) into implicit评估规则.
+
+期望描述"想要什么"，评估规则是"怎么判定"——期望派生的隐式规则与显式规则
+叠加生效（CONTEXT.md）。此翻译是纯逻辑，独立于规则执行与持久化，便于单测。
+"""
+
+from agenteval.models import EvalRuleConfig, Expectation
+
+
+def derive_implicit_rules(expectation: Expectation) -> list[EvalRuleConfig]:
+    """Build the implicit rule configs a case's expectation implies."""
+    rules: list[EvalRuleConfig] = []
+    if expectation.response_time_max_ms:
+        rules.append(
+            EvalRuleConfig(
+                type="response_time",
+                params={"max_ms": expectation.response_time_max_ms},
+            )
+        )
+    if expectation.keywords_include or expectation.keywords_exclude:
+        rules.append(
+            EvalRuleConfig(
+                type="keyword_match",
+                params={
+                    "keywords": expectation.keywords_include,
+                    "exclude_keywords": expectation.keywords_exclude,
+                },
+            )
+        )
+    return rules

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"],
             }
@@ -108,8 +103,10 @@ def generate_report(run_id: str, session=None) -> dict[str, Any]:
     passed_cases = summary.passed_cases
     connectivity_count = sum(1 for c in cases if c["connectivity"])
     judged_total = total_cases - connectivity_count
-    # 连通用例按引擎口径计通过，判定型通过数 = 总通过数 - 连通用例数
+    # 判定型通过率由 build_run_summary 入库，读路径只读；老 run 缺字段时按同一口径回退近似
-    judged_pass_rate = round((passed_cases - connectivity_count) / judged_total, 4) if judged_total > 0 else None
+    judged_pass_rate = summary.judged_pass_rate
+    if judged_pass_rate is None and judged_total > 0:
+        judged_pass_rate = round((passed_cases - connectivity_count) / judged_total, 4)
 
     return {
         "run_id": run.id,

backend/agenteval/evaluation/run_summary.py

@@ -0,0 +1,53 @@
+"""Single-run summary aggregation — the pure落点 for one run's口径.
+
+Parallel to ``metrics.aggregate_runs`` (cross-run) and
+``judgement.combine_case_outcome`` (case-level): given the authoritative
+per-case outcomes plus raw latency/rule material, compute the run's
+summary口径 once. No IO — the engine collects material and calls this; DB
+writes and event emits stay in the caller. See CONTEXT.md (通过率) / ADR-0002.
+"""
+
+from typing import Any, Optional, Sequence
+
+from agenteval.evaluation.judgement import CaseOutcome
+from agenteval.models import CaseOutcomeSummary, RunSummary
+
+
+def build_run_summary(
+    *,
+    case_outcomes: dict[str, CaseOutcome],
+    latencies: Sequence[float],
+    rule_passes: Sequence[bool],
+    case_errors: Optional[list[dict[str, str]]] = None,
+    model_configs: Optional[dict[str, Any]] = None,
+) -> RunSummary:
+    """Compute a run's summary口径 from its authoritative case outcomes."""
+    total_cases = len(case_outcomes)
+    passed_cases = sum(1 for o in case_outcomes.values() if o.passed)
+    connectivity_count = sum(1 for o in case_outcomes.values() if o.connectivity)
+
+    pass_rate = round(passed_cases / total_cases, 4) if total_cases else 0.0
+    # 连通用例按引擎口径计通过，判定型通过数 = 总通过数 - 连通用例数
+    judged_total = total_cases - connectivity_count
+    judged_pass_rate = (
+        round((passed_cases - connectivity_count) / judged_total, 4) if judged_total > 0 else None
+    )
+
+    avg_latency_ms = round(sum(latencies) / len(latencies), 1) if latencies else None
+
+    return RunSummary(
+        total_cases=total_cases,
+        passed_cases=passed_cases,
+        failed_cases=total_cases - passed_cases,
+        total_rules=len(rule_passes),
+        passed_rules=sum(1 for p in rule_passes if p),
+        pass_rate=pass_rate,
+        judged_pass_rate=judged_pass_rate,
+        avg_latency_ms=avg_latency_ms,
+        case_outcomes={
+            case_id: CaseOutcomeSummary(passed=o.passed, connectivity=o.connectivity)
+            for case_id, o in case_outcomes.items()
+        },
+        case_errors=case_errors or [],
+        model_configs=model_configs or {},
+    )

backend/agenteval/models.py

@@ -187,6 +187,8 @@ class RunSummary(BaseModel):
     passed_rules: int = 0
     # 用例级通过率，含执行失败（ADR-0002）；失败/取消的 run 无此值
     pass_rate: Optional[float] = None
+    # 判定型通过率：连通用例从分子分母双双剔除；无判定型用例时为空
+    judged_pass_rate: Optional[float] = None
     avg_latency_ms: Optional[float] = None
     case_outcomes: dict[str, CaseOutcomeSummary] = Field(default_factory=dict)
     case_errors: list[dict[str, str]] = Field(default_factory=list)
@@ -244,10 +246,37 @@ class CampaignPlanEntry(BaseModel):
     count: int = Field(default=1, ge=1)
 
 
+class SchedulerState(BaseModel):
+    """Durable scheduler progress for a campaign — restart-safe (ADR-0003).
+
+    ``spawned_indices`` are the plan entries already派生 into child Runs;
+    ``errors`` records entries whose spawn failed (marked spawned to avoid
+    infinite retry).
+    """
+
+    spawned_indices: list[int] = Field(default_factory=list)
+    errors: list[dict[str, Any]] = Field(default_factory=list)
+
+
+class CampaignSummary(BaseModel):
+    """Typed value of ``Campaign.summary`` — mirrors RunSummary's treatment.
+
+    Unknown top-level keys are preserved (extra=allow) so summaries written by
+    older versions keep parsing and survive read-modify-write.
+    """
+
+    model_config = {"extra": "allow"}
+
+    scheduler: SchedulerState = Field(default_factory=SchedulerState)
+
+
 class Campaign(BaseModel):
     """An evaluation campaign: a service-cycle window over a single target,
     driving many child Runs from a static plan (ADR-0003)."""
 
+    # summary 以属性赋值写入（scheduler loop）；赋值时即校验成 CampaignSummary
+    model_config = {"validate_assignment": True}
+
     id: Optional[str] = None
     name: str
     target_id: str
@@ -258,7 +287,7 @@ class Campaign(BaseModel):
     started_at: Optional[datetime] = None
     completed_at: Optional[datetime] = None
     created_at: Optional[datetime] = None
-    summary: Optional[dict[str, Any]] = None
+    summary: Optional[CampaignSummary] = None
 
 
 class Turn(BaseModel):

backend/agenteval/storage/db.py

@@ -44,6 +44,16 @@ def new_uuid() -> str:
     return str(uuid.uuid4())
 
 
+def _json_dumps(value: Any) -> str:
+    """Serialize a JSON column value. ensure_ascii=False keeps CJK readable
+    in the stored text — the single serialization口径 for all JSON columns."""
+    return json.dumps(value, ensure_ascii=False)
+
+
+def _json_loads(raw: str) -> Any:
+    return json.loads(raw)
+
+
 class EvalTargetDB(SQLModel, table=True):
     """Database table for evaluation targets."""
 
@@ -65,10 +75,10 @@ class EvalTargetDB(SQLModel, table=True):
     )
 
     def get_config(self) -> dict[str, Any]:
-        return json.loads(self.channel_config)
+        return _json_loads(self.channel_config)
 
     def set_config(self, config: dict[str, Any]) -> None:
-        self.channel_config = json.dumps(config, ensure_ascii=False)
+        self.channel_config = _json_dumps(config)
 
 
 class ScenarioDB(SQLModel, table=True):
@@ -92,22 +102,22 @@ class ScenarioDB(SQLModel, table=True):
     )
 
     def get_tags(self) -> list[str]:
-        return json.loads(self.tags)
+        return _json_loads(self.tags)
 
     def set_tags(self, tags: list[str]) -> None:
-        self.tags = json.dumps(tags, ensure_ascii=False)
+        self.tags = _json_dumps(tags)
 
     def get_cases(self) -> list[dict[str, Any]]:
-        return json.loads(self.cases)
+        return _json_loads(self.cases)
 
     def set_cases(self, cases: list[dict[str, Any]]) -> None:
-        self.cases = json.dumps(cases, ensure_ascii=False)
+        self.cases = _json_dumps(cases)
 
     def get_llm_config(self) -> Optional[dict[str, Any]]:
-        return json.loads(self.llm_config) if self.llm_config else None
+        return _json_loads(self.llm_config) if self.llm_config else None
 
     def set_llm_config(self, config: Optional[dict[str, Any]]) -> None:
-        self.llm_config = json.dumps(config, ensure_ascii=False) if config else None
+        self.llm_config = _json_dumps(config) if config else None
 
 
 class ModelConfigDB(SQLModel, table=True):
@@ -140,14 +150,14 @@ class ModelConfigDB(SQLModel, table=True):
     updated_at: Optional[datetime] = Field(default_factory=utc_now)
 
     def get_input_modalities(self) -> list[str]:
-        return json.loads(self.input_modalities)
+        return _json_loads(self.input_modalities)
 
     def get_output_modalities(self) -> list[str]:
-        return json.loads(self.output_modalities)
+        return _json_loads(self.output_modalities)
 
     def set_modalities(self, input_modalities: list[str], output_modalities: list[str]) -> None:
-        self.input_modalities = json.dumps(input_modalities, ensure_ascii=True)
+        self.input_modalities = _json_dumps(input_modalities)
-        self.output_modalities = json.dumps(output_modalities, ensure_ascii=True)
+        self.output_modalities = _json_dumps(output_modalities)
 
 
 class ScenarioModelBindingDB(SQLModel, table=True):
@@ -178,16 +188,16 @@ class CampaignDB(SQLModel, table=True):
     summary: Optional[str] = None
 
     def get_plan(self) -> list[dict[str, Any]]:
-        return json.loads(self.plan)
+        return _json_loads(self.plan)
 
     def set_plan(self, plan: list[dict[str, Any]]) -> None:
-        self.plan = json.dumps(plan, ensure_ascii=False)
+        self.plan = _json_dumps(plan)
 
     def get_summary(self) -> Optional[dict[str, Any]]:
-        return json.loads(self.summary) if self.summary else None
+        return _json_loads(self.summary) if self.summary else None
 
     def set_summary(self, summary: dict[str, Any]) -> None:
-        self.summary = json.dumps(summary, ensure_ascii=False)
+        self.summary = _json_dumps(summary)
 
 
 class EvalRunDB(SQLModel, table=True):
@@ -218,10 +228,10 @@ class EvalRunDB(SQLModel, table=True):
     )
 
     def get_summary(self) -> Optional[dict[str, Any]]:
-        return json.loads(self.summary) if self.summary else None
+        return _json_loads(self.summary) if self.summary else None
 
     def set_summary(self, summary: dict[str, Any]) -> None:
-        self.summary = json.dumps(summary, ensure_ascii=False)
+        self.summary = _json_dumps(summary)
 
 
 class TurnDB(SQLModel, table=True):
@@ -243,16 +253,16 @@ class TurnDB(SQLModel, table=True):
     run: Optional[EvalRunDB] = Relationship(back_populates="turns")
 
     def get_sent_message(self) -> dict[str, Any]:
-        return json.loads(self.sent_message)
+        return _json_loads(self.sent_message)
 
     def set_sent_message(self, message: dict[str, Any]) -> None:
-        self.sent_message = json.dumps(message, ensure_ascii=False)
+        self.sent_message = _json_dumps(message)
 
     def get_reply(self) -> Optional[dict[str, Any]]:
-        return json.loads(self.reply) if self.reply else None
+        return _json_loads(self.reply) if self.reply else None
 
     def set_reply(self, reply: Optional[dict[str, Any]]) -> None:
-        self.reply = json.dumps(reply, ensure_ascii=False) if reply else None
+        self.reply = _json_dumps(reply) if reply else None
 
 
 class EvalResultDB(SQLModel, table=True):

backend/agenteval/storage/repository.py

@@ -1,6 +1,6 @@
 """Repository layer for database access."""
 
-from typing import Optional
+from typing import Generic, Optional, TypeVar
 
 from sqlmodel import Session, select
 
@@ -18,131 +18,56 @@ from agenteval.storage.db import (
 )
 from agenteval.storage.model_config_repository import ScenarioModelBindingRepository
 
-
+M = TypeVar("M")  # domain model
-def _target_to_db(target: EvalTarget) -> EvalTargetDB:
+DB = TypeVar("DB")  # persisted table row
-    db = EvalTargetDB(
-        id=target.id,
-        name=target.name,
-        description=target.description,
-        platform=target.platform.value,
-        channel_type=target.channel_type.value,
-        status=target.status.value,
-        created_at=target.created_at,
-        updated_at=target.updated_at or utc_now(),
-    )
-    db.set_config(target.channel_config)
-    return db
 
 
-def _target_from_db(db: EvalTargetDB) -> EvalTarget:
+class BaseRepository(Generic[M, DB]):
-    return EvalTarget(
+    """Shared CRUD skeleton for id-keyed entity repositories.
-        id=db.id,
-        name=db.name,
-        description=db.description,
-        platform=db.platform,
-        channel_type=db.channel_type,
-        channel_config=db.get_config(),
-        status=db.status,
-        created_at=db.created_at,
-        updated_at=db.updated_at,
-    )
 
+    Subclasses declare the table (``_table``) and the ``list_all`` ordering
+    column name (``_order_by``, newest-first), and implement the ``_to_db`` /
+    ``_from_db`` converter pair. The converters are instance methods so a
+    subclass whose ``_from_db`` needs cross-table reads (e.g. Scenario's model
+    bindings) can reach ``self.session``. Entities with bespoke create/update
+    (binding validation, versioning) override just those methods.
+    """
 
-def _scenario_to_db(scenario: Scenario) -> ScenarioDB:
+    _table: type
-    db = ScenarioDB(
+    _order_by: str
-        id=scenario.id,
-        name=scenario.name,
-        description=scenario.description,
-        created_at=scenario.created_at,
-        updated_at=scenario.updated_at or utc_now(),
-    )
-    db.set_tags(scenario.tags)
-    # mode="json" 与 update() 的考纲比较保持同一序列化形态，避免假升版
-    db.set_cases([case.model_dump(mode="json") for case in scenario.cases])
-    db.set_llm_config(scenario.llm_config)
-    return db
 
+    def __init__(self, session: Optional[Session] = None):
+        self.session = session or get_session()
 
-def _scenario_from_db(db: ScenarioDB, session: Session) -> Scenario:
+    def _to_db(self, obj: M) -> DB:
-    bindings = ScenarioModelBindingRepository(session).get_for_scenario(db.id or "")
+        raise NotImplementedError
-    return Scenario(
-        id=db.id,
-        name=db.name,
-        description=db.description,
-        tags=db.get_tags(),
-        cases=[Case(**case) for case in db.get_cases()],
-        model_bindings=bindings,
-        llm_config=db.get_llm_config(),
-        version=db.version or 1,
-        created_at=db.created_at,
-        updated_at=db.updated_at,
-    )
 
+    def _from_db(self, db: DB) -> M:
+        raise NotImplementedError
 
-def _run_to_db(run: EvalRun) -> EvalRunDB:
+    def list_all(self) -> list[M]:
-    db = EvalRunDB(
+        column = getattr(self._table, self._order_by)
-        id=run.id,
+        statement = select(self._table).order_by(column.desc())
-        target_id=run.target_id,
+        return [self._from_db(r) for r in self.session.exec(statement).all()]
-        scenario_id=run.scenario_id,
-        scenario_version=run.scenario_version,
-        campaign_id=run.campaign_id,
-        status=run.status.value,
-        triggered_by=run.triggered_by.value,
-        started_at=run.started_at,
-        completed_at=run.completed_at,
-    )
-    if run.summary is not None:
-        db.set_summary(run.summary.model_dump(mode="json"))
-    return db
 
+    def get(self, entity_id: str) -> Optional[M]:
+        db = self.session.get(self._table, entity_id)
+        return self._from_db(db) if db else None
 
-def _run_from_db(db: EvalRunDB) -> EvalRun:
+    def create(self, obj: M) -> M:
-    return EvalRun(
+        db = self._to_db(obj)
-        id=db.id,
+        self.session.add(db)
-        target_id=db.target_id,
+        self.session.commit()
-        scenario_id=db.scenario_id,
+        self.session.refresh(db)
-        scenario_version=db.scenario_version or 1,
+        return self._from_db(db)
-        campaign_id=db.campaign_id,
-        status=db.status,
-        triggered_by=db.triggered_by or "manual",
-        started_at=db.started_at,
-        completed_at=db.completed_at,
-        summary=db.get_summary(),
-    )
 
-
+    def delete(self, entity_id: str) -> bool:
-def _campaign_to_db(campaign: Campaign) -> CampaignDB:
+        db = self.session.get(self._table, entity_id)
-    db = CampaignDB(
+        if not db:
-        id=campaign.id,
+            return False
-        name=campaign.name,
+        self.session.delete(db)
-        target_id=campaign.target_id,
+        self.session.commit()
-        window_seconds=campaign.window_seconds,
+        return True
-        time_scale=campaign.time_scale,
-        status=campaign.status.value,
-        started_at=campaign.started_at,
-        completed_at=campaign.completed_at,
-        created_at=campaign.created_at,
-    )
-    db.set_plan([entry.model_dump(mode="json") for entry in campaign.plan])
-    if campaign.summary:
-        db.set_summary(campaign.summary)
-    return db
-
-
-def _campaign_from_db(db: CampaignDB) -> Campaign:
-    return Campaign(
-        id=db.id,
-        name=db.name,
-        target_id=db.target_id,
-        window_seconds=db.window_seconds,
-        time_scale=db.time_scale,
-        plan=db.get_plan(),
-        status=db.status,
-        started_at=db.started_at,
-        completed_at=db.completed_at,
-        created_at=db.created_at,
-        summary=db.get_summary(),
-    )
 
 
 def _result_to_db(result: EvalResult) -> EvalResultDB:
@@ -171,26 +96,38 @@ def _result_from_db(db: EvalResultDB) -> EvalResult:
     )
 
 
-class TargetRepository:
+class TargetRepository(BaseRepository[EvalTarget, EvalTargetDB]):
     """Repository for evaluation targets."""
 
-    def __init__(self, session: Optional[Session] = None):
+    _table = EvalTargetDB
-        self.session = session or get_session()
+    _order_by = "created_at"
 
-    def list_all(self) -> list[EvalTarget]:
+    def _to_db(self, target: EvalTarget) -> EvalTargetDB:
-        statement = select(EvalTargetDB).order_by(EvalTargetDB.created_at.desc())
+        db = EvalTargetDB(
-        return [_target_from_db(r) for r in self.session.exec(statement).all()]
+            id=target.id,
+            name=target.name,
+            description=target.description,
+            platform=target.platform.value,
+            channel_type=target.channel_type.value,
+            status=target.status.value,
+            created_at=target.created_at,
+            updated_at=target.updated_at or utc_now(),
+        )
+        db.set_config(target.channel_config)
+        return db
 
-    def get(self, target_id: str) -> Optional[EvalTarget]:
+    def _from_db(self, db: EvalTargetDB) -> EvalTarget:
-        db = self.session.get(EvalTargetDB, target_id)
+        return EvalTarget(
-        return _target_from_db(db) if db else None
+            id=db.id,
-
+            name=db.name,
-    def create(self, target: EvalTarget) -> EvalTarget:
+            description=db.description,
-        db = _target_to_db(target)
+            platform=db.platform,
-        self.session.add(db)
+            channel_type=db.channel_type,
-        self.session.commit()
+            channel_config=db.get_config(),
-        self.session.refresh(db)
+            status=db.status,
-        return _target_from_db(db)
+            created_at=db.created_at,
+            updated_at=db.updated_at,
+        )
 
     def update(self, target: EvalTarget) -> Optional[EvalTarget]:
         existing = self.session.get(EvalTargetDB, target.id)
@@ -206,33 +143,46 @@ class TargetRepository:
         self.session.add(existing)
         self.session.commit()
         self.session.refresh(existing)
-        return _target_from_db(existing)
+        return self._from_db(existing)
-
-    def delete(self, target_id: str) -> bool:
-        db = self.session.get(EvalTargetDB, target_id)
-        if not db:
-            return False
-        self.session.delete(db)
-        self.session.commit()
-        return True
 
 
-class ScenarioRepository:
+class ScenarioRepository(BaseRepository[Scenario, ScenarioDB]):
     """Repository for evaluation scenarios."""
 
-    def __init__(self, session: Optional[Session] = None):
+    _table = ScenarioDB
-        self.session = session or get_session()
+    _order_by = "created_at"
 
-    def list_all(self) -> list[Scenario]:
+    def _to_db(self, scenario: Scenario) -> ScenarioDB:
-        statement = select(ScenarioDB).order_by(ScenarioDB.created_at.desc())
+        db = ScenarioDB(
-        return [_scenario_from_db(r, self.session) for r in self.session.exec(statement).all()]
+            id=scenario.id,
+            name=scenario.name,
+            description=scenario.description,
+            created_at=scenario.created_at,
+            updated_at=scenario.updated_at or utc_now(),
+        )
+        db.set_tags(scenario.tags)
+        # mode="json" 与 update() 的考纲比较保持同一序列化形态，避免假升版
+        db.set_cases([case.model_dump(mode="json") for case in scenario.cases])
+        db.set_llm_config(scenario.llm_config)
+        return db
 
-    def get(self, scenario_id: str) -> Optional[Scenario]:
+    def _from_db(self, db: ScenarioDB) -> Scenario:
-        db = self.session.get(ScenarioDB, scenario_id)
+        bindings = ScenarioModelBindingRepository(self.session).get_for_scenario(db.id or "")
-        return _scenario_from_db(db, self.session) if db else None
+        return Scenario(
+            id=db.id,
+            name=db.name,
+            description=db.description,
+            tags=db.get_tags(),
+            cases=[Case(**case) for case in db.get_cases()],
+            model_bindings=bindings,
+            llm_config=db.get_llm_config(),
+            version=db.version or 1,
+            created_at=db.created_at,
+            updated_at=db.updated_at,
+        )
 
     def create(self, scenario: Scenario) -> Scenario:
-        db = _scenario_to_db(scenario)
+        db = self._to_db(scenario)
         bindings = {purpose.value: config_id for purpose, config_id in scenario.model_bindings.items()}
         try:
             ModelConfigService(self.session).validate_bindings(bindings)
@@ -244,7 +194,7 @@ class ScenarioRepository:
         except Exception:
             self.session.rollback()
             raise
-        return _scenario_from_db(db, self.session)
+        return self._from_db(db)
 
     def update(self, scenario: Scenario) -> Optional[Scenario]:
         existing = self.session.get(ScenarioDB, scenario.id)
@@ -277,7 +227,7 @@ class ScenarioRepository:
         except Exception:
             self.session.rollback()
             raise
-        return _scenario_from_db(existing, self.session)
+        return self._from_db(existing)
 
     def delete(self, scenario_id: str) -> bool:
         db = self.session.get(ScenarioDB, scenario_id)
@@ -293,15 +243,41 @@ class ScenarioRepository:
         return True
 
 
-class RunRepository:
+class RunRepository(BaseRepository[EvalRun, EvalRunDB]):
     """Repository for evaluation runs."""
 
-    def __init__(self, session: Optional[Session] = None):
+    _table = EvalRunDB
-        self.session = session or get_session()
+    _order_by = "started_at"
 
-    def list_all(self) -> list[EvalRun]:
+    def _to_db(self, run: EvalRun) -> EvalRunDB:
-        statement = select(EvalRunDB).order_by(EvalRunDB.started_at.desc())
+        db = EvalRunDB(
-        return [_run_from_db(r) for r in self.session.exec(statement).all()]
+            id=run.id,
+            target_id=run.target_id,
+            scenario_id=run.scenario_id,
+            scenario_version=run.scenario_version,
+            campaign_id=run.campaign_id,
+            status=run.status.value,
+            triggered_by=run.triggered_by.value,
+            started_at=run.started_at,
+            completed_at=run.completed_at,
+        )
+        if run.summary is not None:
+            db.set_summary(run.summary.model_dump(mode="json"))
+        return db
+
+    def _from_db(self, db: EvalRunDB) -> EvalRun:
+        return EvalRun(
+            id=db.id,
+            target_id=db.target_id,
+            scenario_id=db.scenario_id,
+            scenario_version=db.scenario_version or 1,
+            campaign_id=db.campaign_id,
+            status=db.status,
+            triggered_by=db.triggered_by or "manual",
+            started_at=db.started_at,
+            completed_at=db.completed_at,
+            summary=db.get_summary(),
+        )
 
     def list_by_campaign(self, campaign_id: str) -> list[EvalRun]:
         statement = (
@@ -309,18 +285,7 @@ class RunRepository:
             .where(EvalRunDB.campaign_id == campaign_id)
             .order_by(EvalRunDB.started_at)
         )
-        return [_run_from_db(r) for r in self.session.exec(statement).all()]
+        return [self._from_db(r) for r in self.session.exec(statement).all()]
-
-    def get(self, run_id: str) -> Optional[EvalRun]:
-        db = self.session.get(EvalRunDB, run_id)
-        return _run_from_db(db) if db else None
-
-    def create(self, run: EvalRun) -> EvalRun:
-        db = _run_to_db(run)
-        self.session.add(db)
-        self.session.commit()
-        self.session.refresh(db)
-        return _run_from_db(db)
 
     def mark_orphans_failed(self) -> int:
         """服务启动时清理：把遗留的 running/pending 运行标记为 failed。
@@ -357,7 +322,7 @@ class RunRepository:
         self.session.add(existing)
         self.session.commit()
         self.session.refresh(existing)
-        return _run_from_db(existing)
+        return self._from_db(existing)
 
     def get_turns(self, run_id: str) -> list[TurnDB]:
         statement = select(TurnDB).where(TurnDB.run_id == run_id).order_by(TurnDB.sent_at)
@@ -367,36 +332,44 @@ class RunRepository:
         statement = select(EvalResultDB).where(EvalResultDB.run_id == run_id)
         return [_result_from_db(r) for r in self.session.exec(statement).all()]
 
-    def delete(self, run_id: str) -> bool:
-        """Delete a run. ORM-level cascade removes associated turns/results."""
-        db = self.session.get(EvalRunDB, run_id)
-        if not db:
-            return False
-        self.session.delete(db)
-        self.session.commit()
-        return True
 
-
+class CampaignRepository(BaseRepository[Campaign, CampaignDB]):
-class CampaignRepository:
     """Repository for evaluation campaigns (评估活动)."""
 
-    def __init__(self, session: Optional[Session] = None):
+    _table = CampaignDB
-        self.session = session or get_session()
+    _order_by = "created_at"
 
-    def list_all(self) -> list[Campaign]:
+    def _to_db(self, campaign: Campaign) -> CampaignDB:
-        statement = select(CampaignDB).order_by(CampaignDB.created_at.desc())
+        db = CampaignDB(
-        return [_campaign_from_db(r) for r in self.session.exec(statement).all()]
+            id=campaign.id,
+            name=campaign.name,
+            target_id=campaign.target_id,
+            window_seconds=campaign.window_seconds,
+            time_scale=campaign.time_scale,
+            status=campaign.status.value,
+            started_at=campaign.started_at,
+            completed_at=campaign.completed_at,
+            created_at=campaign.created_at,
+        )
+        db.set_plan([entry.model_dump(mode="json") for entry in campaign.plan])
+        if campaign.summary:
+            db.set_summary(campaign.summary.model_dump(mode="json"))
+        return db
 
-    def get(self, campaign_id: str) -> Optional[Campaign]:
+    def _from_db(self, db: CampaignDB) -> Campaign:
-        db = self.session.get(CampaignDB, campaign_id)
+        return Campaign(
-        return _campaign_from_db(db) if db else None
+            id=db.id,
-
+            name=db.name,
-    def create(self, campaign: Campaign) -> Campaign:
+            target_id=db.target_id,
-        db = _campaign_to_db(campaign)
+            window_seconds=db.window_seconds,
-        self.session.add(db)
+            time_scale=db.time_scale,
-        self.session.commit()
+            plan=db.get_plan(),
-        self.session.refresh(db)
+            status=db.status,
-        return _campaign_from_db(db)
+            started_at=db.started_at,
+            completed_at=db.completed_at,
+            created_at=db.created_at,
+            summary=db.get_summary(),
+        )
 
     def update(self, campaign: Campaign) -> Optional[Campaign]:
         existing = self.session.get(CampaignDB, campaign.id)
@@ -411,11 +384,11 @@ class CampaignRepository:
         existing.started_at = campaign.started_at
         existing.completed_at = campaign.completed_at
         if campaign.summary is not None:
-            existing.set_summary(campaign.summary)
+            existing.set_summary(campaign.summary.model_dump(mode="json"))
         self.session.add(existing)
         self.session.commit()
         self.session.refresh(existing)
-        return _campaign_from_db(existing)
+        return self._from_db(existing)
 
 
 class ResultRepository:

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_build_run_summary.py

@@ -0,0 +1,116 @@
+"""Unit tests for the pure single-run summary seam.
+
+build_run_summary is the single reusable落点 for one run's口径 (pass_rate /
+judged_pass_rate / avg_latency / connectivity split), parallel to
+metrics.aggregate_runs (cross-run) and judgement.combine_case_outcome
+(case-level). No IO — fed constructed material, asserted directly.
+"""
+
+from agenteval.evaluation.judgement import CaseOutcome
+from agenteval.evaluation.run_summary import build_run_summary
+
+
+def _outcomes(**kw: CaseOutcome) -> dict[str, CaseOutcome]:
+    return dict(kw)
+
+
+def test_all_judged_passing():
+    summary = build_run_summary(
+        case_outcomes=_outcomes(
+            a=CaseOutcome(passed=True, connectivity=False),
+            b=CaseOutcome(passed=True, connectivity=False),
+        ),
+        latencies=[100.0, 300.0],
+        rule_passes=[True, True, True],
+    )
+    assert summary.total_cases == 2
+    assert summary.passed_cases == 2
+    assert summary.failed_cases == 0
+    assert summary.pass_rate == 1.0
+    assert summary.judged_pass_rate == 1.0
+    assert summary.total_rules == 3
+    assert summary.passed_rules == 3
+    assert summary.avg_latency_ms == 200.0
+
+
+def test_mixed_pass_fail():
+    summary = build_run_summary(
+        case_outcomes=_outcomes(
+            a=CaseOutcome(passed=True, connectivity=False),
+            b=CaseOutcome(passed=False, connectivity=False),
+            c=CaseOutcome(passed=False, connectivity=False),
+        ),
+        latencies=[],
+        rule_passes=[True, False],
+    )
+    assert summary.total_cases == 3
+    assert summary.passed_cases == 1
+    assert summary.failed_cases == 2
+    assert summary.pass_rate == 0.3333
+    assert summary.judged_pass_rate == 0.3333
+    assert summary.passed_rules == 1
+    assert summary.total_rules == 2
+    assert summary.avg_latency_ms is None
+
+
+def test_connectivity_excluded_from_judged_rate():
+    # 连通用例计入总通过率，但从判定型口径的分子分母双双剔除（report.py 现口径）
+    summary = build_run_summary(
+        case_outcomes=_outcomes(
+            conn=CaseOutcome(passed=True, connectivity=True),
+            judged_pass=CaseOutcome(passed=True, connectivity=False),
+            judged_fail=CaseOutcome(passed=False, connectivity=False),
+        ),
+        latencies=[50.0],
+        rule_passes=[True, False],
+    )
+    assert summary.total_cases == 3
+    assert summary.passed_cases == 2  # conn + judged_pass
+    assert summary.pass_rate == 0.6667
+    # judged: 分母 = 3 - 1 连通 = 2；分子 = 2 通过 - 1 连通 = 1
+    assert summary.judged_pass_rate == 0.5
+
+
+def test_all_connectivity_yields_no_judged_rate():
+    summary = build_run_summary(
+        case_outcomes=_outcomes(
+            a=CaseOutcome(passed=True, connectivity=True),
+            b=CaseOutcome(passed=True, connectivity=True),
+        ),
+        latencies=[10.0],
+        rule_passes=[],
+    )
+    assert summary.pass_rate == 1.0
+    assert summary.judged_pass_rate is None  # judged_total == 0
+
+
+def test_empty_run():
+    summary = build_run_summary(case_outcomes={}, latencies=[], rule_passes=[])
+    assert summary.total_cases == 0
+    assert summary.pass_rate == 0.0
+    assert summary.judged_pass_rate is None
+    assert summary.avg_latency_ms is None
+
+
+def test_case_errors_and_model_configs_passed_through():
+    errors = [{"case_id": "x", "message": "boom"}]
+    configs = {"judge": {"model": "gpt-4o"}}
+    summary = build_run_summary(
+        case_outcomes=_outcomes(x=CaseOutcome(passed=False, connectivity=False)),
+        latencies=[],
+        rule_passes=[],
+        case_errors=errors,
+        model_configs=configs,
+    )
+    assert summary.case_errors == errors
+    assert summary.model_configs == configs
+
+
+def test_case_outcomes_persisted_as_verdict_snapshot():
+    summary = build_run_summary(
+        case_outcomes=_outcomes(a=CaseOutcome(passed=True, connectivity=True)),
+        latencies=[],
+        rule_passes=[],
+    )
+    assert summary.case_outcomes["a"].passed is True
+    assert summary.case_outcomes["a"].connectivity is True

tests/unit/test_campaign_summary.py

@@ -0,0 +1,74 @@
+"""CampaignSummary VO — typed campaign scheduler state, restart-safe (ADR-0003).
+
+Mirrors RunSummary's typed treatment: Campaign.summary is no longer a bare
+dict. Legacy dict summaries coerce; unknown top-level keys survive
+(extra=allow) so older records keep parsing.
+"""
+
+from agenteval.models import (
+    Campaign,
+    CampaignPlanEntry,
+    CampaignSummary,
+    SchedulerState,
+)
+from agenteval.storage.repository import CampaignRepository, TargetRepository
+from tests.unit.test_repository import _make_target
+
+
+def _campaign(**summary_kw) -> Campaign:
+    kw = dict(
+        name="c",
+        target_id="t-1",
+        window_seconds=3600,
+        plan=[CampaignPlanEntry(scenario_id="s-1", offset_seconds=0, count=1)],
+    )
+    kw.update(summary_kw)
+    return Campaign(**kw)
+
+
+def test_summary_defaults_are_empty():
+    s = CampaignSummary()
+    assert s.scheduler.spawned_indices == []
+    assert s.scheduler.errors == []
+
+
+def test_scheduler_state_holds_progress():
+    s = SchedulerState(spawned_indices=[0, 2], errors=[{"index": 1, "error": "boom"}])
+    assert s.spawned_indices == [0, 2]
+    assert s.errors[0]["error"] == "boom"
+
+
+def test_legacy_dict_summary_coerces():
+    c = _campaign(summary={"scheduler": {"spawned_indices": [0, 1], "errors": [{"index": 2, "error": "x"}]}})
+    assert isinstance(c.summary, CampaignSummary)
+    assert c.summary.scheduler.spawned_indices == [0, 1]
+    assert c.summary.scheduler.errors[0]["error"] == "x"
+
+
+def test_assigning_dict_coerces_to_vo():
+    c = _campaign()
+    c.summary = {"scheduler": {"spawned_indices": [3]}}
+    assert isinstance(c.summary, CampaignSummary)
+    assert c.summary.scheduler.spawned_indices == [3]
+
+
+def test_unknown_top_level_keys_survive():
+    c = _campaign(summary={"scheduler": {}, "future_axis": {"availability": 0.9}})
+    assert isinstance(c.summary, CampaignSummary)
+    assert c.summary.model_extra["future_axis"] == {"availability": 0.9}
+
+
+def test_summary_survives_repository_round_trip(db_session):
+    TargetRepository(db_session).create(_make_target())
+    repo = CampaignRepository(db_session)
+    repo.create(
+        _campaign(
+            id="cp-1",
+            summary=CampaignSummary(scheduler=SchedulerState(spawned_indices=[0, 1], errors=[{"index": 2}])),
+        )
+    )
+    fetched = repo.get("cp-1")
+    assert fetched is not None
+    assert isinstance(fetched.summary, CampaignSummary)
+    assert fetched.summary.scheduler.spawned_indices == [0, 1]
+    assert fetched.summary.scheduler.errors == [{"index": 2}]

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)

tests/unit/test_implicit_rules.py

@@ -0,0 +1,45 @@
+"""Unit tests for derive_implicit_rules — 期望→隐式规则的纯翻译."""
+
+from agenteval.evaluation.implicit_rules import derive_implicit_rules
+from agenteval.models import Expectation
+
+
+def test_empty_expectation_yields_no_rules():
+    assert derive_implicit_rules(Expectation()) == []
+
+
+def test_intent_and_coherence_alone_yield_no_rules():
+    # intent / coherence_min_score 不派生隐式规则（无对应规则类型）
+    exp = Expectation(intent="问诊分流", coherence_min_score=0.8)
+    assert derive_implicit_rules(exp) == []
+
+
+def test_response_time_yields_response_time_rule():
+    rules = derive_implicit_rules(Expectation(response_time_max_ms=2000))
+    assert len(rules) == 1
+    assert rules[0].type == "response_time"
+    assert rules[0].params == {"max_ms": 2000}
+
+
+def test_keywords_include_yields_keyword_rule():
+    rules = derive_implicit_rules(Expectation(keywords_include=["挂号", "门诊"]))
+    assert len(rules) == 1
+    assert rules[0].type == "keyword_match"
+    assert rules[0].params == {"keywords": ["挂号", "门诊"], "exclude_keywords": []}
+
+
+def test_keywords_exclude_alone_yields_keyword_rule():
+    rules = derive_implicit_rules(Expectation(keywords_exclude=["投诉"]))
+    assert len(rules) == 1
+    assert rules[0].type == "keyword_match"
+    assert rules[0].params == {"keywords": [], "exclude_keywords": ["投诉"]}
+
+
+def test_both_expectations_yield_two_rules_in_order():
+    exp = Expectation(
+        response_time_max_ms=1500,
+        keywords_include=["预约"],
+        keywords_exclude=["取消"],
+    )
+    rules = derive_implicit_rules(exp)
+    assert [r.type for r in rules] == ["response_time", "keyword_match"]

tests/unit/test_repository.py

@@ -0,0 +1,113 @@
+"""CRUD + serialization round-trip characterization for the repository layer.
+
+These lock the create → get → list_all → delete contract shared by the
+Target / Run / Campaign repositories (the BaseRepository skeleton), plus the
+summary/plan JSON round-trip, so the generic-base refactor stays behaviour-
+preserving. Scenario's bespoke create/update (binding validation, versioning)
+is covered by its own tests.
+"""
+
+from datetime import datetime, timezone
+
+from agenteval.models import (
+    Campaign,
+    CampaignPlanEntry,
+    ChannelType,
+    EvalRun,
+    EvalTarget,
+    PlatformType,
+    RunStatus,
+    RunSummary,
+    TargetStatus,
+)
+from agenteval.storage.repository import (
+    CampaignRepository,
+    RunRepository,
+    TargetRepository,
+)
+
+
+def _make_target(tid: str = "t-1") -> EvalTarget:
+    return EvalTarget(
+        id=tid,
+        name="target",
+        platform=PlatformType.AI_DIGITAL_EMPLOYEE,
+        channel_type=ChannelType.TUTU_API,
+        channel_config={
+            "base_url": "x",
+            "token": "x",
+            "tenant": "x",
+            "chat_channel_id": "x",
+            "chat_contact_id": "x",
+        },
+    )
+
+
+def test_target_crud_round_trip(db_session):
+    repo = TargetRepository(db_session)
+    repo.create(_make_target())
+
+    fetched = repo.get("t-1")
+    assert fetched is not None
+    assert fetched.name == "target"
+    assert fetched.channel_config["base_url"] == "x"
+    assert fetched.status == TargetStatus.ACTIVE or fetched.status is not None
+
+    assert [t.id for t in repo.list_all()] == ["t-1"]
+
+    assert repo.delete("t-1") is True
+    assert repo.get("t-1") is None
+    assert repo.delete("t-1") is False
+
+
+def test_target_list_all_newest_first(db_session):
+    repo = TargetRepository(db_session)
+    older = _make_target("t-old")
+    older.created_at = datetime(2026, 1, 1, tzinfo=timezone.utc)
+    newer = _make_target("t-new")
+    newer.created_at = datetime(2026, 6, 1, tzinfo=timezone.utc)
+    repo.create(older)
+    repo.create(newer)
+    assert [t.id for t in repo.list_all()] == ["t-new", "t-old"]
+
+
+def test_run_summary_json_round_trip(db_session):
+    TargetRepository(db_session).create(_make_target())
+    repo = RunRepository(db_session)
+    repo.create(
+        EvalRun(
+            id="r-1",
+            target_id="t-1",
+            scenario_id="s-1",
+            status=RunStatus.COMPLETED,
+            summary=RunSummary(total_cases=2, passed_cases=1, pass_rate=0.5),
+        )
+    )
+    fetched = repo.get("r-1")
+    assert fetched is not None
+    assert fetched.summary is not None
+    assert fetched.summary.total_cases == 2
+    assert fetched.summary.pass_rate == 0.5
+
+
+def test_campaign_crud_and_plan_round_trip(db_session):
+    TargetRepository(db_session).create(_make_target())
+    repo = CampaignRepository(db_session)
+    repo.create(
+        Campaign(
+            id="cp-1",
+            name="campaign",
+            target_id="t-1",
+            window_seconds=3600,
+            plan=[CampaignPlanEntry(scenario_id="s-1", offset_seconds=0, count=2)],
+        )
+    )
+    fetched = repo.get("cp-1")
+    assert fetched is not None
+    assert fetched.plan[0].scenario_id == "s-1"
+    assert fetched.plan[0].count == 2
+    assert [c.id for c in repo.list_all()] == ["cp-1"]
+
+    # Campaign inherits the shared delete() from the base repository.
+    assert repo.delete("cp-1") is True
+    assert repo.get("cp-1") is None