benchmark: 落地决策层(τ/η/P)、质量(Q_quality)、通信遥测；关闭 #10 #23

四块互相交织的 benchmark 覆盖增量，统一提交：

1) 通信遥测（#23）：orchestrator 路由 peer 消息时按 correlation_id 计请求/应答到
   SwarmRun.collaboration（内部状态，不进 Manager 事件流）；collector 算 s_communication。
   治理计数由 run.approvals 派生（合规/总数）→ s_governance。

2) Q_quality 掩码归一（v2.1 裁定）：metrics.quality_score 改为对 present 输入加权归一，
   非编码任务自动忽略 TestPassRate，全缺 → NaN（不伪造）。

3) 质量插桩 / Group B：新增 Pod 内代码测试沙箱（orchestrator/sandbox.py，环境清洗 +
   超时强杀 + 资源限额 + 路径越界校验，门控 ENABLE_QUALITY_EVAL）与 held-out fixture
   （benchmark/fixtures/）；run 完成时用留出测试评分得 TestPassRate → Q_quality →
   collector 合成 reward。安全边界见 docs/integration/security-boundary.md §8.1。

4) 决策引擎 / Group A（#10，Option A score-at-pull）：新增 orchestrator/decision_engine.py
   —— 信息素 τ（Redis 持久、(role,agent) 键控、冷启动 0.5、ρ 蒸发、夹紧、学习常开）+
   η 启发式评分 + ε-greedy 概率采样；每次 dispatch 产一条 DecisionTrace →
   SwarmRun.decisions；collector 算 tau/eta/p_decision。概率选择门控 ENABLE_ACO_DISPATCH
   （默认关，CI 用 ACO_SEED 固定）。

覆盖：单次 run 真实可算字段由 4 提升至最多 10/15（新增 communication/reward/tau/eta/
p_decision，外加 governance 有条件）。

测试：新增 test-sandbox / test-quality / test-decision-engine；扩充 collector/metrics 用例；
CI 纳入全部 benchmark 套件 + flag-on 的 ACO e2e。本地 11 项 gate 全绿。

诚实边界（未越界声称）：
- Group A 为单边匹配（Option B 待 Group C）；概率派发优于贪心未证；默认关闭。
- reward 的 CodeReview/UserAcceptance 未采集（掩码忽略）；P_risk 为审批派生低估。
- s_gain/s_swarm/g_e/g_e_cost/benchmark 仍 NaN —— 需基线（#21/#13），本 PR 不动验收。

影响范围：Swarm（orchestrator + benchmark + docs + CI）。不改 Manager↔Swarm 事件契约
（遥测均为运行时内部状态）；不影响 Client/计费/密钥/发布链路。新增 ENABLE_QUALITY_EVAL /
ENABLE_ACO_DISPATCH 两个开关，默认关闭。

Closes #10
Closes #23

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

benchmark/README.md

@@ -7,7 +7,9 @@
 | 模块 | 内容 | 状态 |
 |---|---|---|
 | `metrics.py` | v2.0 `SwarmMetrics`（15 字段）+ **纯公式**（`pheromone`/`heuristic`/`p_decision`/`reward`/`completion`/`collaboration`/`communication`/`cost`/`robustness`/`governance`/`swarm_score`/`emergence_gain`/`swarm_cost`/`cost_normalized_gain`/`benchmark_agent`）+ v2.0 推荐权重常量 | ✅ 已实现、可单测（`test-benchmark-metrics.py`） |
