From b5cc68c977e9926e481dda54df1d2e94b75f309d Mon Sep 17 00:00:00 2001 From: Fasthei <167957975+Fasthei@users.noreply.github.com> Date: Fri, 12 Jun 2026 17:38:19 +0800 Subject: [PATCH] =?UTF-8?q?feat(benchmark):=20=E8=90=BD=E5=9C=B0=E8=87=AA?= =?UTF-8?q?=E8=AF=81=E9=87=87=E9=9B=86=E5=99=A8=EF=BC=88=E9=98=B6=E6=AE=B5?= =?UTF-8?q?0+1=EF=BC=89=E2=80=94=20S=5Fgain=E2=89=A1G=5FE=20=E6=8E=A5?= =?UTF-8?q?=E9=80=9A=20S=5Fswarm=20+=20leaderboard?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 阶段0（定口径，docs/benchmark/emergence-evaluation.md §6 v2.1-impl）： - S_gain ≡ G_E（差值，不强制归一 [0,100]，与标准「见涌现增益」字面一致）。 - 聚合 S_gain 取对最强基线（Q_base 最大）的 G_E（最保守，避免挑弱基线虚高）。 - Q ≡ Q_quality；swarm_valid 仍要求对全部基线 G_E>0 且 G_E,c>0。 阶段1（采集器）： - 新增 benchmark/collectors/selfcert_collector.py：把套件 5 份 BenchmarkRunRecord （swarm+4基线）+ 可选活体 SwarmMetrics 合流，经 baselines.compare 算 G_E/G_E,c， 补全 run_collector 无法自算的 s_gain/g_e/g_e_cost/s_swarm，可能时产出 Benchmark_Agent。 - benchmark/leaderboard：实现排行榜聚合+渲染（标准 §11 字段）。 - run-benchmark-suite.py 接入自证 + leaderboard 输出。 诚实纪律（组织规则 #9）：缺真实输入一律 NaN+coverage False，不伪造。 - O（可观测性）标准无公式 → 恒 NaN；Gov 计数器未实现 → 无活体治理则 NaN。 - 故完整 Benchmark_Agent 数字仍待 O 公式 + Gov 计数器（阶段2），采集器明列缺口。 验证：新增 test-benchmark-selfcert.py（17 项）+ 现有 benchmark 测试（metrics/ collector/comparison/runners）+ offline suite + 契约冒烟（runtime/merge/freeze）全 PASS。 影响范围：仅 agent_swarm benchmark 模块 + docs；不改 Manager↔Swarm 契约/计费/审计/密钥/发布链路。 Co-Authored-By: Claude Opus 4.8 (1M context) --- benchmark/collectors/selfcert_collector.py | 202 +++++++++++++++++++++ benchmark/leaderboard/__init__.py | 64 ++++++- docs/benchmark/emergence-evaluation.md | 15 +- scripts/run-benchmark-suite.py | 13 ++ scripts/test-benchmark-selfcert.py | 104 +++++++++++ 5 files changed, 392 insertions(+), 6 deletions(-) create mode 100644 benchmark/collectors/selfcert_collector.py create mode 100644 scripts/test-benchmark-selfcert.py diff --git a/benchmark/collectors/selfcert_collector.py b/benchmark/collectors/selfcert_collector.py new file mode 100644 index 0000000..8a66c75 --- /dev/null +++ b/benchmark/collectors/selfcert_collector.py @@ -0,0 +1,202 @@ +"""SelfCertCollector — 把 benchmark 套件记录（swarm + 4 基线）+ 可选活体 run 指标， +合流成一次**自证**：填上单次 run 无法自算的增益类字段（`s_gain`/`g_e`/`g_e_cost`/`s_swarm`）， +并在可能时产出 `Benchmark_Agent`。 + +为什么需要它（对比 run_collector）： +- `SwarmRunMetricsCollector`（collectors/run_collector.py）从一次活体 run 算 completion/collaboration/ + cost/robustness/communication/governance/reward——但 `s_gain` 是**对比指标**，单 run 无法自算， + 故那里 `s_gain/s_swarm/g_e/g_e_cost/benchmark` 恒为 NaN。 +- 本采集器在**基线管线已就位**（benchmark/runners + baselines.compare）后，把 swarm 与 4 基线的 + `BenchmarkRunRecord` 经 `compare` 算出 `G_E`/`G_E,c`，据此补全 `s_gain` 与 `s_swarm`。 + +口径（docs/benchmark/emergence-evaluation.md §6，v2.1-impl）： +- `S_gain ≡ G_E`（差值，不强制归一到 [0,100]）。 +- 聚合 `S_gain` 取**对最强基线**（Q_base 最大者）的 `G_E`——最保守，避免挑弱基线虚高。 +- 逐基线 `G_E_i`/`G_E,c_i` 全量保留（leaderboard 用）。 + +诚实原则（组织规则 #9）：任何缺真实输入的量返回 `NaN` 并在 `coverage` 标 `False`，绝不伪造分值。 +- `O`（可观测性）：标准**无公式** → 恒 NaN + coverage False（见 telemetry-architecture.md）。 +- `Gov`：取活体 run 的 `s_governance`（需治理计数器，见 governance-score.md）；无则 NaN。 +- 故只有同时具备 {完整 S_swarm, G_E, Reward, O, Gov} 时 `Benchmark_Agent` 才有值，否则 NaN + 列出缺口。 +""" +from __future__ import annotations + +import math +from dataclasses import dataclass, field +from typing import Dict, List, Optional + +from ..baselines import BenchmarkRunRecord, compare, evaluate, quality, cost_efficiency +from ..metrics import ( + SwarmMetrics, swarm_score, benchmark_agent, completion_score, robustness_score, cost_score, +) + + +def _present(x) -> bool: + """True if x is a real number (not None / NaN).""" + return isinstance(x, (int, float)) and not (isinstance(x, float) and math.isnan(x)) + + +# Components of S_swarm and their source (record vs live run). +_SWARM_COMPONENTS = ("completion", "gain", "collaboration", "communication", + "cost", "robustness", "governance") + + +@dataclass +class SelfCertResult: + """One self-certification outcome for the swarm on a scenario/task set.""" + scenario: str + task_set_id: str + metrics: SwarmMetrics + coverage: Dict[str, bool] + comparisons: List[dict] # per-baseline G_E / G_E,c (from baselines.compare) + strongest_baseline: str # baseline used for the aggregate S_gain (highest Q_base) + swarm_valid: bool # beats ALL baselines on G_E and G_E,c + benchmark_available: bool + gaps: List[str] = field(default_factory=list) # unmet inputs blocking full Benchmark_Agent + + def to_dict(self) -> dict: + m = self.metrics + def num(x): + return None if (isinstance(x, float) and math.isnan(x)) else x + return { + "scenario": self.scenario, + "task_set_id": self.task_set_id, + "swarm_valid": self.swarm_valid, + "benchmark_available": self.benchmark_available, + "gaps": self.gaps, + "strongest_baseline": self.strongest_baseline, + "metrics": { + "s_completion": num(m.s_completion), "s_gain": num(m.s_gain), + "s_collaboration": num(m.s_collaboration), "s_communication": num(m.s_communication), + "s_cost": num(m.s_cost), "s_robustness": num(m.s_robustness), + "s_governance": num(m.s_governance), "s_swarm": num(m.s_swarm), + "reward": num(m.reward), "g_e": num(m.g_e), "g_e_cost": num(m.g_e_cost), + "benchmark": num(m.benchmark), + }, + "coverage": self.coverage, + "comparisons": self.comparisons, + } + + +class SelfCertCollector: + """Assemble a self-certification from suite records (+ optional live SwarmMetrics).""" + + def __init__(self, records: Dict[str, BenchmarkRunRecord], *, + live: Optional[SwarmMetrics] = None, + live_coverage: Optional[Dict[str, bool]] = None, + symmetric_cost: bool = True): + if "swarm" not in records: + raise ValueError("records must include a 'swarm' record") + baselines = {s: r for s, r in records.items() if s != "swarm"} + if not baselines: + raise ValueError("need at least one baseline record to compute emergence gain") + self.swarm = records["swarm"] + self.baselines = baselines + self.live = live + self.live_cov = live_coverage or {} + self.symmetric_cost = symmetric_cost + self.coverage: Dict[str, bool] = {} + + def _live(self, field_name: str): + """Pull a live-run metric only if its coverage flag says it's real.""" + if self.live is not None and self.live_cov.get(field_name): + return getattr(self.live, field_name) + return math.nan + + def collect(self) -> SelfCertResult: + swarm = self.swarm + + # --- emergence vs each baseline (defined: baselines.compare) --- + comparisons = [compare(swarm, b, symmetric_cost=self.symmetric_cost) + for b in self.baselines.values()] + ev = evaluate(swarm, list(self.baselines.values()), symmetric_cost=self.symmetric_cost) + + # Aggregate S_gain = G_E vs the STRONGEST baseline (max Q_base → min G_E, most conservative). + strongest_sys = max(self.baselines, key=lambda s: quality(self.baselines[s])) + strongest = next(c for c in comparisons if c["base_system"] == strongest_sys) + g_e = strongest["g_e"] + g_e_cost = strongest["g_e_cost"] + s_gain = g_e # v2.1-impl: S_gain ≡ G_E (emergence-evaluation §6) + self.coverage["g_e"] = self.coverage["g_e_cost"] = self.coverage["s_gain"] = True + + # --- S_swarm components --- + s_completion = completion_score(swarm.completed_tasks, swarm.total_tasks) + self.coverage["s_completion"] = swarm.total_tasks > 0 + + if swarm.actual_cost_usd > 0: + s_cost = cost_score(swarm.budget_usd, swarm.actual_cost_usd) + self.coverage["s_cost"] = True + else: + s_cost, self.coverage["s_cost"] = math.nan, False + + s_robustness = robustness_score(swarm.recovered_failures, swarm.total_failures) + self.coverage["s_robustness"] = True + + # Collaboration / communication / governance come ONLY from a live run (topology record + # has no handoff/peer/approval data). Absent → NaN + coverage False (no fabrication). + s_collaboration = self._live("s_collaboration") + self.coverage["s_collaboration"] = _present(s_collaboration) + s_communication = self._live("s_communication") + self.coverage["s_communication"] = _present(s_communication) + s_governance = self._live("s_governance") + self.coverage["s_governance"] = _present(s_governance) + + # S_swarm only if every component is real (NaN propagates → honest NaN aggregate). + comp_values = { + "completion": s_completion, "gain": s_gain, "collaboration": s_collaboration, + "communication": s_communication, "cost": s_cost, "robustness": s_robustness, + "governance": s_governance, + } + if all(_present(v) for v in comp_values.values()): + s_swarm = swarm_score(**comp_values) + self.coverage["s_swarm"] = True + else: + s_swarm, self.coverage["s_swarm"] = math.nan, False + + # --- Reward: from the live run (topology record lacks g_gov/p_risk/p_rework inputs) --- + reward_value = self._live("reward") + self.coverage["reward"] = _present(reward_value) + + # --- carry through decision-layer signals if the live run had them --- + tau = self._live("tau"); eta = self._live("eta"); p_decision = self._live("p_decision") + for k, v in (("tau", tau), ("eta", eta), ("p_decision", p_decision)): + self.coverage[k] = _present(v) + + # --- Benchmark_Agent inputs O and Gov --- + observability = math.nan # standard gives NO formula for O (telemetry-architecture) + self.coverage["observability"] = False + gov = s_governance # Gov ≈ governance capability (governance-score.md) + self.coverage["gov"] = _present(gov) + + # --- Benchmark_Agent: only when ALL five λ-inputs are real --- + gaps: List[str] = [] + if not _present(s_swarm): + missing = [k for k, v in comp_values.items() if not _present(v)] + gaps.append(f"S_swarm 不完整（缺分量: {', '.join(missing)}；需活体 run）") + if not _present(reward_value): + gaps.append("Reward 缺（需绑定 fixture 的活体 run）") + if not _present(observability): + gaps.append("O 可观测性：标准无公式（telemetry-architecture）") + if not _present(gov): + gaps.append("Gov：治理计数器未实现（governance-score.md）") + if not gaps and _present(g_e): + benchmark = benchmark_agent(s_swarm=s_swarm, g_e=g_e, reward=reward_value, + observability=observability, governance=gov) + self.coverage["benchmark"] = True + benchmark_available = True + else: + benchmark, self.coverage["benchmark"] = math.nan, False + benchmark_available = False + + metrics = SwarmMetrics( + tau=tau, eta=eta, p_decision=p_decision, reward=reward_value, + s_completion=s_completion, s_gain=s_gain, s_collaboration=s_collaboration, + s_communication=s_communication, s_cost=s_cost, s_robustness=s_robustness, + s_governance=s_governance, s_swarm=s_swarm, g_e=g_e, g_e_cost=g_e_cost, + benchmark=benchmark, + ) + return SelfCertResult( + scenario=swarm.scenario, task_set_id=swarm.task_set_id, metrics=metrics, + coverage=self.coverage, comparisons=comparisons, strongest_baseline=strongest_sys, + swarm_valid=ev["swarm_valid"], benchmark_available=benchmark_available, gaps=gaps, + ) diff --git a/benchmark/leaderboard/__init__.py b/benchmark/leaderboard/__init__.py index c6a1707..7933304 100644 --- a/benchmark/leaderboard/__init__.py +++ b/benchmark/leaderboard/__init__.py @@ -1,6 +1,62 @@ -"""Benchmark leaderboard — 排行榜聚合与展示字段。 +"""Benchmark leaderboard — 排行榜聚合与展示（标准 §11）。 -状态：**未落地（骨架）**。展示字段（标准 §11）：Benchmark_Agent / S_swarm / G_E / G_E,c / -Reward / Cost Efficiency；场景：Coding / Refactoring / Architecture / DevOps / Bug Fix。 -需定义 leaderboard schema 与持久化。 +展示字段：`Benchmark_Agent` / `S_swarm` / `G_E` / `G_E,c` / `Reward` / `Cost Efficiency`； +按场景（Coding / Refactoring / Architecture / DevOps / Bug Fix）分组。 + +输入为一组 `SelfCertResult`（见 collectors/selfcert_collector.py）。缺失/未覆盖的字段以 `None` +表示并在渲染时显示「—」，**不伪造分值**（组织规则 #9）。排序按 `Benchmark_Agent`（不可用者沉底）。 """ +from __future__ import annotations + +import math +from typing import List + +LEADERBOARD_FIELDS = ("benchmark_agent", "s_swarm", "g_e", "g_e_cost", "reward", "cost_efficiency") +SCENARIOS = ("coding", "refactoring", "architecture", "devops", "bugfix") + + +def _num(x): + return None if (x is None or (isinstance(x, float) and math.isnan(x))) else x + + +def build_leaderboard(results: List) -> dict: + """Turn SelfCertResult list into leaderboard rows (one per scenario/task_set).""" + rows = [] + for r in results: + m = r.metrics + # cost_efficiency = swarm CostEfficiency (= s_cost 口径, 100·Budget/ActualCost) + rows.append({ + "scenario": r.scenario, + "task_set_id": r.task_set_id, + "swarm_valid": r.swarm_valid, + "benchmark_available": r.benchmark_available, + "benchmark_agent": _num(m.benchmark), + "s_swarm": _num(m.s_swarm), + "g_e": _num(m.g_e), + "g_e_cost": _num(m.g_e_cost), + "reward": _num(m.reward), + "cost_efficiency": _num(m.s_cost), + }) + # Sort by Benchmark_Agent desc; unavailable (None) sink to the bottom. + rows.sort(key=lambda x: (x["benchmark_agent"] is not None, x["benchmark_agent"] or 0.0), + reverse=True) + return {"fields": list(LEADERBOARD_FIELDS), "rows": rows} + + +def render_markdown(board: dict) -> str: + lines = ["# Benchmark Leaderboard", "", + "| scenario | task_set | valid | Benchmark_Agent | S_swarm | G_E | G_E,c | Reward | CostEff |", + "|---|---|---|---|---|---|---|---|---|"] + + def cell(v): + return "—" if v is None else (round(v, 4) if isinstance(v, float) else v) + + for r in board["rows"]: + lines.append( + f"| {r['scenario']} | `{r['task_set_id']}` | {'✅' if r['swarm_valid'] else '🔴'} " + f"| {cell(r['benchmark_agent'])} | {cell(r['s_swarm'])} | {cell(r['g_e'])} " + f"| {cell(r['g_e_cost'])} | {cell(r['reward'])} | {cell(r['cost_efficiency'])} |") + lines.append("") + lines.append("> `—` = 未覆盖/不可用（缺真实输入，未伪造）。`Benchmark_Agent` 不可用通常因 " + "`O`（标准无公式）或 `Gov`（计数器未实现）——见各 run 的 gaps。") + return "\n".join(lines) diff --git a/docs/benchmark/emergence-evaluation.md b/docs/benchmark/emergence-evaluation.md index 6b94f5b..c48c8c4 100644 --- a/docs/benchmark/emergence-evaluation.md +++ b/docs/benchmark/emergence-evaluation.md @@ -61,8 +61,19 @@ Q = Q_quality = 0.4·TestPassRate + 0.3·CodeReviewScore + 0.3·UserAcceptance > 在以上落地前，**不得宣称已验证蜂群涌现能力**（重大能力缺口）。 -## 6. 待对齐 +## 6. 实现口径（v2.1-impl，2026-06-12 落地自证采集器时锁定） + +标准 §5.1 写 `S_gain` 「见第 6 节涌现增益」，但未给从 `G_E`（差值）到 `S_swarm` 分量的换算。落地采集器（`benchmark/collectors/selfcert_collector.py`）按以下口径执行，**不引入标准外的归一化/魔法系数**： + +- **`S_gain ≡ G_E`**：直接取涌现增益值。`S_swarm` 的若干分量（`V_speed`/`E_cost`/`S_cost`）本就可超过 100，故 `S_gain` 不强制归一到 `[0,100]`，与标准字面「见涌现增益」一致。 +- **聚合基线 = 最强基线**：当对 4 类基线分别得 `G_E_i` 时，进入 `S_swarm` 的单一 `S_gain` 取**对最强基线**（`Q_base` 最大者）的 `G_E`，即 `min_i G_E_i`——最保守口径，避免挑弱基线虚高。逐基线 `G_E_i / G_E,c_i` 仍全量保留在 leaderboard。 +- **`Q ≡ Q_quality`**（标准 §5.3，掩码归一见 `swarm-metrics-schema §4`）：与 §3 一致。 +- **成立硬条件不变**：`swarm_valid` 要求对**全部**基线 `G_E>0 且 G_E,c>0`（见 §2 与 `baselines.evaluate`）。 + +> ⚠️ 该口径为**实现级裁定**，已与 owner 对齐（「标准见 docs/benchmark/，S_gain 见涌现增益」）。若后续标准 v2.x 给出不同换算，以标准为准并同步本节。 + +## 7. 待对齐 -- `Q` 的唯一口径。 - 四类基线的标准实现（尤其 Strong Agent / Chain Agent / Sub-Agent 的定义边界）与统一数据集。 - 运行次数、方差/显著性门槛。 +- `Benchmark_Agent` 仍缺两分量：`O`（可观测性，标准无公式）、`Gov`（计数器未实现，见 governance-score.md）——采集器对二者诚实置 `NaN + coverage=False`，故完整 `Benchmark_Agent` 数字待这两项落地。 diff --git a/scripts/run-benchmark-suite.py b/scripts/run-benchmark-suite.py index 1ae4a4b..98e1c02 100644 --- a/scripts/run-benchmark-suite.py +++ b/scripts/run-benchmark-suite.py @@ -27,6 +27,8 @@ from benchmark.tasksets import load_taskset, available_tasksets from benchmark.runners import run_all, OfflineBackend, OpenAIBackend from benchmark.reports import build_report, render_markdown from benchmark.replay import save_archive +from benchmark.collectors.selfcert_collector import SelfCertCollector +from benchmark.leaderboard import build_leaderboard, render_markdown as render_leaderboard def main() -> int: @@ -51,6 +53,17 @@ def main() -> int: report = build_report(records, backend_name=args.backend, n_runs=args.n_runs) print(render_markdown(report)) + + # Self-certification: wire baselines into SwarmMetrics (S_gain ≡ G_E) → leaderboard. + # Without a live run, collaboration/communication/governance are NaN, so S_swarm and + # Benchmark_Agent stay honestly NaN (gaps printed) — this is the pipeline view. + selfcert = SelfCertCollector(records).collect() + print("\n" + render_leaderboard(build_leaderboard([selfcert]))) + if selfcert.gaps: + print("\nBenchmark_Agent 未产出，缺口：") + for g in selfcert.gaps: + print(f" - {g}") + if args.archive: path = save_archive(args.archive, args.run_label, records=records, report=report) print(f"\narchived → {path}") diff --git a/scripts/test-benchmark-selfcert.py b/scripts/test-benchmark-selfcert.py new file mode 100644 index 0000000..b71e39a --- /dev/null +++ b/scripts/test-benchmark-selfcert.py @@ -0,0 +1,104 @@ +"""Test SelfCertCollector + leaderboard (阶段1 自证采集器). + +Hermetic, no Redis / no model. Verifies: + - S_gain ≡ G_E vs the STRONGEST baseline (most conservative), g_e/g_e_cost wired. + - Without a live run: collaboration/communication/governance NaN → S_swarm NaN, + Benchmark_Agent NOT available, gaps list the honest blockers. + - With a live run covering those: S_swarm + Reward real, but Benchmark_Agent STILL NaN + because O (observability) has no formula in the standard → gaps == [O]. + - leaderboard builds and sorts; nothing is fabricated (NaN stays NaN). + +Run: python scripts/test-benchmark-selfcert.py +""" +import math +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + +from benchmark.baselines import BenchmarkRunRecord, quality +from benchmark.metrics import SwarmMetrics +from benchmark.collectors.selfcert_collector import SelfCertCollector +from benchmark.leaderboard import build_leaderboard, render_markdown + +failures = [] + + +def check(name, cond): + print(("PASS" if cond else "FAIL"), "-", name) + if not cond: + failures.append(name) + + +def rec(system, *, test_pass_rate, n_agent, budget, actual, total=4, completed=4): + return BenchmarkRunRecord( + system=system, scenario="coding", task_set_id="coding-set-1", n_agent=n_agent, + completed_tasks=completed, total_tasks=total, test_pass_rate=test_pass_rate, + budget_usd=budget, actual_cost_usd=actual, model_tokens=1000, + target_time_s=10.0, actual_time_s=8.0, recovered_failures=1, total_failures=2, + ) + + +def main(): + # swarm beats all baselines on raw quality; strongest baseline = 'strong' (Q=85). + records = { + "swarm": rec("swarm", test_pass_rate=90, n_agent=5, budget=15, actual=10), + "single": rec("single", test_pass_rate=70, n_agent=1, budget=12, actual=10), + "chain": rec("chain", test_pass_rate=75, n_agent=1, budget=12, actual=10), + "sub": rec("sub", test_pass_rate=80, n_agent=2, budget=12, actual=10), + "strong": rec("strong", test_pass_rate=85, n_agent=1, budget=12, actual=10), + } + + # --- 1. no live run: gain wired, but S_swarm/benchmark honestly NaN --- + res = SelfCertCollector(records).collect() + m = res.metrics + check("strongest baseline = strong", res.strongest_baseline == "strong") + # S_gain ≡ G_E vs strongest = 90 - 85 = 5 + check("s_gain == G_E vs strongest (5.0)", round(m.s_gain, 4) == 5.0 and round(m.g_e, 4) == 5.0) + check("g_e/g_e_cost covered", res.coverage["g_e"] and res.coverage["g_e_cost"]) + check("s_completion real (100)", round(m.s_completion, 1) == 100.0 and res.coverage["s_completion"]) + check("s_collaboration NaN (no live)", math.isnan(m.s_collaboration) and res.coverage["s_collaboration"] is False) + check("s_swarm NaN (missing live components)", math.isnan(m.s_swarm) and res.coverage["s_swarm"] is False) + check("benchmark NOT available", res.benchmark_available is False and math.isnan(m.benchmark)) + check("gaps mention O (no formula) and Gov", any("O " in g or "可观测性" in g for g in res.gaps) + and any("Gov" in g for g in res.gaps)) + + # --- 2. with a live run covering collaboration/communication/governance/reward --- + live = SwarmMetrics( + tau=math.nan, eta=math.nan, p_decision=math.nan, reward=72.0, + s_completion=math.nan, s_gain=math.nan, s_collaboration=88.0, s_communication=92.0, + s_cost=math.nan, s_robustness=math.nan, s_governance=100.0, + s_swarm=math.nan, g_e=math.nan, g_e_cost=math.nan, benchmark=math.nan, + ) + live_cov = {"s_collaboration": True, "s_communication": True, "s_governance": True, "reward": True} + res2 = SelfCertCollector(records, live=live, live_coverage=live_cov).collect() + m2 = res2.metrics + check("with live: s_swarm REAL", not math.isnan(m2.s_swarm) and res2.coverage["s_swarm"]) + check("with live: reward REAL (72)", round(m2.reward, 1) == 72.0 and res2.coverage["reward"]) + check("with live: benchmark STILL NaN (O has no formula)", + res2.benchmark_available is False and math.isnan(m2.benchmark)) + check("with live: ONLY remaining gap is O", len(res2.gaps) == 1 and ("O " in res2.gaps[0] or "可观测性" in res2.gaps[0])) + + # --- 3. offline-style: swarm == baselines → G_E=0, swarm not valid --- + flat = {s: rec(s, test_pass_rate=80, n_agent=(5 if s == "swarm" else 1), budget=12, actual=10) + for s in ("swarm", "single", "chain", "sub", "strong")} + res3 = SelfCertCollector(flat).collect() + check("flat quality → G_E=0", round(res3.metrics.g_e, 6) == 0.0) + check("flat quality → swarm NOT valid", res3.swarm_valid is False) + + # --- 4. leaderboard builds + sorts (unavailable sinks) --- + board = build_leaderboard([res, res2, res3]) + check("leaderboard has 3 rows", len(board["rows"]) == 3) + check("leaderboard renders", "Benchmark Leaderboard" in render_markdown(board)) + check("benchmark_agent shown as None (unavailable)", all(r["benchmark_agent"] is None for r in board["rows"])) + + print() + if failures: + print(f"{len(failures)} self-cert check(s) FAILED: {failures}") + return 1 + print("all benchmark self-cert checks passed") + return 0 + + +if __name__ == "__main__": + sys.exit(main())