Merge pull request 'Feat/queen convergence' (#23) from feat/queen-convergence into main

Reviewed-on: #23

agent/git_operations.py

@@ -79,15 +79,33 @@ class GitOperations:
                     self._clear_workspace_dir(workspace)
 
             clone_url = self._authenticated_repo_url(repo_url)
-            result = await self._run_git_command([
-                "git", "clone", clone_url, self.workspace_dir
-            ])
+            base_commit = (os.getenv("GIT_BASE_COMMIT") or "").strip()
 
-            if result.returncode != 0:
-                logger.error(f"Git clone failed: {result.stderr}")
-                return False
+            if base_commit:
+                # SWE-bench/benchmark: work from the EXACT base_commit, not default HEAD. Shallow-fetch
+                # just that commit (fast + small even for huge repos) and check it out.
+                workspace.mkdir(parents=True, exist_ok=True)
+                steps = [
+                    ["git", "init", "-q", self.workspace_dir],
+                    ["git", "-C", self.workspace_dir, "remote", "add", "origin", clone_url],
+                    ["git", "-C", self.workspace_dir, "fetch", "--depth", "1", "origin", base_commit],
+                    ["git", "-C", self.workspace_dir, "checkout", "-q", "FETCH_HEAD"],
+                ]
+                for step in steps:
+                    r = await self._run_git_command(step)
+                    if r.returncode != 0:
+                        logger.error(f"base_commit shallow checkout failed at {step[1:4]}: {r.stderr}")
+                        return False
+                logger.info(f"Checked out base_commit {base_commit[:12]} (shallow) at {self.workspace_dir}")
+            else:
+                result = await self._run_git_command([
+                    "git", "clone", clone_url, self.workspace_dir
+                ])
+                if result.returncode != 0:
+                    logger.error(f"Git clone failed: {result.stderr}")
+                    return False
+                logger.info(f"Successfully cloned repository to {self.workspace_dir}")
 
-            logger.info(f"Successfully cloned repository to {self.workspace_dir}")
             await self._configure_git_user()
             await self._ensure_baseline_commit()
             return True

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]}"
                         )

agent/requirements.txt

@@ -1,5 +1,8 @@
 openai==1.55.3
 websockets==13.1
-pydantic==2.9.2
+pydantic==2.13.4
 python-dotenv==1.0.1
 prometheus-client==0.20.0
+# Standard MCP SDK — agent connects to Jina MCP (search/read_url) over StreamableHTTP and
+# exposes the tools to the model via OpenAI function-calling. Pulls httpx transitively.
+mcp==1.28.0

agent/task_executor.py

@@ -54,6 +54,11 @@ class TaskExecutor:
         self.usage = self._empty_usage()
         self.current_context: dict = {}
 
