diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 777516a..e233795 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -70,3 +70,19 @@ jobs: - name: Baseline comparison env: { REDIS_FAKE: "1" } run: python scripts/test-baseline-comparison.py + + - name: Code sandbox (in-pod test runner) + run: python scripts/test-sandbox.py + + - name: Quality instrumentation (Group B) + env: { REDIS_FAKE: "1" } + run: python scripts/test-quality.py + + - name: ACO decision engine (Group A) + env: { REDIS_FAKE: "1" } + run: python scripts/test-decision-engine.py + + # Same e2e workflow, but through the probabilistic ACO dispatch path (seeded). + - name: End-to-end workflow test (ACO dispatch on) + env: { REDIS_FAKE: "1", ENABLE_ACO_DISPATCH: "1", ACO_SEED: "42" } + run: python scripts/test-workflow-e2e.py diff --git a/CLAUDE.md b/CLAUDE.md index 777a22f..db2e93c 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -22,7 +22,9 @@ ## 架构与关键约束（便于定位） - **orchestrator/**：FastAPI 编排器。Manager 面接口、HMAC 签名回调、审批链**必须保持契约**。Redis 为权威存储；内存回退仅限 `REDIS_FAKE` / `ALLOW_MEMORY_STORE`（开发/CI）。 - **agent/**：执行单元，**OpenAI 兼容**模型；保留计费/审计归属（`usage` 与 `X-Agent/X-Agnet` 头）。 -- **工作流开关默认关闭**：`ENABLE_PLANNER_FALLBACK`、`ENABLE_REVIEW_LOOP`、`ENABLE_SUBTASK_HANDOFF`。 +- **工作流开关默认关闭**：`ENABLE_PLANNER_FALLBACK`、`ENABLE_REVIEW_LOOP`、`ENABLE_SUBTASK_HANDOFF`、`ENABLE_QUALITY_EVAL`、`ENABLE_ACO_DISPATCH`。 +- **ACO 决策引擎**（`orchestrator/decision_engine.py`）：信息素**学习常开**（被动观察，不改行为）；**概率选择仅在** `ENABLE_ACO_DISPATCH=1` 时生效（改变派发顺序，CI 用 `ACO_SEED` 固定随机数）。设计见 `docs/benchmark/decision-engine.md`。 +- **代码测试沙箱**（`orchestrator/sandbox.py`）：会**执行模型生成代码**，OS 级隔离边界 = K8s Pod；默认关闭，仅 `ENABLE_QUALITY_EVAL=1` 且在隔离 Pod 内启用。安全模型见 `docs/integration/security-boundary.md §8.1`。**不得在隔离 Pod 之外开启。** - 提交前必须本地通过： ``` python scripts/test-runtime-contract.py diff --git a/benchmark/README.md b/benchmark/README.md index 75ca00d..ffe6dc0 100644 --- a/benchmark/README.md +++ b/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/` | 排行榜聚合 | 🔴 未落地 | diff --git a/benchmark/collectors/run_collector.py b/benchmark/collectors/run_collector.py index 8cb3292..7b24c2f 100644 --- a/benchmark/collectors/run_collector.py +++ b/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, diff --git a/benchmark/fixtures/__init__.py b/benchmark/fixtures/__init__.py new file mode 100644 index 0000000..8ec78fa --- /dev/null +++ b/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//fixture.json + /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 /) + } +""" +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, + ) diff --git a/benchmark/fixtures/add_function/fixture.json b/benchmark/fixtures/add_function/fixture.json new file mode 100644 index 0000000..b5e6ca4 --- /dev/null +++ b/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"] +} diff --git a/benchmark/fixtures/add_function/tests/test_add.py b/benchmark/fixtures/add_function/tests/test_add.py new file mode 100644 index 0000000..2bc799d --- /dev/null +++ b/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) diff --git a/benchmark/metrics.py b/benchmark/metrics.py index 8889d84..80938e1 100644 --- a/benchmark/metrics.py +++ b/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: diff --git a/docs/benchmark/IMPORTANT-metric-coverage-gaps.md b/docs/benchmark/IMPORTANT-metric-coverage-gaps.md index 47f0d81..12ca440 100644 --- a/docs/benchmark/IMPORTANT-metric-coverage-gaps.md +++ b/docs/benchmark/IMPORTANT-metric-coverage-gaps.md @@ -8,24 +8,28 @@ 本仓最初是**多 Agent 工作流执行器**，不是**被插桩的基准测量目标**。能算出来的指标是「执行过程本就会产生」的副产物；其余指标各自缺少执行器从不需要产生的东西（基线 / 计数器 / 决策机制 / 输入）。**v2.0 已给定全部权重（τ/η/reward/λ）**，但这不改变「输入/机制缺失」的根因。 -## 2. 覆盖现状（v2.0 `SwarmMetrics` 共 15 字段，4 个真实可算） +## 2. 覆盖现状（v2.0 `SwarmMetrics` 共 15 字段） + +无条件真实可算 3 项（completion/collaboration/robustness）；有条件真实可算 7 项（cost/governance/communication/reward/tau/eta/p_decision，仅在该 run 走过对应路径时）。即单次 run 最多 10 项真实可算，其余仍 NaN。 | 字段 | 状态 | 数据来源 / 缺口 | |---|---|---| | `s_completion` | ✅ 真实 | 任务状态统计 | | `s_collaboration` | ✅ 真实 | `handoff.*` 事件 + `depends_on` + `assigned_agent_id` | -| `s_cost` | ✅ 真实 | 每任务 `usage.model_cost_usd` + 请求 `budget.max_cost_usd` | +| `s_cost` | 🟡 部分 | 每任务 `usage.model_cost_usd` + 请求 `budget.max_cost_usd`；缺预算/用量 → NaN | | `s_robustness` | ✅ 真实 | `retry_count` + 任务状态 | | `s_governance` | 🟡 部分 | 仅审批可派生；无审批 → NaN | +| `s_communication` | 🟡 部分 | peer 消息内部计数（请求→应答率，按 `correlation_id`）；无 peer 通信 → NaN | +| `reward` | 🟡 部分 | 绑定 fixture 时：`Q_quality`=沙箱评测留出测试的 TestPassRate（掩码归一）、`V_speed`=fixture target_time vs 实际时长、`E_cost`/`R_robust`/`G_gov`/`P_rework` 取自 run、`P_risk` 由审批风险派生（无受治理操作→0，见下注）；未绑定 fixture → NaN | +| `tau` / `eta` / `p_decision` | 🟡 部分 | ACO 决策引擎（`ENABLE_ACO_DISPATCH` 开启时按采样决策记录均值；学习常开但选择门控）；无决策记录 → NaN | | `s_gain` | 🔴 NaN | 需基线 | -| `s_communication` | 🔴 NaN | 无消息计数 | -| `tau` / `eta` / `p_decision` | 🔴 NaN | 无 τ/η 决策引擎 | -| `reward` | 🔴 NaN | 权重已给定，**输入**未采集 | -| `s_swarm` / `g_e` / `g_e_cost` / `benchmark` | 🔴 NaN | 依赖上述（任一 NaN → 聚合 NaN） | +| `s_swarm` / `g_e` / `g_e_cost` / `benchmark` | 🔴 NaN | 依赖上述（`gain` 仍 NaN → 聚合 NaN） | + +> ⚠️ `reward` 的 `P_risk` 由审批 `risk_level` 派生，**无受治理操作时记 0**——这是与治理覆盖缺口绑定的**已知低估**（未经审批门的风险面未被计入）。`P_rework` 目前仅由 `retry_count>0` 派生，未含评审重开计数。两者均在 collector 与本表中显式标注，不作隐藏假设。 > 不可算的指标返回 `NaN` 并在 `coverage` 标记 `False`，**不伪造 0/100 分值**（组织规则 #9）。 -## 3. 为什么这 4 个能算 +## 3. 为什么基础几项能算 它们都是执行器正常运行的副产物，已存在于 orchestrator 状态： - `completion` ← 队列本就跟踪任务状态。 @@ -33,23 +37,29 @@ - `cost` ← 计费归因本就记录每任务 `model_cost_usd`；预算在请求里。 - `robustness` ← 重试逻辑本就维护 `retry_count` 与状态。 -## 4. 为什么这 5 个算不出（逐项根因） +`communication`/`reward` 不是天然副产物，而是本仓新增插桩（通信计数器 / fixture 沙箱评测）后才可算——见 §4 顶部「已关闭」。 + +## 4. 为什么这几个算不出（逐项根因） + +> ✅ **已关闭（本仓）**： +> - `communication` —— peer 消息原先只路由不计数，现已在 orchestrator 路由处按 `correlation_id` 计请求/应答（`SwarmRun.collaboration` 内部计数，不进 Manager 事件流），collector 据此算 `SuccessfulMessages/TotalMessages×100`；无 peer 通信的 run 仍 NaN（不伪造）。 +> - `reward`（Group B）—— 已建**留出测试沙箱**（`orchestrator/sandbox.py`，Pod 内执行 + 环境清洗 + 超时/限额）与 **fixture**（`benchmark/fixtures/`）：绑定 fixture 的 run 在完成时用留出测试评分得 `TestPassRate` → 经掩码 `quality_score` 得 `Q_quality`，再由 collector 合成 `reward`。`CodeReview`/`UserAcceptance` 仍未采集（掩码自动忽略）；`P_risk` 为审批派生的已知低估。未绑定 fixture 的 run 仍 NaN。 +> - `tau`/`eta`/`p_decision`（Group A）—— 已建 **ACO 决策引擎**（`orchestrator/decision_engine.py`）：信息素 trail（Redis 持久、按 `(role, agent)` 键控、**学习常开**）+ η 启发式评分 + ε-greedy 概率采样（**选择门控** `ENABLE_ACO_DISPATCH`，默认关）。决策遥测落 `SwarmRun.decisions`，collector 取均值。**单边匹配（Option A）**、τ 的 quality 输入暂 = success、**决策质量未证（需 Group C）**——均已在 decision-engine.md 标注。 | 指标 | 根因（缺什么） | 类别 | 关闭成本 | |---|---|---|---| | `gain`（`G_E = Q_swarm − Q_base`） | **本质是对比指标**，单次 swarm run 无法自算；缺 baseline 运行器（Single/Strong/Chain/Sub-Agent）、对比 harness 与数据集 | 缺**基线** | 高（跨团队/基础设施/设计决策） | -| `communication`（`Successful/Total messages`） | peer/WS 消息只**路由**、从不**计数**；无成功/总数计数器，数据流过但未插桩 | 缺**计数器** | 低（加计数器） | | `governance`（`Compliant/Total ops`） | 有审批机制，但无「受治理/敏感操作 vs 合规」计数；更广的治理面（tool/MCP 权限、allowed_paths 强制）未实现，**没有受治理操作记录可计数**；无审批的 run → 0 操作 → 空集 → NaN | 缺**计数器 + 策略强制点** | 中（计数器 + 部分强制点） | -| `p_decision`（`τ^α·η^β·100`） | v2.0 已给公式，但派发仍是**确定性贪心能力匹配**（`can_agent_run_task`）；ACO 决策模型**未实现**——无信息素 `τ`（历史有效性追踪）、无启发式 `η` 评分；没有可测的概率决策 | 缺**整套机制** | 高（新建决策引擎 + 历史库） | -| `reward`（`w₁·S_task + … − w₈·P_rework`） | v2.0 **已给定 `w₁..w₈`**；但**输入**未采集（`Q_quality` 需 TestPass/CodeReview/UserAcceptance、`P_risk`、`P_rework`）→ 仍不可算 | 缺**输入**（权重已定） | 中（质量/CI/风险/返工接入） | +| `reward` 的剩余短板 | 主输入 `Q_quality` 已通过 fixture 沙箱采集；但 `CodeReview`/`UserAcceptance` 仍缺（需评审/验收信号源），`P_risk` 仅审批派生（低估），fixture 只有 1 个示例、统一任务集未建 | 缺**评审/验收信号 + 统一任务集** | 中（评审/验收接入 + 任务集扩充） | +| `tau`/`eta`/`p_decision` 的剩余短板 | 机制已建但**单边**（任务对 Agent 的全局路由 = Option B 未实现）；τ 的 quality/acceptance/time 输入无信号暂 0 或 = success；**概率派发是否优于贪心未证**（需 Group C 对比） | 缺**双边匹配 + 决策质量验证** | 中-高（matchmaker + 对比 harness） | ## 5. 关闭路径（按成本排序） -1. **低成本（本仓可做）**：`s_communication`、`s_governance` — 在 peer/WS 发送处与受治理操作处加计数器，按 `swarm_id` 聚合。可把真实可算项从 4 提到 6。 -2. **中成本**：`reward` — 权重 v2.0 已定；接入 `Q_quality`（CI/评审/验收）与 risk/rework 输入即可算。 +1. **低成本（本仓可做）**：~~`s_communication`~~（✅ 已关闭）、`s_governance` — 通信计数器已落地；治理仍需在受治理操作处加计数器 + 策略强制点，按 `swarm_id` 聚合。无条件真实项 3 项，有条件已增至 4 项（cost/governance/communication/reward）。 +2. **中成本**：~~`reward`~~（✅ 已关闭，本仓）— `Q_quality` 主输入已由 fixture 沙箱采集；剩余为**扩充统一任务集** + 接入 `CodeReview`/`UserAcceptance` 评审/验收信号（部分需 Product/Manager）。 3. **高成本**： - - `p_decision` — 设计并实现 `τ/η/P` 决策引擎与历史有效性追踪（改变派发机制）。 - - `gain`（及由其驱动的 `Benchmark_Agent`、`G_E,c`、「Swarm > baselines」验收）— 实现 4 类基线运行器 + 统一数据集 + 对比/显著性，属跨团队与基础设施工作（见 baseline-comparison）。 + - ~~`p_decision`~~（✅ 已关闭，本仓，Option A）— `τ/η/P` 决策引擎与信息素历史库已落地（`ENABLE_ACO_DISPATCH` 门控）；剩余为 Option B 双边匹配与决策质量验证（依赖 Group C）。 + - `gain`（及由其驱动的 `Benchmark_Agent`、`G_E,c`、「Swarm > baselines」验收）— 实现 4 类基线运行器 + 统一数据集 + 对比/显著性，属跨团队与基础设施工作（见 baseline-comparison）。**这是最后一块，也是唯一动验收的一块。** ## 6. 影响 diff --git a/docs/benchmark/decision-engine.md b/docs/benchmark/decision-engine.md new file mode 100644 index 0000000..3c1e7dc --- /dev/null +++ b/docs/benchmark/decision-engine.md @@ -0,0 +1,71 @@ +# ACO 决策引擎（Benchmark Group A）：τ / η / P_decision + +让标准 §3 的决策层指标由 NaN 转为可测：派发从「确定性贪心首配」变为「信息素 × 启发式的概率采样」。本文档是该机制的唯一入口。 + +## 1. 方案选型：Option A（score-at-pull，单边） + +派发是双边匹配（任务挑 Agent / Agent 挑任务）。本实现取 **Option A**： + +- 候选集 = **「当前空闲 Agent 视角下的全部就绪任务」**。Agent 由到达顺序固定，任务由 `P_i = τ_i^α·η_i^β / Σ` **采样**选出（ε-greedy 探索，ε=0.10）。 +- **已知局限（单边）**：τ 影响的是「该 Agent 先做哪个任务」，而非「该任务给谁做」——强专家若晚到，仍可能输给先拉取的弱者。全局双边匹配（Option B：matchmaker 调度 tick + hold 策略）**推迟到 Group C 基线就绪后**再评估，因为「更优匹配」是一个**当前无法度量**的优化命题。 +- A 是 B 的严格子集：信息素库、η 评分、采样、遥测、collector 全部复用，B 只需更换选择触发点。 + +## 2. 双半开关（核心设计） + +| 半边 | 开关 | 默认 | 行为 | +|---|---|---|---| +| **学习**（信息素沉积） | 无（常开） | 开 | 任务到达终态即更新 τ trail。纯被动观察，**不改变任何派发行为**；为未来开启选择积累「实测声誉」而非冷启动先验 | +| **行动**（概率选择） | `ENABLE_ACO_DISPATCH` | **关** | 关闭时派发与原贪心逐字节一致；开启时走采样路径并记录决策遥测 | + +CI 同时跑两条：默认全套（flag off，证不变性）+ `ENABLE_ACO_DISPATCH=1 ACO_SEED=42` 的 e2e（证开启后工作流仍完整收敛）。 + +## 3. 机制 + +### 3.1 信息素 τ（`pheromone:{agent_role}:{agent_id}`，Redis） + +- 更新（标准 §3.1）：`τ(t+1) = (1−ρ)·τ(t) + Δτ`，ρ=0.10，夹紧到 `[0.05, 1.0]`（正性保证 `τ^α` 良定义；上界即 trail 饱和）。 +- `Δτ = metrics.pheromone(...)`，**只喂真实信号**：success（1/0）、quality（暂 = success，run 级 fixture 评分发生在任务完成之后、沉积之时不可得）、cost（任务成本/run 预算，未知则 0）。acceptance/time/risk/rollback **无信号 → 0**，不伪造（规则 #9）。 +- 冷启动 τ₀ = 0.5（中性先验）；**单次 run 的 τ 多为先验**，需多 run 积累才是「挣来的声誉」——已如实标注。 +- trail 按 `(agent_role, agent_id)` 键控、跨 run 持久（这正是 τ 的意义）。 + +### 3.2 启发式 η（纯函数，无新状态） + +`metrics.heuristic(...)` 输入全部来自编排器既有状态：match（required∩caps 的 **Jaccard**，奖励专精）、urgency（任务等待时长，300s 饱和）、dependency（下游依赖数，3 饱和）、resource（Agent 空闲槽位）、confidence=0.5（无信号 → 对所有候选同值中性常数，不扭曲排序）、risk/budget_pressure=0（无信号）。下限夹紧 0.05。 + +### 3.3 选择与遥测 + +- `P_i = τ^α·η^β / Σ`（α=1, β=2），ε-greedy 均匀探索防饿死（低 τ Agent 仍偶得任务，从而有机会重建 trail）。 +- 每次采样记录 `{task_id, agent_id, tau, eta, p_norm, p_score, explored}` 到 `SwarmRun.decisions`（**内部状态，不进 Manager 事件流**，上限 1000 条）。 +- `p_score = τ^α·η^β·100`（标准 §3.3 的 P_decision 公式）；`p_norm` 为该候选集内的归一化概率，两者并存以免混淆。 +- collector：run 级 `tau`/`eta`/`p_decision` = 决策记录的均值；无记录 → NaN + `coverage=False`。 + +### 3.4 确定性与 CI + +随机源可经 `ACO_SEED` 固定（CI 用），生产不固定。超参可经 `ACO_ALPHA/ACO_BETA/ACO_RHO/ACO_EPSILON` 覆盖，默认取标准 §7.2 推荐值。 + +## 4. 文件 + +| 文件 | 职责 | +|---|---| +| `orchestrator/decision_engine.py` | τ 存取/沉积、η 评分、ε-greedy 采样、`Decision` 遥测载体 | +| `orchestrator/task_queue.py: get_ready_pending_tasks` | 候选枚举（不出队；选中后再 `remove_pending_task`） | +| `orchestrator/main.py` | 派发环 ACO 分支 + 四个任务终态处的 `deposit_pheromone` + 决策落库 | +| `orchestrator/swarm_runtime.py: SwarmRun.decisions / record_decision` | 决策遥测存储 | +| `benchmark/collectors/run_collector.py` | `tau`/`eta`/`p_decision` 聚合 | +| `scripts/test-decision-engine.py` | 22 项断言（trail 数学/夹紧/键控、η 排序、种子化采样分布、探索、遥测→collector、枚举不出队） | + +## 5. 与 #9（dispatch scoring）的边界 + +Issue #9（调度评分）与 #10（τ/η/P 决策模型）耦合但分工明确： + +- **#9 出「打分输入」**：候选 Agent×任务的可解释评分维度（capability_match、load、budget_pressure、risk、historical_success、estimated_cost/time），以及每次派发的候选列表与排除原因（`dispatch.decision_made`）。 +- **#10 出「概率决策」**：把这些维度归一为 τ（历史）与 η（先验），按 `P=τ^α·η^β/Σ` **概率采样**，并产出可回放的 `DecisionTrace`。 + +本仓 `decision_engine.py` 已实现 #10 的概率决策层与 τ/η 的最小喂入；#9 的「完整打分 schema + 候选/排除可解释性」为独立增量（`docs/scheduling/dispatch-score-schema.md`、`dispatch.decision_made` 事件），不在 #10 关闭范围内。 + +## 6. 诚实边界（本机制**未**关闭的） + +- **决策质量未证**：概率派发是否优于贪心是**经验命题**，需 Group C 的对比 harness 才能回答；在此之前生产默认关闭。 +- τ 的 quality 输入暂 = success（无每任务质量信号）；acceptance/time/risk 无信号 → 0。 +- 单边匹配（见 §1）；Option B 推迟。 +- `gain`/`s_swarm`/`g_e`/`g_e_cost`/`benchmark` **仍 NaN**——Group A 不动验收的针。 diff --git a/docs/benchmark/quality-instrumentation.md b/docs/benchmark/quality-instrumentation.md new file mode 100644 index 0000000..cd1895d --- /dev/null +++ b/docs/benchmark/quality-instrumentation.md @@ -0,0 +1,67 @@ +# 质量插桩（Benchmark Group B）：TestPassRate → Q_quality → reward + +把「生成代码」变成可计算的质量分，进而让 `reward` 由 NaN 转为真实可算。本文档是该链路的唯一入口。 + +## 1. 链路总览 + +``` +specialist agent 生成文件 ──┐ + ├─(完成时, 门控)─► 沙箱执行 fixture 留出测试 ─► TestPassRate +fixture 留出测试 (held-out) ─┘ │ + ▼ + quality_score(掩码归一) ─► Q_quality ─► collector ─► reward +``` + +- **谁生成代码**：swarm 自身的执行单元（`agent/task_executor.py`，OpenAI 兼容模型），其输出含 `files:[{path,action,content}]` 写入工作区。 +- **谁评分**：**fixture 的留出测试**（held-out），swarm 看不到 → 避免「自己出卷自己改」。swarm 自带的 `test_*.py` 仅作**协作/鲁棒性信号**（`agent_test_pass_rate`），**不计入** Q_quality（Owner 裁定）。 + +## 2. 组件与文件 + +| 组件 | 文件 | 职责 | +|---|---|---| +| 代码沙箱 | `orchestrator/sandbox.py` + `sandbox_runner.py` | 在临时工作目录的子进程内运行测试；超时强杀、资源限额、环境清洗、路径越界校验；从 JSON 读取计数 | +| Fixture | `benchmark/fixtures/__init__.py` + `/fixture.json` + `/tests/test_*.py` | 任务目标 + 留出测试 + `target_time_seconds` + `required_capabilities` | +| 质量评测 | `orchestrator/quality.py` | 收集生成文件（区分 impl/测试）、沙箱评分、合成 `Q_quality` | +| 运行时存储 | `SwarmRun.quality`（`swarm_runtime.py`） | 存评测结果，供 collector 读取 | +| 触发点 | `main.py: refresh_swarm_run_status`（run 完成时） | 门控 + 绑定 fixture 时调用评测并落库 | +| 采集 | `benchmark/collectors/run_collector.py` | 由 `quality` + run 数据合成 `reward` | +| 公式 | `benchmark/metrics.py: quality_score / reward` | 纯函数（掩码归一 / 加权和） | + +## 3. Q_quality：掩码归一（v2.1 裁定） + +`Q_quality = Σ_{i∈present} wᵢ·xᵢ / Σ_{i∈present} wᵢ`，`i∈{TestPassRate, CodeReview, UserAcceptance}`，默认 `w=0.4/0.3/0.3`。 + +- 不适用项传 `None`，同时退出分子与分母 → 分值恒在 `[0,100]` 且**跨任务类型可比**。 +- 非编码任务（`required_capabilities` 不含 `{python,code_generation,testing,pytest,...}`）→ `TestPassRate=None` 自动忽略。 +- 当前只采集到 `TestPassRate`；`CodeReview`/`UserAcceptance` 未接入 → `Q_quality` 退化为 `TestPassRate`。 +- 三项全无 → `NaN`（规则 #9，不伪造 0）。 + +## 4. reward 何时为真 + +collector 在以下**全部满足**时计算 `reward`，否则 NaN + `coverage=False`： +- 绑定 fixture 且 `Q_quality` 为真实数（沙箱产出了 TestPassRate）； +- `V_speed` 可算（fixture 给 `target_time_seconds`，run 时长 > 0）； +- `E_cost` 可算（预算 + 用量）。 + +派生口径：`s_task=完成率`、`r_robust=鲁棒分`、`g_gov=审批合规（无受治理操作→100）`、`p_rework=retry>0 的任务占比`、`p_risk=审批高危占比（无受治理操作→0）`。 + +> ⚠️ **已知低估**：`p_risk` 仅由审批派生、`p_rework` 仅由 retry 派生（未含评审重开）。两者与治理覆盖缺口绑定，已在 collector 与 coverage 文档显式标注，不作隐藏假设。 + +## 5. 安全（执行不可信代码） + +详见 [`../integration/security-boundary.md §8.1`](../integration/security-boundary.md)。要点： +- **OS 级隔离边界 = K8s Pod**（非 root/只读根/NetworkPolicy/限额/seccomp，部署侧强制，非本仓）。 +- 沙箱在 Pod 内做**纵深防御**：临时目录即用即删、超时强杀、资源限额、**环境清洗（密钥/Token/代理变量不入子进程）**、路径越界校验、输出截断。 +- 默认**关闭**：仅 `ENABLE_QUALITY_EVAL=1` 且在隔离 Pod 内启用。**不构成独立安全边界，不替代 Pod 层强化沙箱。** + +## 6. 诚实边界（本链路**未**关闭的） + +- 仅 1 个示例 fixture（`add_function`）；**统一任务集未建**。 +- `CodeReview`/`UserAcceptance` 未接入（需评审/验收信号源，部分属 Product/Manager）。 +- `Q_quality` 真 ≠ `G_E`/`Benchmark_Agent` 真：**`gain` 仍需基线（Group C）**，故 `s_swarm`/`g_e`/`g_e_cost`/`benchmark` 仍 NaN。 +- 本链路**不构成 benchmark 主链路验收**。 + +## 7. 测试 + +- `scripts/test-sandbox.py`：沙箱真实执行、计数、超时、**环境清洗**、路径越界（无需模型 key）。 +- `scripts/test-quality.py`：fixture 评分 → `Q_quality=100` → `reward` 转真；并断言 `gain`/`benchmark` 仍 NaN（不越界声称）。 diff --git a/docs/benchmark/swarm-metrics-schema.md b/docs/benchmark/swarm-metrics-schema.md index 5ab72a9..4b6ca92 100644 --- a/docs/benchmark/swarm-metrics-schema.md +++ b/docs/benchmark/swarm-metrics-schema.md @@ -41,17 +41,20 @@ P_decision = τ^α · η^β · 100 # v2.0 §3.3（变更） **τ 权重（v2.0）**：Success 0.25 · Quality 0.20 · **Acceptance 0.20（提升为一级因子）** · Cost 0.10 · Time 0.10 · Risk 0.08 · Rollback 0.07。 **η 权重（v2.0）**：Match 0.25 · Urgency 0.15 · Dependency 0.15 · Resource 0.15 · **Confidence 0.10（新增）** · Risk 0.10 · BudgetPressure 0.10。 -| 因子 | 数据来源（标准） | 本仓可采集 | +> ✅ **τ/η/P 引擎已落地**（`orchestrator/decision_engine.py`，Group A / Option A 单边）：信息素 trail 持久于 Redis（学习常开），概率选择门控 `ENABLE_ACO_DISPATCH`（默认关）。下表为**沉积/评分时各因子的真实喂入状态**——无信号的因子喂 0 或中性常数，不伪造（设计与局限见 decision-engine.md）。 + +| 因子 | 数据来源（标准） | 本仓沉积/评分时喂入 | |---|---|---| -| τ.Success | Task 完成记录 | ✅ | -| τ.Quality | 测试 / CI/CD | 🔴 无 CI 接入 | -| τ.Acceptance | 人工验收日志 | 🟡 评审 accepted | -| τ.Cost / Time | 资源监控 / Runtime 日志 | 🟡 usage / 时间戳 | -| τ.Risk / Rollback | 安全审计 / Git·部署 | 🔴 | -| η.Match / Dependency | 能力画像 / DAG | ✅ | -| η.Confidence | Agent 自评接口 | 🔴 未实现 | -| η.Urgency / Resource / Risk / BudgetPressure | 优先级 / 授权 / 风险引擎 / 预算 | 🔴 / 🟡 / 🔴 / 🟡 | -| P_decision | 上述综合 | 🔴 无 τ/η 引擎 | +| τ.Success | Task 完成记录 | ✅ 1/0 | +| τ.Quality | 测试 / CI/CD | 🟡 run 级 fixture 评分已有（Group B），但晚于沉积时点 → 暂 = success | +| τ.Acceptance | 人工验收日志 | 🔴 无每任务验收信号 → 0 | +| τ.Cost | 资源监控 | ✅ 任务成本 / run 预算（未知 → 0） | +| τ.Time / Risk / Rollback | Runtime / 审计 / 部署 | 🔴 无每任务目标/信号 → 0 | +| η.Match / Dependency | 能力画像 / DAG | ✅ Jaccard 匹配 / 下游依赖数 | +| η.Urgency / Resource | 任务等待时长 / Agent 空闲槽位 | ✅ | +| η.Confidence | Agent 自评接口 | 🔴 未实现 → 中性 0.5（全候选同值，不扭曲排序） | +| η.Risk / BudgetPressure | 风险引擎 / 预算 | 🔴 无信号 → 0 | +| P_decision | 上述综合 | ✅ ε-greedy 采样 + 每决策遥测（`SwarmRun.decisions`）；**单边匹配，决策质量未证（需 Group C）** | ## 2. 执行层（标准 §4） @@ -68,12 +71,13 @@ P_rework = ReworkCount/TotalTasks×100 | 指标 | 本仓可采集 | |---|---| | `S_task` | ✅ | -| `Q_quality` | 🔴 需 TestPass/CodeReview/UserAcceptance（无 CI/评分） | -| `V_speed` / `E_cost` | 🟡 Actual 有；Target/Expected 需基线 | -| `R_robust` | 🟡 重试/恢复未计数 | -| `G_gov` / `P_risk` / `P_rework` | 🔴 无敏感操作/风险/返工计数 | +| `Q_quality` | 🟡 TestPass 已接入（fixture 留出测试 + 沙箱，Group B）；CodeReview/UserAcceptance 缺 → 掩码归一 | +| `V_speed` | 🟡 绑定 fixture 时（`target_time_seconds`）可算 | +| `E_cost` | 🟡 预算 + 用量齐备时可算 | +| `R_robust` | ✅ `retry_count` + 任务状态（collector 已算） | +| `G_gov` / `P_risk` / `P_rework` | 🟡 审批合规 / 审批高危占比 / retry 派生——均为**已知低估口径**（见 metric-coverage-gaps 注） | -> v2.0 已给定全部 `w*`/`γ*` 权重，但多数**输入**仍未采集 → `τ`/`η`/`reward`/`p_decision` 暂不可算。 +> v2.0 已给定全部 `w*`/`γ*` 权重；输入现为**有条件采集**：绑定 fixture 的 run 可算 `reward`，开启 `ENABLE_ACO_DISPATCH` 的 run 可算 `τ`/`η`/`p_decision`；条件不满足 → NaN。 ## 3. 蜂群层（标准 §5，权重未变） @@ -90,14 +94,25 @@ S_governance = CompliantOperations/TotalOperations×100 | 指标 | 状态 | |---|---| -| `s_completion` / `s_collaboration` / `s_cost` / `s_robustness` | ✅ 真实可算（采集器） | +| `s_completion` / `s_collaboration` / `s_robustness` | ✅ 真实可算（采集器，无条件） | +| `s_cost` | 🟡 有条件（需预算 + 用量） | +| `s_communication` | 🟡 有条件（peer 消息请求→应答率，按 `correlation_id` 内部计数；无 peer 通信 → NaN） | +| `s_governance` | 🟡 有条件（仅审批可派生） | | `s_gain` | 🔴 需基线（见 emergence-evaluation） | -| `s_communication` | 🔴 未计数（无消息 telemetry） | -| `s_governance` | 🟡 仅审批可派生 | -| `s_swarm` | 🔴 含 gain/communication NaN → 暂为 NaN | +| `s_swarm` | 🔴 含 gain NaN → 暂为 NaN | ## 4. 说明与待对齐 - **成本口径统一（v2.1）**：`S_cost`（§3）、`E_cost`（§2）、`CostEfficiency`（见 cost-normalized-gain）为**同一量** `100×Budget/ActualCost`。 +- **`Q_quality` 掩码归一（v2.1 裁定）**：`Q_quality` 是对 `{TestPassRate, CodeReviewScore, UserAcceptance}` 的**加权均值，但只对“该任务适用”的项计权并归一化**——不适用项（如非编码任务无 `TestPassRate`）同时退出分子与分母，权重按比例重分配给其余项，使分值恒在 `[0,100]` 且**跨任务类型可比**： + + ``` + Q_quality = Σ_{i∈present} wᵢ·xᵢ / Σ_{i∈present} wᵢ （默认 w = 0.4/0.3/0.3） + ``` + + - 「是否编码任务」由任务 `required_capabilities` 是否含 `{python, code_generation, testing, pytest}` 派生，无需额外输入。 + - 三项**全不适用 → NaN**（规则 #9，不伪造 0）。三项**全适用**时退化为 v2.0 的 `0.4/0.3/0.3` 混合。 + - 实现：`benchmark/metrics.py:quality_score`（纯函数，缺项传 `None`）；单测见 `scripts/test-benchmark-metrics.py`。 + - ⚠️ 该函数为**纯公式**；其输入（TestPassRate 等）仍**未采集**（见 metric-coverage-gaps `reward` 行），故 `reward` 仍 NaN。 - `α/β/ρ/N_agent/ε` 取值：标准 §7.2 给推荐初值（1.0 / 2.0 / 0.10 / 5 / 0.10），调优口径待定。 - `Q_quality` 三项来源、`SuccessfulMessages/TotalMessages`、`Recovered/Total`、`Compliant/Total`、`Rework/Total` 的精确计数定义。 diff --git a/docs/benchmark/telemetry-architecture.md b/docs/benchmark/telemetry-architecture.md index 59f40db..89ea23e 100644 --- a/docs/benchmark/telemetry-architecture.md +++ b/docs/benchmark/telemetry-architecture.md @@ -10,9 +10,9 @@ |---|---|---|---| | 任务完成 | Task Logs、Handoff Logs | Elasticsearch | ✅ 任务状态 + 事件流（按 `swarm_id`）；未汇入 ES | | 成本 / Token | Runtime Metrics | Prometheus | 🟡 usage 已采集；Prometheus 指标可抓取 | -| 质量 | CI/CD Results、Code Review | 各 CI 平台 | 🔴 未接入 | +| 质量 | CI/CD Results、Code Review | 各 CI 平台 | 🟡 TestPassRate 已接入（fixture 留出测试 + 沙箱执行，`ENABLE_QUALITY_EVAL` 门控）；CodeReview/UserAcceptance 仍缺 | | 治理合规 | Audit Logs | OpenTelemetry | 🟡 事件流可作审计源，未独立留存 | -| 通信 | WebSocket Logs | ClickHouse | 🟡 连接事件有日志；消息成功率未计数 | +| 通信 | WebSocket Logs | ClickHouse | 🟡 连接事件有日志；peer 消息已按 `correlation_id` 内部计数（请求→应答率，存 `SwarmRun.collaboration`），尚未落 ClickHouse | | 基础设施 | Infrastructure Metrics | Prometheus | 🔴 未接入（cpu/memory/Pod） | ## 3. 本仓已发出的信号 diff --git a/docs/integration/security-boundary.md b/docs/integration/security-boundary.md index 7cbd81c..608f040 100644 --- a/docs/integration/security-boundary.md +++ b/docs/integration/security-boundary.md @@ -67,6 +67,15 @@ - 当前执行单元为进程 / K8s Pod，隔离强度依赖部署（namespace/资源限额）。 - 🟡 待接入：强化沙箱（seccomp/只读根/网络策略/能力裁剪），由 Infra/Security Team 定义。 +### 8.1 代码测试沙箱（benchmark Group B，`orchestrator/sandbox.py`） + +为给「生成代码」算真实 TestPassRate，需**执行模型生成的代码**。安全模型与边界如下： + +- **OS 级隔离边界 = K8s Pod / 容器**（非 root、只读根文件系统、NetworkPolicy 出网拒绝、CPU/内存/pids 限额、seccomp），由部署侧（Manager/release 清单）强制，**非本仓可改**。按 Owner 裁定，在该隔离 Pod 内运行测试代码可接受。 +- **Pod 内纵深防御（本模块新增）**：每次运行用临时工作目录（结束即删）；wall-clock 超时 + 进程组强杀；POSIX 资源限额（CPU 时间 / 地址空间 / 文件大小 / 子进程数，见 `SANDBOX_*` 环境变量）；**环境变量清洗**（不向子进程泄漏任何 API Key/Token/云凭据/代理变量，仅放行 `PATH`/语言区域等白名单）；写入路径**越界校验**（拒绝绝对路径与 `..` 逃逸）；输出截断；计数从 JSON 结果文件读取，不信任 stdout。 +- **门控**：默认**关闭**，仅 `ENABLE_QUALITY_EVAL=1` 且在上述隔离 Pod 内启用；评测仅用 `benchmark/fixtures//` 的**留出测试**（held-out）作权威评分。 +- **明确不构成**：本模块**不是**独立安全边界，**不替代** §8 的强化沙箱（seccomp/只读根/网络策略仍由 Infra/Security 在 Pod 层强制）。不得在隔离 Pod 之外开启代码执行。 + ## 9. 覆盖与缺口 | 边界 | 状态 | @@ -77,7 +86,8 @@ | 短期凭证派生注入、Workload Identity | 🟡 AM/K8s 侧 | | Tool/MCP 权限引擎 | 🔴 未实现 | | allowed_paths 运行时强制、强隔离 | 🔴 未实现 | -| 强化执行沙箱 | 🔴 未实现 | +| 代码测试沙箱（Pod 内纵深防御 + 环境清洗 + 超时/限额，门控 `ENABLE_QUALITY_EVAL`） | ✅ 已实现（本仓，§8.1）；OS 级隔离仍依赖 Pod | +| 强化执行沙箱（seccomp/只读根/网络策略/能力裁剪，Pod 层） | 🔴 未实现（Infra/Security） | | 租户隔离 | ⛔ 不在本仓（归因按 user/channelId） | ## 10. 待对齐对象 diff --git a/docs/peer-communication-logistics.md b/docs/peer-communication-logistics.md index 527c214..4b41963 100644 --- a/docs/peer-communication-logistics.md +++ b/docs/peer-communication-logistics.md @@ -46,7 +46,7 @@ Agent 出站连编排器 WebSocket，彼此不直连。编排器收到 `peer_mes | 触发窄 | 仅 testing/documentation 角色发起；实现角色不主动协作 | | 单轮 | 一问一答，无多轮/澄清/协商线程 | | 无广播 | 只能点对点 `target_agent_id`，无按角色/能力的群发或发现 | -| 无遥测 | 消息未计数（成功/失败/总数）→ `S_communication` 不可算（见 metric-coverage-gaps） | +| ~~无遥测~~ ✅ 已计数 | orchestrator 路由 peer 消息时按 `correlation_id` 记请求/应答与投递失败到 `SwarmRun.collaboration`（内部计数，不进 Manager 事件流）→ collector 据此算 `S_communication`（请求→应答率）；无 peer 通信的 run 仍 NaN（见 metric-coverage-gaps） | | 弱冲突处理 | 仅"以实现为准"的提示约定，无结构化协商/升级协议 | | 无前端可见 | peer 消息未作为事件暴露，前端协作时间线无数据 | diff --git a/orchestrator/decision_engine.py b/orchestrator/decision_engine.py new file mode 100644 index 0000000..bda62a3 --- /dev/null +++ b/orchestrator/decision_engine.py @@ -0,0 +1,216 @@ +"""ACO decision engine (benchmark Group A, Option A: score-at-pull). + +Implements the standard's §3 decision layer over the EXISTING pull dispatch: +when an idle agent asks for work, score all eligible ready tasks with +P_i = τ_i^α·η_i^β / Σ and SAMPLE one (ε-greedy exploration), instead of taking +the first match. The agent side of the pairing is fixed by arrival order — this +is the documented one-sided limitation of Option A (full two-sided matching = +Option B, deferred until Group C can measure whether it helps). + +Two independently-switched halves: + - LEARNING (pheromone deposits) is ALWAYS ON: every task terminal state updates + the trail. Pure passive observation — flag-off dispatch behavior is unchanged, + but history accrues so enabling the flag later acts on earned reputation, not + the cold-start prior. + - ACTING (probabilistic selection) is gated by ENABLE_ACO_DISPATCH (default OFF). + +Honesty rules baked in (org rule #9 — no fabricated signals): + - τ deposit inputs we DON'T have contribute 0, never a made-up value: + acceptance (no acceptance signal), time (no per-task target), risk/rollback + (no signals). quality reuses success until per-task grading exists (run-level + fixture grading happens AFTER completion, too late for the deposit). + - η inputs without a source are a documented neutral constant (confidence=0.5, + identical for every candidate → no ranking distortion) or 0 (risk, + budget_pressure when unknown). + +Determinism: the RNG is seedable via ACO_SEED so CI can assert exact behavior; +production leaves it unseeded. +""" +from __future__ import annotations + +import json +import logging +import os +import random +import time +from dataclasses import dataclass +from typing import Dict, List, Optional + +from benchmark.metrics import THETA_DEFAULTS, heuristic, p_decision, pheromone + +from .redis_client import redis_client + +logger = logging.getLogger(__name__) + +PHEROMONE_KEY_PREFIX = "pheromone:" + +# Trail bounds: keep τ strictly positive (τ^α must stay well-defined) and capped +# (the standard's literal update τ←(1−ρ)τ+Δτ saturates; the cap is the saturation). +TAU_INITIAL = 0.5 +TAU_MIN = 0.05 +TAU_MAX = 1.0 +ETA_MIN = 0.05 + +URGENCY_NORM_SECONDS = 300.0 # task age at which urgency saturates to 1.0 +DEPENDENCY_NORM = 3.0 # dependent-count at which criticality saturates to 1.0 +RESOURCE_NORM_SLOTS = 2.0 # free slots at which resource fit saturates to 1.0 +CONFIDENCE_NEUTRAL = 0.5 # no confidence signal → same neutral for all candidates +MAX_RECORDED_DECISIONS = 1000 # cap on per-run decision telemetry + + +def aco_dispatch_enabled() -> bool: + return os.getenv("ENABLE_ACO_DISPATCH", "false").lower() in {"1", "true", "yes"} + + +def _hyper(name: str, default: float) -> float: + try: + return float(os.getenv(name, str(default))) + except ValueError: + return default + + +@dataclass +class Decision: + """One sampled assignment, with everything the benchmark layer needs.""" + task_id: str + agent_id: str + agent_role: str + tau: float + eta: float + p_norm: float # normalized selection probability over this candidate set + p_score: float # standard §3.3 P_decision = τ^α·η^β·100 (what the metric reports) + explored: bool # True when the ε-greedy branch picked uniformly + + def telemetry(self) -> dict: + return { + "task_id": self.task_id, + "agent_id": self.agent_id, + "agent_role": self.agent_role, + "tau": round(self.tau, 6), + "eta": round(self.eta, 6), + "p_norm": round(self.p_norm, 6), + "p_score": round(self.p_score, 6), + "explored": self.explored, + } + + +class DecisionEngine: + """Pheromone store + η scorer + ε-greedy probabilistic selection.""" + + def __init__(self, rng: Optional[random.Random] = None): + seed = os.getenv("ACO_SEED") + self.rng = rng or (random.Random(int(seed)) if seed else random.Random()) + + # --- pheromone trail (τ) ------------------------------------------------- + @staticmethod + def _trail_key(agent_role: str, agent_id: str) -> str: + return f"{PHEROMONE_KEY_PREFIX}{agent_role}:{agent_id}" + + async def get_tau(self, agent_role: str, agent_id: str) -> float: + raw = await redis_client.get(self._trail_key(agent_role, agent_id)) + if not raw: + return TAU_INITIAL + try: + return float(json.loads(raw).get("tau", TAU_INITIAL)) + except Exception: + return TAU_INITIAL + + async def deposit( + self, + *, + agent_role: str, + agent_id: str, + success: bool, + cost_ratio: float = 0.0, + ) -> float: + """Update the trail after a task reaches a terminal state (always on). + + Standard §3.1: τ(t+1) = (1−ρ)·τ(t) + Δτ, clamped to [TAU_MIN, TAU_MAX]. + Δτ comes from metrics.pheromone() with only the signals we actually have: + success (1/0), quality (=success — see module docstring), cost (task cost + as a fraction of the run budget, 0 when unknown). Absent signals are 0. + """ + rho = _hyper("ACO_RHO", THETA_DEFAULTS["rho"]) + s = 1.0 if success else 0.0 + delta = pheromone( + success=s, quality=s, acceptance=0.0, + cost=max(0.0, min(1.0, cost_ratio)), time=0.0, risk=0.0, rollback=0.0, + ) + old = await self.get_tau(agent_role, agent_id) + new = max(TAU_MIN, min(TAU_MAX, (1.0 - rho) * old + delta)) + await redis_client.set( + self._trail_key(agent_role, agent_id), + json.dumps({"tau": new, "updated_at": time.time()}), + ) + logger.debug("pheromone %s:%s %.3f -> %.3f (Δ=%.3f)", agent_role, agent_id, old, new, delta) + return new + + # --- heuristic (η) ------------------------------------------------------- + @staticmethod + def compute_eta(task, agent_capabilities: List[str], *, free_slots: int, + dependents_count: int, now: Optional[float] = None) -> float: + """A-priori desirability of (agent, task) from signals we actually hold.""" + now = now or time.time() + required = set(task.required_capabilities or []) + caps = set(agent_capabilities or []) + union = required | caps + match = (len(required & caps) / len(union)) if union else 1.0 # Jaccard: rewards focus + urgency = min(1.0, max(0.0, now - task.created_at) / URGENCY_NORM_SECONDS) + dependency = min(1.0, dependents_count / DEPENDENCY_NORM) + resource = min(1.0, max(0, free_slots) / RESOURCE_NORM_SLOTS) + eta = heuristic( + match=match, urgency=urgency, dependency=dependency, resource=resource, + confidence=CONFIDENCE_NEUTRAL, risk=0.0, budget_pressure=0.0, + ) + return max(ETA_MIN, eta) + + # --- selection (P) ------------------------------------------------------- + async def select(self, agent_id: str, agent_capabilities: List[str], + candidates: List, *, free_slots: int, + dependents_counts: Dict[str, int]) -> Optional[Decision]: + """Sample one task from the eligible candidates with P_i = τ^α·η^β / Σ. + + ε-greedy: with probability ε pick uniformly (exploration keeps low-τ + agents able to rebuild their trail and prevents starvation). + """ + if not candidates: + return None + alpha = _hyper("ACO_ALPHA", THETA_DEFAULTS["alpha"]) + beta = _hyper("ACO_BETA", THETA_DEFAULTS["beta"]) + epsilon = _hyper("ACO_EPSILON", THETA_DEFAULTS["epsilon"]) + + scored = [] + for task in candidates: + tau = await self.get_tau(task.agent_role, agent_id) + eta = self.compute_eta( + task, agent_capabilities, free_slots=free_slots, + dependents_count=dependents_counts.get(task.task_id, 0), + ) + scored.append((task, tau, eta, (tau ** alpha) * (eta ** beta))) + + total = sum(w for *_xs, w in scored) + explored = self.rng.random() < epsilon + if explored or total <= 0: + task, tau, eta, weight = scored[self.rng.randrange(len(scored))] + else: + pick = self.rng.random() * total + cum = 0.0 + task, tau, eta, weight = scored[-1] + for cand in scored: + cum += cand[3] + if pick <= cum: + task, tau, eta, weight = cand + break + return Decision( + task_id=task.task_id, + agent_id=agent_id, + agent_role=task.agent_role, + tau=tau, + eta=eta, + p_norm=(weight / total) if total > 0 else 1.0 / len(scored), + p_score=p_decision(tau, eta, alpha, beta), + explored=explored, + ) + + +decision_engine = DecisionEngine() diff --git a/orchestrator/main.py b/orchestrator/main.py index 6b289d1..4b8a617 100644 --- a/orchestrator/main.py +++ b/orchestrator/main.py @@ -25,6 +25,8 @@ from .task_queue import task_queue, TaskStatus from .swarm_runtime import RuntimeValidationError, swarm_runtime from .planner import planner from .master_agent import master_agent +from .quality import evaluate_run_quality +from .decision_engine import decision_engine, aco_dispatch_enabled # Configure logging logging.basicConfig( @@ -69,13 +71,16 @@ class ConnectionManager: del self.active_connections[agent_id] logger.info(f"Agent {agent_id} disconnected") - async def send_message(self, agent_id: str, message: dict): - """Send message to specific agent.""" + async def send_message(self, agent_id: str, message: dict) -> bool: + """Send message to specific agent. Returns True if it was delivered.""" if agent_id in self.active_connections: try: await self.active_connections[agent_id].send_json(message) + return True except Exception as e: logger.error(f"Failed to send message to agent {agent_id}: {e}") + return False + return False async def broadcast(self, message: dict): """Broadcast message to all connected agents.""" @@ -107,6 +112,34 @@ def agent_has_capacity(agent_id: str) -> bool: return AGENT_SLOTS.get(agent_id, 1) > 0 +async def deposit_pheromone(task, agent_id: str, *, success: bool, result=None, run=None): + """Group A learning (always on): update the agent's τ trail for this task's role. + + Passive observation — never changes dispatch behavior by itself (selection is gated + separately by ENABLE_ACO_DISPATCH) and never fails the caller. cost_ratio is the task's + model cost as a fraction of the run budget when both are known, else 0 (no fabricated + penalty — see decision_engine module docstring). + """ + if not task or not agent_id: + return + try: + cost_ratio = 0.0 + if run is not None and isinstance(result, dict): + budget = ((run.request_body or {}).get("orchestration_plan") or {}).get("budget") or {} + max_cost = budget.get("max_cost_usd") + if isinstance(max_cost, (int, float)) and max_cost > 0: + cost = float((result.get("usage") or {}).get("model_cost_usd") or 0.0) + cost_ratio = cost / float(max_cost) + await decision_engine.deposit( + agent_role=task.agent_role, + agent_id=agent_id, + success=success, + cost_ratio=cost_ratio, + ) + except Exception as exc: + logger.debug(f"pheromone deposit skipped for {getattr(task, 'task_id', '?')}: {exc}") + + # Background task for failure detection async def failure_detection_loop(): """Periodically check for failed agents and reassign their tasks.""" @@ -272,9 +305,31 @@ async def task_dispatch_loop(): ] for agent in connected_idle_agents: - task = await task_queue.get_ready_pending_task(agent.capabilities) - if not task: - break + decision = None + if aco_dispatch_enabled(): + # Group A (Option A, score-at-pull): enumerate ALL eligible ready tasks for + # this agent and SAMPLE one with P=τ^α·η^β/Σ instead of taking the first + # match. One-sided by design: the agent is fixed by arrival order. + candidates = await task_queue.get_ready_pending_tasks(agent.capabilities) + if not candidates: + break + dependents_counts: Dict[str, int] = {} + for t in await task_queue.get_all_tasks(): + for dep in t.depends_on: + dependents_counts[dep] = dependents_counts.get(dep, 0) + 1 + decision = await decision_engine.select( + agent.agent_id, + agent.capabilities, + candidates, + free_slots=AGENT_SLOTS.get(agent.agent_id, 1), + dependents_counts=dependents_counts, + ) + task = next(t for t in candidates if t.task_id == decision.task_id) + await task_queue.remove_pending_task(task.task_id) + else: + task = await task_queue.get_ready_pending_task(agent.capabilities) + if not task: + break success = await task_queue.assign_task(task.task_id, agent.agent_id) if not success: @@ -296,6 +351,9 @@ async def task_dispatch_loop(): "agent_id": agent.agent_id, }, ) + if decision is not None: + # Group A telemetry: feeds tau/eta/p_decision in the collector. + await swarm_runtime.record_decision(run, decision.telemetry()) dispatch_context = ( await build_dispatch_context(run, task) if run else task.context @@ -387,6 +445,16 @@ async def refresh_swarm_run_status(run): run.metadata["final_summary"] = final_summary await swarm_runtime.save_run(run) + # Benchmark Group B: grade the run's generated code against its held-out fixture tests in + # the sandbox. Gated (ENABLE_QUALITY_EVAL) + fixture-bound; a no-op otherwise. Never fails + # the run — a grading error just leaves quality unrecorded (reward stays NaN). + try: + quality = await evaluate_run_quality(run, tasks) + if quality: + await swarm_runtime.record_quality(run, quality) + except Exception as exc: + logger.warning("quality eval failed for run %s: %s", run.swarm_id, exc) + await swarm_runtime.emit_event( run, "deployment.status_changed", @@ -1988,6 +2056,8 @@ async def websocket_endpoint(websocket: WebSocket, agent_id: str): completed = await task_queue.complete_task(task_id, result_text) task = await task_queue.get_task(task_id) run = await swarm_runtime.get_run_for_task(task_id) + if completed and task: + await deposit_pheromone(task, agent_id, success=True, result=result, run=run) if completed and run and task: await emit_task_completion_events(run, task, agent_id, result) await emit_usage_event(run, task, agent_id, result) @@ -2018,6 +2088,8 @@ async def websocket_endpoint(websocket: WebSocket, agent_id: str): completed = await task_queue.complete_task(task_id, result_text) task = await task_queue.get_task(task_id) run = await swarm_runtime.get_run_for_task(task_id) + if completed and task: + await deposit_pheromone(task, agent_id, success=True, result=result, run=run) if completed and run and task: await emit_task_completion_events(run, task, agent_id, result) await emit_usage_event(run, task, agent_id, result) @@ -2038,6 +2110,8 @@ async def websocket_endpoint(websocket: WebSocket, agent_id: str): failed = await task_queue.fail_task(task_id, reason) task = await task_queue.get_task(task_id) run = await swarm_runtime.get_run_for_task(task_id) + if failed and task: + await deposit_pheromone(task, agent_id, success=False, run=run) if failed and run and task: event_type = "task.retried" if task.status == TaskStatus.PENDING else "task.failed" await swarm_runtime.emit_event( @@ -2076,6 +2150,8 @@ async def websocket_endpoint(websocket: WebSocket, agent_id: str): failed = await task_queue.fail_task(task_id, reason) task = await task_queue.get_task(task_id) run = await swarm_runtime.get_run_for_task(task_id) + if failed and task: + await deposit_pheromone(task, agent_id, success=False, run=run) if failed and run and task: event_type = "task.retried" if task.status == TaskStatus.PENDING else "task.failed" await swarm_runtime.emit_event( @@ -2158,13 +2234,26 @@ async def websocket_endpoint(websocket: WebSocket, agent_id: str): # Route peer collaboration messages between agents; stamp the sender so the # recipient knows where to send its reply. target_agent_id = message.get("target_agent_id") + delivered = False if target_agent_id: - await manager.send_message( + delivered = await manager.send_message( target_agent_id, {**message, "from_agent_id": agent_id}, ) else: logger.warning(f"peer_message from {agent_id} missing target_agent_id") + # Record internal communication telemetry (feeds benchmark S_communication). + # Not emitted to the Manager event stream — it is not a registered HM event. + peer_task_id = message.get("task_id") + if peer_task_id: + peer_run = await swarm_runtime.get_run_for_task(peer_task_id) + if peer_run: + await swarm_runtime.record_peer_message( + peer_run, + correlation_id=message.get("correlation_id"), + is_reply=bool(message.get("is_reply")), + delivered=delivered, + ) else: logger.warning(f"Unknown message type from {agent_id}: {message_type}") diff --git a/orchestrator/quality.py b/orchestrator/quality.py new file mode 100644 index 0000000..0f9cd93 --- /dev/null +++ b/orchestrator/quality.py @@ -0,0 +1,126 @@ +"""Quality instrumentation (benchmark Group B): turn a finished run's generated code into a real +TestPassRate, then into Q_quality via the masked/renormalized quality_score. + +Pipeline: + 1. gather the files the specialist agents generated (from task results) — split implementation + files from the swarm's own test files. + 2. grade against the fixture's HELD-OUT tests (authoritative) in the sandbox → TestPassRate. + 3. separately run the swarm's OWN tests as a non-grading signal (collaboration/robustness only). + 4. Q_quality = quality_score(test_pass_rate=, code_review_score=None, user_acceptance=None). + CodeReview / UserAcceptance are not collected here, so the masking rule renormalizes Q_quality + onto the one present input (rule #9: absent ≠ fabricated 0). + +SECURITY: step 2/3 execute model-generated code. They run ONLY when ENABLE_QUALITY_EVAL is set, +and only inside the isolated pod (see orchestrator/sandbox.py security model). Default OFF. +""" +from __future__ import annotations + +import asyncio +import json +import logging +import os +from typing import Any, Dict, List, Optional + +from benchmark.fixtures import Fixture, load_fixture +from benchmark.metrics import quality_score +from .sandbox import SandboxFile, run_tests + +logger = logging.getLogger(__name__) + + +def quality_eval_enabled() -> bool: + return os.getenv("ENABLE_QUALITY_EVAL", "false").lower() in {"1", "true", "yes"} + + +def _is_test_file(path: str) -> bool: + base = os.path.basename(path or "") + return base.startswith("test_") and base.endswith(".py") + + +def collect_generated_files(tasks) -> Dict[str, List[SandboxFile]]: + """Reconstruct files the agents wrote, from task results. Last writer wins per path. + + Returns {"impl": [...non-test files...], "agent_tests": [...the swarm's own test_*.py...]}. + """ + impl: Dict[str, str] = {} + agent_tests: Dict[str, str] = {} + for task in tasks: + result = getattr(task, "result", None) + if not result: + continue + try: + data = json.loads(result) if isinstance(result, str) else result + except Exception: + continue + subtasks = data.get("subtasks") if isinstance(data, dict) else None + file_groups = [data] + list(subtasks or []) if isinstance(data, dict) else [] + for group in file_groups: + for f in (group.get("files") or []): + if not isinstance(f, dict): + continue + if (f.get("action") or "write") == "delete": + continue + path = f.get("path") + content = f.get("content") + if not path or content is None: + continue + (agent_tests if _is_test_file(path) else impl)[path] = content + return { + "impl": [SandboxFile(p, c) for p, c in impl.items()], + "agent_tests": [SandboxFile(os.path.basename(p), c) for p, c in agent_tests.items()], + } + + +async def evaluate_run_quality(run, tasks) -> Optional[Dict[str, Any]]: + """Grade a completed run's code against its fixture. Returns a quality dict or None. + + None means: quality eval disabled, no fixture bound, or nothing to grade — caller leaves + Q_quality/reward NaN (coverage=False). Never fabricates a score. + """ + if not quality_eval_enabled(): + return None + fixture_id = (run.metadata or {}).get("benchmark_fixture_id") + if not fixture_id: + return None + try: + fixture: Fixture = load_fixture(fixture_id) + except Exception as exc: + logger.warning("quality eval: cannot load fixture %s: %s", fixture_id, exc) + return None + + files = collect_generated_files(tasks) + impl = files["impl"] + agent_tests = files["agent_tests"] + + quality: Dict[str, Any] = { + "fixture_id": fixture.id, + "expects_code": fixture.expects_code, + "test_pass_rate": None, + "code_review_score": None, # not collected (no reviewer wired) + "user_acceptance": None, # not collected (external Manager/human signal) + "agent_test_pass_rate": None, # SIGNAL ONLY — never part of the grade + "target_time_seconds": fixture.target_time_seconds, + } + + # 1) Authoritative grade: held-out fixture tests against the implementation files. + if fixture.expects_code and fixture.test_files and impl: + fix_files = [SandboxFile(t.path, t.content) for t in fixture.test_files] + result = await asyncio.to_thread(run_tests, impl, fix_files) + quality["test_pass_rate"] = result.pass_rate + quality["fixture_total"] = result.total + quality["fixture_passed"] = result.passed + if result.error: + quality["fixture_error"] = result.error + + # 2) Signal only: the swarm's own tests (collaboration/robustness, NOT the grade). + if agent_tests and impl: + sig = await asyncio.to_thread(run_tests, impl, agent_tests) + quality["agent_test_pass_rate"] = sig.pass_rate + + # 3) Q_quality via the masked/renormalized blend (absent inputs drop out). + quality["q_quality"] = quality_score( + test_pass_rate=quality["test_pass_rate"], + code_review_score=quality["code_review_score"], + user_acceptance=quality["user_acceptance"], + ) + return quality diff --git a/orchestrator/sandbox.py b/orchestrator/sandbox.py new file mode 100644 index 0000000..e2729a1 --- /dev/null +++ b/orchestrator/sandbox.py @@ -0,0 +1,200 @@ +"""Secure-within-pod code sandbox for testing specialist agents' generated code. + +SECURITY MODEL — read before changing anything here: + The OS-level isolation boundary is the **Kubernetes pod / container** this process runs in: + non-root user, read-only root filesystem, NetworkPolicy egress-deny, CPU/memory/pids limits, + and seccomp — all enforced at deploy time (see docs/integration/security-boundary.md and the + Manager/release deployment manifests; NOT editable from this repo). Per Owner ruling, running + generated test code inside that pod is acceptable. + + This module adds **in-pod defense in depth** on top of the pod boundary: + - ephemeral per-run workdir (tempfile), always removed in `finally` + - wall-clock timeout with process-group kill + - POSIX resource limits (CPU time, address space, file size, subprocess count) where available + - environment scrub: no API keys / tokens / cloud creds / proxy vars leak into the child + - path-traversal guard on every written file (no abs paths, no `..` escape) + - stdout/stderr size caps; counts are read from a JSON file, never parsed from stdout + + It is NOT a standalone security boundary. It MUST run only inside the isolated pod and is gated + by ENABLE_QUALITY_EVAL upstream (orchestrator/quality.py). Never enable code execution outside + that pod. +""" +from __future__ import annotations + +import json +import os +import shutil +import subprocess +import sys +import tempfile +from dataclasses import dataclass, field +from pathlib import Path +from typing import List, Optional + +try: # POSIX only; absent on Windows dev boxes (prod is a Linux pod) + import resource # type: ignore +except Exception: # pragma: no cover - platform dependent + resource = None # type: ignore + +_RUNNER = Path(__file__).resolve().parent / "sandbox_runner.py" + +# Env vars always allowed through to the child (everything else is dropped). +_ENV_ALLOWLIST = {"PATH", "SYSTEMROOT", "SystemRoot", "WINDIR", "LANG", "LC_ALL", "LC_CTYPE", "TZ", "COMSPEC", "PATHEXT"} + +DEFAULT_TIMEOUT_SECONDS = float(os.getenv("SANDBOX_TIMEOUT_SECONDS", "30")) +DEFAULT_CPU_SECONDS = int(os.getenv("SANDBOX_CPU_SECONDS", "20")) +DEFAULT_MEM_BYTES = int(os.getenv("SANDBOX_MEM_BYTES", str(512 * 1024 * 1024))) +DEFAULT_FSIZE_BYTES = int(os.getenv("SANDBOX_FSIZE_BYTES", str(32 * 1024 * 1024))) +DEFAULT_NPROC = int(os.getenv("SANDBOX_NPROC", "64")) +_OUTPUT_CAP = 16 * 1024 + + +@dataclass +class SandboxFile: + path: str + content: str + + +@dataclass +class SandboxResult: + total: int = 0 + passed: int = 0 + failed: int = 0 + errored: int = 0 + timed_out: bool = False + exit_code: Optional[int] = None + stdout: str = "" + stderr: str = "" + error: Optional[str] = None + details: List[dict] = field(default_factory=list) + + @property + def pass_rate(self) -> Optional[float]: + """passed / total * 100, or None when nothing ran (caller decides NaN/coverage).""" + if self.total <= 0: + return None + return 100.0 * self.passed / self.total + + +def _safe_join(root: Path, rel: str) -> Path: + # Reject absolute paths and any `..` escape; resolve strictly under root. + candidate = (root / rel).resolve() + if not str(candidate).startswith(str(root.resolve())): + raise ValueError(f"path escapes sandbox: {rel!r}") + return candidate + + +def _write_files(root: Path, files: List[SandboxFile]) -> None: + for f in files: + if not f.path or os.path.isabs(f.path): + raise ValueError(f"unsafe file path: {f.path!r}") + dest = _safe_join(root, f.path) + dest.parent.mkdir(parents=True, exist_ok=True) + dest.write_text(f.content, encoding="utf-8") + + +def _child_env(workdir: Path) -> dict: + env = {k: v for k, v in os.environ.items() if k in _ENV_ALLOWLIST} + # Confine writes and bytecode; isolate HOME/TMP to the ephemeral dir. + env["HOME"] = str(workdir) + env["TMPDIR"] = str(workdir) + env["TEMP"] = str(workdir) + env["TMP"] = str(workdir) + env["PYTHONDONTWRITEBYTECODE"] = "1" + env["PYTHONNOUSERSITE"] = "1" + env["PYTHONPATH"] = "" + return env + + +def _preexec(): # pragma: no cover - POSIX only, runs in the child before exec + if resource is not None: + resource.setrlimit(resource.RLIMIT_CPU, (DEFAULT_CPU_SECONDS, DEFAULT_CPU_SECONDS)) + try: + resource.setrlimit(resource.RLIMIT_AS, (DEFAULT_MEM_BYTES, DEFAULT_MEM_BYTES)) + except (ValueError, OSError): + pass + resource.setrlimit(resource.RLIMIT_FSIZE, (DEFAULT_FSIZE_BYTES, DEFAULT_FSIZE_BYTES)) + try: + resource.setrlimit(resource.RLIMIT_NPROC, (DEFAULT_NPROC, DEFAULT_NPROC)) + except (ValueError, OSError): + pass + os.setsid() # own process group, so a timeout can kill the whole tree + + +def run_tests( + source_files: List[SandboxFile], + test_files: List[SandboxFile], + *, + timeout_seconds: float = DEFAULT_TIMEOUT_SECONDS, +) -> SandboxResult: + """Write source + test files into an ephemeral workdir and run the tests in a child process. + + Returns a SandboxResult with pass/fail counts (read from the runner's JSON output). The + workdir is always removed. Blocking/CPU-bound — callers should offload via asyncio.to_thread. + """ + workdir = Path(tempfile.mkdtemp(prefix="swarm-sbx-")) + is_posix = os.name == "posix" + try: + _write_files(workdir, list(source_files) + list(test_files)) + shutil.copyfile(_RUNNER, workdir / "_runner.py") + + popen_kwargs = dict( + cwd=str(workdir), + env=_child_env(workdir), + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + ) + if is_posix: + popen_kwargs["preexec_fn"] = _preexec + else: # Windows dev: new process group so we can signal the tree + popen_kwargs["creationflags"] = getattr(subprocess, "CREATE_NEW_PROCESS_GROUP", 0) + + proc = subprocess.Popen([sys.executable, "-I", "_runner.py"], **popen_kwargs) + result = SandboxResult() + try: + out, err = proc.communicate(timeout=timeout_seconds) + result.exit_code = proc.returncode + result.stdout = (out or "")[:_OUTPUT_CAP] + result.stderr = (err or "")[:_OUTPUT_CAP] + except subprocess.TimeoutExpired: + result.timed_out = True + _kill(proc, is_posix) + out, err = proc.communicate() + result.stdout = (out or "")[:_OUTPUT_CAP] + result.stderr = (err or "")[:_OUTPUT_CAP] + result.error = f"timeout after {timeout_seconds}s" + return result + + report = workdir / "_result.json" + if report.exists(): + try: + data = json.loads(report.read_text(encoding="utf-8")) + result.total = int(data.get("total", 0)) + result.passed = int(data.get("passed", 0)) + result.failed = int(data.get("failed", 0)) + result.errored = int(data.get("errored", 0)) + result.details = data.get("details", []) or [] + if data.get("fatal"): + result.error = str(data["fatal"]) + except Exception as exc: + result.error = f"unparseable result: {exc!r}" + else: + result.error = "no result produced by sandbox runner" + return result + finally: + shutil.rmtree(workdir, ignore_errors=True) + + +def _kill(proc: "subprocess.Popen", is_posix: bool) -> None: # pragma: no cover - timing dependent + try: + if is_posix: + import signal + os.killpg(os.getpgid(proc.pid), signal.SIGKILL) + else: + proc.kill() + except Exception: + try: + proc.kill() + except Exception: + pass diff --git a/orchestrator/sandbox_runner.py b/orchestrator/sandbox_runner.py new file mode 100644 index 0000000..b1211dd --- /dev/null +++ b/orchestrator/sandbox_runner.py @@ -0,0 +1,96 @@ +"""In-sandbox test harness — runs INSIDE the isolated workdir as a child process. + +Stdlib only (no pytest dependency): collects both `unittest.TestCase` tests and bare +`test_*` functions from every `test_*.py` in the working directory, runs them, and writes a +machine-readable `_result.json` ({total, passed, failed, errored, details}). The parent +(orchestrator/sandbox.py) reads that file; it never trusts stdout for counts. + +This file is copied into the ephemeral sandbox workdir at run time and executed there with the +workdir as CWD. It must stay self-contained and import nothing outside the stdlib. +""" +import importlib.util +import json +import os +import sys +import unittest + +RESULT_FILE = "_result.json" + + +def _load_module(path: str): + name = "sbx_" + os.path.splitext(os.path.basename(path))[0] + spec = importlib.util.spec_from_file_location(name, path) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) # may raise on import/collection error + return module + + +def main() -> None: + workdir = os.getcwd() + sys.path.insert(0, workdir) + test_files = sorted( + f for f in os.listdir(workdir) if f.startswith("test_") and f.endswith(".py") + ) + + total = passed = failed = errored = 0 + details = [] + suite = unittest.TestSuite() + bare_funcs = [] # (label, callable) + + for tf in test_files: + try: + module = _load_module(os.path.join(workdir, tf)) + except Exception as exc: # import-time failure counts as one errored test + total += 1 + errored += 1 + failed += 1 + details.append({"test": tf, "status": "error", "error": repr(exc)}) + continue + # unittest.TestCase-style tests + suite.addTests(unittest.defaultTestLoader.loadTestsFromModule(module)) + # bare pytest-style functions defined in this module + for attr in dir(module): + if not attr.startswith("test_"): + continue + obj = getattr(module, attr) + if callable(obj) and not isinstance(obj, type) and getattr(obj, "__module__", None) == module.__name__: + bare_funcs.append((f"{tf}::{attr}", obj)) + + # Run the unittest suite (TestCase subclasses). + ut_result = unittest.TestResult() + suite.run(ut_result) + ut_total = ut_result.testsRun + ut_failed = len(ut_result.failures) + len(ut_result.errors) + total += ut_total + failed += ut_failed + errored += len(ut_result.errors) + passed += ut_total - ut_failed + + # Run bare test_* functions. + for label, fn in bare_funcs: + total += 1 + try: + fn() + passed += 1 + details.append({"test": label, "status": "passed"}) + except AssertionError as exc: + failed += 1 + details.append({"test": label, "status": "failed", "error": str(exc)}) + except Exception as exc: + failed += 1 + errored += 1 + details.append({"test": label, "status": "error", "error": repr(exc)}) + + with open(os.path.join(workdir, RESULT_FILE), "w", encoding="utf-8") as fh: + json.dump( + {"total": total, "passed": passed, "failed": failed, "errored": errored, "details": details}, + fh, + ) + + +if __name__ == "__main__": + try: + main() + except Exception as exc: # never leave the parent without a result file + with open(RESULT_FILE, "w", encoding="utf-8") as fh: + json.dump({"total": 0, "passed": 0, "failed": 0, "errored": 1, "fatal": repr(exc)}, fh) diff --git a/orchestrator/swarm_runtime.py b/orchestrator/swarm_runtime.py index 20685ce..57213c0 100644 --- a/orchestrator/swarm_runtime.py +++ b/orchestrator/swarm_runtime.py @@ -40,6 +40,24 @@ class SwarmRun(BaseModel): callback: CallbackConfig = Field(default_factory=CallbackConfig) task_ids: List[str] = Field(default_factory=list) approvals: Dict[str, Dict[str, Any]] = Field(default_factory=dict) + # Internal peer-communication telemetry (NOT a Manager event — kept off the HM event + # registry on purpose). Feeds benchmark S_communication. correlation_ids are de-duplicated + # so a retried/echoed message is not double counted. + collaboration: Dict[str, Any] = Field( + default_factory=lambda: { + "request_correlations": [], # correlation_ids of peer requests routed + "reply_correlations": [], # correlation_ids of peer replies routed + "delivery_failures": 0, # peer messages dropped (target not connected) + } + ) + # Benchmark Group B: quality grade of the run's generated code (fixture TestPassRate → + # Q_quality). Populated only when ENABLE_QUALITY_EVAL is set and a fixture is bound. Feeds + # benchmark reward(); empty dict means not graded (collector leaves reward NaN). + quality: Dict[str, Any] = Field(default_factory=dict) + # Benchmark Group A: per-assignment ACO decision telemetry ({task_id, agent_id, tau, eta, + # p_norm, p_score, explored}). Internal state (NOT a Manager event). Appended only when + # ENABLE_ACO_DISPATCH selected the assignment; empty → tau/eta/p_decision stay NaN. + decisions: List[Dict[str, Any]] = Field(default_factory=list) metadata: Dict[str, Any] = Field(default_factory=dict) request_body: Dict[str, Any] = Field(default_factory=dict) created_at: float = Field(default_factory=time.time) @@ -324,6 +342,42 @@ class SwarmRuntime: return None return await self.get_run(swarm_id) + async def record_peer_message( + self, + run: SwarmRun, + *, + correlation_id: Optional[str], + is_reply: bool, + delivered: bool, + ) -> None: + """Record one routed peer-collaboration message for S_communication. + + Internal telemetry only: counters live on the run, not on the Manager event stream. + Requests and replies are tracked by correlation_id so the collector can compute the + share of requests that received an answer (a real communication-success rate). + """ + collab = run.collaboration or {} + requests = collab.setdefault("request_correlations", []) + replies = collab.setdefault("reply_correlations", []) + if correlation_id: + bucket = replies if is_reply else requests + if correlation_id not in bucket: + bucket.append(correlation_id) + if not delivered: + collab["delivery_failures"] = int(collab.get("delivery_failures", 0)) + 1 + run.collaboration = collab + await self.save_run(run) + + async def record_quality(self, run: SwarmRun, quality: Dict[str, Any]) -> None: + """Store the benchmark quality grade for a completed run (Group B).""" + run.quality = quality or {} + await self.save_run(run) + + async def record_decision(self, run: SwarmRun, decision: Dict[str, Any]) -> None: + """Append one ACO assignment decision (Group A telemetry). Bounded list.""" + run.decisions = (run.decisions or [])[-999:] + [decision] + await self.save_run(run) + async def attach_task(self, run: SwarmRun, task_id: str): """Associate a queue task with a swarm run.""" if task_id not in run.task_ids: diff --git a/orchestrator/task_queue.py b/orchestrator/task_queue.py index 49d5dea..d788532 100644 --- a/orchestrator/task_queue.py +++ b/orchestrator/task_queue.py @@ -137,6 +137,36 @@ class TaskQueue: return None + async def get_ready_pending_tasks( + self, + agent_capabilities: Optional[List[str]] = None, + ) -> List[Task]: + """Return ALL dispatchable tasks for an agent WITHOUT dequeuing any. + + Candidate enumeration for the ACO decision engine (score-at-pull): the caller + scores/samples one and removes it via remove_pending_task. Dead/stale queue + entries are cleaned up the same way get_ready_pending_task does. + """ + pending_ids = await redis_client.lrange(self.PENDING_QUEUE_KEY, 0, -1) + capabilities = set(agent_capabilities or []) + candidates: List[Task] = [] + + for task_id in pending_ids: + task = await self.get_task(task_id) + if not task: + await self.remove_pending_task(task_id) + continue + if task.status != TaskStatus.PENDING: + await self.remove_pending_task(task_id) + continue + if not await self.is_task_ready(task): + continue + if not self.can_agent_run_task(task, capabilities): + continue + candidates.append(task) + + return candidates + async def is_task_ready(self, task: Task) -> bool: """Return True when all dependencies are terminal and successful.""" if task.status != TaskStatus.PENDING: diff --git a/scripts/test-benchmark-collector.py b/scripts/test-benchmark-collector.py index d6cfbe8..58b3780 100644 --- a/scripts/test-benchmark-collector.py +++ b/scripts/test-benchmark-collector.py @@ -64,6 +64,12 @@ async def main(): await swarm_runtime.emit_event(run, "handoff.requested", payload={"task_id": "t1", "from_role": "implementation", "to_role": "testing"}) await swarm_runtime.emit_event(run, "handoff.completed", payload={"task_id": "t1", "from_role": "implementation", "to_role": "testing"}) + # Peer communication: 3 requests routed, 2 of them answered (matching correlation_ids). + for cid, delivered in [("c1", True), ("c2", True), ("c3", True)]: + await swarm_runtime.record_peer_message(run, correlation_id=cid, is_reply=False, delivered=delivered) + for cid in ("c1", "c2"): # c3 never gets a reply + await swarm_runtime.record_peer_message(run, correlation_id=cid, is_reply=True, delivered=True) + collector = SwarmRunMetricsCollector(run.swarm_id) metrics = await collector.collect() cov = collector.coverage @@ -78,12 +84,32 @@ async def main(): check("s_cost = 200.0", round(metrics.s_cost, 1) == 200.0 and cov["s_cost"]) # governance: no approvals -> NaN, coverage False check("s_governance NaN + coverage False", math.isnan(metrics.s_governance) and cov["s_governance"] is False) + # s_communication = 2 answered / 3 requests * 100 = 66.67 + check("s_communication = 66.67", round(metrics.s_communication, 2) == 66.67 and cov["s_communication"]) # not-yet-collectable -> NaN + coverage False check("uncollectable metrics NaN + coverage False", all(math.isnan(getattr(metrics, k)) and cov[k] is False - for k in ("tau", "eta", "p_decision", "reward", "s_gain", "s_communication", + for k in ("tau", "eta", "p_decision", "reward", "s_gain", "s_swarm", "g_e", "g_e_cost", "benchmark"))) + # --- governance telemetry counter: REAL when the run had governed ops (#23) --- + # CompliantOperations/TotalOperations: 2 approvals decided (approved+rejected) of 3 governed + # ops -> 66.67. Proves the governance counter is wired, not just the empty/NaN path above. + gov_run, _ = await swarm_runtime.get_or_create_run( + body={**body, "metadata": {"manager_deployment_id": "m-gov"}}, + idempotency_key=None, correlation_id="cg") + gov_run.approvals = { + "a1": {"approval_id": "a1", "decision": "approved"}, + "a2": {"approval_id": "a2", "decision": "rejected"}, + "a3": {"approval_id": "a3", "decision": "pending"}, + } + await add_task(gov_run, "g-implementation", status=TaskStatus.COMPLETED, agent="A", cost=1.0) + await swarm_runtime.save_run(gov_run) + gov_collector = SwarmRunMetricsCollector(gov_run.swarm_id) + gm = await gov_collector.collect() + check("s_governance = 66.67 (2 decided / 3 governed ops)", + round(gm.s_governance, 2) == 66.67 and gov_collector.coverage["s_governance"] is True) + print() if failures: print(f"{len(failures)} collector check(s) FAILED: {failures}") diff --git a/scripts/test-benchmark-metrics.py b/scripts/test-benchmark-metrics.py index 23f6602..2dde113 100644 --- a/scripts/test-benchmark-metrics.py +++ b/scripts/test-benchmark-metrics.py @@ -2,6 +2,7 @@ Run from agent_swarm_v6: python scripts/test-benchmark-metrics.py """ +import math import sys from pathlib import Path @@ -41,6 +42,16 @@ check("action_probability normalized", round(ap, 4) == round(4.0 / 5.0, 4)) check("reward positive-only = 88.0", round(m.reward(s_task=100, q_quality=100, v_speed=100, e_cost=100, r_robust=100, g_gov=100, p_risk=0, p_rework=0), 4) == 88.0) check("quality_score", m.quality_score(100, 100, 100) == 100.0) +# masked & renormalized (v2.1): all present reduces to the 0.4/0.3/0.3 blend +check("quality_score full blend = 0.4/0.3/0.3", m.quality_score(50, 100, 100) == 80.0) +# non-code task: test_pass_rate=None drops out; renorm over {review 0.3, acceptance 0.3} -> 0.5/0.5 +check("quality_score non-code renormalized", m.quality_score(None, 80, 60) == 70.0) +# renormalization preserves scale: missing input never deflates a perfect score +check("quality_score stays on [0,100] when masked", m.quality_score(None, 100, 100) == 100.0) +# a single present input renormalizes to weight 1.0 +check("quality_score single input", m.quality_score(None, 90, None) == 90.0) +# no applicable inputs -> NaN (rule #9: no fabricated 0) +check("quality_score all-absent = NaN", math.isnan(m.quality_score(None, None, None))) # --- emergence + cost-normalized (v2.0 CHANGED to difference + dynamic C_swarm) --- check("emergence_gain", m.emergence_gain(90, 80) == 10) diff --git a/scripts/test-decision-engine.py b/scripts/test-decision-engine.py new file mode 100644 index 0000000..73e689c --- /dev/null +++ b/scripts/test-decision-engine.py @@ -0,0 +1,186 @@ +"""Tests for the ACO decision engine (benchmark Group A, Option A: score-at-pull). + +Hermetic (REDIS_FAKE) and deterministic (seeded RNG over a probabilistic mechanism). +Run from agent_swarm_v6: python scripts/test-decision-engine.py +""" +import asyncio +import math +import os +import random +import sys +import time +from pathlib import Path + +os.environ["REDIS_FAKE"] = "1" +os.environ["ENABLE_ACO_DISPATCH"] = "1" +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + +from orchestrator.redis_client import redis_client +from orchestrator import swarm_runtime as sr_mod +from orchestrator.swarm_runtime import swarm_runtime +from orchestrator.task_queue import task_queue +from orchestrator.decision_engine import ( + DecisionEngine, TAU_INITIAL, TAU_MAX, TAU_MIN, aco_dispatch_enabled, +) +from benchmark.collectors.run_collector import SwarmRunMetricsCollector + +failures = [] + + +def check(name, cond): + print(("PASS" if cond else "FAIL"), "-", name) + if not cond: + failures.append(name) + + +async def _noop(self, *a, **k): + return None + + +async def main(): + await redis_client.connect() + sr_mod.SwarmRuntime._post_callback = _noop + engine = DecisionEngine(rng=random.Random(42)) + + check("flag readable", aco_dispatch_enabled() is True) + + # --- pheromone trail (τ) --- + check("cold start tau = 0.5", await engine.get_tau("testing", "agent-X") == TAU_INITIAL) + + up = await engine.deposit(agent_role="testing", agent_id="agent-X", success=True) + check("success deposit raises tau", up > TAU_INITIAL) + down_start = await engine.get_tau("testing", "agent-Y") + down = await engine.deposit(agent_role="testing", agent_id="agent-Y", success=False) + check("failure deposit lowers tau", down < down_start) + + # repeated success saturates at the cap; repeated failure floors (clamps hold) + for _ in range(20): + hi = await engine.deposit(agent_role="testing", agent_id="agent-X", success=True) + lo = await engine.deposit(agent_role="testing", agent_id="agent-Y", success=False) + check("tau capped at TAU_MAX", hi <= TAU_MAX) + check("tau floored at TAU_MIN", lo >= TAU_MIN) + + # trails are per-(role, agent): agent-X's testing trail does not bleed into other roles + check("trail keyed by role", await engine.get_tau("documentation", "agent-X") == TAU_INITIAL) + + # cost eats into the deposit: same outcome, higher cost_ratio → lower tau + a = await engine.deposit(agent_role="r2", agent_id="cheap", success=True, cost_ratio=0.0) + b = await engine.deposit(agent_role="r2", agent_id="pricey", success=True, cost_ratio=1.0) + check("higher cost ratio -> lower deposit", b < a) + + # --- heuristic (η) --- + t_old = await task_queue.create_task(task_id="d-implementation", description="x", + agent_role="implementation", + required_capabilities=["python"], enqueue=False) + t_old.created_at = time.time() - 600 # old → urgency saturated + t_new = await task_queue.create_task(task_id="d2-implementation", description="y", + agent_role="implementation", + required_capabilities=["python"], enqueue=False) + eta_old = DecisionEngine.compute_eta(t_old, ["python"], free_slots=2, dependents_count=3) + eta_new = DecisionEngine.compute_eta(t_new, ["python"], free_slots=2, dependents_count=0) + check("older/critical task scores higher eta", eta_old > eta_new) + # focused specialist (exact caps) beats generalist with many unrelated caps (Jaccard match) + eta_spec = DecisionEngine.compute_eta(t_new, ["python"], free_slots=1, dependents_count=0) + eta_gen = DecisionEngine.compute_eta(t_new, ["python", "a", "b", "c", "d"], free_slots=1, dependents_count=0) + check("specialist match > generalist match", eta_spec > eta_gen) + check("eta floored positive", eta_new > 0) + + # --- probabilistic selection --- + # agent Z earned a strong testing trail; tasks compete for Z's pull. + for _ in range(10): + await engine.deposit(agent_role="strong", agent_id="agent-Z", success=True) + for _ in range(10): + await engine.deposit(agent_role="weak", agent_id="agent-Z", success=False) + strong_t = await task_queue.create_task(task_id="s-strong", description="s", agent_role="strong", + required_capabilities=["python"], enqueue=False) + weak_t = await task_queue.create_task(task_id="w-weak", description="w", agent_role="weak", + required_capabilities=["python"], enqueue=False) + picks = {"s-strong": 0, "w-weak": 0} + sel_engine = DecisionEngine(rng=random.Random(7)) + for _ in range(200): + d = await sel_engine.select("agent-Z", ["python"], [strong_t, weak_t], + free_slots=1, dependents_counts={}) + picks[d.task_id] += 1 + check("high-tau role wins most pulls (seeded)", picks["s-strong"] > picks["w-weak"]) + check("epsilon keeps exploring the weak trail", picks["w-weak"] > 0) + + # decision payload sanity: p_norm normalized, p_score = standard formula + d = await sel_engine.select("agent-Z", ["python"], [strong_t, weak_t], + free_slots=1, dependents_counts={}) + check("p_norm in (0,1]", 0 < d.p_norm <= 1.0) + check("p_score = tau^a*eta^b*100 > 0", d.p_score > 0) + + # same seed → identical pick sequence (CI determinism) + seq1 = [ + (await DecisionEngine(rng=random.Random(99)).select( + "agent-Z", ["python"], [strong_t, weak_t], free_slots=1, dependents_counts={})).task_id + for _ in range(5) + ] + e2 = DecisionEngine(rng=random.Random(99)) + seq2 = [ + (await e2.select("agent-Z", ["python"], [strong_t, weak_t], + free_slots=1, dependents_counts={})).task_id + for _ in range(1) + ] + check("seeded selection deterministic", seq1[0] == seq2[0]) + + # --- telemetry -> collector (tau/eta/p_decision become real) --- + body = { + "mode": "swarm", + "requirement": {"objective": "decision test"}, + "callback": {"url": "http://127.0.0.1:9/cb", "subscribed_events": []}, + "metadata": {"manager_deployment_id": "m-aco"}, + } + run, _ = await swarm_runtime.get_or_create_run(body=body, idempotency_key=None, correlation_id="ca") + await swarm_runtime.record_decision(run, d.telemetry()) + t = await task_queue.create_task(task_id="t-implementation", description="t", + agent_role="implementation", enqueue=False) + from orchestrator.task_queue import TaskStatus + t.status = TaskStatus.COMPLETED + t.assigned_agent_id = "agent-Z" + await task_queue._save_task(t) + await swarm_runtime.attach_task(run, t.task_id) + + collector = SwarmRunMetricsCollector(run.swarm_id) + m = await collector.collect() + check("tau real in collector", not math.isnan(m.tau) and collector.coverage["tau"]) + check("eta real in collector", not math.isnan(m.eta) and collector.coverage["eta"]) + check("p_decision real in collector", not math.isnan(m.p_decision) and collector.coverage["p_decision"]) + # Group A must NOT fake the still-blocked aggregates + check("gain/benchmark still NaN", math.isnan(m.s_gain) and math.isnan(m.benchmark)) + + # run with no decisions -> NaN + coverage False (flag-off / non-ACO runs are honest) + run2, _ = await swarm_runtime.get_or_create_run( + body={**body, "metadata": {"manager_deployment_id": "m-aco2"}}, + idempotency_key=None, correlation_id="cb") + t2 = await task_queue.create_task(task_id="t2-implementation", description="t2", + agent_role="implementation", enqueue=False) + t2.status = TaskStatus.COMPLETED + t2.assigned_agent_id = "A" + await task_queue._save_task(t2) + await swarm_runtime.attach_task(run2, t2.task_id) + c2 = SwarmRunMetricsCollector(run2.swarm_id) + m2 = await c2.collect() + check("no decisions -> tau/eta/p_decision NaN", + math.isnan(m2.tau) and math.isnan(m2.eta) and math.isnan(m2.p_decision) + and c2.coverage["tau"] is False) + + # --- queue helper: enumeration does not dequeue --- + q_task = await task_queue.create_task(task_id="q-implementation", description="q", + agent_role="implementation", + required_capabilities=["python"]) + before = await task_queue.get_pending_count() + cands = await task_queue.get_ready_pending_tasks(["python"]) + after = await task_queue.get_pending_count() + check("get_ready_pending_tasks enumerates without dequeue", + any(c.task_id == "q-implementation" for c in cands) and before == after) + + print() + if failures: + print(f"{len(failures)} decision-engine check(s) FAILED: {failures}") + sys.exit(1) + print("all ACO decision-engine checks passed") + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/scripts/test-quality.py b/scripts/test-quality.py new file mode 100644 index 0000000..f902855 --- /dev/null +++ b/scripts/test-quality.py @@ -0,0 +1,112 @@ +"""Integration test for benchmark Group B: fixture grading -> Q_quality -> reward. + +Boots the in-memory store, seeds a completed run whose tasks carry generated code, grades it +against the held-out `add_function` fixture in the sandbox, and asserts Q_quality and reward +become real. Hermetic, no model key. Run from agent_swarm_v6: python scripts/test-quality.py +""" +import asyncio +import json +import math +import os +import sys +from pathlib import Path + +os.environ["REDIS_FAKE"] = "1" +os.environ["ENABLE_QUALITY_EVAL"] = "1" # gate ON for this test (executes code in the sandbox) +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + +from orchestrator.redis_client import redis_client +from orchestrator import swarm_runtime as sr_mod +from orchestrator.swarm_runtime import swarm_runtime +from orchestrator.task_queue import task_queue, TaskStatus +from orchestrator.quality import evaluate_run_quality, collect_generated_files +from benchmark.collectors.run_collector import SwarmRunMetricsCollector + +failures = [] + + +def check(name, cond): + print(("PASS" if cond else "FAIL"), "-", name) + if not cond: + failures.append(name) + + +async def _noop(self, *a, **k): + return None + + +# A correct implementation + the swarm's OWN test (which must NOT be the grader). +IMPL = "def add(a, b):\n return a + b\n" +RESULT = { + "success": True, + "subtasks": [ + {"status": "completed", "files": [{"path": "calc.py", "action": "write", "content": IMPL}]}, + {"status": "completed", "files": [{"path": "test_calc.py", "action": "write", + "content": "from calc import add\ndef test_self():\n assert add(1, 1) == 2\n"}]}, + ], + "usage": {"model_cost_usd": 2.0}, +} + + +async def add_completed_task(run, task_id, result): + t = await task_queue.create_task(task_id=task_id, description=task_id, + agent_role=task_id.split("-")[-1], depends_on=[], enqueue=False) + t.status = TaskStatus.COMPLETED + t.assigned_agent_id = "A" + t.retry_count = 0 + t.result = json.dumps(result) + await task_queue._save_task(t) + await swarm_runtime.attach_task(run, t.task_id) + return t + + +async def main(): + await redis_client.connect() + sr_mod.SwarmRuntime._post_callback = _noop + + body = { + "mode": "swarm", + "requirement": {"objective": "Write add(a,b)"}, + "orchestration_plan": {"budget": {"max_cost_usd": 10}}, + "callback": {"url": "http://127.0.0.1:9/cb", "subscribed_events": []}, + "metadata": {"manager_deployment_id": "m-q", "benchmark_fixture_id": "add_function"}, + } + run, _ = await swarm_runtime.get_or_create_run(body=body, idempotency_key=None, correlation_id="cq") + impl_task = await add_completed_task(run, "t-implementation", RESULT) + tasks = [impl_task] + + # file separation: impl vs the swarm's own tests + files = collect_generated_files(tasks) + check("collect splits impl vs agent tests", + [f.path for f in files["impl"]] == ["calc.py"] and len(files["agent_tests"]) == 1) + + # grade against held-out fixture tests in the sandbox + quality = await evaluate_run_quality(run, tasks) + check("quality produced", quality is not None) + check("fixture TestPassRate = 100 (4/4 held-out)", quality and quality["test_pass_rate"] == 100.0) + check("Q_quality = 100 (masked: only test_pass present)", quality and quality["q_quality"] == 100.0) + check("agent self-test kept as separate signal", quality and quality["agent_test_pass_rate"] == 100.0) + check("code_review/user_acceptance remain uncollected (None)", + quality and quality["code_review_score"] is None and quality["user_acceptance"] is None) + + # record + simulate a 30s run so V_speed is computable + await swarm_runtime.record_quality(run, quality) + run.created_at = run.updated_at - 30.0 + await swarm_runtime.save_run(run) + + collector = SwarmRunMetricsCollector(run.swarm_id) + m = await collector.collect() + check("reward is now REAL (not NaN)", not math.isnan(m.reward) and collector.coverage["reward"] is True) + # gain/p_decision/benchmark must still be NaN — Group B does not close them + check("gain still NaN (needs baselines)", math.isnan(m.s_gain) and collector.coverage["s_gain"] is False) + check("benchmark still NaN (aggregate)", math.isnan(m.benchmark) and collector.coverage["benchmark"] is False) + + print() + if failures: + print(f"{len(failures)} quality check(s) FAILED: {failures}") + sys.exit(1) + print("all Group B quality checks passed") + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/scripts/test-sandbox.py b/scripts/test-sandbox.py new file mode 100644 index 0000000..4c4e9ed --- /dev/null +++ b/scripts/test-sandbox.py @@ -0,0 +1,82 @@ +"""Test the in-pod code sandbox: real subprocess execution, counting, timeout, isolation. + +Hermetic, no model key. Run from agent_swarm_v6: python scripts/test-sandbox.py +""" +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + +from orchestrator.sandbox import SandboxFile, run_tests + +failures = [] + + +def check(name, cond): + print(("PASS" if cond else "FAIL"), "-", name) + if not cond: + failures.append(name) + + +# --- 1. bare pytest-style tests against a generated module: 2/3 pass --- +impl = [SandboxFile("calc.py", "def add(a, b):\n return a + b\n")] +tests = [SandboxFile( + "test_calc.py", + "from calc import add\n" + "def test_add_pos():\n assert add(1, 2) == 3\n" + "def test_add_zero():\n assert add(0, 0) == 0\n" + "def test_wrong():\n assert add(1, 1) == 3\n", +)] +r = run_tests(impl, tests) +check("runs bare functions: total=3", r.total == 3) +check("counts passed=2", r.passed == 2) +check("counts failed=1", r.failed == 1) +check("pass_rate ~= 66.67", r.pass_rate is not None and round(r.pass_rate, 2) == 66.67) + +# --- 2. unittest.TestCase style is also collected --- +ut = [SandboxFile( + "test_ut.py", + "import unittest\n" + "class T(unittest.TestCase):\n" + " def test_ok(self):\n self.assertEqual(2 + 2, 4)\n" + " def test_bad(self):\n self.assertEqual(2 + 2, 5)\n", +)] +r2 = run_tests([], ut) +check("unittest TestCase collected: total=2", r2.total == 2) +check("unittest passed=1 failed=1", r2.passed == 1 and r2.failed == 1) + +# --- 3. import error in test counts as errored, not a crash --- +bad = [SandboxFile("test_imp.py", "import this_module_does_not_exist_xyz\n")] +r3 = run_tests([], bad) +check("import error -> errored>=1, no crash", r3.errored >= 1 and r3.error is None) + +# --- 4. timeout is enforced and reported (no hang) --- +loop = [SandboxFile("test_loop.py", "def test_spin():\n while True:\n pass\n")] +r4 = run_tests([], loop, timeout_seconds=3) +check("infinite loop times out", r4.timed_out is True) + +# --- 5. secrets are NOT visible to sandboxed code (env scrub) --- +import os as _os +_os.environ["OPENAI_API_KEY"] = "sk-should-not-leak" +_os.environ["AWS_SECRET_ACCESS_KEY"] = "should-not-leak" +leak = [SandboxFile( + "test_leak.py", + "import os\n" + "def test_no_openai():\n assert os.environ.get('OPENAI_API_KEY') is None\n" + "def test_no_aws():\n assert os.environ.get('AWS_SECRET_ACCESS_KEY') is None\n", +)] +r5 = run_tests([], leak) +check("env scrub: no secrets leak into sandbox", r5.total == 2 and r5.passed == 2) + +# --- 6. path traversal is rejected --- +try: + run_tests([SandboxFile("../escape.py", "x=1")], []) + check("path traversal rejected", False) +except ValueError: + check("path traversal rejected", True) + +print() +if failures: + print(f"{len(failures)} sandbox check(s) FAILED: {failures}") + sys.exit(1) +print("all sandbox checks passed")