feat(comparison): unify read model and validation for period comparison

周期对比读模型升位为单一出口（load_comparison_view），GET/POST/markdown
三处调用点统一走同一 view 投影，消除「取数三件套」重复。校验逻辑收敛到
validate_comparison_request，router 捕获映射 400，执行器捕获落 failed 行，
校验顺序权威不再漂移。

- 新增 load_comparison_view：无行返回 status=none + auto_baseline，有行
  返回完整 comparison dict（含 model_name 标签）+ metric_diff
- 新增 validate_comparison_request：活动终态 → 模型 → 基线 → 分析，违
  规抛 ComparisonError
- execute_campaign_comparison 内联校验替换为 validate_comparison_request
  调用，catch ComparisonError 落 failed 行
- router 三处迁移：GET /comparison、POST /comparison、markdown 导出
- 删除 build_comparison_payload（已吸收进 load_comparison_view）
- 8 个新测试覆盖读模型三态 + 校验五错

backend/agenteval/evaluation/comparison.py

@@ -15,8 +15,9 @@ from sqlmodel import Session
 
 from agenteval.evaluation.analysis import ChatClient, gateway_chat_client, resolve_analysis_model
 from agenteval.evaluation.report import load_campaign_report
-from agenteval.models import Campaign
+from agenteval.models import Campaign, CampaignStatus
 from agenteval.storage.db import get_session, iso_utc, utc_now
