orchestrator/main.py

@@ -37,6 +37,7 @@ from . import autonomous_tasks as autonomous_mod
 from . import task_competition as competition_mod
 from . import cross_review as cross_review_mod
 from . import guard as guard_mod
+from . import queen as queen_mod
 from . import agent_launcher
 
 # Configure logging
@@ -721,6 +722,18 @@ 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.
+        try:
+            queen_summary = await queen_mod.aggregate_run(run, tasks)
+            run.metadata["queen"] = queen_summary
+            if isinstance(deliverable, dict) and queen_summary.get("winner"):
+                deliverable["selected"] = queen_summary["winner"]
+                deliverable["candidate_count"] = queen_summary.get("candidate_count")
+        except Exception as exc:
+            logger.warning("queen aggregation failed for run %s: %s", run.swarm_id, exc)
         await swarm_runtime.save_run(run)
 
         # Benchmark Group B: grade the run's generated code against its held-out fixture tests in

orchestrator/queen.py

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

scripts/test-queen.py

@@ -0,0 +1,49 @@
+"""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")
+
+from orchestrator.queen import Candidate, select_best
+
+_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")
+
+if _failures:
+    print(f"\nFAILED: {len(_failures)} check(s): {_failures}")
+    raise SystemExit(1)
+print("\nALL PASSED")