-| `collectors/` | base `SwarmMetricsCollector` + `SwarmRunMetricsCollector`（从真实 run 计算 `s_completion`/`s_collaboration`/`s_cost`/`s_robustness`；其余标记 NaN + `coverage=False`） | 🟡 部分落地（4/15 字段真实可算） |
+| `collectors/` | base `SwarmMetricsCollector` + `SwarmRunMetricsCollector`（无条件计算 `s_completion`/`s_collaboration`/`s_robustness`，有条件计算 `s_cost`/`s_governance`/`s_communication`/`reward`/`tau`/`eta`/`p_decision`；其余标记 NaN + `coverage=False`） | 🟡 部分落地（单次 run 最多 10/15 字段真实可算） |
+| `fixtures/` | 留出（held-out）任务 fixture + 加载器；`reward` 的 `Q_quality` 评分源 | 🟡 1 个示例 fixture（`add_function`），统一任务集待扩充 |
+| 决策引擎 | `orchestrator/decision_engine.py`（τ trail + η 评分 + ε-greedy 采样，`ENABLE_ACO_DISPATCH` 门控；见 docs/benchmark/decision-engine.md） | 🟡 Option A 单边已落地；Option B 与决策质量验证待 Group C |
 | `baselines/` | Single / Chain / Sub-Agent / Strong 基线运行器 | 🔴 未落地 |
 | `replay/` | 执行回放 | 🔴 未落地 |
 | `leaderboard/` | 排行榜聚合 | 🔴 未落地 |

benchmark/collectors/run_collector.py