+from agenteval.storage.model_config_repository import ModelConfigRepository
 from agenteval.storage.repository import (
     CampaignAnalysisRepository,
     CampaignPeriodComparisonRepository,
@@ -152,6 +153,118 @@ def compute_metric_diff(
     return {"overall": overall, "scenarios": scenarios}
 
 
+def validate_comparison_request(
+    session: Session,
+    campaign: Campaign,
+    *,
+    explicit_baseline_id: Optional[str] = None,
+) -> Campaign:
+    """共享校验入口：活动终态 → 模型 → 基线 → 分析。违规抛 ComparisonError。
+
+    router 触发端点捕获映射 400；执行器捕获落 failed 行。校验顺序权威，
+    两处不再漂移。
+    """
+    if campaign.status in (CampaignStatus.PLANNED, CampaignStatus.RUNNING):
+        raise ComparisonError("活动完成后才能生成周期对比")
+    if resolve_analysis_model(campaign, session) is None:
+        raise ComparisonError(
+            "未配置分析模型：请在模型配置中心将某个 chat 配置设为「分析默认」，或为该活动指定分析模型"
+        )
+    current_analysis = CampaignAnalysisRepository(session).get_by_campaign(campaign.id)
+    if current_analysis is None or current_analysis.status != "completed":
+        raise ComparisonError("请先生成本期活动的智能分析")
+
+    if explicit_baseline_id:
+        baseline = CampaignRepository(session).get(explicit_baseline_id)
+        if baseline is None:
+            raise ComparisonError(f"基线活动不存在: {explicit_baseline_id}")
+    else:
+        baseline = resolve_auto_baseline(campaign, session)
+        if baseline is None:
+            raise ComparisonError(
+                "未找到自动基线：历史活动中没有同计划指纹且已完成分析的活动，可手动选择基线活动"
+            )
+
+    baseline_analysis = CampaignAnalysisRepository(session).get_by_campaign(baseline.id)
+    if baseline_analysis is None or baseline_analysis.status != "completed":
+        raise ComparisonError("基线活动没有已完成的智能分析，请先生成基线活动的分析")
+
+    return baseline
+
+
+def load_comparison_view(session: Session, campaign: Campaign) -> dict[str, Any]:
+    """周期对比读模型单一出口：返回 GET /comparison 完整响应形状。
+
+    无行时 status=none + auto_baseline；有行时含 comparison dict（含 model_name
+    标签）+ 对生效基线的 metric_diff。markdown 导出从同一 view 投影。
+    """
+    payload = _build_auto_baseline_payload(campaign, session)
+    row = CampaignPeriodComparisonRepository(session).get_by_campaign(campaign.id)
+    if row is None:
+        return {"status": "none", "comparison": None, **payload}
+
+    effective_baseline = CampaignRepository(session).get(row.baseline_campaign_id)
+    metric_diff = (
+        compute_metric_diff(
+            load_campaign_report(session, effective_baseline),
+            load_campaign_report(session, campaign),
+        )
+        if effective_baseline is not None
+        else None
+    )
+    model_cfg = (
+        ModelConfigRepository(session).get(row.model_config_id) if row.model_config_id else None
+    )
+    model_label = (
+        f"{model_cfg.name}（{model_cfg.model_name}）"
+        if model_cfg and model_cfg.model_name
+        else (model_cfg.name if model_cfg else None)
+    )
+    comparison = {
+        "baseline_campaign_id": row.baseline_campaign_id,
+        "baseline": (
+            {
+                "id": effective_baseline.id,
+                "name": effective_baseline.name,
+                "completed_at": iso_utc(effective_baseline.completed_at),
+            }
+            if effective_baseline
+            else None
+        ),
+        "result": row.get_result(),
+        "error": row.error,
+        "model_config_id": row.model_config_id,
+        "model_name": model_label,
+        "triggered_by": row.triggered_by,
+        "updated_at": iso_utc(row.updated_at),
+    }
+    return {
+        "status": row.status,
+        "comparison": comparison,
+        "auto_baseline": payload["auto_baseline"],
+        "metric_diff": metric_diff,
+    }
+
+
+def _build_auto_baseline_payload(campaign: Campaign, session: Session) -> dict[str, Any]:
+    """自动基线信息 + 机械 diff（无基线时两者均为 null）。"""
+    baseline = resolve_auto_baseline(campaign, session)
+    if baseline is None:
+        return {"auto_baseline": None, "metric_diff": None}
+    diff = compute_metric_diff(
+        load_campaign_report(session, baseline),
+        load_campaign_report(session, campaign),
+    )
+    return {
+        "auto_baseline": {
+            "id": baseline.id,
+            "name": baseline.name,
+            "completed_at": iso_utc(baseline.completed_at),
+        },
+        "metric_diff": diff,
+    }
+
+
 def build_comparison_payload(campaign: Campaign, session: Session) -> dict[str, Any]:
     """GET 返回体：自动基线信息 + 机械 diff（无基线时两者均为 null）。"""
     baseline = resolve_auto_baseline(campaign, session)
@@ -281,64 +394,28 @@ async def execute_campaign_comparison(
     session = get_session()
     try:
         comparisons = CampaignPeriodComparisonRepository(session)
-        analyses = CampaignAnalysisRepository(session)
         campaign = CampaignRepository(session).get(campaign_id)
         if not campaign:
             return
 
+        try:
+            baseline = validate_comparison_request(
+                session,
+                campaign,
+                explicit_baseline_id=baseline_campaign_id,
+            )
+        except ComparisonError as exc:
+            comparisons.upsert(
+                campaign_id,
+                status="failed",
+                triggered_by=triggered_by,
+                error=str(exc),
+            )
+            return
+
         runtime = resolve_analysis_model(campaign, session)
-        if runtime is None:
+        baseline_analysis = CampaignAnalysisRepository(session).get_by_campaign(baseline.id)
-            comparisons.upsert(
+        current_analysis = CampaignAnalysisRepository(session).get_by_campaign(campaign_id)
-                campaign_id,
-                status="failed",
-                triggered_by=triggered_by,
-                error="未配置分析模型：请在模型配置中心将某个 chat 配置设为「分析默认」",
-            )
-            return
-
-        if baseline_campaign_id:
-            baseline = CampaignRepository(session).get(baseline_campaign_id)
-            if baseline is None:
-                comparisons.upsert(
-                    campaign_id,
-                    status="failed",
-                    triggered_by=triggered_by,
-                    error=f"基线活动不存在: {baseline_campaign_id}",
-                )
-                return
-        else:
-            baseline = resolve_auto_baseline(campaign, session)
-        if baseline is None:
-            comparisons.upsert(
-                campaign_id,
-                status="failed",
-                triggered_by=triggered_by,
-                error="未找到可用的基线活动（计划指纹不一致或历史活动无已完成分析）",
-            )
-            return
-
-        baseline_analysis = analyses.get_by_campaign(baseline.id)
-        if baseline_analysis is None or baseline_analysis.status != "completed":
-            comparisons.upsert(
-                campaign_id,
-                status="failed",
-                baseline_campaign_id=baseline.id,
-                model_config_id=runtime.id,
-                triggered_by=triggered_by,
-                error="基线活动没有已完成的智能分析，请先生成基线活动的分析",
-            )
-            return
-        current_analysis = analyses.get_by_campaign(campaign_id)
-        if current_analysis is None or current_analysis.status != "completed":
-            comparisons.upsert(
-                campaign_id,
-                status="failed",
-                baseline_campaign_id=baseline.id,
-                model_config_id=runtime.id,
-                triggered_by=triggered_by,
-                error="本期活动没有已完成的智能分析，请先生成智能分析",
-            )
-            return
 
         comparisons.upsert(
             campaign_id,

backend/agenteval/web/routers/campaigns.py

@@ -14,10 +14,10 @@ from sqlmodel import Session
 from agenteval.evaluation.analysis import resolve_analysis_model, start_campaign_analysis
 from agenteval.evaluation.campaign_runner import campaign_progress, request_cancel, start_campaign
 from agenteval.evaluation.comparison import (
-    build_comparison_payload,
+    ComparisonError,
-    compute_metric_diff,
+    load_comparison_view,
-    resolve_auto_baseline,
     start_campaign_comparison,
+    validate_comparison_request,
 )
 from agenteval.evaluation.report import (
     build_campaign_timeline,
@@ -32,7 +32,6 @@ from agenteval.storage.db import iso_utc, utc_now
 from agenteval.storage.model_config_repository import ModelConfigRepository
 from agenteval.storage.repository import (
     CampaignAnalysisRepository,
-    CampaignPeriodComparisonRepository,
     CampaignRepository,
     RunRepository,
     ScenarioRepository,
@@ -148,28 +147,17 @@ async def get_campaign_report_markdown(campaign_id: str, session: Session = Depe
     target_name = target.name if target else None
 
     comparison = None
-    cmp_row = CampaignPeriodComparisonRepository(session).get_by_campaign(campaign_id)
+    view = load_comparison_view(session, campaign)
-    if cmp_row is not None and cmp_row.status == "completed" and cmp_row.get_result():
+    if view["status"] == "completed" and view["comparison"] and view["comparison"].get("result"):
-        baseline = CampaignRepository(session).get(cmp_row.baseline_campaign_id)
+        cmp = view["comparison"]
-        metric_diff = (
+        baseline = cmp.get("baseline") or {}
-            compute_metric_diff(
-                load_campaign_report(session, baseline),
-                load_campaign_report(session, campaign),
-            )
-            if baseline is not None
-            else None
-        )
-        model_cfg = ModelConfigRepository(session).get(cmp_row.model_config_id) if cmp_row.model_config_id else None
-        model_label = f"{model_cfg.name}（{model_cfg.model_name}）" if model_cfg and model_cfg.model_name else (
-            model_cfg.name if model_cfg else None
-        )
         comparison = {
-            "result": cmp_row.get_result(),
+            "result": cmp["result"],
-            "baseline_name": baseline.name if baseline else None,
+            "baseline_name": baseline.get("name"),
-            "baseline_completed_at": iso_utc(baseline.completed_at) if baseline else None,
+            "baseline_completed_at": baseline.get("completed_at"),
-            "model_name": model_label,
+            "model_name": cmp.get("model_name"),
-            "updated_at": iso_utc(cmp_row.updated_at),
+            "updated_at": cmp.get("updated_at"),
-            "metric_diff": metric_diff,
+            "metric_diff": view["metric_diff"],
         }
 
     md = render_campaign_markdown(
@@ -241,43 +229,7 @@ async def get_campaign_comparison(campaign_id: str, session: Session = Depends(g
     if not campaign:
         raise HTTPException(status_code=404, detail="campaign not found")
 
-    payload = build_comparison_payload(campaign, session)
+    return load_comparison_view(session, campaign)
-    row = CampaignPeriodComparisonRepository(session).get_by_campaign(campaign_id)
-    if row is None:
-        return {"status": "none", "comparison": None, **payload}
-
-    baseline = CampaignRepository(session).get(row.baseline_campaign_id)
-    metric_diff = (
-        compute_metric_diff(
-            load_campaign_report(session, baseline),
-            load_campaign_report(session, campaign),
-        )
-        if baseline is not None
-        else None
-    )
-    comparison = {
-        "baseline_campaign_id": row.baseline_campaign_id,
-        "baseline": (
-            {
-                "id": baseline.id,
-                "name": baseline.name,
-                "completed_at": iso_utc(baseline.completed_at),
-            }
-            if baseline
-            else None
-        ),
-        "result": row.get_result(),
-        "error": row.error,
-        "model_config_id": row.model_config_id,
-        "triggered_by": row.triggered_by,
-        "updated_at": iso_utc(row.updated_at),
-    }
-    return {
-        "status": row.status,
-        "comparison": comparison,
-        "auto_baseline": payload["auto_baseline"],
-        "metric_diff": metric_diff,
-    }
 
 
 class GenerateComparisonRequest(BaseModel):
@@ -293,33 +245,12 @@ async def trigger_campaign_comparison(
     campaign = CampaignRepository(session).get(campaign_id)
     if not campaign:
         raise HTTPException(status_code=404, detail="campaign not found")
-    if campaign.status in (CampaignStatus.PLANNED, CampaignStatus.RUNNING):
-        raise HTTPException(status_code=400, detail="活动完成后才能生成周期对比")
-    analyses = CampaignAnalysisRepository(session)
-    current_analysis = analyses.get_by_campaign(campaign_id)
-    if current_analysis is None or current_analysis.status != "completed":
-        raise HTTPException(status_code=400, detail="请先生成本期活动的智能分析")
-    if resolve_analysis_model(campaign, session) is None:
-        raise HTTPException(
-            status_code=400,
-            detail="未配置分析模型：请在模型配置中心将某个 chat 配置设为「分析默认」，或为该活动指定分析模型",
-        )
 
     explicit_baseline_id = request.baseline_campaign_id if request else None
-    if explicit_baseline_id:
+    try:
-        baseline = CampaignRepository(session).get(explicit_baseline_id)
+        baseline = validate_comparison_request(session, campaign, explicit_baseline_id=explicit_baseline_id)
-        if baseline is None:
+    except ComparisonError as exc:
-            raise HTTPException(status_code=400, detail=f"基线活动不存在: {explicit_baseline_id}")
+        raise HTTPException(status_code=400, detail=str(exc))
-        baseline_analysis = analyses.get_by_campaign(explicit_baseline_id)
-        if baseline_analysis is None or baseline_analysis.status != "completed":
-            raise HTTPException(status_code=400, detail="基线活动没有已完成的智能分析，请先生成基线活动的分析")
-    else:
-        baseline = resolve_auto_baseline(campaign, session)
-        if baseline is None:
-            raise HTTPException(
-                status_code=400,
-                detail="未找到自动基线：历史活动中没有同计划指纹且已完成分析的活动，可手动选择基线活动",
-            )
 
     start_campaign_comparison(campaign_id, triggered_by="manual", baseline_campaign_id=baseline.id)
     return {"status": "generating"}

tests/unit/test_comparison_read_model.py

@@ -0,0 +1,178 @@
+"""周期对比读模型与校验合一直测（架构保养第二轮候选 3）。
+
+load_comparison_view 单一出口返回 GET /comparison 完整响应形状；
+validate_comparison_request 共享校验入口抛 ComparisonError，router
+映射 400、执行器落 failed 行。
+"""
+
+from datetime import datetime, timezone
+from uuid import uuid4
+
+import pytest
+from agenteval.evaluation.comparison import (
+    ComparisonError,
+    load_comparison_view,
+    validate_comparison_request,
+)
+from agenteval.models import (
+    Campaign,
+    CampaignPlanEntry,
+    CampaignStatus,
+    EvalTarget,
+    Scenario,
+    Case,
+    CaseType,
+)
+from agenteval.storage.repository import (
+    CampaignAnalysisRepository,
+    CampaignPeriodComparisonRepository,
+    CampaignRepository,
+    ScenarioRepository,
+    TargetRepository,
+)
+from sqlmodel import Session, SQLModel, create_engine
+
+T0 = datetime(2026, 1, 1, 0, 0, 0, tzinfo=timezone.utc)
+
+
+@pytest.fixture()
+def db_session(tmp_path):
+    from agenteval.storage.db import (  # noqa: F401
+        CampaignAnalysisDB,
+        CampaignDB,
+        CampaignPeriodComparisonDB,
+        EvalRunDB,
+        EvalTargetDB,
+        ScenarioDB,
+    )
+
+    engine = create_engine(
+        f"sqlite:///{tmp_path / 'comparison.db'}",
+        connect_args={"check_same_thread": False},
+    )
+    SQLModel.metadata.create_all(engine)
+    session = Session(engine)
+    try:
+        yield session
+    finally:
+        session.close()
+        engine.dispose()
+
+
+def _seed_campaign(db_session, *, status=CampaignStatus.COMPLETED):
+    target = TargetRepository(db_session).create(
+        EvalTarget(id=f"t-{uuid4().hex[:8]}", name="数字员工")
+    )
+    scenario = ScenarioRepository(db_session).create(
+        Scenario(id=f"s-{uuid4().hex[:8]}", name="查账单", target_id=target.id, cases=[Case(id="c-1", type=CaseType.SINGLE, messages=["你好"])])
+    )
+    campaign = CampaignRepository(db_session).create(
+        Campaign(
+            name="cycle",
+            target_id=target.id,
+            window_seconds=3600,
+            time_scale=1.0,
+            plan=[CampaignPlanEntry(scenario_id=scenario.id, offset_seconds=0, count=1)],
+            status=status,
+            started_at=T0,
+        )
+    )
+    return campaign
+
+
+def test_load_comparison_view_no_row_returns_none_status(db_session):
+    campaign = _seed_campaign(db_session)
+    view = load_comparison_view(db_session, campaign)
+    assert view["status"] == "none"
+    assert view["comparison"] is None
+    assert "auto_baseline" in view
+    assert "metric_diff" in view
+
+
+def test_load_comparison_view_generating_row(db_session):
+    campaign = _seed_campaign(db_session)
+    CampaignPeriodComparisonRepository(db_session).upsert(
+        campaign.id,
+        status="generating",
+        baseline_campaign_id="baseline-1",
+        triggered_by="manual",
+    )
+    view = load_comparison_view(db_session, campaign)
+    assert view["status"] == "generating"
+    assert view["comparison"] is not None
+    assert view["comparison"]["baseline_campaign_id"] == "baseline-1"
+
+
+def test_load_comparison_view_completed_row_includes_model_name(db_session):
+    campaign = _seed_campaign(db_session)
+    CampaignPeriodComparisonRepository(db_session).upsert(
+        campaign.id,
+        status="completed",
+        baseline_campaign_id="baseline-1",
+        result={"trend": "improving"},
+        model_config_id="mc-1",
+        triggered_by="manual",
+    )
+    view = load_comparison_view(db_session, campaign)
+    assert view["status"] == "completed"
+    assert view["comparison"]["result"] == {"trend": "improving"}
+    assert "model_name" in view["comparison"]
+
+
+def test_validate_rejects_non_completed_campaign(db_session):
+    campaign = _seed_campaign(db_session, status=CampaignStatus.RUNNING)
+    with pytest.raises(ComparisonError, match="活动完成后"):
+        validate_comparison_request(db_session, campaign)
+
+
+def test_validate_rejects_missing_analysis_model(db_session, monkeypatch):
+    campaign = _seed_campaign(db_session)
+    monkeypatch.setattr(
+        "agenteval.evaluation.comparison.resolve_analysis_model",
+        lambda *a, **kw: None,
+    )
+    with pytest.raises(ComparisonError, match="未配置分析模型"):
+        validate_comparison_request(db_session, campaign)
+
+
+def test_validate_rejects_incomplete_current_analysis(db_session, monkeypatch):
+    campaign = _seed_campaign(db_session)
+    monkeypatch.setattr(
+        "agenteval.evaluation.comparison.resolve_analysis_model",
+        lambda *a, **kw: object(),
+    )
+    CampaignAnalysisRepository(db_session).upsert(
+        campaign.id, status="generating", triggered_by="auto"
+    )
+    with pytest.raises(ComparisonError, match="请先生成本期活动的智能分析"):
+        validate_comparison_request(db_session, campaign)
+
+
+def test_validate_rejects_missing_explicit_baseline(db_session, monkeypatch):
+    campaign = _seed_campaign(db_session)
+    monkeypatch.setattr(
+        "agenteval.evaluation.comparison.resolve_analysis_model",
+        lambda *a, **kw: object(),
+    )
+    CampaignAnalysisRepository(db_session).upsert(
+        campaign.id, status="completed", result={}, triggered_by="auto"
+    )
+    with pytest.raises(ComparisonError, match="基线活动不存在"):
+        validate_comparison_request(db_session, campaign, explicit_baseline_id="missing")
+
+
+def test_validate_rejects_incomplete_baseline_analysis(db_session, monkeypatch):
+    campaign = _seed_campaign(db_session)
+    monkeypatch.setattr(
+        "agenteval.evaluation.comparison.resolve_analysis_model",
+        lambda *a, **kw: object(),
+    )
+    baseline = _seed_campaign(db_session)
+    CampaignAnalysisRepository(db_session).upsert(
+        campaign.id, status="completed", result={}, triggered_by="auto"
+    )
+    CampaignAnalysisRepository(db_session).upsert(
+        baseline.id, status="generating", triggered_by="auto"
+    )
+    with pytest.raises(ComparisonError, match="基线活动没有已完成的智能分析"):
+        validate_comparison_request(db_session, campaign, explicit_baseline_id=baseline.id)