Merge pull request '蜂后收敛闭环 P0+M2-M4(防抢run/选最优/质量门打回/失败隔离)' (#19) from feat/queen-convergence into main

Reviewed-on: #19

Dockerfile.orchestrator

@@ -4,7 +4,7 @@ FROM python:3.11-slim
 # to create/teardown agent Pods + per-swarm key Secrets. Without it the k8s backend fails (0 agents).
 # Pinned to the cluster minor (AKS 1.34) per kubectl skew policy. (linux/amd64 — AKS default node arch.)
 RUN apt-get update \
-    && apt-get install -y --no-install-recommends curl ca-certificates \
+    && apt-get install -y --no-install-recommends curl ca-certificates git \
     && KUBECTL_VERSION="$(curl -fsSL https://dl.k8s.io/release/stable-1.34.txt)" \
     && curl -fsSL "https://dl.k8s.io/release/${KUBECTL_VERSION}/bin/linux/amd64/kubectl" -o /usr/local/bin/kubectl \
     && chmod +x /usr/local/bin/kubectl \

docs/swarm/convergence-protocol.md

@@ -127,3 +127,20 @@ consensus = 100 × (未被任何冲突牵连的已完成任务数 / 已完成任
 - **质量/预算/风险输入有条件**：`quality` 依赖 Group B fixture 评分（绑定 fixture 时）；`budget`/`usage` 需 run 提供；`risks` 需上游注入。缺失时相应原因不触发，回退 `tasks_completed`（不伪造）。
 - **消解为保守首版**：不做自动 merge / 自动选胜，硬冲突留待重做或人工。
 - **收敛事件不进 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`）
+
+fan-out 后的「收口」由**蜂后（Queen）**承担——一个**不执行、不分配任务**的终态仲裁层，把「任务完成即停」升级为「质量驱动收敛」。对齐公开 Swarm 范式的「感知→决策→交互→更新」迭代循环 + 终止「任务完成 ∨ 质量阈值 ∨ 预算/轮次」。
+
+**职责（已实现）**：
+- **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`）。
+
+**诚实差距**：
+- **落 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`（防抢）。

docs/swarm/decentralized-rework-plan.md

@@ -57,6 +57,7 @@
 - **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 率，客观验证涌现是否超越）。
 
 ---
 

orchestrator/main.py

@@ -37,6 +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 agent_launcher
 
 # Configure logging
@@ -188,6 +189,23 @@ def max_agents_per_user(body: Optional[Dict[str, Any]] = None) -> int:
     return _env_max_agents_per_user()
 
 
+def extract_swarm_from_agent(agent_id: str) -> Optional[str]:
+    """Launcher-minted agent ids are `{swarm_id}-agent-{N}`; return the `swarm_id` prefix, or None
+    for externally-supplied / malformed ids. Used for run-boundary isolation (no cross-run task
+    grab, agent_swarm#8) and the per-user cap lookup."""
+    marker = "-agent-"
+    idx = (agent_id or "").rfind(marker)
+    return agent_id[:idx] if idx > 0 else None
+
+
+def _agent_belongs_to_run(agent_id: str, run) -> bool:
+    """True if `agent_id` is a launcher-minted agent of `run`, or its swarm prefix can't be
+    resolved (fail-soft for externally-supplied agents). Blocks cross-run competition (agent_swarm#8):
+    the Queen rejects any bid/yield/takeover from an agent that does not belong to the task's run."""
+    swarm = extract_swarm_from_agent(agent_id)
+    return swarm is None or swarm == run.swarm_id
+
+
 async def _per_user_cap_for_agent(agent_id: str) -> int:
     """Per-user cap to enforce at WS registration for `agent_id`.
 
@@ -577,6 +595,20 @@ async def swarm_dispatch(idle_agents):
         ready = await task_queue.get_ready_pending_tasks(agent.capabilities)
         if not ready:
             continue
+        # P0 run-boundary isolation (agent_swarm#8): an agent may only self-select tasks that
+        # belong to ITS OWN run. Without this, an idle agent of run A can pull run B's pending task
+        # from the shared queue (cross-run grab → billing misattribution / result cross-contamination).
+        # Fail-soft: externally-supplied agents (no resolvable swarm prefix) keep the global behavior.
+        agent_swarm = extract_swarm_from_agent(agent.agent_id)
+        if agent_swarm:
+            owned = []
+            for t in ready:
+                trun = await swarm_runtime.get_run_for_task(t.task_id)
+                if trun is not None and trun.swarm_id == agent_swarm:
+                    owned.append(t)
+            ready = owned
+            if not ready:
+                continue
         candidates = []
         for t in ready:
             tau = await decision_engine.get_tau(t.agent_role, agent.agent_id)
@@ -646,7 +678,13 @@ async def refresh_swarm_run_status(run):
                 await swarm_runtime.save_run(run)
             return
 
-    next_status = "failed" if any(task.status == TaskStatus.FAILED for task in tasks) else "completed"
+    # Failure isolation (SC-12): a single failed task must NOT sink the whole run (e.g. one task
+    # tripped a transient LLM-gateway 404). Declare the run failed ONLY when there is no completed
+    # work to deliver; otherwise take the completed path and let the Queen / convergence judge
+    # acceptability (the failure is still surfaced in convergence termination_reason).
+    failed_tasks = [t for t in tasks if t.status == TaskStatus.FAILED]
+    completed_tasks = [t for t in tasks if t.status == TaskStatus.COMPLETED]
+    next_status = "failed" if (failed_tasks and not completed_tasks) else "completed"
 
     # Master review-and-iterate gate: before declaring success, optionally run the critic and
     # send rejected work back to the specialists. Off unless ENABLE_REVIEW_LOOP is set, so the
@@ -660,6 +698,12 @@ 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):
+            return
 
     if run.status == next_status:
         return
@@ -690,6 +734,27 @@ 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)
         await swarm_runtime.save_run(run)
 
         # Benchmark Group B: grade the run's generated code against its held-out fixture tests in
@@ -1108,6 +1173,46 @@ 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")
+    if raw is None:
+        raw = os.getenv("QUEEN_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
+        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] = {}
@@ -1127,6 +1232,10 @@ async def handle_task_bid(agent_id: str, message: Dict[str, Any]) -> Dict[str, A
     run = await swarm_runtime.get_run_for_task(task_id) if task_id else None
     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)",
+                       agent_id, run.swarm_id, task_id)
+        return {"recorded": False, "reason": "cross_run_denied"}
     bid = competition_mod.TaskBid(
         task_id=task_id, agent_id=agent_id,
         confidence=message.get("confidence", 0.5),
@@ -1182,6 +1291,10 @@ async def handle_task_yield(agent_id: str, message: Dict[str, Any]) -> Dict[str,
     run = await swarm_runtime.get_run_for_task(task_id) if task_id else None
     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)",
+                       agent_id, run.swarm_id, task_id)
+        return {"released": False, "reason": "cross_run_denied"}
     yield_msg = competition_mod.TaskYield(
         task_id=task_id, agent_id=agent_id,
         release_with_reason=message.get("reason", message.get("release_with_reason", "")),
@@ -1207,6 +1320,10 @@ async def handle_task_takeover(agent_id: str, message: Dict[str, Any]) -> Dict[s
     run = await swarm_runtime.get_run_for_task(task_id) if task_id else None
     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)",
+                       agent_id, run.swarm_id, task_id)
+        return {"taken_over": False, "reason": "cross_run_denied"}
     incumbent_id = task.assigned_agent_id
     requester_bid = competition_mod.TaskBid(
         task_id=task_id, agent_id=agent_id,
@@ -1833,6 +1950,18 @@ async def create_swarm_run_from_request(
     if created and run.status == "running":
         await create_tasks_for_run(run, body)
         await launch_swarm_agents(run, body)
+        # Persist the git grant REFERENCE (repo_url + secret_ref — both non-secret; the actual
+        # credential is resolved from secret_ref at promote time and never stored) so the Queen can
+        # push the winning artifact to the repo's main at run terminal (SC-7). Never store creds.
+        _git_grant = agent_launcher._first_git_grant(body)
+        if _git_grant:
+            _meta = _git_grant.get("metadata") or {}
+            run.metadata["git_grant"] = {
+                "repo_url": _meta.get("repo_url") or _git_grant.get("repo_url"),
+                "secret_ref": _git_grant.get("secret_ref"),
+                "base_branch": _meta.get("base_branch", "main"),
+            }
+            await swarm_runtime.save_run(run)
 
     return {
         "success": True,

orchestrator/queen.py

@@ -0,0 +1,237 @@
+"""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}"}

scripts/test-queen.py

@@ -0,0 +1,82 @@
+"""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")

scripts/test-run-isolation.py

@@ -0,0 +1,48 @@
+"""P0 run-boundary isolation (agent_swarm#8 / Queen anti-grab).
+
+Verifies the Queen's run-ownership guard: a launcher-minted agent is bound to its own run by the
+`{swarm_id}-agent-{N}` id prefix, so cross-run task grab (bid/yield/takeover/self-select) is denied.
+Externally-supplied agents (no resolvable prefix) keep the global behavior (fail-soft).
+"""
+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")
+
+from orchestrator.main import extract_swarm_from_agent, _agent_belongs_to_run
+
+_failures = []
+
+
+def check(name, cond):
+    print(("PASS" if cond else "FAIL") + " - " + name)
+    if not cond:
+        _failures.append(name)
+
+
+class _Run:
+    def __init__(self, swarm_id):
+        self.swarm_id = swarm_id
+
+
+# extract_swarm_from_agent: strip the -agent-N suffix
+check("extract swarm prefix", extract_swarm_from_agent("swarm-abc-agent-1") == "swarm-abc")
+check("extract multi-digit N", extract_swarm_from_agent("swarm-xyz123-agent-16") == "swarm-xyz123")
+check("extract None for external id", extract_swarm_from_agent("external-worker") is None)
+check("extract None for empty", extract_swarm_from_agent("") is None)
+
+run_a = _Run("swarm-abc")
+# same run → allowed
+check("own-run agent allowed", _agent_belongs_to_run("swarm-abc-agent-3", run_a) is True)
+# cross run → DENIED (the core anti-grab guarantee)
+check("cross-run agent denied", _agent_belongs_to_run("swarm-xyz-agent-1", run_a) is False)
+# external (unresolvable prefix) → fail-soft allow
+check("external agent fail-soft allowed", _agent_belongs_to_run("external-worker", run_a) is True)
+
+if _failures:
+    print(f"\nFAILED: {len(_failures)} check(s): {_failures}")
+    raise SystemExit(1)
+print("\nALL PASSED")