+        # Jina MCP — loaded once at startup; empty list means no tools available
+        self.jina_api_key = os.getenv("JINA_API_KEY", "")
+        self._jina_tools: list[dict] = []  # OpenAI-format tool schemas
+        self._jina_tools_loaded = False
+
     async def execute_task(
         self,
         task_id: str,
@@ -143,6 +148,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 +188,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.
@@ -237,18 +281,49 @@ class TaskExecutor:
         # (gateway rejects >65536 with HTTP 400 InvalidParameter), so default to the full 65536
         # so large-task breakdown JSON never truncates. It's a cap, not a target.
         max_tokens = int(os.getenv("AGENT_PLAN_MAX_TOKENS", "65536") or 65536)
-        prompt = f"""You are a task planning assistant. Break down the following programming task into concrete, actionable subtasks.
+        prompt = f"""You are decomposing a software change for a DECENTRALIZED SWARM of autonomous coding agents.
+How the swarm executes (this CONSTRAINS what a valid subtask is — read carefully):
+
+- The target repository is ALREADY checked out locally at the correct base commit. Each agent can
+  READ any file in the repo and its ONLY output is the COMPLETE new content of the file(s) it owns.
+- Subtasks go into a shared pool and are picked up by DIFFERENT agents IN PARALLEL. Agents do NOT
+  share a filesystem and there is NO guaranteed order — every subtask must be independently doable
+  straight from the base commit, without seeing any other agent's output.
+- All agents' file outputs are merged BY PATH (one file = one owner) and diffed against the base
+  commit to produce the final patch.
+
+So every subtask MUST be a concrete CODE CHANGE scoped to specific file(s) that yields complete file
+content. A subtask must NOT be a process/meta step.
+
+FORBIDDEN subtasks (an agent literally cannot do these — they will fail and waste the run):
+  exploring/surveying the repo, running or executing tests, moving/renaming/deleting files as an
+  \"action\", reviewing or verifying other agents' work, setting up the environment, installing
+  dependencies, committing/pushing, \"ensure/confirm/check that ...\".
+ALLOWED subtask shape: \"Implement/modify <specific change> in <file(s)>, outputting the complete
+  updated file.\" Writing a NEW source or test FILE is allowed (it produces content); \"run the
+  tests\" is not.
+
+Rules:
+1. Partition by FILE OWNERSHIP: each file that must change is owned by EXACTLY ONE subtask. Never let
+   two subtasks edit the same file (parallel agents would clobber each other).
+2. One subtask may own several files only if they must change together; keep ownership disjoint.
+3. Make the SMALLEST change that satisfies the task. If it fits in one file, return EXACTLY ONE
+   subtask — do not invent extra steps to look thorough.
+4. Ground every subtask in the ACTUAL files present at the base commit, not assumptions.
+5. dependencies = real content dependencies only (e.g. a test file depends on the intended interface
+   of an impl file); each subtask must still be writable independently from the base commit.
 
 Task: {description}
 
 Context: {json.dumps(context, indent=2)}
 
-Return a JSON array of subtasks, where each subtask has:
-- description: Clear description of what needs to be done
-- complexity: \"low\", \"medium\", or \"high\"
-- estimated_time: Estimated time in minutes
-- dependencies: List of subtask indices this depends on (empty if none)
-- required_capabilities: List of capabilities needed
+Return ONLY a JSON array. Each element:
+- description: imperative and file-scoped — WHAT to change, in WHICH file(s), and that it must output
+  the complete final file content.
+- target_files: array of repo-relative paths this subtask OWNS (disjoint across subtasks).
+- complexity: \"low\" | \"medium\" | \"high\"
+- dependencies: array of subtask indices (content dependencies only; empty if none)
+- required_capabilities: e.g. [\"python\"], [\"python\",\"testing\"]
 
 Return ONLY the JSON array, no other text."""
         last_err = None
@@ -298,6 +373,8 @@ Return ONLY the JSON array, no other text."""
         content = ""
         try:
             description = subtask["description"]
+            logger.info("════════ [任务开始] task=%s role=%s\n  描述: %s",
+                        task_id, context.get("specialist_role", "general"), description)
             workspace_files = self._summarize_workspace()
             workspace_context = self._collect_workspace_context()
             user_prompt = context.get("user_prompt") or context.get("run_goal") or context.get("root_task_description") or ""
@@ -373,6 +450,16 @@ Relevant workspace file contents:
 Peer specialist input:
 {json.dumps(peer_context, indent=2)}
 
+Workspace & tool boundaries (IMPORTANT):
+- The target repository is ALREADY checked out locally in your workspace at the correct base commit.
+  The file contents above ARE that local code — treat them as the single source of truth. Read and
+  edit these local files; return complete modified file content in `files`.
+- The web search / read_url tools are for EXTERNAL knowledge ONLY (library/framework docs, language
+  features, error-message lookups). NEVER use them to fetch THIS repository's own source — the web
+  copy is a DIFFERENT version than your local base commit and will corrupt your patch.
+- Do not pass local paths or file:// URLs to read_url; it only fetches public web URLs. To see a
+  repo file not shown above, infer from the listed files rather than fetching it from the web.
+
 Alignment requirements:
 - If your role is testing, align your tests with the implementation artifacts and their stated error semantics.
 - If your role is documentation, align your docs with both implementation and testing artifacts.
@@ -393,6 +480,10 @@ Return your response as JSON with this structure:
 Return ONLY the JSON, no other text."""
             content = await self._complete(prompt, max_tokens=4000)
             result = self._parse_json_response(content)
+            logger.info("──────── [任务产出] task=%s status=%s 文件=%s\n  说明: %s",
+                        task_id, result.get("status"),
+                        [f.get("path") for f in (result.get("files") or [])],
+                        (result.get("changes") or result.get("summary") or "")[:600])
             if result.get("status") == "completed":
                 apply_result = await self._apply_file_changes(result.get("files", []))
                 result["files_modified"] = apply_result["files_modified"]
@@ -427,12 +518,124 @@ Return ONLY the JSON, no other text."""
                 "summary": f"Failed to execute: {e}",
             }
 
+    # Jina MCP endpoint (StreamableHTTP). Read-only web tools (search_web/read_url/…).
+    _JINA_MCP_URL = "https://mcp.jina.ai/v1"
+
+    def _jina_mcp_headers(self) -> dict:
+        return {"Authorization": f"Bearer {self.jina_api_key}"}
+
+    async def _load_jina_tools(self) -> list[dict]:
+        """Fetch tool schemas from Jina MCP via the standard `mcp` SDK (StreamableHTTP transport).
+        Cached after first call. Returns OpenAI function-calling tool specs. The SDK handles the
+        MCP handshake, SSE framing, and session — no hand-rolled JSON-RPC/SSE parsing."""
+        if self._jina_tools_loaded:
+            return self._jina_tools
+        self._jina_tools_loaded = True
+        if not self.jina_api_key:
+            return []
+        try:
+            from mcp import ClientSession
+            from mcp.client.streamable_http import streamablehttp_client
+
+            async with streamablehttp_client(
+                self._JINA_MCP_URL, headers=self._jina_mcp_headers()
+            ) as (read, write, _):
+                async with ClientSession(read, write) as session:
+                    await session.initialize()
+                    tools = (await session.list_tools()).tools
+                    # Constrain the toolset: the agent only needs web SEARCH + fetch a page's full
+                    # CONTENT. Exposing all ~21 Jina tools bloats the prompt and tempts the model to
+                    # misuse them (e.g. read_url on local file:// paths, or fetching the repo's own
+                    # source from the web = WRONG version vs the local base_commit checkout).
+                    allow = {t.strip() for t in os.getenv(
+                        "JINA_TOOL_ALLOWLIST", "search_web,read_url").split(",") if t.strip()}
+                    self._jina_tools = [
+                        {
+                            "type": "function",
+                            "function": {
+                                "name": t.name,
+                                "description": t.description or "",
+                                "parameters": t.inputSchema or {"type": "object", "properties": {}},
+                            },
+                        }
+                        for t in tools
+                        if not allow or t.name in allow
+                    ]
+            logger.info("Loaded %d Jina MCP tools (allowlist=%s, of %d offered)",
+                        len(self._jina_tools), sorted(allow) or "*", len(tools))
+        except Exception as exc:
+            logger.warning("Failed to load Jina MCP tools: %s", exc)
+            self._jina_tools = []
+        return self._jina_tools
+
+    async def _call_jina_tool(self, tool_name: str, arguments: dict) -> str:
+        """Invoke a single Jina MCP tool via the standard `mcp` SDK and return its text result."""
+        try:
+            from mcp import ClientSession
+            from mcp.client.streamable_http import streamablehttp_client
+
+            async with streamablehttp_client(
+                self._JINA_MCP_URL, headers=self._jina_mcp_headers()
+            ) as (read, write, _):
+                async with ClientSession(read, write) as session:
+                    await session.initialize()
+                    result = await session.call_tool(tool_name, arguments)
+                    parts = [
+                        c.text for c in result.content
+                        if getattr(c, "type", None) == "text"
+                    ]
+                    return "\n".join(parts) or str(result.content)
+        except Exception as exc:
+            return f"[tool error: {exc}]"
+
     async def _complete(self, prompt: str, max_tokens: int) -> str:
+        """Call the LLM with optional Jina MCP tools; handles the tool-call loop."""
         extra_headers = self._model_attribution_headers()
+        tools = await self._load_jina_tools()
+        messages = [{"role": "user", "content": prompt}]
+
+        for round_i in range(8):  # max 8 tool-call rounds
+            kwargs: dict = dict(
+                model=self.model,
+                messages=messages,
+                max_tokens=max_tokens,
+                extra_headers=extra_headers or None,
+            )
+            if tools:
+                kwargs["tools"] = tools
+                kwargs["tool_choice"] = "auto"
+            response = await self.client.chat.completions.create(**kwargs)
+            self._record_openai_usage(response)
+            msg = response.choices[0].message
+
+            # 记录这一轮 LLM 的"思考/回复"(可观测 agent 怎么想的)
+            if msg.content:
+                logger.info("[LLM·第%d轮] 思考/回复:\n%s", round_i + 1, msg.content.strip()[:1200])
+
+            if not msg.tool_calls:
+                return msg.content or ""
+
+            # Execute each tool call and feed results back
+            messages.append(msg.model_dump(exclude_unset=True))
+            for tc in msg.tool_calls:
+                args = json.loads(tc.function.arguments or "{}")
+                # 记录调用了哪个 tool、query 是什么
+                logger.info("[TOOL·调用] %s 参数=%s", tc.function.name,
+                            json.dumps(args, ensure_ascii=False)[:400])
+                result = await self._call_jina_tool(tc.function.name, args)
+                # 记录 tool 返回了什么(搜索结果内容,截断)
+                logger.info("[TOOL·返回] %s (%d字):\n%s", tc.function.name, len(result),
+                            (result or "").strip()[:1000])
+                messages.append({
+                    "role": "tool",
+                    "tool_call_id": tc.id,
+                    "content": result,
+                })
+
+        # Fallback: ask for a final answer without tools
+        messages.append({"role": "user", "content": "Please provide your final answer now."})
         response = await self.client.chat.completions.create(
-            model=self.model,
-            messages=[{"role": "user", "content": prompt}],
-            max_tokens=max_tokens,
+            model=self.model, messages=messages, max_tokens=max_tokens,
             extra_headers=extra_headers or None,
         )
         self._record_openai_usage(response)
@@ -583,6 +786,10 @@ Return ONLY JSON:
         return sorted(files)
 
     def _collect_workspace_context(self, max_files: int = 20, max_bytes_per_file: int = 6000) -> dict[str, str]:
+        # Real-repo edits need the FULL file (the LLM rewrites complete content; truncated input →
+        # truncated/hallucinated rewrite → wrong diff). Raise budgets, env-tunable for big repos.
+        max_files = int(os.getenv("AGENT_CTX_MAX_FILES", str(max_files)))
+        max_bytes_per_file = int(os.getenv("AGENT_CTX_MAX_BYTES", str(max_bytes_per_file)))
         root = Path(self.workspace_dir)
         if not root.exists():
             return {}

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()

benchmark/collectors/run_collector.py

@@ -140,7 +140,11 @@ class SwarmRunMetricsCollector(SwarmMetricsCollector):
         have_quality = isinstance(q_quality, (int, float)) and not math.isnan(q_quality)
         have_speed = isinstance(target_time, (int, float)) and target_time and duration > 0
         if have_quality and have_speed and self.coverage["s_cost"]:
-            rework_count = sum(1 for t in tasks if getattr(t, "retry_count", 0) > 0)
+            # P_rework counts BOTH transient retries AND quality-driven review/queen reopens
+            # (reopen_task doesn't bump retry_count — it's a quality decision, not a failure;
+            # SC-10 fixes the prior under-count). Reopen tally accumulated on run.metadata.
+            rework_count = (sum(1 for t in tasks if getattr(t, "retry_count", 0) > 0)
+                            + int((run.metadata or {}).get("rework_reopens", 0) or 0))
             risky = sum(1 for a in approvals
                         if str(a.get("risk_level", "")).lower() in {"high", "critical"})
             p_risk = (100.0 * risky / len(approvals)) if approvals else 0.0

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`）。此为竞争路径的隔离保证，与聚合主路径并存。

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 率，客观验证涌现是否超越）。
 
 ---
 

k8s/orchestrator-heicode-test.yaml

@@ -0,0 +1,186 @@
+# 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: heicodetest.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: "heicodetest.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 }
+            # Jina MCP key — launcher 透传给 agent pod(SENSITIVE → per-swarm Secret),
+            # agent 用它连 mcp.jina.ai 取 search_web/read_url 工具做 function-calling。
+            - name: JINA_API_KEY
+              valueFrom:
+                secretKeyRef: { name: swarm-jina-key, key: JINA_API_KEY }
+            # --- agent 池窗口（弹性）---
+            # 初始只起 2 个；autoscale loop 按 PENDING 积压按需扩到 MAX/per-user 上限。
+            - name: ENABLE_AGENT_AUTOSCALING
+              value: "1"
+            - name: AGENT_LAUNCH_MIN_POOL
+              value: "2"
+            - name: AGENT_LAUNCH_POOL_SIZE
+              value: "2"
+            - name: AGENT_LAUNCH_MAX_POOL
+              value: "64"
+            - name: MAX_AGENTS_PER_USER
+              value: "64"
+            - name: AGENT_PROPOSAL_BUDGET
+              value: "12"
+            # plan/分解 token 上限：降到 32768 避开模型网关 >32768 → 400(launcher 透传给 agent pod)
+            - name: AGENT_PLAN_MAX_TOKENS
+              value: "32768"
+            # 心跳超时：真实仓 clone/长模型调用会让 agent >30s 不心跳被误杀,放宽到 120s
+            - name: AGENT_HEARTBEAT_TIMEOUT
+              value: "120"
+            # 真实仓上下文预算（launcher 透传给 agent pod）：让 LLM 看到完整文件,避免重写截断
+            - name: AGENT_CTX_MAX_FILES
+              value: "40"
+            - name: AGENT_CTX_MAX_BYTES
+              value: "40000"
+            # 协作聚合质量门：pass_rate<70 打回迭代（对齐本地已验证收敛行为；未设=门关=不打回）
+            - name: AGGREGATE_ACCEPTANCE_THRESHOLD
+              value: "70"
+            # SWE-bench repo clone 凭据（dev secret 解析：grant.secret_ref 末段 swe_git_1 → 读此 env）
+            # JSON {"git_username":...,"git_password":...}，走 Secret，绝不明文。
+            - name: HEICODE_SECRET_swe_git_1
+              valueFrom:
+                secretKeyRef: { name: swarm-git-grant, key: HEICODE_SECRET_swe_git_1 }
+            # --- 隔离沙箱双门控（heicode-test 已确认隔离集群）---
+            - name: ENABLE_QUALITY_EVAL
+              value: "1"
+            - name: HEICODE_SANDBOX_ISOLATED
+              value: "1"
+            # 代码沙箱后端：daytona = 在 Daytona 云沙箱执行模型生成代码(不在本 Pod 跑,隔离更强)；
+            # 出错 fail-soft 回退 Pod 内 subprocess。双门控不变。
+            - name: SANDBOX_BACKEND
+              value: "daytona"
+            - name: DAYTONA_API_URL
+              value: "https://app.daytona.io/api"
+            - name: DAYTONA_API_KEY
+              valueFrom:
+                secretKeyRef: { name: swarm-daytona-key, key: DAYTONA_API_KEY }
+            # --- 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
+# ---------------------------------------------------------------------------

k8s/orchestrator-local.yaml

@@ -87,6 +87,19 @@ spec:
                   name: swarm-model-key
                   key: OPENAI_API_KEY
                   optional: true
+            # Sandbox: default in-pod code-execution quality evaluation, fail-closed double gate
+            # (ENABLE_QUALITY_EVAL + HEICODE_SANDBOX_ISOLATED). Matches the deployed environment.
+            - name: ENABLE_QUALITY_EVAL
+              value: "1"
+            - name: HEICODE_SANDBOX_ISOLATED
+              value: "1"
+            # Jina MCP key — passed through to launched agent pods via agent_launcher
+            - name: JINA_API_KEY
+              valueFrom:
+                secretKeyRef:
+                  name: swarm-jina-key
+                  key: JINA_API_KEY
+                  optional: true
           resources:
             requests: { memory: "256Mi", cpu: "200m" }
             limits: { memory: "512Mi", cpu: "500m" }

orchestrator/agent_launcher.py

@@ -322,6 +322,11 @@ def resolve_git_grant(body: Dict[str, Any]) -> Optional[Dict[str, str]]:
     branch = (meta.get("base_branch") or meta.get("branch") or "").strip()
     if branch:
         env["GIT_BASE_BRANCH"] = branch
+    # SWE-bench / benchmark: pin the exact commit the agent must work from. The agent shallow-fetches
+    # and checks out this commit so reads/edits are against the correct baseline (not default HEAD).
+    base_commit = (meta.get("base_commit") or grant.get("base_commit") or "").strip()
+    if base_commit:
+        env["GIT_BASE_COMMIT"] = base_commit
     secret_ref = (grant.get("secret_ref") or grant.get("ref") or "").strip()
     if secret_ref.startswith("azkv://"):
         creds = _resolve_git_secret_ref(secret_ref)
@@ -460,6 +465,8 @@ def plan_launch_specs(
     orchestrator_url: str,
     user_id: Optional[str] = None,
     git_env: Optional[Dict[str, str]] = None,
+    count_override: Optional[int] = None,
+    id_start: int = 0,
 ) -> List[AgentLaunchSpec]:
     """Pure: plan the agent pool for a run (count capped by the per-user limit) + each agent's env.
 
@@ -467,18 +474,24 @@ def plan_launch_specs(
     the configured pool so the pool is capability-diverse (the seed task has no required caps, so
     any agent can claim it; subtasks proposed later route by capability). `git_env` (from
     git_launch_env) carries GIT_REPO_URL so each agent clones the bound repo into its workspace.
+
+    Elastic top-up: pass ``count_override`` (the delta to add) + ``id_start`` (current agent count)
+    to plan ADDITIONAL agents with non-colliding ids (``agent-{id_start+i+1}``). The caller is
+    responsible for capping the delta against MAX_POOL and the per-user limit (the autoscale loop).
     """
-    count = launch_count(pool_size=pool_size, limit=limit, connected_user_agents=connected_user_agents)
+    count = count_override if count_override is not None else launch_count(
+        pool_size=pool_size, limit=limit, connected_user_agents=connected_user_agents)
     caps = pool_capabilities() or ["general"]
     base = model_api_base()
     mid = model_id(body)
     task_timeout = resolve_task_timeout(body)  # #70:透传执行超时，与 run budget.duration_seconds 取较小
     specs: List[AgentLaunchSpec] = []
     for i in range(count):
-        cap_csv = caps[i % len(caps)]
+        idx = id_start + i
+        cap_csv = caps[idx % len(caps)]
         env = {
             "ORCHESTRATOR_URL": orchestrator_url,
-            "AGENT_ID": f"{run.swarm_id}-agent-{i+1}",
+            "AGENT_ID": f"{run.swarm_id}-agent-{idx+1}",
             "AGENT_CAPABILITIES": cap_csv,
             "OPENAI_API_BASE": base,
             # Non-secret execution-timeout knob; inline like the other config env (never a Secret).
@@ -490,12 +503,27 @@ def plan_launch_specs(
             env["OPENAI_MODEL"] = mid
         if user_id:
             env["HEICODE_USER_ID"] = user_id
+        # Jina MCP key — passed through from orchestrator env so launched agents can call Jina
+        # search/read tools. Secret-classed (SENSITIVE_ENV_KEYS) → delivered via per-swarm Secret.
+        jina_key = os.getenv("JINA_API_KEY")
+        if jina_key:
+            env["JINA_API_KEY"] = jina_key
         if git_env:
             env.update(git_env)
         # LOCAL TEST ONLY (not for upstream): propagate the agent-side subtask-handoff switch so
         # launched agents decompose + hand off subtasks to peers (else each agent does its task alone).
         if os.getenv("ENABLE_SUBTASK_HANDOFF", "false").lower() in {"1", "true", "yes"}:
             env["ENABLE_SUBTASK_HANDOFF"] = "true"
+        # 透传 plan/分解 token 上限到 agent pod：部分模型网关上限 32768，>之 400 → 分解调用失败、
+        # fan-out 断。deployment 设 AGENT_PLAN_MAX_TOKENS 时透传；未设则不传，agent 用自身默认。
+        plan_max_tokens = os.getenv("AGENT_PLAN_MAX_TOKENS")
+        if plan_max_tokens:
+            env["AGENT_PLAN_MAX_TOKENS"] = plan_max_tokens
+        # 透传真实仓上下文预算(读多少文件/每文件多少字节)给 agent pod
+        for _ctx in ("AGENT_CTX_MAX_FILES", "AGENT_CTX_MAX_BYTES"):
+            _v = os.getenv(_ctx)
+            if _v:
+                env[_ctx] = _v
         specs.append(AgentLaunchSpec(agent_id=env["AGENT_ID"], capabilities=cap_csv, env=env))
     return specs
 
@@ -580,7 +608,7 @@ _k8s_swarms: set = set()
 #: Env keys that carry secrets — delivered via the per-swarm k8s Secret (`secretKeyRef`), NEVER
 #: inlined into the PodSpec (would leak into etcd / `kubectl get pod -o yaml`). `GIT_USERNAME` and
 #: `GIT_REPO_URL` are non-secret and stay inline.
-SENSITIVE_ENV_KEYS = ("OPENAI_API_KEY", "GIT_PASSWORD")
+SENSITIVE_ENV_KEYS = ("OPENAI_API_KEY", "GIT_PASSWORD", "JINA_API_KEY")
 
 
 def pod_image() -> str:
@@ -653,6 +681,7 @@ def build_pod_manifest(spec: AgentLaunchSpec, *, namespace: str, swarm_id: str,
         "containers": [{
             "name": "agent",
             "image": image or pod_image(),
+            "imagePullPolicy": "Always",  # always pull the latest :heicode-test tag (else node caches stale)
             "command": ["python", "-m", "agent.main"],
             "env": env_list,
             "resources": resources or pod_resources(),

orchestrator/agent_registry.py

@@ -1,5 +1,6 @@
 """Agent registry with Redis-backed state management."""
 import json
+import os
 import time
 import logging
 from typing import Dict, List, Optional
@@ -30,7 +31,10 @@ class AgentMetadata(BaseModel):
 class AgentRegistry:
     """Manages agent registration and heartbeat tracking."""
 
-    HEARTBEAT_TIMEOUT = 30  # seconds
+    # Seconds without a heartbeat before an agent is marked FAILED. 30s is too tight when an agent
+    # is busy cloning a large real repo or in a long model call (event loop can stall) — that spuriously
+    # failed real SWE tasks. Configurable; default raised to 120. Set AGENT_HEARTBEAT_TIMEOUT to tune.
+    HEARTBEAT_TIMEOUT = int(os.getenv("AGENT_HEARTBEAT_TIMEOUT", "120"))  # seconds
     AGENT_KEY_PREFIX = "agent:"
 
     def __init__(self):

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
@@ -449,6 +449,69 @@ async def finalize_dispatch(agent, task, *, decision=None, dispatch_event=None)
     return True
 
 
+async def _maybe_topup_agents(run, backlog: int) -> None:
+    """Elastic top-up for one run: if its PENDING backlog exceeds idle connected agents and the
+    pool is below the cap, launch the deficit (capped by MAX_POOL and the run's per-user limit).
+    New agents get non-colliding ids (id_start = current pool size). Fail-soft."""
+    swarm_id = run.swarm_id
+    prefix = f"{swarm_id}-agent-"
+    live = sum(1 for aid in manager.active_connections if aid.startswith(prefix))
+    launched = run.metadata.get("launched_agents") or []
+    pool = max(live, len(launched))  # avoid re-launching agents still connecting
+    idle = sum(1 for a in await agent_registry.get_idle_agents()
+               if a.agent_id.startswith(prefix) and a.agent_id in manager.active_connections
+               and agent_has_capacity(a.agent_id))
+    body = run.request_body or {}
+    cap = min(agent_launcher.max_pool_size(), max_agents_per_user(body))
+    if backlog <= idle or pool >= cap:
+        return
+    delta = min(backlog - idle, cap - pool)
+    if delta <= 0:
+        return
+    user_id = ((body.get("metadata") or {}).get("runtime_headers") or {}).get("x_user_id")
+    specs = agent_launcher.plan_launch_specs(
+        run, body, connected_user_agents=0, limit=max_agents_per_user(body),
+        pool_size=delta, model_key=agent_launcher.resolve_model_key(body),
+        orchestrator_url=agent_launcher.orchestrator_ws_url(), user_id=user_id,
+        git_env=agent_launcher.resolve_git_grant(body), count_override=delta, id_start=pool)
+    launched_ids = await agent_launcher.launch(specs, swarm_id=swarm_id)
+    if launched_ids:
+        run.metadata["launched_agents"] = launched + launched_ids
+        await swarm_runtime.save_run(run)
+        logger.info("autoscale: run %s +%d agents (backlog=%d idle=%d pool=%d cap=%d)",
+                    swarm_id, len(launched_ids), backlog, idle, pool, cap)
+
+
+async def agent_autoscale_loop():
+    """Elastic agent pool (demand-driven). Initial launch stays small (AGENT_LAUNCH_MIN_POOL); this
+    loop grows each run's pool only as its PENDING-task backlog demands, up to MAX_POOL / per-user
+    cap — so trivial runs use few agents and heavy runs scale out. Gated by ENABLE_AGENT_AUTOSCALING;
+    no-op when the launch backend is 'none'. Fail-soft (never raises into the event loop)."""
+    if os.getenv("ENABLE_AGENT_AUTOSCALING", "false").lower() not in {"1", "true", "yes"}:
+        return
+    if agent_launcher.launch_backend() == "none":
+        return
+    interval = float(os.getenv("AUTOSCALE_INTERVAL_SECONDS", "5"))
+    while True:
+        try:
+            await asyncio.sleep(interval)
+            if await task_queue.get_pending_count() == 0:
+                continue
+            backlog: Dict[str, list] = {}  # swarm_id -> [run, count]
+            for task in await task_queue.get_all_tasks():
+                if task.status != TaskStatus.PENDING:
+                    continue
+                run = await swarm_runtime.get_run_for_task(task.task_id)
+                if not run or run.status != "running":
+                    continue
+                entry = backlog.setdefault(run.swarm_id, [run, 0])
+                entry[1] += 1
+            for run, n in backlog.values():
+                await _maybe_topup_agents(run, n)
+        except Exception as e:
+            logger.error(f"Error in agent autoscale loop: {e}")
+
+
 async def compute_convergence_report(run, tasks, *, terminal: bool):
     """Build a run-state snapshot and evaluate the convergence report (#12, shadow by default).
 
@@ -698,12 +761,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 +814,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
@@ -846,6 +914,7 @@ async def lifespan(app: FastAPI):
     # Start background tasks
     failure_task = asyncio.create_task(failure_detection_loop())
     dispatch_task = asyncio.create_task(task_dispatch_loop())
+    autoscale_task = asyncio.create_task(agent_autoscale_loop())
 
     yield
 
@@ -853,6 +922,7 @@ async def lifespan(app: FastAPI):
     logger.info("Shutting down orchestrator...")
     failure_task.cancel()
     dispatch_task.cancel()
+    autoscale_task.cancel()
     await redis_client.disconnect()
 
 
@@ -1156,6 +1226,8 @@ async def run_cross_review(run, tasks) -> bool:
         await swarm_runtime.save_run(run)
         return False
     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) + len(reopened)
     run.metadata["review_summary"] = verdict.summary
     run.status = "running"
     await swarm_runtime.save_run(run)
@@ -1173,46 +1245,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
-        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] = {}
@@ -1233,7 +1277,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(
@@ -1292,7 +1336,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(
@@ -1321,7 +1365,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
@@ -1960,6 +2004,8 @@ async def create_swarm_run_from_request(
                 "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"),
+                # benchmark: emit_patch must diff against the exact base_commit, not default HEAD
+                "base_commit": _meta.get("base_commit") or _git_grant.get("base_commit"),
             }
             await swarm_runtime.save_run(run)
 

orchestrator/queen.py

@@ -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}"}

orchestrator/requirements.txt

@@ -3,18 +3,25 @@ uvicorn[standard]==0.32.0
 websockets==13.1
 redis==5.2.0
 openai==1.55.3
-pydantic==2.9.2
+# pydantic 2.11+ required by the Daytona SDK (daytona-api-client); 2.13.4 matches the agent pin.
+pydantic==2.13.4
 python-dotenv==1.0.1
 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
-opentelemetry-api==1.24.0
-opentelemetry-sdk==1.24.0
-opentelemetry-exporter-otlp==1.24.0
-opentelemetry-instrumentation-redis==0.45b0
-opentelemetry-instrumentation-requests==0.45b0
-opentelemetry-instrumentation-logging==0.45b0
+# 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
+# otel bumped 1.24→1.42 to satisfy the Daytona SDK's otel floor (instrumentation-aiohttp-client
+# >=0.59b0); instrumentation pins move 0.45b0→0.63b1 in lockstep. Import + contract/security/
+# aggregator tests verified green under this set.
+opentelemetry-api==1.42.1
+opentelemetry-sdk==1.42.1
+opentelemetry-exporter-otlp==1.42.1
+opentelemetry-instrumentation-redis==0.63b1
+opentelemetry-instrumentation-requests==0.63b1
+opentelemetry-instrumentation-logging==0.63b1
 # benchmark metrics export targets (lazy-imported; only used when BENCHMARK_EXPORT_TARGET set)
 azure-cosmos==4.7.0
 azure-storage-blob==12.23.1
@@ -22,3 +29,7 @@ azure-identity==1.19.0
 # model-key resolution from Key Vault via Pod workload identity (lazy-imported in
 # agent_launcher._resolve_from_keyvault; only when AZURE_FEDERATED_TOKEN_FILE / SECRET_RESOLVER=azkv)
 azure-keyvault-secrets==4.9.0
+# Daytona cloud sandbox backend (orchestrator/sandbox_daytona.py): runs model-generated tests in a
+# dedicated external sandbox (code never executes in this pod). Lazy-imported, only when
+# SANDBOX_BACKEND=daytona. Drives the otel/pydantic bumps above.
+daytona==0.189.0

orchestrator/result_aggregator.py

@@ -0,0 +1,401 @@
+"""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=600)
+
+    # Heuristic: a 40-hex (or short-hex) ref is a commit SHA → shallow-fetch just it (fast/small even
+    # for huge repos). A branch name → shallow clone that branch. Falls back to full clone on failure.
+    def _looks_like_sha(ref: str) -> bool:
+        return bool(ref) and len(ref) >= 7 and all(c in "0123456789abcdef" for c in ref.lower())
+
+    try:
+        if base_ref and _looks_like_sha(base_ref):
+            os.makedirs(repo_dir, exist_ok=True)
+            ok = (git("init", "-q", "repo").returncode == 0
+                  and git("remote", "add", "origin", auth_url, cwd=repo_dir).returncode == 0
+                  and git("fetch", "--depth", "1", "origin", base_ref, cwd=repo_dir).returncode == 0
+                  and git("checkout", "-q", "FETCH_HEAD", cwd=repo_dir).returncode == 0)
+            if not ok:
+                return {"emitted": False, "reason": "shallow_checkout_failed"}
+        else:
+            if git("clone", auth_url, "repo").returncode != 0:
+                return {"emitted": False, "reason": "clone_failed"}
+            if base_ref and 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}

orchestrator/sandbox.py

@@ -22,6 +22,7 @@ SECURITY MODEL — read before changing anything here:
 from __future__ import annotations
 
 import json
+import logging
 import os
 import shutil
 import subprocess
@@ -31,6 +32,8 @@ from dataclasses import dataclass, field
 from pathlib import Path
 from typing import List, Optional
 
+logger = logging.getLogger(__name__)
+
 try:  # POSIX only; absent on Windows dev boxes (prod is a Linux pod)
     import resource  # type: ignore
 except Exception:  # pragma: no cover - platform dependent
@@ -164,6 +167,17 @@ def run_tests(
     is spawned.
     """
     assert_isolated()
+
+    # Backend selection. `daytona` runs the code in a dedicated external Daytona sandbox (code never
+    # touches this pod). Fail-soft: any Daytona/transport error falls back to the in-pod child below
+    # — same fail-closed gate already passed (assert_isolated above), so this is safe, just noisier.
+    if os.getenv("SANDBOX_BACKEND", "local").strip().lower() == "daytona":
+        try:
+            from .sandbox_daytona import run_tests_daytona
+            return run_tests_daytona(source_files, test_files, timeout_seconds=timeout_seconds)
+        except Exception as exc:
+            logger.warning("sandbox: daytona backend failed (%s); falling back to in-pod child", exc)
+
     workdir = Path(tempfile.mkdtemp(prefix="swarm-sbx-"))
     is_posix = os.name == "posix"
     try:

orchestrator/sandbox_daytona.py

@@ -0,0 +1,141 @@
+"""Daytona-backed code sandbox — runs model-generated tests in a Daytona cloud sandbox.
+
+WHY THIS EXISTS (read with orchestrator/sandbox.py security model and
+docs/integration/security-boundary.md §8.1):
+  The default backend executes generated code in a child process **inside this pod**. Daytona
+  runs it in a **dedicated external sandbox** (its own isolated VM/container) — exactly the
+  "专用 sandbox worker/job" evolution path §8.1 names. Code never touches this pod's process
+  space, and we send ONLY the generated files + the test runner (never our env / secrets).
+
+GATING IS UNCHANGED. This module is reached only from sandbox.run_tests(), which already called
+  assert_isolated() (HEICODE_SANDBOX_ISOLATED) under the ENABLE_QUALITY_EVAL feature gate. Picking
+  the Daytona backend does NOT weaken the fail-closed dual gate — it only changes where code runs.
+
+CONTRACT: run_tests_daytona() returns the SAME SandboxResult shape as the local backend, by
+  reusing sandbox_runner.py verbatim inside the Daytona sandbox and reading its _result.json.
+  Raises on any Daytona/transport error so the caller can fail-soft to the local backend.
+"""
+from __future__ import annotations
+
+import json
+import logging
+import os
+from pathlib import Path
+from typing import List
+
+from .sandbox import SandboxFile, SandboxResult, DEFAULT_TIMEOUT_SECONDS
+
+logger = logging.getLogger(__name__)
+
+_RUNNER = Path(__file__).resolve().parent / "sandbox_runner.py"
+_SBX_SUBDIR = "swarm_sbx"
+_RESULT_SENTINEL = "___SBX_RESULT_JSON___"
+_OUTPUT_CAP = 16 * 1024
+
+
+def daytona_backend_selected() -> bool:
+    return os.getenv("SANDBOX_BACKEND", "local").strip().lower() == "daytona"
+
+
+def _safe_rel(path: str) -> str:
+    """Reject absolute paths / `..` escape — same guard as the local backend's _safe_join."""
+    if not path or os.path.isabs(path):
+        raise ValueError(f"unsafe file path: {path!r}")
+    norm = os.path.normpath(path)
+    if norm.startswith("..") or norm.startswith("/"):
+        raise ValueError(f"path escapes sandbox: {path!r}")
+    return norm
+
+
+def _client():
+    """Build a Daytona client from env. Raises if the SDK is absent or no api_key is set."""
+    from daytona import Daytona, DaytonaConfig  # lazy: only when daytona backend is on
+
+    api_key = os.getenv("DAYTONA_API_KEY")
+    if not api_key:
+        raise RuntimeError("DAYTONA_API_KEY is not set")
+    cfg_kwargs = {"api_key": api_key}
+    api_url = os.getenv("DAYTONA_API_URL")
+    if api_url:
+        cfg_kwargs["api_url"] = api_url
+    target = os.getenv("DAYTONA_TARGET")
+    if target:
+        cfg_kwargs["target"] = target
+    return Daytona(DaytonaConfig(**cfg_kwargs))
+
+
+def _parse_result(stdout: str) -> SandboxResult:
+    """Extract the runner's JSON (after the sentinel) and map it to a SandboxResult."""
+    result = SandboxResult()
+    result.stdout = (stdout or "")[:_OUTPUT_CAP]
+    if not stdout or _RESULT_SENTINEL not in stdout:
+        result.error = "no result produced by daytona sandbox runner"
+        return result
+    raw = stdout.rsplit(_RESULT_SENTINEL, 1)[1].strip()
+    try:
+        data = json.loads(raw)
+    except Exception as exc:
+        result.error = f"unparseable daytona result: {exc!r}"
+        return result
+    result.total = int(data.get("total", 0))
+    result.passed = int(data.get("passed", 0))
+    result.failed = int(data.get("failed", 0))
+    result.errored = int(data.get("errored", 0))
+    result.details = data.get("details", []) or []
+    if data.get("fatal"):
+        result.error = str(data["fatal"])
+    return result
+
+
+def run_tests_daytona(
+    source_files: List[SandboxFile],
+    test_files: List[SandboxFile],
+    *,
+    timeout_seconds: float = DEFAULT_TIMEOUT_SECONDS,
+) -> SandboxResult:
+    """Run source + test files in a Daytona cloud sandbox; return a SandboxResult.
+
+    Uploads the generated files + sandbox_runner.py into an ephemeral dir in a freshly-created
+    Daytona sandbox, runs the runner (pytest, stdlib fallback), and reads its _result.json. The
+    sandbox is always deleted. Raises on Daytona/transport failure (caller fails soft to local).
+    """
+    from daytona import FileUpload  # lazy
+
+    runner_src = _RUNNER.read_text(encoding="utf-8")
+    exec_timeout = int(timeout_seconds) + 120  # SDK exec timeout: test wall-clock + setup overhead
+
+    daytona = _client()
+    sandbox = daytona.create()
+    try:
+        root = (sandbox.get_user_root_dir() or "/home/daytona").rstrip("/")
+        workdir = f"{root}/{_SBX_SUBDIR}"
+
+        uploads = [FileUpload(source=runner_src.encode("utf-8"), destination=f"{workdir}/_runner.py")]
+        for f in list(source_files) + list(test_files):
+            rel = _safe_rel(f.path)
+            uploads.append(FileUpload(source=f.content.encode("utf-8"),
+                                      destination=f"{workdir}/{rel}"))
+        sandbox.fs.upload_files(uploads)
+
+        # Best-effort pytest install; runner falls back to stdlib if it's unavailable. Quiet, and
+        # never fatal — a non-zero pip exit still lets the runner produce a result.
+        try:
+            sandbox.process.exec("python -m pip install -q pytest 2>/dev/null || true",
+                                 timeout=exec_timeout)
+        except Exception as exc:  # pragma: no cover - network dependent
+            logger.info("daytona: pytest install best-effort failed (runner will fall back): %s", exc)
+
+        # Run the runner, then print _result.json after a sentinel so we can parse it out of the
+        # combined stdout/stderr stream (we never trust loose stdout for counts).
+        cmd = (f"cd {workdir} && python -I _runner.py >/dev/null 2>&1; "
+               f"echo {_RESULT_SENTINEL}; cat _result.json")
+        resp = sandbox.process.exec(cmd, timeout=exec_timeout)
+        result = _parse_result(getattr(resp, "result", "") or "")
+        if result.exit_code is None:
+            result.exit_code = getattr(resp, "exit_code", None)
+        return result
+    finally:
+        try:
+            sandbox.delete()
+        except Exception as exc:  # pragma: no cover - cleanup best-effort
+            logger.warning("daytona: sandbox delete failed (may leak a sandbox): %s", exc)

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__":

scripts/test-queen.py

@@ -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")

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")