@@ -4,12 +4,16 @@ Wires benchmark.metrics formulas to a run's tasks + event stream (orchestrator s
 Only metrics with real data sources are computed; the rest are returned as NaN and flagged
 in `coverage` (False) — we do NOT fake a 0/100 score for uncollected metrics (rule #9).
 
+Collected when the run exercises the relevant path (else NaN + coverage False):
+- completion, collaboration, robustness → always, from tasks + handoff events
+- cost          → when the plan has a budget and tasks report usage
+- governance    → when the run had governed ops / approvals
+- communication → when peers exchanged messages (request→reply rate, internal telemetry)
+- reward        → when the run was graded against a fixture (Q_quality real) + speed/cost present
+- tau/eta/p_decision → when ENABLE_ACO_DISPATCH sampled assignments (recorded decisions)
+
 Uncollected today (need work flagged in docs/benchmark/swarm-metrics-schema.md):
 - gain          → needs baselines (emergence-evaluation)
-- communication → needs agent message telemetry (not counted yet)
-- p_decision    → needs τ/η decision scoring (not implemented)
-- reward        → needs weights (standard gives no numeric w*)
-- governance    → only derivable from approvals; NaN when a run has no governed ops
 """
 from __future__ import annotations
 
@@ -19,8 +23,8 @@ from collections import Counter
 
 from . import SwarmMetricsCollector
 from ..metrics import (
-    SwarmMetrics, completion_score, collaboration_score, cost_score, robustness_score,
-    governance_score,
+    SwarmMetrics, completion_score, collaboration_score, communication_score, cost_score,
+    robustness_score, governance_score, reward, speed_score, rework_penalty,
 )
 
 
@@ -110,23 +114,85 @@ class SwarmRunMetricsCollector(SwarmMetricsCollector):
             s_governance = math.nan
             self.coverage["s_governance"] = False
 
+        # --- s_communication (real only if the run exchanged peer messages) ---
+        # successful = peer requests that received a matching reply (by correlation_id);
+        # total = distinct peer requests routed. A run with no peer collaboration → NaN.
+        collab = run.collaboration or {}
+        requests = set(collab.get("request_correlations") or [])
+        replies = set(collab.get("reply_correlations") or [])
+        if requests:
+            answered = len(requests & replies)
+            s_communication = communication_score(answered, len(requests))
+            self.coverage["s_communication"] = True
+        else:
+            s_communication = math.nan
+            self.coverage["s_communication"] = False
+
+        # --- reward (real only when the run was graded against a fixture: Group B) ---
+        # Needs Q_quality (fixture TestPassRate, masked-renormalized), a V_speed target_time, and a
+        # real E_cost. r_robust/g_gov derive from the run; p_rework from retries; p_risk from the
+        # approval risk levels (0 when no governed op was observed — a known under-approximation tied
+        # to the governance coverage gap, see docs/benchmark/metric-coverage-gaps.md).
+        quality = run.quality or {}
+        q_quality = quality.get("q_quality")
+        target_time = quality.get("target_time_seconds")
+        duration = max(0.0, float(run.updated_at or 0) - float(run.created_at or 0))
+        have_quality = isinstance(q_quality, (int, float)) and not math.isnan(q_quality)
+        have_speed = isinstance(target_time, (int, float)) and target_time and duration > 0
+        if have_quality and have_speed and self.coverage["s_cost"]:
+            rework_count = sum(1 for t in tasks if getattr(t, "retry_count", 0) > 0)
+            risky = sum(1 for a in approvals
+                        if str(a.get("risk_level", "")).lower() in {"high", "critical"})
+            p_risk = (100.0 * risky / len(approvals)) if approvals else 0.0
+            g_gov = governance_score(
+                sum(1 for a in approvals if a.get("decision") in ("approved", "rejected")),
+                len(approvals),
+            )  # empty -> 100 (no governed-op violations observed)
+            reward_value = reward(
+                s_task=s_completion,
+                q_quality=float(q_quality),
+                v_speed=speed_score(float(target_time), duration),
+                e_cost=s_cost,
+                r_robust=s_robustness,
+                g_gov=g_gov,
+                p_risk=p_risk,
+                p_rework=rework_penalty(rework_count, total),
+            )
+            self.coverage["reward"] = True
+        else:
+            reward_value = math.nan
+            self.coverage["reward"] = False
+
+        # --- tau / eta / p_decision (real only when ACO dispatch recorded decisions: Group A) ---
+        # Run-level value = mean over the run's sampled assignments. p_decision uses the
+        # standard's §3.3 score (τ^α·η^β·100), recorded per decision as p_score. No decisions
+        # (flag off, or no ACO-dispatched task) → NaN.
+        decisions = [d for d in (run.decisions or [])
+                     if isinstance(d.get("tau"), (int, float)) and isinstance(d.get("eta"), (int, float))]
+        if decisions:
+            tau = sum(d["tau"] for d in decisions) / len(decisions)
+            eta = sum(d["eta"] for d in decisions) / len(decisions)
+            p_dec = sum(float(d.get("p_score") or 0.0) for d in decisions) / len(decisions)
+            self.coverage["tau"] = self.coverage["eta"] = self.coverage["p_decision"] = True
+        else:
+            tau = eta = p_dec = math.nan
+            self.coverage["tau"] = self.coverage["eta"] = self.coverage["p_decision"] = False
+
         # --- not yet collectable (see docs/benchmark/metric-coverage-gaps.md) ---
-        # s_gain needs baselines; s_communication needs message telemetry; tau/eta/p_decision need
-        # the decision-layer signals; reward needs quality/risk/rework inputs; s_swarm/g_e/g_e_cost/
-        # benchmark depend on the above (any NaN component → NaN aggregate).
-        for k in ("tau", "eta", "p_decision", "reward", "s_gain", "s_communication",
-                  "s_swarm", "g_e", "g_e_cost", "benchmark"):
+        # s_gain needs baselines; s_swarm/g_e/g_e_cost/benchmark depend on it
+        # (any NaN component → NaN aggregate).
+        for k in ("s_gain", "s_swarm", "g_e", "g_e_cost", "benchmark"):
             self.coverage[k] = False
 
         return SwarmMetrics(
-            tau=math.nan,
-            eta=math.nan,
-            p_decision=math.nan,
-            reward=math.nan,
+            tau=tau,
+            eta=eta,
+            p_decision=p_dec,
+            reward=reward_value,
             s_completion=s_completion,
             s_gain=math.nan,
             s_collaboration=s_collaboration,
-            s_communication=math.nan,
+            s_communication=s_communication,
             s_cost=s_cost,
             s_robustness=s_robustness,
             s_governance=s_governance,

benchmark/fixtures/__init__.py

@@ -0,0 +1,88 @@
+"""Benchmark task fixtures with HELD-OUT acceptance tests.
+
+A fixture pairs a task objective with authoritative tests that the swarm never sees. These tests
+— not the swarm's own testing-agent output — are what produce TestPassRate (Owner ruling: avoid
+self-grading; the swarm's own tests are kept as a separate signal in quality.py).
+
+Layout per fixture:  benchmark/fixtures/<id>/fixture.json  +  <id>/tests/test_*.py
+
+`fixture.json` schema:
+    {
+      "id": "add_function",
+      "objective": "...",                 # what the swarm is asked to build
+      "required_capabilities": [...],      # used to decide expects_code()
+      "entrypoint": "calc.py",             # the module the tests import (informational)
+      "target_time_seconds": 60,           # V_speed target for reward()
+      "test_files": ["tests/test_add.py"]  # held-out acceptance tests (relative to <id>/)
+    }
+"""
+from __future__ import annotations
+
+import json
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import List, Optional
+
+_FIXTURE_ROOT = Path(__file__).resolve().parent
+
+# Capabilities that imply the deliverable is runnable/testable code → TestPassRate applies.
+CODE_CAPABILITIES = {"python", "code_generation", "coding", "testing", "pytest", "implementation"}
+
+
+@dataclass
+class FixtureTest:
+    path: str
+    content: str
+
+
+@dataclass
+class Fixture:
+    id: str
+    objective: str
+    required_capabilities: List[str] = field(default_factory=list)
+    entrypoint: Optional[str] = None
+    target_time_seconds: Optional[float] = None
+    test_files: List[FixtureTest] = field(default_factory=list)
+
+    @property
+    def expects_code(self) -> bool:
+        return expects_code(self.required_capabilities)
+
+
+def expects_code(required_capabilities) -> bool:
+    """A task is code-shaped (TestPassRate applies) if it requires any coding capability."""
+    return bool(CODE_CAPABILITIES & {str(c).lower() for c in (required_capabilities or [])})
+
+
+def fixture_dir(fixture_id: str) -> Path:
+    return _FIXTURE_ROOT / fixture_id
+
+
+def available_fixtures() -> List[str]:
+    return sorted(
+        p.name for p in _FIXTURE_ROOT.iterdir()
+        if p.is_dir() and (p / "fixture.json").exists()
+    )
+
+
+def load_fixture(fixture_id: str) -> Fixture:
+    base = fixture_dir(fixture_id)
+    meta_path = base / "fixture.json"
+    if not meta_path.exists():
+        raise FileNotFoundError(f"unknown fixture: {fixture_id}")
+    meta = json.loads(meta_path.read_text(encoding="utf-8"))
+    tests = []
+    for rel in meta.get("test_files", []):
+        tpath = base / rel
+        if not tpath.exists():
+            raise FileNotFoundError(f"fixture {fixture_id} missing test file: {rel}")
+        # Flatten to a basename so the sandbox runner (which scans CWD for test_*.py) finds it.
+        tests.append(FixtureTest(path=Path(rel).name, content=tpath.read_text(encoding="utf-8")))
+    return Fixture(
+        id=meta.get("id", fixture_id),
+        objective=meta.get("objective", ""),
+        required_capabilities=meta.get("required_capabilities", []),
+        entrypoint=meta.get("entrypoint"),
+        target_time_seconds=meta.get("target_time_seconds"),
+        test_files=tests,
+    )

benchmark/fixtures/add_function/fixture.json

@@ -0,0 +1,8 @@
+{
+  "id": "add_function",
+  "objective": "Implement a Python function add(a, b) in calc.py that returns the sum of two numbers.",
+  "required_capabilities": ["python", "code_generation"],
+  "entrypoint": "calc.py",
+  "target_time_seconds": 60,
+  "test_files": ["tests/test_add.py"]
+}

benchmark/fixtures/add_function/tests/test_add.py

@@ -0,0 +1,22 @@
+"""Held-out acceptance tests for the add_function fixture.
+
+The swarm never sees these — they are the authoritative grader for TestPassRate. The tests import
+the module the implementation specialist is asked to produce (calc.py with add(a, b)).
+"""
+from calc import add
+
+
+def test_add_positive():
+    assert add(2, 3) == 5
+
+
+def test_add_with_zero():
+    assert add(0, 7) == 7
+
+
+def test_add_negative():
+    assert add(-4, -6) == -10
+
+
+def test_add_commutative():
+    assert add(5, 9) == add(9, 5)

benchmark/metrics.py

@@ -7,7 +7,9 @@ provided as defaults below.
 """
 from __future__ import annotations
 
+import math
 from dataclasses import dataclass
+from typing import Optional
 
 # --- v2.0 recommended weights / hyperparameters ---
 TAU_WEIGHTS = {"success": 0.25, "quality": 0.20, "acceptance": 0.20,
@@ -16,6 +18,8 @@ ETA_WEIGHTS = {"match": 0.25, "urgency": 0.15, "dependency": 0.15, "resource": 0
                "confidence": 0.10, "risk": 0.10, "budget_pressure": 0.10}
 REWARD_WEIGHTS = {"s_task": 0.20, "q_quality": 0.20, "v_speed": 0.12, "e_cost": 0.13,
                   "r_robust": 0.13, "g_gov": 0.10, "p_risk": 0.07, "p_rework": 0.05}
+# Sub-weights inside Q_quality (v2.0 §4.1). Masked & renormalized per task — see quality_score().
+QUALITY_WEIGHTS = {"test_pass_rate": 0.4, "code_review_score": 0.3, "user_acceptance": 0.3}
 SWARM_WEIGHTS = {"completion": 0.25, "gain": 0.20, "collaboration": 0.15, "communication": 0.10,
                  "cost": 0.10, "robustness": 0.10, "governance": 0.10}
 LAMBDA_WEIGHTS = {"lambda1": 0.30, "lambda2": 0.20, "lambda3": 0.25, "lambda4": 0.15, "lambda5": 0.10}  # Σ = 1.0
@@ -87,8 +91,36 @@ def p_decision(tau: float, eta: float, alpha: float = THETA_DEFAULTS["alpha"],
 
 
 # --- 执行层（v2.0 §4） ---
-def quality_score(test_pass_rate: float, code_review_score: float, user_acceptance: float) -> float:
-    return 0.4 * test_pass_rate + 0.3 * code_review_score + 0.3 * user_acceptance
+def quality_score(
+    test_pass_rate: Optional[float] = None,
+    code_review_score: Optional[float] = None,
+    user_acceptance: Optional[float] = None,
+    *,
+    weights: dict = QUALITY_WEIGHTS,
+) -> float:
+    """Q_quality as a MASKED, RENORMALIZED weighted mean over the inputs that apply (v2.1 ruling).
+
+    An input that does not apply to the task is passed as ``None`` and drops out of BOTH the
+    weighted sum and the weight normalizer, so the score stays on [0,100] and remains comparable
+    across task types. Canonical case: a non-code prompt has no tests → ``test_pass_rate=None`` →
+    Q_quality is renormalized over {code_review_score, user_acceptance}. The same masking applies
+    to any absent input, not just test_pass_rate.
+
+        Q_quality = Σ_{i∈present} wᵢ·xᵢ / Σ_{i∈present} wᵢ
+
+    All inputs absent → NaN (org rule #9: never fabricate a 0/100 for an uncollected quantity).
+    With all three present and the default weights this reduces to the v2.0 0.4/0.3/0.3 blend.
+    """
+    inputs = {
+        "test_pass_rate": test_pass_rate,
+        "code_review_score": code_review_score,
+        "user_acceptance": user_acceptance,
+    }
+    num = sum(weights[k] * x for k, x in inputs.items() if x is not None)
+    denom = sum(weights[k] for k, x in inputs.items() if x is not None)
+    if not denom:
+        return math.nan
+    return num / denom
 
 
 def speed_score(target_time: float, actual_time: float) -> float: