diff --git a/tests/unit/test_campaign_report_format.py b/tests/unit/test_campaign_report_format.py
new file mode 100644
index 0000000..e088ee5
--- /dev/null
+++ b/tests/unit/test_campaign_report_format.py
@@ -0,0 +1,235 @@
+"""Report aggregation logic & output format validation (报告聚合与格式校验).
+
+补充 test_campaign_report.py（数值口径）与 test_metrics.py（单点聚合）之外：
+- 聚合边界：用户取消不计分母（ADR-0004）、窗口外/窗口前运行钳制、
+  SQLite naive 时间戳与 aware 等价；
+- 格式校验：报告 dict 的键集合、类型、取值范围、时间序列连续性、
+  能力面排序与 ISO-UTC 时间戳格式。generate_campaign_report 是纯函数，
+  直接构造模型对象，不落库。
+"""
+
+from datetime import datetime, timedelta, timezone
+
+from agenteval.evaluation.metrics import aggregate_runs
+from agenteval.evaluation.report import generate_campaign_report
+from agenteval.models import Campaign, CampaignPlanEntry, CampaignStatus, EvalRun, RunStatus
+
+T0 = datetime(2026, 1, 1, 0, 0, 0, tzinfo=timezone.utc)
+
+
+def _campaign(*, window_seconds: int = 12, time_scale: float = 1.0, started_at=T0, completed_at=None) -> Campaign:
+    return Campaign(
+        id="camp-fmt",
+        name="格式校验活动",
+        target_id="t-1",
+        window_seconds=window_seconds,
+        time_scale=time_scale,
+        plan=[CampaignPlanEntry(scenario_id="s-a", offset_seconds=0, count=1)],
+        status=CampaignStatus.RUNNING,
+        started_at=started_at,
+        completed_at=completed_at,
+    )
+
+
+def _run(
+    scenario_id: str = "s-a",
+    *,
+    status: RunStatus = RunStatus.COMPLETED,
+    offset_seconds: float = 0.0,
+    pass_rate: float | None = None,
+    latency: float | None = None,
+    error: dict | None = None,
+    started_at: datetime | None = None,
+) -> EvalRun:
+    summary = None
+    if pass_rate is not None or latency is not None or error is not None:
+        summary = {}
+        if pass_rate is not None:
+            summary["pass_rate"] = pass_rate
+        if latency is not None:
+            summary["avg_latency_ms"] = latency
+        if error is not None:
+            summary["error"] = error
+    return EvalRun(
+        target_id="t-1",
+        scenario_id=scenario_id,
+        campaign_id="camp-fmt",
+        status=status,
+        started_at=started_at if started_at is not None else T0 + timedelta(seconds=offset_seconds),
+        summary=summary,
+    )
+
+
+def _report(runs, **kwargs):
+    return generate_campaign_report(_campaign(), runs, **kwargs)
+
+
+# ── 格式校验：顶层结构 ────────────────────────────────────────────────
+
+
+def test_top_level_schema_keys_and_types():
+    report = _report([_run(pass_rate=1.0, latency=100)])
+    assert set(report) == {
+        "campaign_id",
+        "name",
+        "target_id",
+        "status",
+        "window_seconds",
+        "time_scale",
+        "started_at",
+        "completed_at",
+        "summary",
+        "time_trend",
+        "capability_summary",
+    }
+    assert report["campaign_id"] == "camp-fmt"
+    assert isinstance(report["window_seconds"], int) and report["window_seconds"] > 0
+    assert isinstance(report["time_scale"], float)
+    assert report["status"] == CampaignStatus.RUNNING.value
+    assert isinstance(report["time_trend"], list)
+    assert isinstance(report["capability_summary"], list)
+
+
+def test_summary_schema_types_and_bounds():
+    report = _report(
+        [
+            _run(pass_rate=1.0, latency=100),
+            _run(scenario_id="s-b", status=RunStatus.FAILED),
+        ]
+    )
+    summary = report["summary"]
+    assert set(summary) == {
+        "total_runs",
+        "completed_runs",
+        "overall_pass_rate",
+        "overall_availability",
+        "avg_latency_ms",
+    }
+    assert isinstance(summary["total_runs"], int) and summary["total_runs"] == 2
+    assert isinstance(summary["completed_runs"], int) and summary["completed_runs"] == 1
+    for key in ("overall_pass_rate", "overall_availability"):
+        value = summary[key]
+        assert value is None or (isinstance(value, float) and 0.0 <= value <= 1.0), key
+    assert summary["avg_latency_ms"] is None or summary["avg_latency_ms"] >= 0
+
+
+def test_summary_matches_single_aggregation_seam():
+    """summary 数值必须与单点聚合口径 aggregate_runs 完全一致（ADR-0004）。"""
+    runs = [
+        _run(pass_rate=1.0, latency=100),
+        _run(pass_rate=0.5, latency=300),
+        _run(status=RunStatus.FAILED),
+    ]
+    report = _report(runs)
+    agg = aggregate_runs(runs)
+    summary = report["summary"]
+    assert summary["overall_pass_rate"] == agg["pass_rate"]
+    assert summary["overall_availability"] == agg["availability"]
+    assert summary["avg_latency_ms"] == agg["avg_latency_ms"]
+
+
+# ── 格式校验：时间趋势序列 ────────────────────────────────────────────
+
+
+def test_time_trend_contiguous_buckets_with_custom_count():
+    report = generate_campaign_report(_campaign(window_seconds=12), [_run(pass_rate=1.0)], bucket_count=4)
+    trend = report["time_trend"]
+    assert len(trend) == 4
+    for idx, bucket in enumerate(trend):
+        assert bucket["bucket_index"] == idx
+        assert bucket["start_seconds"] == idx * 3.0  # window / bucket_count
+        assert bucket["end_seconds"] == (idx + 1) * 3.0
+        assert {"run_count", "pass_rate", "availability", "avg_latency_ms"} <= set(bucket)
+    # 相邻桶首尾相接
+    for prev, cur in zip(trend, trend[1:]):
+        assert prev["end_seconds"] == cur["start_seconds"]
+
+
+def test_time_trend_bucket_types_and_bounds():
+    report = _report([_run(pass_rate=0.8, latency=120)])
+    for bucket in report["time_trend"]:
+        assert isinstance(bucket["run_count"], int) and bucket["run_count"] >= 0
+        if bucket["run_count"] == 0:
+            assert bucket["pass_rate"] is None and bucket["availability"] is None
+        else:
+            assert 0.0 <= bucket["pass_rate"] <= 1.0
+            assert 0.0 <= bucket["availability"] <= 1.0
+
+
+# ── 格式校验：能力面汇总 ──────────────────────────────────────────────
+
+
+def test_capability_summary_schema_and_sorted_by_run_count_desc():
+    runs = [
+        _run("s-a", pass_rate=1.0),
+        _run("s-a", pass_rate=0.5),
+        _run("s-a", pass_rate=0.0),
+        _run("s-b", pass_rate=1.0),
+    ]
+    report = _report(runs, scenario_names={"s-a": "夜间问诊"})
+    caps = report["capability_summary"]
+    assert [c["scenario_id"] for c in caps] == ["s-a", "s-b"]  # run_count 降序
+    for cap in caps:
+        assert {"scenario_id", "scenario_name", "run_count", "pass_rate", "availability", "avg_latency_ms"} <= set(cap)
+        assert cap["run_count"] >= 1
+    assert caps[0]["scenario_name"] == "夜间问诊"
+    assert caps[1]["scenario_name"] == "s-b"[:8]  # 无映射时回落 id 前缀
+
+
+# ── 格式校验：时间戳 ISO-UTC ──────────────────────────────────────────
+
+
+def test_timestamps_serialize_as_iso_utc_or_none():
+    report = generate_campaign_report(_campaign(completed_at=T0 + timedelta(hours=1)), [_run()])
+    assert report["started_at"].endswith("Z")
+    assert report["completed_at"].endswith("Z")
+    assert "T" in report["started_at"]
+
+    unfinished = generate_campaign_report(_campaign(), [_run()])
+    assert unfinished["completed_at"] is None
+
+
+def test_naive_campaign_timestamps_still_serialize_utc():
+    naive_start = datetime(2026, 1, 1, 0, 0, 0)  # SQLite round-trip drops tzinfo
+    report = generate_campaign_report(_campaign(started_at=naive_start), [_run()])
+    assert report["started_at"].endswith("Z")
+
+
+# ── 聚合边界：取消 / 钳制 / naive 时间 ────────────────────────────────
+
+
+def test_cancelled_run_excluded_from_summary_but_counted():
+    runs = [
+        _run(pass_rate=1.0, latency=100),
+        _run(status=RunStatus.FAILED, error={"code": "cancelled_by_user", "message": "stop"}),
+    ]
+    summary = _report(runs)["summary"]
+    assert summary["total_runs"] == 2  # 发生过的事仍然可见
+    assert summary["completed_runs"] == 1
+    assert summary["overall_pass_rate"] == 1.0  # 取消不进分母（ADR-0004）
+    assert summary["overall_availability"] == 1.0
+
+
+def test_run_started_before_window_clamped_to_first_bucket():
+    report = _report([_run(pass_rate=1.0, started_at=T0 - timedelta(seconds=10))])
+    assert report["time_trend"][0]["run_count"] == 1
+
+
+def test_run_started_after_window_clamped_to_last_bucket():
+    report = _report([_run(pass_rate=1.0, started_at=T0 + timedelta(seconds=100))])
+    assert report["time_trend"][-1]["run_count"] == 1
+    assert sum(b["run_count"] for b in report["time_trend"]) == 1
+
+
+def test_naive_run_timestamps_bucket_same_as_aware():
+    aware = _run(pass_rate=1.0, started_at=T0 + timedelta(seconds=6))
+    naive = _run(pass_rate=1.0, started_at=datetime(2026, 1, 1, 0, 0, 6))  # SQLite 读回的 naive UTC
+    trend_aware = _report([aware])["time_trend"]
+    trend_naive = _report([naive])["time_trend"]
+    assert trend_aware == trend_naive
+    assert trend_naive[6]["run_count"] == 1  # window 12s / 12 buckets → offset 6 落第 6 桶
+
+
+def test_unstarted_run_sits_at_offset_zero():
+    report = _report([_run(pass_rate=1.0, started_at=None)])
+    assert report["time_trend"][0]["run_count"] == 1