diff --git a/agent/main.py b/agent/main.py
index f45d169..05bc2f4 100644
--- a/agent/main.py
+++ b/agent/main.py
@@ -403,6 +403,15 @@ class Agent:
         except Exception as e:
             logger.error(f"Failed to propose task: {e}")
 
+    @staticmethod
+    def _git_push_enabled() -> bool:
+        # New convergence architecture (feat/queen-convergence): worker agents no longer push a
+        # per-task git WORK branch into the delivery repo — intermediate artifacts ride in
+        # task.result["files"] and the orchestrator's aggregation node integrates them onto main.
+        # Default OFF. The legacy "commit + push agent/<id>/<task> branch" path runs ONLY when an
+        # operator explicitly opts in with AGENT_GIT_PUSH_ENABLED=true (e.g. for debugging).
+        return os.getenv("AGENT_GIT_PUSH_ENABLED", "false").lower() in {"1", "true", "yes"}
+
     async def execute_assignment(self, assignment: TaskAssignment):
         # Execute one accepted assignment, manage workspace/git flow, and publish lifecycle updates.
         async with self.task_semaphore:
@@ -426,6 +435,7 @@ class Agent:
 
             git_enabled = False
             task_git = None
+            git_push_enabled = self._git_push_enabled()
             try:
                 # Each task runs in its OWN git worktree cut from this agent's clone, so an agent can
                 # run several tasks concurrently without sharing a checkout/index. The worktree holds
@@ -433,7 +443,13 @@ class Agent:
                 # source there and the per-task GitOperations (task_git) commits/pushes that branch.
                 # This fixes the prior empty per-task subdir → "Empty workspace: no files detected".
                 # Falls back to the repo root (no isolated branch) only if worktree creation fails.
-                if await self.workspace_git.is_git_workspace():
+                #
+                # Convergence default (AGENT_GIT_PUSH_ENABLED unset/false): we do NOT cut a per-task
+                # result branch at all — there is nothing to commit/push because artifacts are
+                # returned in task.result["files"]. The task simply executes against the shared repo
+                # root for read context. The isolated worktree branch is created only in the legacy
+                # opt-in push path below.
+                if git_push_enabled and await self.workspace_git.is_git_workspace():
                     async with self._git_admin_lock:  # serialize shared-repo git plumbing only
                         result_branch = await self.workspace_git.add_task_worktree(str(task_workspace), task_id)
                     if result_branch:
@@ -468,7 +484,14 @@ class Agent:
                 if result.get("success"):
                     self.last_summary = self._summarize_result(result) or self.last_summary
                 if result.get("success") and not awaiting_handoff:
-                    if git_enabled and task_git:
+                    if not git_push_enabled:
+                        # Convergence default: artifacts are carried in result["files"]; the
+                        # orchestrator aggregation node integrates them. No work branch is pushed.
+                        result["git_skipped"] = (
+                            "git push disabled (AGENT_GIT_PUSH_ENABLED=false); "
+                            "artifacts returned in task.result.files"
+                        )
+                    elif git_enabled and task_git:
                         commit_sha = await task_git.commit_changes(
                             message=f"Task {task_id}: {description[:50]}"
                         )
diff --git a/agent/task_executor.py b/agent/task_executor.py
index 4e19ab7..e862185 100644
--- a/agent/task_executor.py
+++ b/agent/task_executor.py
@@ -143,6 +143,14 @@ class TaskExecutor:
                 "agent_id": self.agent_id,
                 "usage": self._usage_payload(time.time() - started_at),
             }
+            # Hoist every subtask's generated files to a top-level `files` array so the produced
+            # artifacts travel in task.result and reach the orchestrator's aggregation node WITHOUT
+            # a git push of a work branch (the swarm's new convergence path; orchestrator
+            # quality.collect_generated_files consumes result['files']). Each entry keeps the frozen
+            # shape {path, content, action} with the COMPLETE file content; last writer wins per path
+            # so a later subtask that rewrites a file supersedes an earlier one. Deletes are carried
+            # through as {path, action:"delete"} (no content) for the aggregator to honor.
+            payload["files"] = self._aggregate_subtask_files(results)
             if not success:
                 # Surface the model's OWN failure explanation (what it saw / what was missing) as a
                 # top-level `error`, so the orchestrator/Manager records WHY instead of a generic
@@ -175,6 +183,37 @@ class TaskExecutor:
         finally:
             self.current_context = {}
 
+    @staticmethod
+    def _aggregate_subtask_files(results: list[dict]) -> list[dict]:
+        """Flatten the per-subtask file specs into one ordered, de-duplicated `files` list.
+
+        Each subtask result carries the model's `files` (the same {path, action, content} specs the
+        executor applied to the workspace). We re-emit them at the top level so the artifacts travel
+        in task.result instead of a pushed git branch. Last writer wins per path (a later subtask
+        rewriting/deleting a file supersedes an earlier write); ordering follows last occurrence.
+        Only writes with non-None content and deletes are kept; malformed entries are skipped.
+        """
+        by_path: dict[str, dict] = {}
+        for r in results:
+            if not isinstance(r, dict):
+                continue
+            for f in (r.get("files") or []):
+                if not isinstance(f, dict):
+                    continue
+                path = f.get("path")
+                if not path:
+                    continue
+                action = f.get("action", "write")
+                if action == "delete":
+                    by_path[path] = {"path": path, "action": "delete"}
+                elif action == "write":
+                    content = f.get("content")
+                    if content is None:
+                        continue  # an incomplete write (no content) is not a usable artifact
+                    by_path[path] = {"path": path, "content": content, "action": "write"}
+                # unknown actions are ignored (the orchestrator only consumes write/delete)
+        return list(by_path.values())
+
     async def _maybe_propose_subtasks(self, task_id: str, description: str, context: dict,
                                       proposal_callback: Optional[Callable]) -> int:
         """agent_swarm#7: decompose a top-level seed and propose each subtask to the shared pool.
diff --git a/benchmark/collect_cli.py b/benchmark/collect_cli.py
new file mode 100644
index 0000000..5a57136
--- /dev/null
+++ b/benchmark/collect_cli.py
@@ -0,0 +1,198 @@
+"""北极星采集器 CLI — 从运行中的 orchestrator 采集一次 swarm run 的真实 metrics,持久化到本地。
+
+设计:纯 stdlib + benchmark.metrics 公式,**不 import orchestrator 重依赖**,宿主直接跑、直接落桌面,
+无需重构建镜像。数据源 = orchestrator HTTP API(tasks/events/summary)。
+
+诚实(组织规则 #9):只采有真实数据源的 metrics;无数据的标 NaN + coverage False,绝不伪造 0/100。
+持久化:① sqlite(~/Desktop/swarm-benchmark.db,可累积/可查询)② JSON(~/Desktop/)。
+涌现增益 s_gain = Q_swarm - Q_base:给 --baseline 时用两个 run 的质量分(--q-swarm/--q-base 优先,
+否则回退 completion 作代理并显式标注 proxy)。
+
+用法:
+  python -m benchmark.collect_cli <deployment_id> \
+      [--baseline <deployment_id>] [--q-swarm 80] [--q-base 60] \
+      [--orchestrator http://localhost:8000] [--label humaneval-bon] [--db PATH]
+"""
+from __future__ import annotations
+
+import argparse
+import json
+import math
+import os
+import sqlite3
+import time
+import urllib.request
+from collections import Counter
+
+from .metrics import (
+    completion_score, collaboration_score, robustness_score, communication_score,
+    cost_score, governance_score, emergence_gain,
+)
+
+
+def _get(base: str, path: str) -> dict:
+    req = urllib.request.Request(base.rstrip("/") + path)
+    with urllib.request.urlopen(req, timeout=20) as r:
+        return json.loads(r.read().decode() or "{}")
+
+
+def _is_status(t: dict, name: str) -> bool:
+    return str(t.get("status", "")).lower() == name
+
+
+def collect(base: str, dep_id: str) -> dict:
+    """采集一次 run 的真实 metrics + coverage(诚实标注)。"""
+    tasks = (_get(base, f"/api/swarms/{dep_id}/tasks").get("data") or {}).get("tasks") or []
+    events = (_get(base, f"/api/swarms/{dep_id}/events").get("data") or {}).get("events") or []
+    summary = _get(base, f"/api/swarms/{dep_id}").get("data") or {}
+    et = Counter(e.get("event_type") for e in events)
+    cov: dict = {}
+
+    # completion(真实)
+    total = len(tasks)
+    done = sum(1 for t in tasks if _is_status(t, "completed"))
+    s_completion = completion_score(done, total) if total else math.nan
+    cov["s_completion"] = total > 0
+
+    # collaboration(真实):handoff 成功率 + 依赖解析率 + 负载均衡
+    req_h, comp_h = et.get("handoff.requested", 0), et.get("handoff.completed", 0)
+    handoff = (100.0 * comp_h / req_h) if req_h else 100.0
+    deps = [t for t in tasks if t.get("depends_on")]
+    done_ids = {t["task_id"] for t in tasks if _is_status(t, "completed")}
+    resolved = [t for t in deps if all(d in done_ids for d in t["depends_on"])]
+    dep_res = (100.0 * len(resolved) / len(deps)) if deps else 100.0
+    per_agent = Counter(t.get("assigned_agent_id") for t in tasks if t.get("assigned_agent_id"))
+    bal = (100.0 * min(per_agent.values()) / max(per_agent.values())) if per_agent else 100.0
+    s_collaboration = collaboration_score(handoff, dep_res, bal) if total else math.nan
+    cov["s_collaboration"] = total > 0
+
+    # robustness(真实):从失败中恢复
+    failures = [t for t in tasks if (t.get("retry_count") or 0) > 0 or _is_status(t, "failed")]
+    recovered = [t for t in failures if _is_status(t, "completed")]
+    s_robustness = robustness_score(len(recovered), len(failures))
+    cov["s_robustness"] = True
+
+    # communication(有 peer 消息才采)
+    msg_req = et.get("agent.message.request", 0) or et.get("message.request", 0)
+    msg_rep = et.get("agent.message.reply", 0) or et.get("message.reply", 0)
+    if msg_req:
+        s_communication = communication_score(min(msg_rep, msg_req), msg_req)
+        cov["s_communication"] = True
+    else:
+        s_communication = math.nan
+        cov["s_communication"] = False
+
+    # governance(有审批才采)
+    approvals = summary.get("approvals") or {}
+    appr_list = list(approvals.values()) if isinstance(approvals, dict) else (approvals or [])
+    if appr_list:
+        compliant = sum(1 for a in appr_list if a.get("decision") in ("approved", "rejected"))
+        s_governance = governance_score(compliant, len(appr_list))
+        cov["s_governance"] = True
+    else:
+        s_governance = math.nan
+        cov["s_governance"] = False
+
+    # cost(需 budget+usage,HTTP summary 一般不给 → 诚实 NaN)
+    s_cost = math.nan
+    cov["s_cost"] = False
+
+    quality = summary.get("quality") or {}
+    return {
+        "deployment_id": dep_id,
+        "swarm_id": summary.get("swarm_id"),
+        "status": summary.get("status") or summary.get("state"),
+        "n_tasks": total,
+        "n_completed": done,
+        "n_agents": len(per_agent),
+        "metrics": {
+            "s_completion": s_completion,
+            "s_collaboration": s_collaboration,
+            "s_robustness": s_robustness,
+            "s_communication": s_communication,
+            "s_governance": s_governance,
+            "s_cost": s_cost,
+        },
+        "coverage": cov,
+        "q_quality_run": quality.get("q_quality"),
+        "event_counts": dict(et),
+    }
+
+
+def _persist_sqlite(db_path: str, row: dict):
+    os.makedirs(os.path.dirname(db_path), exist_ok=True)
+    con = sqlite3.connect(db_path)
+    con.execute("""CREATE TABLE IF NOT EXISTS runs(
+        ts INTEGER, label TEXT, deployment_id TEXT, swarm_id TEXT, status TEXT,
+        n_tasks INTEGER, n_completed INTEGER, n_agents INTEGER,
+        s_completion REAL, s_collaboration REAL, s_robustness REAL,
+        s_gain REAL, q_swarm REAL, q_base REAL,
+        coverage_json TEXT, raw_json TEXT)""")
+    m = row["metrics"]
+    con.execute("INSERT INTO runs VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)", (
+        row["ts"], row.get("label", ""), row["deployment_id"], row.get("swarm_id"),
+        row.get("status"), row["n_tasks"], row["n_completed"], row["n_agents"],
+        m["s_completion"], m["s_collaboration"], m["s_robustness"],
+        row.get("s_gain", math.nan), row.get("q_swarm", math.nan), row.get("q_base", math.nan),
+        json.dumps(row["coverage"]), json.dumps(row, default=str),
+    ))
+    con.commit(); con.close()
+
+
+def main():
+    ap = argparse.ArgumentParser(description="北极星采集器:采集 swarm run metrics 并持久化")
+    ap.add_argument("deployment_id")
+    ap.add_argument("--baseline", help="单 agent 基线 run 的 deployment_id(算涌现增益 s_gain)")
+    ap.add_argument("--q-swarm", type=float, help="蜂群质量分(0-100,如测试 pass rate);省略则用 completion 代理")
+    ap.add_argument("--q-base", type=float, help="基线质量分(0-100)")
+    ap.add_argument("--orchestrator", default=os.getenv("ORCH_URL", "http://localhost:8000"))
+    ap.add_argument("--label", default="")
+    ap.add_argument("--db", default=os.path.expanduser("~/Desktop/swarm-benchmark.db"))
+    a = ap.parse_args()
+
+    ts = int(time.time())
+    row = collect(a.orchestrator, a.deployment_id)
+    row["ts"] = ts
+    row["label"] = a.label
+
+    # 涌现增益 s_gain = Q_swarm - Q_base
+    s_gain = math.nan; q_s = a.q_swarm; q_b = a.q_base; proxy = False
+    if a.baseline:
+        base = collect(a.orchestrator, a.baseline)
+        if q_s is None:
+            q_s = row["metrics"]["s_completion"]; proxy = True
+        if q_b is None:
+            q_b = base["metrics"]["s_completion"]; proxy = True
+        if not (math.isnan(q_s) or math.isnan(q_b)):
+            s_gain = emergence_gain(q_s, q_b)
+        row["baseline"] = base
+    row["s_gain"] = s_gain; row["q_swarm"] = q_s if q_s is not None else math.nan
+    row["q_base"] = q_b if q_b is not None else math.nan
+    row["s_gain_proxy"] = proxy
+
+    _persist_sqlite(a.db, row)
+    json_path = os.path.join(os.path.dirname(a.db),
+                             f"swarm-benchmark-{a.label or a.deployment_id}-{ts}.json")
+    with open(json_path, "w", encoding="utf-8") as f:
+        json.dump(row, f, ensure_ascii=False, indent=2, default=str)
+
+    def fmt(v):
+        return "NaN(未采)" if isinstance(v, float) and math.isnan(v) else f"{v:.1f}" if isinstance(v, float) else v
+    m = row["metrics"]
+    print(f"=== 采集 {a.deployment_id} (label={a.label}) ===")
+    print(f"  任务 {row['n_completed']}/{row['n_tasks']} 完成 | agent 数 {row['n_agents']} | 状态 {row['status']}")
+    print(f"  s_completion    = {fmt(m['s_completion'])}  [cov={row['coverage']['s_completion']}]")
+    print(f"  s_collaboration = {fmt(m['s_collaboration'])}  [cov={row['coverage']['s_collaboration']}]")
+    print(f"  s_robustness    = {fmt(m['s_robustness'])}  [cov={row['coverage']['s_robustness']}]")
+    print(f"  s_communication = {fmt(m['s_communication'])}  [cov={row['coverage']['s_communication']}]")
+    print(f"  s_governance    = {fmt(m['s_governance'])}  [cov={row['coverage']['s_governance']}]")
+    print(f"  s_cost          = {fmt(m['s_cost'])}  [cov={row['coverage']['s_cost']}]")
+    if a.baseline:
+        tag = "(completion 代理)" if proxy else "(质量分)"
+        print(f"  涌现增益 s_gain = {fmt(s_gain)}  = Q_swarm({fmt(q_s)}) - Q_base({fmt(q_b)}) {tag}")
+    print(f"\n  持久化 → sqlite: {a.db}")
+    print(f"  持久化 → json:   {json_path}")
+
+
+if __name__ == "__main__":
+    main()
diff --git a/docs/swarm/convergence-protocol.md b/docs/swarm/convergence-protocol.md
index 85f656e..cf4102a 100644
--- a/docs/swarm/convergence-protocol.md
+++ b/docs/swarm/convergence-protocol.md
@@ -125,22 +125,64 @@ consensus = 100 × (未被任何冲突牵连的已完成任务数 / 已完成任
 **诚实差距**：
 - **解释性，非权威**：报告**不覆盖** `run.status`（今天 next_status 与报告对 completed/failed 一致）。authoritative 模式（如 `BLOCKED`→置 run `blocked`）改变 Manager 面终态语义，须先过 `scripts/test-runtime-contract.py` + 契约评审——列为后续。
 - **质量/预算/风险输入有条件**：`quality` 依赖 Group B fixture 评分（绑定 fixture 时）；`budget`/`usage` 需 run 提供；`risks` 需上游注入。缺失时相应原因不触发，回退 `tasks_completed`（不伪造）。
-- **消解为保守首版**：不做自动 merge / 自动选胜，硬冲突留待重做或人工。
+- **消解为保守首版**：`convergence.py` 的冲突消解层不做自动选胜，硬冲突留待重做或人工。注意这与 §7 的**协作聚合**是不同层次：聚合节点（§7.2）对**互补**子产物做 LLM 合并属正常收口，不是「冲突消解」；冲突消解针对的是同路径不一致/测试失败等异常。互补聚合不依赖「选胜」。
 - **收敛事件不进 Manager 流**：六个 `convergence.*`/`conflict.*`/`consensus.*` 事件**构建器已实现但不经 `emit_event` 外发**（未在 Manager `agent_callback.go` 注册；与 `swarm.health` 同策略，避免向订阅全部的回调投递未登记事件）。登记后方可启用 Manager 侧发送。`termination_reason` 以**新增可选字段**附在 `timeline.updated`，对旧消费方向后兼容。
 
-## 7. 蜂后收敛闭环（Queen，agent_swarm#8，`orchestrator/queen.py`）
+## 7. 收敛 = 协作聚合（不是 best-of-N 选优）
 
-fan-out 后的「收口」由**蜂后（Queen）**承担——一个**不执行、不分配任务**的终态仲裁层，把「任务完成即停」升级为「质量驱动收敛」。对齐公开 Swarm 范式的「感知→决策→交互→更新」迭代循环 + 终止「任务完成 ∨ 质量阈值 ∨ 预算/轮次」。
+> **理念来源**：去中心化蜂群范式（掘金《蜂群智能多 Agent 框架》理念，舆情分析案例）——「**个体简单、群体智能**」。涌现来自**分工协作 + stigmergy 间接协调**，**不是**多个独立解相互竞争后选一个最优。本节据此把旧的「蜂后 best-of-N 选最优」收敛模型**重写为协作聚合模型**。
 
-**职责（已实现）**：
-- **best-of-N 选最优（M2 / SC-5·6·7）**：`queen.aggregate_run` 收集各 agent 候选产物 → `score_candidates`（共享 test 跑各 impl，复用 `sandbox.run_tests`）→ `select_best`（测试通过率最高，纯函数）。winner 标到 `deliverable.selected`，verdict 存 `run.metadata["queen"]`。这是单模型没有的涌现杠杆。
-- **质量门打回（M3 / SC-9）**：`queen_quality_gate` 在 `run_cross_review` 之后、状态提交之前——最优分 < `QUEEN_ACCEPTANCE_THRESHOLD` 且 `review_cycles < MAX_REVIEW_CYCLES` → `reopen_task` 回灌迭代；达标/触顶 → 收敛。`should_bounce` 为纯函数。
-- **失败隔离（M4 / SC-12）**：单 task 失败不拖垮整个 run；仅「全失败且无完成产物」才 failed，否则交蜂后/convergence 判定。
-- **防跨 run 抢夺（P0 / #8）**：`extract_swarm_from_agent` / `_agent_belongs_to_run`——agent 只能竞争/认领自己 run 的 task；`swarm_dispatch` 过滤、`handle_task_bid/yield/takeover` 拒绝跨 run（`cross_run_denied`）。
+### 7.0 两种收敛语义（先区分，再选默认）
 
-**诚实差距**：
-- **落 main 待端到端**：SC-7 的「git push 最优产物到产物仓 `main`」需 orchestrator 加 git CLI + 凭据持久化；当前先标记 winner（SWE-bench 语境产物是 patch，选最优即够）。
-- **质量门默认禁用**：需 operator 设 `QUEEN_ACCEPTANCE_THRESHOLD` 才打回；评分依赖沙箱隔离（`HEICODE_SANDBOX_ISOLATED`），未确认隔离则不评分（unscored ≠ 0，不打回，规则 #9）。
-- **convergence 全 authoritative（SC-8）**：质量驱动收敛已由 `queen_quality_gate` 实现；让 `ConvergenceReport.status` 完全覆盖 `run.status`（BLOCKED 等）动 Manager 终态语义，列为后续。
-- **北极星（M5，后续）**：在 SWE-bench Pro 50 上对比单 Opus 4.8 的 resolved 率，客观验证涌现是否超越——前提是上述闭环 + 接入真实代码执行环境。
-- **测试**：`scripts/test-queen.py`（select_best + should_bounce）、`scripts/test-run-isolation.py`（防抢）。
+蜂群里「收口」有两种本质不同的语义，必须分开处理：
+
+| 语义 | 何时产生 | 收敛方式 | 是否本仓默认 |
+|---|---|---|---|
+| **互补聚合**（complementary aggregation） | 一个种子任务经**自主分解（#7）**铺成多个**互补**子任务（各做一块，产出不重叠） | **聚合合并**：从共享池读所有子任务产出 → 整合成一份完整产物 | ✅ **主路径** |
+| **多解选优**（best-of-N selection） | 同一个**原子任务**被多 Agent **竞争（#8）**各自给出**可互换**的完整解 | 选优：按质量排序取胜出解 | 仅竞争原子任务时的次要路径 |
+
+> 文章主推**前者**：舆情案例里「情感分析 Agent」「趋势分析 Agent」产出**互补**，由「报告生成 Agent」**汇总所有分析结果**成一份报告——这是合并，不是选优。本仓 fan-out 的产物来自**自主分解的互补子任务**，因此默认走聚合合并。选优只在「同一原子任务有多个可互换完整解」时才适用，属次要路径。
+
+### 7.1 共享结果池：中间产物不进交付仓
+
+对齐文章 `SwarmEnvironment.results`（agent 把产出 `append` 进共享池）：
+
+- 每个子任务完成时，产物以 `task.result.files` 写入**共享结果池**（Redis run 状态里的任务结果，对应文章的 `environment.results`），**而不是**直接提交进交付（git）仓库。
+- 共享池是聚合节点的**唯一输入源**：聚合前没有任何子任务分支落到产物仓。
+- **产物仓永远只有 `main` 一份，无 per-agent 工作分支**——子任务产出停留在共享池（运行时状态）里，落仓是聚合之后的**单次**动作（§7.3）。这也消除了「artifact 碎在各结果分支、需要事后 merge」的旧问题。
+
+### 7.2 聚合节点（ResultAggregator，扮演「报告生成 Agent」）
+
+对齐文章的**专门聚合节点** + `ResultAggregator`（合并结果 / 按质量排序 / 格式转换）+ DAG `mode=wait_all`：
+
+- **wait_all 栅栏**：聚合节点是 DAG 的汇聚点，**等所有上游互补子任务到达终态**后才触发——即文章舆情案例里报告节点 `mode=wait_all` 等情感/趋势分析全部完成。本仓以「该种子下所有互补子任务均完成」作为 wait_all 条件（无活跃 `pending|assigned|in_progress` 互补子任务）。
+- **从共享池读取**：聚合节点从 `SwarmEnvironment.results`（共享结果池）取出该种子下所有子任务的 `task.result.files`，**不**重新执行子任务。
+- **LLM 整合成完整产物**：把互补产出（各子任务的文件/片段）交 LLM **整合**为一份完整、自洽的产物（代码：合并到一致的文件树并消解接口/依赖；文档：汇总成一篇）。这是「报告生成 Agent」职责的实现——`master_agent.synthesize` 作为汇总工具在此复用（仅文字汇总能力；代码整合的边界见 §7.5）。
+- **跑测试验证**：整合后的完整产物跑 held-out / 共享测试（沙箱见下方门控）做验收，而**不是**对每个候选解分别打分选优。
+- **单次落 main**：验证通过后，聚合产物**一次性**提交到产物仓 `main`，并发**单一** `artifact.created`（见 §7.4 契约一致性）。
+
+> `ResultAggregator` 三职责映射：**合并** = LLM 整合互补产出；**按质量排序** = 仅在 §7.0「多解选优」次要路径下对可互换解排序；**格式转换** = 把异构子产物归一为交付格式。主路径用「合并」，不用「排序选优」。
+
+### 7.3 落 main 的单次提交
+
+- 聚合 + 验证通过 → orchestrator 把整合后的完整产物**一次** push 到产物仓 `main`。
+- 全程**无 per-agent 工作分支**、无多分支后置 merge：子任务产出活在共享池，分支层面只有 `main`。
+- 失败隔离仍适用：个别互补子任务失败不必拖垮整个 run；聚合节点对**已到达共享池**的产出做整合，缺失部分按 `tasks_completed` / 冲突语义如实反映（不伪造，规则 #9）。
+
+### 7.4 与冻结契约的一致性（FROZEN v1，不得违背）
+
+聚合收敛流程**完全落在**现有冻结契约内，不新增/不改字段、类型、状态机：
+
+- **单一 artifact**：聚合后**只**发一个 `artifact.created`（整合产物），扁平字段 `{uri, checksum, task_id, size_bytes?, created_at}` 不变。子任务中间产物**不**各发 `artifact.created`（它们在共享池里，不是交付物）。
+- **sequence 递增**：聚合相关回调沿用 per-swarm 严格递增 `sequence`，无空洞。
+- **状态机不变**：聚合是 run 收敛前的内部步骤，不引入新 run/task 状态；终态仍由 §3 终止函数判定（`completed`/`failed` 语义不变）。
+- **13 类客户端事件冻结**：聚合不新增客户端可见事件类型；wait_all/合并属内部编排，对客户端仅体现为既有 `task.*` 与最终 `artifact.created` + `swarm.completed`。
+
+### 7.5 诚实差距（规则 #9，不主张未实现的）
+
+- **本节为目标模型**：上述聚合收敛是按文章理念重写的**协作聚合设计**；与之相对，旧的 **best-of-N 选优**（`queen.aggregate_run` → `score_candidates` → `select_best`，winner 标 `deliverable.selected`）**已废弃为主路径**，仅在 §7.0「多解选优」次要语义下保留（同一原子任务多个可互换解时排序取胜出）。文档以聚合为主路径，不再把选优当作默认收口。
+- **代码整合非纯文字汇总**：`master_agent.synthesize` 当前仅文字汇总；把互补**代码**子产物整合成一致文件树（消解接口/import/依赖冲突）是更强能力，列为后续实现，不冒充已完成。
+- **落 main 待端到端**：聚合产物 push 到 `main` 需 orchestrator 的 git CLI + 凭据持久化打通，端到端待验证。
+- **测试沙箱门控**：聚合后跑验证测试依赖沙箱隔离双门控（`ENABLE_QUALITY_EVAL` + `HEICODE_SANDBOX_ISOLATED`）；未确认隔离则不执行测试、不伪造分数（unscored ≠ 0/100，规则 #9）。
+- **convergence 全 authoritative（后续）**：让 `ConvergenceReport.status` 完全覆盖 `run.status` 动 Manager 终态语义，仍列为后续（同 §6）。
+- **防跨 run 抢夺（P0 / #8，保留）**：`extract_swarm_from_agent` / `_agent_belongs_to_run`——agent 只能竞争/认领自己 run 的 task；`swarm_dispatch` 过滤、`handle_task_bid/yield/takeover` 拒绝跨 run（`cross_run_denied`）。此为竞争路径的隔离保证，与聚合主路径并存。
diff --git a/docs/swarm/decentralized-rework-plan.md b/docs/swarm/decentralized-rework-plan.md
index b691036..e768f58 100644
--- a/docs/swarm/decentralized-rework-plan.md
+++ b/docs/swarm/decentralized-rework-plan.md
@@ -57,7 +57,12 @@
 - **P6 去中心化自选**：✅ 已建（`main.py: swarm_dispatch` + `ENABLE_SWARM_DISPATCH`）：每个空闲 Agent 感知共享池、按 capability+τ+load+budget **自选**最适任务（统一 #9 可解释打分 + #10 信息素 τ），记录可解释 `dispatch.decision_made`。集成测试 `test-swarm-dispatch.py`。cutover 时成为**唯一**派发，删除 greedy/ACO/scored 与各模式开关。
 - **P-cutover 切换为唯一路径**：✅ 已完成。`swarm_dispatch` 为唯一派发（删 greedy/ACO/scored 分支 + `scored_matchmake`）；`build_seed_task_specs` 为唯一任务创建（删 planner-fallback `build_planner_task_specs`/`planner_fallback_enabled`）；删单 critic Master 评审环（`maybe_run_review_cycle`/`review_loop_enabled`），`run_cross_review` 为唯一评审；收敛/提案/竞争/评审原语全部**无条件**（移除全部 `ENABLE_*` 构建开关）；`test-workflow-e2e` 改写为 seed→自选→自主分解→执行→收敛全流程（stub agent 改为感知种子后提案分解），`test-merge-smoke` 的 planner/单评审用例改写为 seeder/cross-review；CI 同步。`master_agent.synthesize` 作为汇总工具保留。
 - **P-guard 守卫（最后）**：✅ 已完成。`orchestrator/guard.py: diagnose`（纯函数）检测 `NO_AGENTS_CONNECTED`/`NO_CAPABLE_AGENT`/`DEPENDENCY_DEADLOCK`/`BUDGET_EXHAUSTED`/`SEED_UNDECOMPOSED` 并给出可读原因；`main.py: assess_swarm_health` 在派发环检测到「有待办却本 tick 无任何分派」时诊断受影响 run，存 `run.metadata["health"]` 并在不健康时发内部 `swarm.health` 事件（仅诊断，不改 run）。测试 `test-swarm-guard.py`。**这是唯一保留的非正常路径处理。**
-- **P-后续 蜂后收敛闭环（agent_swarm#8，已实现 M2–M4）**：fan-out 后的「收口」由 `orchestrator/queen.py` **蜂后**（不执行/不分配的终态仲裁层）承担——best-of-N 选最优（M2）、质量门打回迭代（M3，`QUEEN_ACCEPTANCE_THRESHOLD` 默认禁用）、失败隔离（M4，单 task 失败不拖垮整个 run）、防跨 run 抢夺（P0，`extract_swarm_from_agent`）。设计与诚实差距见 `convergence-protocol.md §7`。**剩余**：SC-7 git push 最优产物落 `main`（待端到端）、SC-8 convergence 全 authoritative、**北极星 M5**（SWE-bench Pro 50 对比单 Opus 4.8 的 resolved 率，客观验证涌现是否超越）。
+- **P-后续 协作聚合收敛闭环（重写，agent_swarm#8）**：fan-out 后的「收口」是**协作聚合**——对齐去中心化蜂群范式（掘金《蜂群智能多 Agent 框架》舆情案例）的「**个体简单、群体智能**」：涌现来自**分工协作 + stigmergy 间接协调**，不是多个独立解竞争选优。
+  - **中间产物走共享结果池**（对应文章 `SwarmEnvironment.results`）：每个互补子任务产出以 `task.result.files` 写入 run 共享态，**不直接进交付（git）仓**；产物仓只有 `main` 一份，**无 per-agent 工作分支**。
+  - **专门聚合节点（ResultAggregator，扮演文章「报告生成 Agent」，DAG `mode=wait_all`）**：等该种子下所有互补子任务到达终态 → 从共享池读全部产出 → LLM **整合**成一份完整产物 → 跑测试验证 → **单次** push 到 `main` 并发**单一** `artifact.created`。
+  - **两种收敛语义**：互补子任务（自主分解 #7 产生）→**聚合合并**（默认主路径）；同一原子任务多个可互换完整解（竞争 #8）→才用**选优**（次要路径）。文章主推前者。
+  - **⚠️ 旧 best-of-N 选优已废弃为主路径**：`orchestrator/queen.py` 的 `aggregate_run`/`score_candidates`/`select_best`（winner 标 `deliverable.selected`）不再作为默认收口，仅保留于「多解选优」次要语义。质量门打回迭代（`QUEEN_ACCEPTANCE_THRESHOLD` 默认禁用）、失败隔离（单 task 失败不拖垮 run）、防跨 run 抢夺（`extract_swarm_from_agent`）仍保留。
+  - 完整设计、与冻结契约一致性、诚实差距见 `convergence-protocol.md §7`。**剩余**：聚合产物 git push 落 `main`（待端到端）、代码级互补整合（非纯文字汇总）、convergence 全 authoritative、**北极星 M5**（SWE-bench Pro 50 对比单 Opus 4.8 的 resolved 率，客观验证涌现是否超越）。
 
 ---
 
diff --git a/k8s/orchestrator-heicode-test.yaml b/k8s/orchestrator-heicode-test.yaml
new file mode 100644
index 0000000..f01b255
--- /dev/null
+++ b/k8s/orchestrator-heicode-test.yaml
@@ -0,0 +1,150 @@
+# heicode-test 云部署清单（SWE-bench 涌现评测 · 生成段）—— 改自 orchestrator-local.yaml。
+# 与本地差异：ACR 镜像（非 local）、Redis 集群（REDIS_URL+REDIS_CLUSTER）、隔离沙箱双门控开、
+# SWARM_EMIT_PATCH 产 diff（不落 main）、Cosmos/Blob 用测试专属 database/container 与生产区分、
+# Service=LoadBalancer 公网裸入（本地测试式，无鉴权）。
+# 所有凭据走 Secret（secretKeyRef），manifest 内零明文。创建 Secret 见文件末尾注释（你自行执行）。
+apiVersion: v1
+kind: Service
+metadata:
+  name: orchestrator-public
+  namespace: swarm-system
+  labels: { app: orchestrator }
+spec:
+  type: LoadBalancer          # 公网入口（无鉴权，仅限隔离测试集群）；建议加 loadBalancerSourceRanges 限 IP
+  ports:
+    - port: 80
+      targetPort: 8000
+      name: http
+  selector: { app: orchestrator }
+---
+apiVersion: v1
+kind: Service
+metadata:
+  name: orchestrator-service    # 集群内 WS 入口（agent pod 连这个），保持与 local 同名
+  namespace: swarm-system
+  labels: { app: orchestrator }
+spec:
+  type: ClusterIP
+  ports:
+    - port: 8000
+      targetPort: 8000
+      name: http
+  selector: { app: orchestrator }
+---
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: orchestrator
+  namespace: swarm-system
+  labels: { app: orchestrator }
+spec:
+  replicas: 1
+  selector:
+    matchLabels: { app: orchestrator }
+  template:
+    metadata:
+      labels: { app: orchestrator }
+    spec:
+      serviceAccountName: swarm-orchestrator
+      # ACR 拉取：AKS 已 attach ACR 时无需此项；否则用 imagePullSecret。
+      # imagePullSecrets: [{ name: acr-pull }]
+      containers:
+        - name: orchestrator
+          image: REPLACE_ACR.azurecr.io/swarm-orchestrator:heicode-test   # ← 填 ACR
+          imagePullPolicy: Always
+          ports:
+            - containerPort: 8000
+              name: http
+          env:
+            # --- Redis 集群（权威存储，非 FAKE）---
+            # REDIS_URL 形如 rediss://:<password>@<host>:<port>/0；含密码 → 走 Secret。
+            - name: REDIS_URL
+              valueFrom:
+                secretKeyRef: { name: swarm-redis, key: REDIS_URL }
+            - name: REDIS_CLUSTER
+              value: "1"                       # Azure Redis Enterprise OSSCluster 协议需要
+            - name: LOG_LEVEL
+              value: "INFO"
+            # --- agent pod 启动（ACR 镜像）---
+            - name: AGENT_LAUNCH_BACKEND
+              value: "kubernetes"
+            - name: AGENT_POD_IMAGE
+              value: "REPLACE_ACR.azurecr.io/swarm-agent:heicode-test"     # ← 填 ACR
+            - name: AGENT_POD_NAMESPACE
+              value: "swarm-system"
+            - name: ORCHESTRATOR_PUBLIC_URL
+              value: "ws://orchestrator-service.swarm-system.svc.cluster.local:8000"
+            # --- 模型网关（OpenAI 兼容）= 你的叮嘱：qwen3.7-max @ api.heicode.cc ---
+            - name: AGENT_OPENAI_API_BASE
+              value: "https://api.heicode.cc/v1"
+            - name: OPENAI_API_BASE
+              value: "https://api.heicode.cc/v1"
+            - name: OPENAI_MODEL
+              value: "qwen3.7-max"
+            - name: MASTER_REVIEW_MODEL          # 聚合器整合也用同模型（锁模型）
+              value: "qwen3.7-max"
+            - name: OPENAI_API_KEY
+              valueFrom:
+                secretKeyRef: { name: swarm-model-key, key: OPENAI_API_KEY }
+            # --- agent 池窗口 ---
+            - name: AGENT_LAUNCH_MIN_POOL
+              value: "16"
+            - name: AGENT_LAUNCH_POOL_SIZE
+              value: "16"
+            - name: AGENT_LAUNCH_MAX_POOL
+              value: "64"
+            - name: MAX_AGENTS_PER_USER
+              value: "64"
+            - name: AGENT_PROPOSAL_BUDGET
+              value: "12"
+            # --- 隔离沙箱双门控（heicode-test 已确认隔离集群）---
+            - name: ENABLE_QUALITY_EVAL
+              value: "1"
+            - name: HEICODE_SANDBOX_ISOLATED
+              value: "1"
+            # --- SWE-bench 评测：产 unified diff，不 push 到 main ---
+            - name: SWARM_EMIT_PATCH
+              value: "1"
+            # --- 采集遥测：与生产 Cosmos/Blob 共用底座，但用测试专属 database/container 区分 ---
+            - name: BENCHMARK_COSMOS_DATABASE
+              value: "benchmark-heicodetest"     # 区分键（生产默认 benchmark）
+            - name: BENCHMARK_COSMOS_CONTAINER
+              value: "selfcert-heicodetest"      # 区分键（生产默认 selfcert）
+            - name: BENCHMARK_BLOB_CONTAINER
+              value: "selfcert-heicodetest"      # 区分键（生产默认 benchmark-selfcert）
+            - name: BENCHMARK_COSMOS_CONNECTION_STRING
+              valueFrom:
+                secretKeyRef: { name: swarm-telemetry, key: COSMOS_CONNECTION_STRING, optional: true }
+            - name: AZURE_STORAGE_CONNECTION_STRING
+              valueFrom:
+                secretKeyRef: { name: swarm-telemetry, key: AZURE_STORAGE_CONNECTION_STRING, optional: true }
+          resources:
+            requests: { memory: "512Mi", cpu: "300m" }
+            limits: { memory: "1Gi", cpu: "1" }
+          livenessProbe:
+            httpGet: { path: /health, port: 8000 }
+            initialDelaySeconds: 20
+            periodSeconds: 10
+            timeoutSeconds: 5
+            failureThreshold: 3
+          readinessProbe:
+            httpGet: { path: /health, port: 8000 }
+            initialDelaySeconds: 8
+            periodSeconds: 5
+            timeoutSeconds: 3
+            failureThreshold: 3
+# ---------------------------------------------------------------------------
+# 创建 Secret（你自行执行，凭据不进任何文件/日志；用 ! 前缀在会话里跑）：
+#   kubectl -n swarm-system create secret generic swarm-model-key \
+#     --from-literal=OPENAI_API_KEY=<模型KEY>
+#   kubectl -n swarm-system create secret generic swarm-redis \
+#     --from-literal=REDIS_URL='rediss://:<password>@<host>:<port>/0'
+#   # 遥测（可选，不重要时可跳过；跳过则 export 自动 no-op）：
+#   kubectl -n swarm-system create secret generic swarm-telemetry \
+#     --from-literal=COSMOS_CONNECTION_STRING='<...>' \
+#     --from-literal=AZURE_STORAGE_CONNECTION_STRING='<...>'
+# 部署：
+#   kubectl create -f k8s/rbac/
+#   kubectl create -f k8s/orchestrator-heicode-test.yaml
+#   kubectl -n swarm-system get svc orchestrator-public -w   # 等 EXTERNAL-IP
+# ---------------------------------------------------------------------------
diff --git a/orchestrator/main.py b/orchestrator/main.py
index 59cc98b..1036f94 100644
--- a/orchestrator/main.py
+++ b/orchestrator/main.py
@@ -37,7 +37,7 @@ from . import autonomous_tasks as autonomous_mod
 from . import task_competition as competition_mod
 from . import cross_review as cross_review_mod
 from . import guard as guard_mod
-from . import queen as queen_mod
+from . import result_aggregator
 from . import agent_launcher
 
 # Configure logging
@@ -698,12 +698,29 @@ async def refresh_swarm_run_status(run):
     if next_status == "completed":
         if await run_cross_review(run, tasks):
             return
-        # Queen quality gate (M3/SC-9): score the candidates and, if the best fails the acceptance
-        # bar (and the review-cycle cap isn't hit), send work BACK for another round instead of
-        # declaring success on substandard output. Disabled by default (no threshold) — keeps
-        # current completion semantics until an operator sets QUEEN_ACCEPTANCE_THRESHOLD.
-        if await queen_quality_gate(run, tasks):
+        # 协作聚合质量门(①合并 ②沙箱验证 ③不达标打回):聚合节点合并各 agent 互补产出 → 跑测试验证 →
+        # 不达标且未到轮次上限则 reopen 打回(run 保持 running,下轮重做),达标/达上限则落 main。
+        # 阈值未设(AGGREGATE_ACCEPTANCE_THRESHOLD)时不打回,保持原完成语义。
+        agg_cycles = int(run.metadata.get("review_cycles", 0) or 0)
+        agg_max = int(os.getenv("MAX_REVIEW_CYCLES", "2") or 2)
+        agg = await result_aggregator.finalize_run(
+            run, tasks, run.objective, _accept_threshold(run), agg_cycles, agg_max)
+        run.metadata["aggregation"] = agg
+        if agg.get("decision") == "bounce":
+            from .quality import collect_generated_files
+            reopened = 0
+            for t in tasks:
+                if collect_generated_files([t]).get("impl"):
+                    if await task_queue.reopen_task(t.task_id):
+                        reopened += 1
+            run.metadata["review_cycles"] = agg_cycles + 1
+            run.metadata["rework_reopens"] = int(run.metadata.get("rework_reopens", 0) or 0) + reopened
+            await swarm_runtime.save_run(run)
+            logger.info("aggregation: bounced run %s — reopened %d for rework (cycle %d, pass_rate=%s)",
+                        run.swarm_id, reopened, agg_cycles + 1,
+                        (agg.get("validation") or {}).get("pass_rate"))
             return
+        await swarm_runtime.save_run(run)
 
     if run.status == next_status:
         return
@@ -734,27 +751,15 @@ async def refresh_swarm_run_status(run):
         results = {t.task_id: {"result": parse_task_result(t) or {}} for t in completed}
         final_summary = await master_agent.synthesize(run.objective, results)
         run.metadata["final_summary"] = final_summary
-        # Queen (agent_swarm#8/#12): aggregate the fan-out agents' artifacts, score each by running
-        # its impl against the swarm's shared tests, and SELECT the single best (best-of-N). Records
-        # the verdict on the run and marks the winner on the deliverable so the result is one
-        # coherent pick, not N scattered branches. Best-effort: never breaks the terminal path.
-        # The Queen verdict was computed by the quality gate above (run.metadata['queen']); mark the
-        # selected winner on the deliverable so the result is one coherent pick, not N branches.
-        queen_summary = run.metadata.get("queen") or {}
-        winner = queen_summary.get("winner") if isinstance(queen_summary, dict) else None
-        if isinstance(deliverable, dict) and winner:
-            deliverable["selected"] = winner
-            deliverable["candidate_count"] = queen_summary.get("candidate_count")
-            # SC-7: promote the winning artifact to the repo's main → one coherent deliverable on
-            # main, not N scattered agent branches. No-op when the run has no git grant.
-            try:
-                promo = await queen_mod.promote_to_main(run, tasks, winner.get("task_id"))
-                run.metadata["queen_promotion"] = promo
-                if isinstance(deliverable, dict) and promo.get("promoted"):
-                    deliverable["promoted_to_main"] = {
-                        "branch": promo.get("branch"), "commit_sha": promo.get("commit_sha")}
-            except Exception as exc:
-                logger.warning("queen promote_to_main failed for run %s: %s", run.swarm_id, exc)
+        # 聚合产物已由上面的协作聚合质量门产出并落 main(run.metadata['aggregation']);折进 deliverable。
+        agg = run.metadata.get("aggregation") or {}
+        if isinstance(deliverable, dict) and agg.get("finalized"):
+            deliverable["aggregated_files"] = agg.get("files")
+            deliverable["merged_file_count"] = agg.get("merged_file_count")
+            promo = agg.get("promotion") or {}
+            if promo.get("promoted"):
+                deliverable["promoted_to_main"] = {
+                    "branch": promo.get("branch"), "commit_sha": promo.get("commit_sha")}
         await swarm_runtime.save_run(run)
 
         # Benchmark Group B: grade the run's generated code against its held-out fixture tests in
@@ -1175,48 +1180,18 @@ async def run_cross_review(run, tasks) -> bool:
     return True
 
 
-def _queen_threshold(run) -> Optional[float]:
-    """Queen acceptance bar (pass_rate 0-100): run.metadata override → QUEEN_ACCEPTANCE_THRESHOLD
-    env → None (gate disabled). None keeps the current task-completion completion semantics."""
-    raw = (run.metadata or {}).get("queen_acceptance_threshold")
+def _accept_threshold(run) -> Optional[float]:
+    """协作聚合验收阈值(pass_rate 0-100):run.metadata 覆盖 → AGGREGATE_ACCEPTANCE_THRESHOLD env →
+    None(门禁用)。None 保持当前完成语义(不打回);设了才启用质量驱动的"不达标打回迭代"。"""
+    raw = (run.metadata or {}).get("aggregate_acceptance_threshold")
     if raw is None:
-        raw = os.getenv("QUEEN_ACCEPTANCE_THRESHOLD")
+        raw = os.getenv("AGGREGATE_ACCEPTANCE_THRESHOLD")
     try:
         return float(raw) if raw not in (None, "") else None
     except (TypeError, ValueError):
         return None
 
 
-async def queen_quality_gate(run, tasks) -> bool:
-    """M3/SC-9: the Queen scores the fan-out candidates and, if the best fails the acceptance bar
-    and the review-cycle cap isn't hit, sends work BACK (reopen impl tasks) for another round
-    instead of declaring success on substandard output. Stores the verdict on run.metadata['queen']
-    (reused by the deliverable). Returns True if it reopened (caller keeps the run RUNNING).
-    Best-effort: never raises, never bounces on a score it couldn't compute (org rule #9)."""
-    try:
-        summary = await queen_mod.aggregate_run(run, tasks)
-        run.metadata["queen"] = summary
-        cycles = int(run.metadata.get("review_cycles", 0) or 0)
-        max_cycles = int(os.getenv("MAX_REVIEW_CYCLES", "2") or 2)
-        if not queen_mod.should_bounce(summary, _queen_threshold(run), cycles, max_cycles):
-            await swarm_runtime.save_run(run)
-            return False
-        reopened = 0
-        for c in summary.get("candidates", []):
-            if await task_queue.reopen_task(c["task_id"]):
-                reopened += 1
-        run.metadata["review_cycles"] = cycles + 1
-        # SC-10: tally quality-driven reopens for P_rework (reopen_task doesn't bump retry_count).
-        run.metadata["rework_reopens"] = int(run.metadata.get("rework_reopens", 0) or 0) + reopened
-        await swarm_runtime.save_run(run)
-        logger.info("queen: quality gate bounced run %s — reopened %d for rework (cycle %d)",
-                    run.swarm_id, reopened, cycles + 1)
-        return reopened > 0
-    except Exception as exc:
-        logger.warning("queen quality gate failed for run %s: %s", run.swarm_id, exc)
-        return False
-
-
 async def _historical_success_map(agent_role: str, agent_ids) -> Dict[str, float]:
     """τ (decision_engine pheromone) per agent, normalized to [0,1] for arbitration."""
     out: Dict[str, float] = {}
@@ -1237,7 +1212,7 @@ async def handle_task_bid(agent_id: str, message: Dict[str, Any]) -> Dict[str, A
     if not run or not task_id:
         return {"recorded": False, "reason": "no_run"}
     if not _agent_belongs_to_run(agent_id, run):
-        logger.warning("queen: rejected cross-run bid — agent %s ∉ run %s (task %s)",
+        logger.warning("run-boundary: rejected cross-run bid — agent %s ∉ run %s (task %s)",
                        agent_id, run.swarm_id, task_id)
         return {"recorded": False, "reason": "cross_run_denied"}
     bid = competition_mod.TaskBid(
@@ -1296,7 +1271,7 @@ async def handle_task_yield(agent_id: str, message: Dict[str, Any]) -> Dict[str,
     if not run or not task_id:
         return {"released": False, "reason": "no_run"}
     if not _agent_belongs_to_run(agent_id, run):
-        logger.warning("queen: rejected cross-run yield — agent %s ∉ run %s (task %s)",
+        logger.warning("run-boundary: rejected cross-run yield — agent %s ∉ run %s (task %s)",
                        agent_id, run.swarm_id, task_id)
         return {"released": False, "reason": "cross_run_denied"}
     yield_msg = competition_mod.TaskYield(
@@ -1325,7 +1300,7 @@ async def handle_task_takeover(agent_id: str, message: Dict[str, Any]) -> Dict[s
     if not task or not run:
         return {"taken_over": False, "reason": "no_task"}
     if not _agent_belongs_to_run(agent_id, run):
-        logger.warning("queen: rejected cross-run takeover — agent %s ∉ run %s (task %s)",
+        logger.warning("run-boundary: rejected cross-run takeover — agent %s ∉ run %s (task %s)",
                        agent_id, run.swarm_id, task_id)
         return {"taken_over": False, "reason": "cross_run_denied"}
     incumbent_id = task.assigned_agent_id
diff --git a/orchestrator/queen.py b/orchestrator/queen.py
deleted file mode 100644
index 7689d62..0000000
--- a/orchestrator/queen.py
+++ /dev/null
@@ -1,237 +0,0 @@
-"""Queen — the swarm's terminal arbitration layer (agent_swarm#8/#12, Queen role).
-
-The Queen does NOT execute or dispatch tasks. She only judges the FINAL result of a run:
-  - aggregate the candidate artifacts produced by the fan-out agents,
-  - score each candidate by running its tests in the sandbox,
-  - SELECT the single best candidate (best-of-N — the emergence lever a single model lacks),
-  - (M2) promote the winner to the artifact repo's `main` as one coherent deliverable,
-  - (M3) if no candidate meets the quality bar, send work BACK for another round (reopen_task),
-  - (P0, in main.py) arbitrate competition and reject cross-run grabs.
-
-This module is intentionally import-light and does NOT import orchestrator.main (avoids an import
-cycle): `aggregate_run` returns a plain summary dict; the caller (refresh_swarm_run_status) folds it
-into the run's deliverable. `select_best` is a pure function so it can be unit-tested without I/O.
-"""
-from __future__ import annotations
-
-import asyncio
-import logging
-from dataclasses import dataclass, field
-from typing import Any, Dict, List, Optional
-
-logger = logging.getLogger(__name__)
-
-
-@dataclass
-class Candidate:
-    """One agent's contribution to the run, with its sandbox score."""
-    task_id: str
-    agent_id: Optional[str]
-    impl_files: List[Any] = field(default_factory=list)   # SandboxFile (impl)
-    score: Optional[float] = None                         # pass_rate 0-100, None = not scored
-    total: int = 0
-    passed: int = 0
-
-
-def select_best(candidates: List[Candidate]) -> Optional[Candidate]:
-    """Pure selection: the candidate with the highest test pass_rate wins. Unscored (None)
-    candidates rank below any scored one; ties and all-unscored fall back to the first candidate
-    that actually carries impl files (deterministic — preserves input order). None if no candidate
-    has impl files."""
-    with_impl = [c for c in candidates if c.impl_files]
-    if not with_impl:
-        return None
-    scored = [c for c in with_impl if c.score is not None]
-    if scored:
-        # max by score; stable on ties (first in input order wins)
-        return max(scored, key=lambda c: (c.score, c.passed, -with_impl.index(c)))
-    return with_impl[0]
-
-
-def should_bounce(summary: Dict[str, Any], threshold: Optional[float],
-                  cycles: int, max_cycles: int) -> bool:
-    """Pure decision (M3/SC-9): should the run be sent BACK for another round?
-
-    True only when the best candidate WAS scored, fell BELOW `threshold`, and the review-cycle cap
-    isn't hit yet. False (= accept / converge) when: no threshold (gate disabled), not scored
-    (honest — don't bounce on a score we couldn't compute, rule #9), already meets the bar, or the
-    cap is reached (convergence then marks MAX_ROUNDS_REACHED on the best-so-far)."""
-    if threshold is None:
-        return False
-    winner = summary.get("winner")
-    if not winner or winner.get("score") is None:
-        return False
-    if winner["score"] >= threshold:
-        return False
-    return cycles < max_cycles
-
-
-def _result_of(task) -> Dict[str, Any]:
-    """task.result → dict (JSON string or dict), {} on failure. Mirrors main.parse_task_result
-    without importing main."""
-    import json
-    raw = getattr(task, "result", None)
-    if isinstance(raw, dict):
-        return raw
-    if isinstance(raw, str):
-        try:
-            return json.loads(raw)
-        except Exception:
-            return {}
-    return {}
-
-
-def collect_candidates(tasks) -> List[Candidate]:
-    """One Candidate per completed task that produced impl files. Reuses quality.collect_generated_files
-    (per task) to split impl vs test_*.py, so the Queen scores each agent's implementation."""
-    from .quality import collect_generated_files
-    out: List[Candidate] = []
-    for t in tasks:
-        files = collect_generated_files([t])
-        impl = files.get("impl") or []
-        if not impl:
-            continue
-        out.append(Candidate(
-            task_id=getattr(t, "task_id", "?"),
-            agent_id=getattr(t, "assigned_agent_id", None) or getattr(t, "agent_role", None),
-            impl_files=impl,
-        ))
-    return out
-
-
-def _shared_tests(tasks) -> List[Any]:
-    """All test_*.py the swarm produced this run — the shared yardstick the Queen scores impls against."""
-    from .quality import collect_generated_files
-    seen: Dict[str, Any] = {}
-    for t in tasks:
-        for tf in (collect_generated_files([t]).get("agent_tests") or []):
-            seen[tf.path] = tf  # de-dup by path, last writer wins
-    return list(seen.values())
-
-
-async def score_candidates(candidates: List[Candidate], tests: List[Any]) -> None:
-    """Score each candidate in-place: run its impl against the shared test set in the sandbox.
-
-    FAIL-CLOSED + honest: if isolation isn't confirmed (sandbox.assert_isolated would refuse) or
-    there are no tests, scores stay None (not 0 — 'not scored' != 'scored zero', org rule #9)."""
-    from .sandbox import run_tests, isolation_confirmed
-    if not tests or not isolation_confirmed():
-        return
-    for c in candidates:
-        try:
-            res = await asyncio.to_thread(run_tests, c.impl_files, tests)
-            c.score = res.pass_rate
-            c.total = res.total
-            c.passed = res.passed
-        except Exception as exc:  # a scoring error leaves this candidate unscored, never crashes
-            logger.warning("queen: scoring candidate %s failed: %s", c.task_id, exc)
-
-
-async def aggregate_run(run, tasks) -> Dict[str, Any]:
-    """Queen entry point at run terminal. Collect → score → select best. Returns a summary dict the
-    caller folds into the deliverable. Never raises (best-effort; a failure leaves winner=None and
-    the caller keeps the legacy per-task deliverable)."""
-    try:
-        candidates = collect_candidates(tasks)
-        if not candidates:
-            return {"winner": None, "candidate_count": 0, "reason": "no_impl_artifacts"}
-        tests = _shared_tests(tasks)
-        await score_candidates(candidates, tests)
-        best = select_best(candidates)
-        return {
-            "winner": (
-                {"task_id": best.task_id, "agent_id": best.agent_id,
-                 "score": best.score, "passed": best.passed, "total": best.total}
-                if best else None
-            ),
-            "candidate_count": len(candidates),
-            "scored": sum(1 for c in candidates if c.score is not None),
-            "candidates": [
-                {"task_id": c.task_id, "agent_id": c.agent_id, "score": c.score}
-                for c in candidates
-            ],
-        }
-    except Exception as exc:  # Queen never breaks the run's terminal path
-        logger.warning("queen: aggregate_run failed for %s: %s", getattr(run, "swarm_id", "?"), exc)
-        return {"winner": None, "candidate_count": 0, "reason": f"error:{exc!r}"}
-
-
-def _auth_url(repo_url: str, user: Optional[str], pw: Optional[str]) -> str:
-    """Embed credentials into an http(s) clone URL. URL-encodes the password (handles '@','/', etc).
-    The result is NEVER logged."""
-    import urllib.parse
-    if not user or not pw or "://" not in repo_url:
-        return repo_url
-    scheme, rest = repo_url.split("://", 1)
-    return f"{scheme}://{urllib.parse.quote(user, safe='')}:{urllib.parse.quote(pw, safe='')}@{rest}"
-
-
-def _git_promote(env: Dict[str, str], base_branch: str, files: List[Any], swarm_id: str) -> Dict[str, Any]:
-    """Sync git: clone base_branch → overwrite with winner files → commit → push base_branch.
-    Blocking (run via asyncio.to_thread). Credentials live only in the clone URL, never logged."""
-    import os
-    import shutil
-    import subprocess
-    import tempfile
-    repo_url = env["GIT_REPO_URL"]
-    auth_url = _auth_url(repo_url, env.get("GIT_USERNAME"), env.get("GIT_PASSWORD"))
-    workdir = tempfile.mkdtemp(prefix="queen-promote-")
-    repo_dir = os.path.join(workdir, "repo")
-
-    def git(*args, cwd=None):
-        return subprocess.run(["git", *args], cwd=cwd or workdir,
-                              capture_output=True, text=True, timeout=180)
-
-    try:
-        if git("clone", "--depth", "1", "--branch", base_branch, auth_url, "repo").returncode != 0:
-            return {"promoted": False, "reason": "clone_failed"}
-        for f in files:
-            dest = os.path.join(repo_dir, f.path)
-            os.makedirs(os.path.dirname(dest) or repo_dir, exist_ok=True)
-            with open(dest, "w", encoding="utf-8") as fh:
-                fh.write(f.content)
-        git("config", "user.email", "queen@heicode.swarm", cwd=repo_dir)
-        git("config", "user.name", "HeiCode Queen", cwd=repo_dir)
-        git("add", "-A", cwd=repo_dir)
-        c = git("commit", "-m", f"queen: promote best swarm artifact ({swarm_id})", cwd=repo_dir)
-        if c.returncode != 0:
-            return {"promoted": False, "reason": "no_changes"}
-        if git("push", "origin", base_branch, cwd=repo_dir).returncode != 0:
-            return {"promoted": False, "reason": "push_failed"}
-        sha = git("rev-parse", "HEAD", cwd=repo_dir).stdout.strip()
-        return {"promoted": True, "commit_sha": sha, "branch": base_branch}
-    except Exception as exc:
-        return {"promoted": False, "reason": f"error:{exc!r}"}
-    finally:
-        shutil.rmtree(workdir, ignore_errors=True)
-
-
-async def promote_to_main(run, tasks, winner_task_id: str) -> Dict[str, Any]:
-    """SC-7: push the winning candidate's files to the artifact repo's base branch (main), so the
-    run delivers ONE coherent artifact, not N scattered agent branches. Credentials are resolved
-    from the grant's secret_ref at call time (never stored/logged). Best-effort: never raises."""
-    try:
-        grant = (run.metadata or {}).get("git_grant") or {}
-        repo_url = grant.get("repo_url")
-        if not repo_url:
-            return {"promoted": False, "reason": "no_git_grant"}
-        wtask = next((t for t in tasks if getattr(t, "task_id", None) == winner_task_id), None)
-        if wtask is None:
-            return {"promoted": False, "reason": "winner_not_found"}
-        from .quality import collect_generated_files
-        cf = collect_generated_files([wtask])
-        files = (cf.get("impl") or []) + (cf.get("agent_tests") or [])
-        if not files:
-            return {"promoted": False, "reason": "winner_no_files"}
-        from .agent_launcher import resolve_git_grant
-        env = resolve_git_grant({"resource_grants": [{
-            "resource_type": "git", "secret_ref": grant.get("secret_ref"),
-            "metadata": {"repo_url": repo_url, "base_branch": grant.get("base_branch", "main")},
-        }]})
-        if not env or "GIT_REPO_URL" not in env:
-            return {"promoted": False, "reason": "grant_unresolved"}
-        return await asyncio.to_thread(_git_promote, env, grant.get("base_branch", "main"),
-                                       files, getattr(run, "swarm_id", "?"))
-    except Exception as exc:
-        logger.warning("queen: promote_to_main failed for %s: %s", getattr(run, "swarm_id", "?"), exc)
-        return {"promoted": False, "reason": f"error:{exc!r}"}
diff --git a/orchestrator/requirements.txt b/orchestrator/requirements.txt
index 01d6f8d..298ef2c 100644
--- a/orchestrator/requirements.txt
+++ b/orchestrator/requirements.txt
@@ -9,6 +9,9 @@ prometheus-client==0.20.0
 httpx==0.28.1
 # dev/CI only: in-memory Redis emulator for the gated REDIS_FAKE/ALLOW_MEMORY_STORE fallback
 fakeredis==2.26.1
+# sandbox test runner: pytest natively runs the test styles agents produce (pytest-style classes,
+# fixtures, parametrize, marks) — the in-pod harness (sandbox_runner.py) prefers it, stdlib fallback.
+pytest==8.3.3
 opentelemetry-api==1.24.0
 opentelemetry-sdk==1.24.0
 opentelemetry-exporter-otlp==1.24.0
diff --git a/orchestrator/result_aggregator.py b/orchestrator/result_aggregator.py
new file mode 100644
index 0000000..39a31cc
--- /dev/null
+++ b/orchestrator/result_aggregator.py
@@ -0,0 +1,389 @@
+"""ResultAggregator — 蜂群的终态聚合节点(对应掘金文章的"报告生成 Agent",mode=wait_all)。
+
+设计依据 https://juejin.cn/post/7603575399255949352 的协作聚合模型,取代旧的"蜂后 best-of-N 选优":
+  - 中间产物存**共享结果池**(task.result.files,经 collect_generated_files 读取),**不进 git 交付仓**;
+  - run 终态(所有子任务完成)→ 从共享池**收集所有 agent 的互补产出**;
+  - 同一文件被多个 agent 各写一部分时,**整合成一份完整产物**:优先 LLM(协作合并),无 LLM 时用
+    **AST 函数级合并**(提取各版本 def/class/import 拼合,而非"取最长版本"糊弄);
+  - 跑测试**验证**(沙箱,fail-closed);
+  - 不达标且未到轮次上限 → **打回迭代**(decision=bounce,调用方 reopen 重做);
+  - 达标/达上限 → **单次**把整合产物 push 到 main —— 产物仓永远只有 main 一份,无 agent 工作分支。
+
+不 import orchestrator.main(避免循环);finalize_run 返回 plain dict,调用方折进 deliverable 并据
+decision 决定是否打回。质量驱动闭环 = 聚合合并 → 验证 → 不过打回 → 达标落库。
+"""
+from __future__ import annotations
+
+import ast
+import asyncio
+import logging
+import os
+import re
+from typing import Any, Dict, List, Optional
+
+logger = logging.getLogger(__name__)
+
+try:
+    from openai import AsyncOpenAI
+except Exception:  # pragma: no cover
+    AsyncOpenAI = None
+
+
+def _aggregator_timeout() -> float:
+    try:
+        return float(os.getenv("AGGREGATOR_TIMEOUT_SECONDS", "90") or 90)
+    except ValueError:
+        return 90.0
+
+
+def _llm():
+    """(client, model) for LLM file integration, or (None, model) when no key — degrades to AST merge."""
+    api_key = os.getenv("OPENAI_API_KEY") or os.getenv("MODEL_API_KEY")
+    api_base = (os.getenv("OPENAI_API_BASE") or os.getenv("MODEL_API_BASE")
+                or "https://api.openai.com/v1")
+    model = (os.getenv("OPENAI_MODEL") or os.getenv("MODEL_NAME") or os.getenv("MODEL_ID")
+             or "gpt-4o-mini")
+    model = os.getenv("MASTER_REVIEW_MODEL", model)
+    client = AsyncOpenAI(api_key=api_key, base_url=api_base) if (api_key and AsyncOpenAI) else None
+    return client, model
+
+
+_FENCE = re.compile(r"^```[a-zA-Z0-9_+-]*\n(.*?)\n```$", re.S)
+
+
+def _strip_fence(text: str) -> str:
+    m = _FENCE.match(text.strip())
+    return m.group(1) if m else text.strip()
+
+
+def _ast_merge_python(contents: List[str]) -> Optional[str]:
+    """确定性整合多个 Python 版本:提取各版本的 import 与顶层 def/class,按名字去重合并(同名后者覆盖)。
+    比"取最长版本"强 —— 真把各 agent 写的不同函数拼成一份完整文件。解析全失败 → None(回退最长)。"""
+    imports: Dict[str, str] = {}
+    defs: Dict[str, str] = {}
+    parsed_any = False
+    for content in contents:
+        try:
+            tree = ast.parse(content)
+        except SyntaxError:
+            continue
+        parsed_any = True
+        for node in tree.body:
+            seg = ast.get_source_segment(content, node)
+            if not seg:
+                continue
+            if isinstance(node, (ast.Import, ast.ImportFrom)):
+                imports[seg.strip()] = seg
+            elif isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)):
+                defs[node.name] = seg
+    if not parsed_any or not defs:
+        return None
+    parts = list(imports.values())
+    if parts:
+        parts.append("")
+    parts.extend(defs.values())
+    return "\n".join(parts).strip() + "\n"
+
+
+def _merge_fallback(path: str, contents: List[str]) -> str:
+    """无 LLM 兜底:Python 走 AST 函数级合并,其他文件取信息量最大版本。"""
+    if path.endswith(".py"):
+        merged = _ast_merge_python(contents)
+        if merged:
+            return merged
+    return max(contents, key=len)
+
+
+# ---------- 收集共享池 ----------
+
+def _collect_per_path(tasks) -> Dict[str, List[Any]]:
+    """从所有子任务的共享池产出,按文件 path 分组保留**每个版本**(不去重 —— 去重会丢互补内容)。
+    返回 {path: [SandboxFile, ...]}。复用 quality.collect_generated_files 逐 task 提取(它已兼容
+    task.result.files 与旧 subtasks 两种结构)。"""
+    from .quality import collect_generated_files
+    by_path: Dict[str, List[Any]] = {}
+    for t in tasks:
+        cf = collect_generated_files([t])
+        for kind in ("impl", "agent_tests"):
+            for f in cf.get(kind) or []:
+                by_path.setdefault(f.path, []).append(f)
+    return by_path
+
+
+# ---------- 同文件互补整合(协作合并) ----------
+
+async def _merge_one_path(path: str, versions: List[Any], objective: str) -> Any:
+    """同一文件多个 agent 版本 → 整合成一份完整产物。
+    单版本直接用;多版本优先 LLM 整合(文章的聚合节点),失败/无 LLM 时用 AST 函数级合并兜底。"""
+    from .sandbox import SandboxFile
+    contents = [v.content for v in versions]
+    if len(contents) == 1:
+        return versions[0]
+    client, model = _llm()
+    if not client:
+        logger.warning("aggregator: no LLM for %s; AST/longest merge of %d versions", path, len(contents))
+        return SandboxFile(path=path, content=_merge_fallback(path, contents))
+    try:
+        joined = "\n\n".join(f"===VERSION {i}===\n{c}" for i, c in enumerate(contents))
+        resp = await client.chat.completions.create(
+            model=model,
+            messages=[
+                {"role": "system", "content": (
+                    "你是蜂群的结果聚合节点。多个 agent 各自实现了同一个文件的不同部分。"
+                    "请把它们整合成一个完整、正确、无重复、可直接运行的文件:合并所有互补的函数/类/导入,"
+                    "去掉重复定义,保持一致风格。只输出该文件的最终完整内容,不要任何解释、不要 markdown 围栏。")},
+                {"role": "user", "content": (
+                    f"目标:{objective}\n文件路径:{path}\n\n"
+                    f"以下是 {len(contents)} 个 agent 各自的版本:\n\n{joined}")},
+            ],
+            timeout=_aggregator_timeout(),
+        )
+        out = _strip_fence((resp.choices[0].message.content or "").strip())
+        return SandboxFile(path=path, content=out or _merge_fallback(path, contents))
+    except Exception as exc:
+        logger.warning("aggregator: LLM merge failed for %s (%s); AST/longest fallback", path, exc)
+        return SandboxFile(path=path, content=_merge_fallback(path, contents))
+
+
+async def merge_artifacts(tasks, objective: str) -> List[Any]:
+    """收集共享池 → 对每个文件整合互补产出 → 返回完整产物文件集(SandboxFile 列表)。"""
+    by_path = _collect_per_path(tasks)
+    merged: List[Any] = []
+    for path, versions in by_path.items():
+        merged.append(await _merge_one_path(path, versions, objective))
+    return merged
+
+
+# ---------- 验证(沙箱,fail-closed) ----------
+
+async def _validate(merged: List[Any]) -> Dict[str, Any]:
+    """跑测试验证整合后的产物。fail-closed:隔离未确认/无测试 → 不评分(诚实 None,非 0)。"""
+    from .sandbox import run_tests, isolation_confirmed
+    from .quality import _is_test_file
+    if not isolation_confirmed():
+        return {"validated": False, "reason": "sandbox_not_isolated", "pass_rate": None}
+    impl = [f for f in merged if not _is_test_file(f.path)]
+    tests = [f for f in merged if _is_test_file(f.path)]
+    if not tests:
+        return {"validated": False, "reason": "no_tests", "pass_rate": None}
+    try:
+        res = await asyncio.to_thread(run_tests, impl, tests)
+        return {"validated": True, "pass_rate": res.pass_rate,
+                "passed": res.passed, "total": res.total}
+    except Exception as exc:
+        return {"validated": False, "reason": f"error:{exc!r}", "pass_rate": None}
+
+
+# ---------- 落 main(单次,复用 git helper) ----------
+
+def _auth_url(repo_url: str, user: Optional[str], pw: Optional[str]) -> str:
+    """凭据嵌入 http(s) clone URL,URL-encode 密码。结果**绝不记日志**。"""
+    import urllib.parse
+    if not user or not pw or "://" not in repo_url:
+        return repo_url
+    scheme, rest = repo_url.split("://", 1)
+    return f"{scheme}://{urllib.parse.quote(user, safe='')}:{urllib.parse.quote(pw, safe='')}@{rest}"
+
+
+def _git_push(env: Dict[str, str], base_branch: str, files: List[Any], swarm_id: str) -> Dict[str, Any]:
+    """同步 git:clone base_branch → 写入整合产物 → commit → push。阻塞,经 to_thread 调用。
+    凭据只活在 clone URL,绝不记日志。"""
+    import shutil
+    import subprocess
+    import tempfile
+    repo_url = env["GIT_REPO_URL"]
+    auth_url = _auth_url(repo_url, env.get("GIT_USERNAME"), env.get("GIT_PASSWORD"))
+    workdir = tempfile.mkdtemp(prefix="aggregate-promote-")
+    repo_dir = os.path.join(workdir, "repo")
+
+    def git(*args, cwd=None):
+        return subprocess.run(["git", *args], cwd=cwd or workdir,
+                              capture_output=True, text=True, timeout=180)
+
+    try:
+        if git("clone", "--depth", "1", "--branch", base_branch, auth_url, "repo").returncode != 0:
+            return {"promoted": False, "reason": "clone_failed"}
+        for f in files:
+            dest = os.path.join(repo_dir, f.path)
+            os.makedirs(os.path.dirname(dest) or repo_dir, exist_ok=True)
+            with open(dest, "w", encoding="utf-8") as fh:
+                fh.write(f.content)
+        git("config", "user.email", "aggregator@heicode.swarm", cwd=repo_dir)
+        git("config", "user.name", "HeiCode Aggregator", cwd=repo_dir)
+        git("add", "-A", cwd=repo_dir)
+        c = git("commit", "-m", f"aggregate: integrated swarm deliverable ({swarm_id})", cwd=repo_dir)
+        if c.returncode != 0:
+            return {"promoted": False, "reason": "no_changes"}
+        if git("push", "origin", base_branch, cwd=repo_dir).returncode != 0:
+            return {"promoted": False, "reason": "push_failed"}
+        sha = git("rev-parse", "HEAD", cwd=repo_dir).stdout.strip()
+        return {"promoted": True, "commit_sha": sha, "branch": base_branch}
+    except Exception as exc:
+        return {"promoted": False, "reason": f"error:{exc!r}"}
+    finally:
+        shutil.rmtree(workdir, ignore_errors=True)
+
+
+def _emit_patch_mode(run) -> bool:
+    """Benchmark 模式(SWE-bench 等):产出 unified diff,不 push 到 main。
+    per-run 门控(run.metadata['emit_patch'])或 per-deployment 门控(env SWARM_EMIT_PATCH)。
+    默认关 —— 生产路径仍走 _promote,行为不变。"""
+    if (run.metadata or {}).get("emit_patch"):
+        return True
+    return os.getenv("SWARM_EMIT_PATCH", "").strip().lower() in {"1", "true", "yes"}
+
+
+def _git_diff(env: Dict[str, str], base_ref: str, files: List[Any], swarm_id: str) -> Dict[str, Any]:
+    """_git_push 的 benchmark 变体:clone → checkout base_ref → 写入整合产物 → `git add -A` →
+    `git diff --cached`(相对 base 的 unified diff)。**只产 diff,不 commit、不 push**。
+    凭据只活在 clone URL,绝不记日志;diff 本身只含代码,无凭据。阻塞,经 to_thread 调用。"""
+    import shutil
+    import subprocess
+    import tempfile
+    repo_url = env["GIT_REPO_URL"]
+    auth_url = _auth_url(repo_url, env.get("GIT_USERNAME"), env.get("GIT_PASSWORD"))
+    workdir = tempfile.mkdtemp(prefix="aggregate-patch-")
+    repo_dir = os.path.join(workdir, "repo")
+
+    def git(*args, cwd=None):
+        return subprocess.run(["git", *args], cwd=cwd or workdir,
+                              capture_output=True, text=True, timeout=300)
+
+    try:
+        # 全量 clone(非 --depth 1):base_ref 可能是任意 base_commit,浅 clone 不含其历史。
+        if git("clone", auth_url, "repo").returncode != 0:
+            return {"emitted": False, "reason": "clone_failed"}
+        if base_ref:
+            if git("checkout", base_ref, cwd=repo_dir).returncode != 0:
+                return {"emitted": False, "reason": "checkout_failed"}
+        for f in files:
+            dest = os.path.join(repo_dir, f.path)
+            os.makedirs(os.path.dirname(dest) or repo_dir, exist_ok=True)
+            with open(dest, "w", encoding="utf-8") as fh:
+                fh.write(f.content)
+        git("add", "-A", cwd=repo_dir)
+        d = git("diff", "--cached", cwd=repo_dir)
+        if d.returncode != 0:
+            return {"emitted": False, "reason": "diff_failed"}
+        if not (d.stdout or "").strip():
+            return {"emitted": False, "reason": "empty_patch"}
+        return {"emitted": True, "patch": d.stdout, "base_ref": base_ref}
+    except Exception as exc:
+        return {"emitted": False, "reason": f"error:{exc!r}"}
+    finally:
+        shutil.rmtree(workdir, ignore_errors=True)
+
+
+async def _emit_patch(run, files: List[Any]) -> Dict[str, Any]:
+    """Benchmark 入口:把整合产物相对 base_commit 产成 unified diff(SWE-bench patch 格式)。
+    凭据按需从 grant.secret_ref 解析,绝不存储/记录。base_ref 优先 grant.base_commit(SWE-bench
+    精确 commit),回退 base_branch。"""
+    try:
+        grant = (run.metadata or {}).get("git_grant") or {}
+        repo_url = grant.get("repo_url")
+        if not repo_url:
+            return {"emitted": False, "reason": "no_git_grant"}
+        if not files:
+            return {"emitted": False, "reason": "no_files"}
+        from .agent_launcher import resolve_git_grant
+        env = resolve_git_grant({"resource_grants": [{
+            "resource_type": "git", "secret_ref": grant.get("secret_ref"),
+            "metadata": {"repo_url": repo_url, "base_branch": grant.get("base_branch", "main")},
+        }]})
+        if not env or "GIT_REPO_URL" not in env:
+            return {"emitted": False, "reason": "grant_unresolved"}
+        base_ref = grant.get("base_commit") or grant.get("base_branch", "main")
+        return await asyncio.to_thread(_git_diff, env, base_ref, files, getattr(run, "swarm_id", "?"))
+    except Exception as exc:
+        logger.warning("aggregator: emit_patch failed for %s: %s", getattr(run, "swarm_id", "?"), exc)
+        return {"emitted": False, "reason": f"error:{exc!r}"}
+
+
+async def _promote(run, files: List[Any]) -> Dict[str, Any]:
+    """把整合产物 push 到产物仓 base 分支(main)。凭据按需从 grant.secret_ref 解析,绝不存储/记录。"""
+    try:
+        grant = (run.metadata or {}).get("git_grant") or {}
+        repo_url = grant.get("repo_url")
+        if not repo_url:
+            return {"promoted": False, "reason": "no_git_grant"}
+        if not files:
+            return {"promoted": False, "reason": "no_files"}
+        from .agent_launcher import resolve_git_grant
+        env = resolve_git_grant({"resource_grants": [{
+            "resource_type": "git", "secret_ref": grant.get("secret_ref"),
+            "metadata": {"repo_url": repo_url, "base_branch": grant.get("base_branch", "main")},
+        }]})
+        if not env or "GIT_REPO_URL" not in env:
+            return {"promoted": False, "reason": "grant_unresolved"}
+        return await asyncio.to_thread(_git_push, env, grant.get("base_branch", "main"),
+                                       files, getattr(run, "swarm_id", "?"))
+    except Exception as exc:
+        logger.warning("aggregator: promote failed for %s: %s", getattr(run, "swarm_id", "?"), exc)
+        return {"promoted": False, "reason": f"error:{exc!r}"}
+
+
+# ---------- 质量决策 ----------
+
+def _decide(validation: Dict[str, Any], threshold: Optional[float],
+            cycles: int, max_cycles: int) -> str:
+    """聚合质量门决策:accept(达标/无门/无法评分/达上限)或 bounce(评分不达标且未到上限)。
+    诚实(规则 #9):无法评分(pass_rate=None)不打回 —— 不在"算不出的分"上重做。"""
+    if threshold is None:
+        return "accept"
+    pass_rate = validation.get("pass_rate")
+    if pass_rate is None:
+        return "accept"
+    if pass_rate >= threshold:
+        return "accept"
+    if cycles >= max_cycles:
+        return "accept"  # 达上限:接受当前最好的,交收敛标 MAX_ROUNDS_REACHED
+    return "bounce"
+
+
+# ---------- 终态入口 ----------
+
+async def finalize_run(run, tasks, objective: str = "", threshold: Optional[float] = None,
+                       cycles: int = 0, max_cycles: int = 2) -> Dict[str, Any]:
+    """聚合节点入口(run 终态调用,此时子任务已 wait_all 完成)。
+    收集共享池 → 整合互补产出 → 验证 → 决策。
+      - decision='bounce':不 promote,调用方 reopen 重做(质量驱动迭代);
+      - decision='accept':单次落 main。
+    Best-effort,绝不让终态路径崩。"""
+    try:
+        merged = await merge_artifacts(tasks, objective or "")
+        if not merged:
+            return {"finalized": False, "decision": "accept", "reason": "no_artifacts",
+                    "merged_file_count": 0}
+        validation = await _validate(merged)
+        decision = _decide(validation, threshold, cycles, max_cycles)
+        if decision == "bounce":
+            return {"finalized": False, "decision": "bounce", "validation": validation,
+                    "merged_file_count": len(merged), "files": [f.path for f in merged]}
+        # Benchmark 模式:产 unified diff 不 push(SWE-bench 等需要 patch,产物仓不能被改);
+        # 生产模式:单次落 main。两条互斥,由 _emit_patch_mode 门控。
+        if _emit_patch_mode(run):
+            patch_res = await _emit_patch(run, merged)
+            return {
+                "finalized": True,
+                "decision": "accept",
+                "mode": "emit_patch",
+                "merged_file_count": len(merged),
+                "files": [f.path for f in merged],
+                "validation": validation,
+                "patch": patch_res.get("patch"),
+                "patch_meta": {k: v for k, v in patch_res.items() if k != "patch"},
+            }
+        promotion = await _promote(run, merged)
+        return {
+            "finalized": True,
+            "decision": "accept",
+            "merged_file_count": len(merged),
+            "files": [f.path for f in merged],
+            "validation": validation,
+            "promotion": promotion,
+        }
+    except Exception as exc:
+        logger.warning("aggregator: finalize_run failed for %s: %s",
+                       getattr(run, "swarm_id", "?"), exc)
+        return {"finalized": False, "decision": "accept", "reason": f"error:{exc!r}",
+                "merged_file_count": 0}
diff --git a/orchestrator/sandbox_runner.py b/orchestrator/sandbox_runner.py
index b1211dd..d441866 100644
--- a/orchestrator/sandbox_runner.py
+++ b/orchestrator/sandbox_runner.py
@@ -1,91 +1,132 @@
 """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.
+Prefers **pytest** (native support for pytest-style classes, fixtures, parametrize, marks AND
+unittest.TestCase) — the test styles agents actually produce. Falls back to a stdlib-only
+collector (unittest.TestCase + bare module-level ``test_*`` functions) when pytest is absent, so
+the harness still works without the dependency. Writes a machine-readable ``_result.json``
+({total, passed, failed, errored, details}); the parent (orchestrator/sandbox.py) reads that
+file and 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.
+Copied into the ephemeral sandbox workdir at run time and executed there with the workdir as CWD.
 """
-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 _run_with_pytest(workdir: str) -> dict:
+    """Run every test under workdir with pytest; collect pass/fail/error via an inline plugin.
+    Native support for pytest classes/fixtures/parametrize/marks + unittest.TestCase."""
+    import pytest
+
+    class _Collector:
+        def __init__(self):
+            self.total = 0
+            self.passed = 0
+            self.failed = 0
+            self.errored = 0
+            self.details = []
+
+        def pytest_runtest_logreport(self, report):
+            text = (getattr(report, "longreprtext", "") or "")[:500]
+            if report.when == "call":
+                self.total += 1
+                if report.outcome == "passed":
+                    self.passed += 1
+                    self.details.append({"test": report.nodeid, "status": "passed"})
+                else:
+                    self.failed += 1
+                    self.details.append({"test": report.nodeid, "status": "failed", "error": text})
+            elif report.when in ("setup", "teardown") and report.outcome == "failed":
+                # setup/teardown failure (e.g. fixture error) counts as one errored test
+                self.total += 1
+                self.failed += 1
+                self.errored += 1
+                self.details.append({"test": report.nodeid, "status": "error", "error": text})
+
+        def pytest_collectreport(self, report):
+            # import/collection failure (e.g. missing dependency) counts as one errored test
+            if report.failed:
+                text = (getattr(report, "longreprtext", "") or "")[:500]
+                self.total += 1
+                self.failed += 1
+                self.errored += 1
+                self.details.append({"test": report.nodeid or "collection",
+                                     "status": "error", "error": text})
+
+    collector = _Collector()
+    # -q quiet, disable cache writes, ignore any repo pytest config so the sandbox is hermetic
+    pytest.main(["-q", "-p", "no:cacheprovider", "--no-header", "-o", "addopts=", workdir],
+                plugins=[collector])
+    return {"total": collector.total, "passed": collector.passed,
+            "failed": collector.failed, "errored": collector.errored, "details": collector.details}
 
 
-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")
-    )
+def _run_with_stdlib(workdir: str) -> dict:
+    """Fallback when pytest is unavailable: unittest.TestCase + bare module-level test_* functions."""
+    import importlib.util
+    import unittest
 
+    def _load_module(path):
+        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)
+        return module
+
+    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)
-
+    bare_funcs = []
     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
+        except Exception as exc:
+            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
+    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.
+    passed += ut_result.testsRun - ut_failed
     for label, fn in bare_funcs:
         total += 1
         try:
-            fn()
-            passed += 1
+            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
+            failed += 1; errored += 1
             details.append({"test": label, "status": "error", "error": repr(exc)})
+    return {"total": total, "passed": passed, "failed": failed, "errored": errored, "details": details}
 
+
+def main() -> None:
+    workdir = os.getcwd()
+    sys.path.insert(0, workdir)
+    try:
+        import pytest  # noqa: F401
+        result = _run_with_pytest(workdir)
+    except ImportError:
+        result = _run_with_stdlib(workdir)
     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,
-        )
+        json.dump(result, fh)
 
 
 if __name__ == "__main__":
diff --git a/scripts/test-queen.py b/scripts/test-queen.py
deleted file mode 100644
index 95bf473..0000000
--- a/scripts/test-queen.py
+++ /dev/null
@@ -1,82 +0,0 @@
-"""Queen aggregation (agent_swarm#8/#12, M2).
-
-Verifies the Queen's best-of-N selection: among the fan-out agents' candidates, the one whose impl
-passes the most tests wins; selection is deterministic; honest fail-closed (unscored != zero).
-"""
-import os
-import sys
-
-sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
-os.environ.setdefault("REDIS_FAKE", "1")
-os.environ.setdefault("ALLOW_MEMORY_STORE", "1")
-os.environ.setdefault("OPENAI_API_KEY", "test-key")
-
-import asyncio
-
-from orchestrator.queen import Candidate, select_best, should_bounce, _auth_url, promote_to_main
-
-_failures = []
-
-
-def check(name, cond):
-    print(("PASS" if cond else "FAIL") + " - " + name)
-    if not cond:
-        _failures.append(name)
-
-
-F = ["impl.py"]  # non-empty impl marker
-
-# highest pass_rate wins (best-of-N)
-check("highest score wins",
-      select_best([Candidate("t1", "a1", F, score=40.0), Candidate("t2", "a2", F, score=90.0)]).task_id == "t2")
-# scored ranks above unscored
-check("scored beats unscored",
-      select_best([Candidate("t1", "a1", F), Candidate("t2", "a2", F, score=10.0)]).task_id == "t2")
-# no impl anywhere → None (nothing to deliver)
-check("no impl → None", select_best([Candidate("t1", "a1", [])]) is None)
-# all unscored → deterministic fallback to first with impl
-check("all unscored → first with impl",
-      select_best([Candidate("t1", "a1", F), Candidate("t2", "a2", F)]).task_id == "t1")
-# tie on score → stable (first input order wins)
-check("score tie → stable first",
-      select_best([Candidate("t1", "a1", F, score=100.0), Candidate("t2", "a2", F, score=100.0)]).task_id == "t1")
-# tie on score, more passed wins
-check("score tie → more passed wins",
-      select_best([Candidate("t1", "a1", F, score=100.0, passed=2, total=2),
-                   Candidate("t2", "a2", F, score=100.0, passed=5, total=5)]).task_id == "t2")
-
-# --- should_bounce (M3/SC-9 quality gate decision) ---
-def _summary(score):
-    return {"winner": {"task_id": "t1", "score": score}, "candidates": [{"task_id": "t1"}]}
-
-# below threshold + under cap → bounce
-check("below bar under cap → bounce", should_bounce(_summary(40.0), 80.0, cycles=0, max_cycles=2) is True)
-# meets bar → accept
-check("meets bar → no bounce", should_bounce(_summary(90.0), 80.0, cycles=0, max_cycles=2) is False)
-# cap reached → accept best-so-far
-check("cap reached → no bounce", should_bounce(_summary(40.0), 80.0, cycles=2, max_cycles=2) is False)
-# no threshold (disabled) → never bounce
-check("no threshold → no bounce", should_bounce(_summary(0.0), None, cycles=0, max_cycles=2) is False)
-# unscored → honest, don't bounce
-check("unscored → no bounce", should_bounce(_summary(None), 80.0, cycles=0, max_cycles=2) is False)
-
-# --- SC-7 promote_to_main ---
-# _auth_url embeds + URL-encodes creds; passthrough when missing
-check("auth url embeds + encodes creds", _auth_url("http://h/r.git", "u", "p@ss") == "http://u:p%40ss@h/r.git")
-check("auth url passthrough w/o creds", _auth_url("http://h/r.git", None, None) == "http://h/r.git")
-
-
-class _RunNoGrant:
-    def __init__(self):
-        self.metadata = {}
-        self.swarm_id = "s1"
-
-
-# promote is a no-op (not an error) when the run has no git grant
-_promo = asyncio.run(promote_to_main(_RunNoGrant(), [], "t1"))
-check("promote without grant → not promoted", _promo.get("promoted") is False and _promo.get("reason") == "no_git_grant")
-
-if _failures:
-    print(f"\nFAILED: {len(_failures)} check(s): {_failures}")
-    raise SystemExit(1)
-print("\nALL PASSED")
diff --git a/scripts/test-result-aggregator.py b/scripts/test-result-aggregator.py
new file mode 100644
index 0000000..286ae5d
--- /dev/null
+++ b/scripts/test-result-aggregator.py
@@ -0,0 +1,105 @@
+"""ResultAggregator(质量驱动协作聚合)测试 — 取代旧 test-queen(best-of-N 选优)。
+
+覆盖文章模型 + ①②③闭环:收集共享池互补产出、AST 函数级整合(无 LLM 兜底不靠"最长版本")、
+质量门决策(达标 accept / 不达标 bounce 打回 / 无法评分诚实不打回)、终态聚合 best-effort。
+"""
+import asyncio
+import json
+import os
+import sys
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+os.environ.setdefault("REDIS_FAKE", "1")
+os.environ.setdefault("ALLOW_MEMORY_STORE", "1")
+for k in ("OPENAI_API_KEY", "MODEL_API_KEY"):  # 无 LLM,走 AST 兜底(确定性,不依赖外部模型)
+    os.environ.pop(k, None)
+
+from orchestrator.result_aggregator import (
+    _strip_fence, _ast_merge_python, _merge_fallback, _decide,
+    _collect_per_path, merge_artifacts, finalize_run, _emit_patch_mode,
+)
+
+_failures = []
+
+
+def check(name, cond):
+    print(("PASS" if cond else "FAIL") + " - " + name)
+    if not cond:
+        _failures.append(name)
+
+
+class FakeTask:
+    def __init__(self, task_id, files):
+        self.task_id = task_id
+        self.assigned_agent_id = "agent-" + task_id
+        self.agent_role = "impl"
+        self.result = json.dumps({"files": files})
+
+
+class FakeRun:
+    def __init__(self, metadata=None):
+        self.metadata = metadata or {}
+        self.swarm_id = "swarm-test"
+
+
+# --- _strip_fence ---
+check("strip fence python", _strip_fence("```python\nx = 1\n```") == "x = 1")
+check("strip fence passthrough", _strip_fence("z = 3") == "z = 3")
+
+# --- ① AST 函数级合并:各 agent 写不同函数 → 合并成含全部函数的一份(不靠最长版本) ---
+v1 = "def is_palindrome(s):\n    return s == s[::-1]\n"
+v2 = "def count_vowels(s):\n    return sum(c in 'aeiou' for c in s)\ndef reverse_words(s):\n    return ' '.join(s.split()[::-1])\n"
+merged_py = _ast_merge_python([v1, v2])
+check("AST 合并含 is_palindrome", merged_py and "def is_palindrome" in merged_py)
+check("AST 合并含 count_vowels", merged_py and "def count_vowels" in merged_py)
+check("AST 合并含 reverse_words", merged_py and "def reverse_words" in merged_py)
+check("AST 合并去重(同名一次)", merged_py and merged_py.count("def is_palindrome") == 1 if merged_py else False)
+# 语法垃圾 → None(回退最长)
+check("AST 解析全失败 → None", _ast_merge_python(["@#$%", "!!!"]) is None)
+check("merge_fallback 非py取最长", _merge_fallback("a.txt", ["short", "longer text"]) == "longer text")
+
+# --- _decide 决策门 ---
+check("decide: 无阈值 → accept", _decide({"pass_rate": 10.0}, None, 0, 2) == "accept")
+check("decide: 无法评分(None) → accept(诚实)", _decide({"pass_rate": None}, 80.0, 0, 2) == "accept")
+check("decide: 达标 → accept", _decide({"pass_rate": 90.0}, 80.0, 0, 2) == "accept")
+check("decide: 不达标且未到上限 → bounce", _decide({"pass_rate": 40.0}, 80.0, 0, 2) == "bounce")
+check("decide: 不达标但达上限 → accept", _decide({"pass_rate": 40.0}, 80.0, 2, 2) == "accept")
+
+# --- 收集共享池:同文件多版本都保留 ---
+t1 = FakeTask("t1", [{"path": "stringutils.py", "content": v1, "action": "write"}])
+t2 = FakeTask("t2", [{"path": "stringutils.py", "content": v2, "action": "write"}])
+t3 = FakeTask("t3", [{"path": "test_stringutils.py", "content": "def test_x():\n    assert True\n", "action": "write"}])
+by_path = _collect_per_path([t1, t2, t3])
+check("collect: 同文件两版本都保留", len(by_path.get("stringutils.py", [])) == 2)
+
+# --- merge_artifacts:无 LLM 用 AST 整合三函数(不丢互补) ---
+merged = asyncio.run(merge_artifacts([t1, t2, t3], "实现 stringutils"))
+su = next(f for f in merged if f.path == "stringutils.py")
+check("merge: 整合后含全部三函数", all(fn in su.content for fn in ("is_palindrome", "count_vowels", "reverse_words")))
+
+# --- finalize_run:无阈值 → accept;无 git grant → 诚实不 promote ---
+res = asyncio.run(finalize_run(FakeRun(), [t1, t2, t3], "实现 stringutils"))
+check("finalize: decision=accept(无阈值)", res.get("decision") == "accept")
+check("finalize: finalized=True", res.get("finalized") is True)
+check("finalize: 无 git grant promoted=False", (res.get("promotion") or {}).get("reason") == "no_git_grant")
+
+# --- finalize_run:无产物 → finalized=False ---
+empty = asyncio.run(finalize_run(FakeRun(), [], "空"))
+check("finalize: 无产物 finalized=False", empty.get("finalized") is False and empty.get("reason") == "no_artifacts")
+
+# --- emit_patch 模式(SWE-bench):门控 + 无 grant 诚实失败 ---
+check("emit_patch_mode: 默认关", _emit_patch_mode(FakeRun()) is False)
+check("emit_patch_mode: per-run flag 开", _emit_patch_mode(FakeRun({"emit_patch": True})) is True)
+os.environ["SWARM_EMIT_PATCH"] = "1"
+check("emit_patch_mode: env 开", _emit_patch_mode(FakeRun()) is True)
+patch_run = asyncio.run(finalize_run(FakeRun(), [t1, t2, t3], "实现 stringutils"))
+check("emit_patch: mode=emit_patch", patch_run.get("mode") == "emit_patch")
+check("emit_patch: 无 git grant patch=None", patch_run.get("patch") is None)
+check("emit_patch: 无 grant reason=no_git_grant",
+      (patch_run.get("patch_meta") or {}).get("reason") == "no_git_grant")
+os.environ.pop("SWARM_EMIT_PATCH", None)
+
+if _failures:
+    print(f"\nFAILED: {len(_failures)} check(s): {_failures}")
+    raise SystemExit(1)
+print("\nALL PASSED")