From c257e7e5627130953e33ea58bb15c589a34e3f5c Mon Sep 17 00:00:00 2001 From: gongzhiyong Date: Fri, 19 Jun 2026 18:42:39 +0800 Subject: [PATCH 1/6] =?UTF-8?q?feat(orchestrator):=20P0=20run=20=E8=BE=B9?= =?UTF-8?q?=E7=95=8C=E9=9A=94=E7=A6=BB=20=E2=80=94=20=E8=9C=82=E5=90=8E?= =?UTF-8?q?=E9=98=B2=E8=B7=A8=20run=20=E6=8A=A2=E5=A4=BA(agent=5Fswarm#8)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit run-boundary isolation:agent 只能竞争/认领属于自己 run 的 task,杜绝跨 run 抢夺 (A run 的 agent 抢 B run 的 task → 计费错账/结果污染)。 - 新增 extract_swarm_from_agent / _agent_belongs_to_run(复用 -agent- 前缀) - swarm_dispatch:候选过滤为本 run 的 task - handle_task_bid/yield/takeover:拒绝跨 run 请求(cross_run_denied) - 测试 scripts/test-run-isolation.py(7 检查全过) 影响:仅 orchestrator 派发/竞争路径;不涉及 Manager 契约/计费字段/发布链路。 Co-Authored-By: Claude Opus 4.8 (1M context) --- orchestrator/main.py | 43 +++++++++++++++++++++++++++++++ scripts/test-run-isolation.py | 48 +++++++++++++++++++++++++++++++++++ 2 files changed, 91 insertions(+) create mode 100644 scripts/test-run-isolation.py diff --git a/orchestrator/main.py b/orchestrator/main.py index 882a293..ec303b7 100644 --- a/orchestrator/main.py +++ b/orchestrator/main.py @@ -188,6 +188,23 @@ def max_agents_per_user(body: Optional[Dict[str, Any]] = None) -> int: return _env_max_agents_per_user() +def extract_swarm_from_agent(agent_id: str) -> Optional[str]: + """Launcher-minted agent ids are `{swarm_id}-agent-{N}`; return the `swarm_id` prefix, or None + for externally-supplied / malformed ids. Used for run-boundary isolation (no cross-run task + grab, agent_swarm#8) and the per-user cap lookup.""" + marker = "-agent-" + idx = (agent_id or "").rfind(marker) + return agent_id[:idx] if idx > 0 else None + + +def _agent_belongs_to_run(agent_id: str, run) -> bool: + """True if `agent_id` is a launcher-minted agent of `run`, or its swarm prefix can't be + resolved (fail-soft for externally-supplied agents). Blocks cross-run competition (agent_swarm#8): + the Queen rejects any bid/yield/takeover from an agent that does not belong to the task's run.""" + swarm = extract_swarm_from_agent(agent_id) + return swarm is None or swarm == run.swarm_id + + async def _per_user_cap_for_agent(agent_id: str) -> int: """Per-user cap to enforce at WS registration for `agent_id`. @@ -577,6 +594,20 @@ async def swarm_dispatch(idle_agents): ready = await task_queue.get_ready_pending_tasks(agent.capabilities) if not ready: continue + # P0 run-boundary isolation (agent_swarm#8): an agent may only self-select tasks that + # belong to ITS OWN run. Without this, an idle agent of run A can pull run B's pending task + # from the shared queue (cross-run grab → billing misattribution / result cross-contamination). + # Fail-soft: externally-supplied agents (no resolvable swarm prefix) keep the global behavior. + agent_swarm = extract_swarm_from_agent(agent.agent_id) + if agent_swarm: + owned = [] + for t in ready: + trun = await swarm_runtime.get_run_for_task(t.task_id) + if trun is not None and trun.swarm_id == agent_swarm: + owned.append(t) + ready = owned + if not ready: + continue candidates = [] for t in ready: tau = await decision_engine.get_tau(t.agent_role, agent.agent_id) @@ -1127,6 +1158,10 @@ async def handle_task_bid(agent_id: str, message: Dict[str, Any]) -> Dict[str, A run = await swarm_runtime.get_run_for_task(task_id) if task_id else None if not run or not task_id: return {"recorded": False, "reason": "no_run"} + if not _agent_belongs_to_run(agent_id, run): + logger.warning("queen: rejected cross-run bid — agent %s ∉ run %s (task %s)", + agent_id, run.swarm_id, task_id) + return {"recorded": False, "reason": "cross_run_denied"} bid = competition_mod.TaskBid( task_id=task_id, agent_id=agent_id, confidence=message.get("confidence", 0.5), @@ -1182,6 +1217,10 @@ async def handle_task_yield(agent_id: str, message: Dict[str, Any]) -> Dict[str, run = await swarm_runtime.get_run_for_task(task_id) if task_id else None if not run or not task_id: return {"released": False, "reason": "no_run"} + if not _agent_belongs_to_run(agent_id, run): + logger.warning("queen: rejected cross-run yield — agent %s ∉ run %s (task %s)", + agent_id, run.swarm_id, task_id) + return {"released": False, "reason": "cross_run_denied"} yield_msg = competition_mod.TaskYield( task_id=task_id, agent_id=agent_id, release_with_reason=message.get("reason", message.get("release_with_reason", "")), @@ -1207,6 +1246,10 @@ async def handle_task_takeover(agent_id: str, message: Dict[str, Any]) -> Dict[s run = await swarm_runtime.get_run_for_task(task_id) if task_id else None if not task or not run: return {"taken_over": False, "reason": "no_task"} + if not _agent_belongs_to_run(agent_id, run): + logger.warning("queen: rejected cross-run takeover — agent %s ∉ run %s (task %s)", + agent_id, run.swarm_id, task_id) + return {"taken_over": False, "reason": "cross_run_denied"} incumbent_id = task.assigned_agent_id requester_bid = competition_mod.TaskBid( task_id=task_id, agent_id=agent_id, diff --git a/scripts/test-run-isolation.py b/scripts/test-run-isolation.py new file mode 100644 index 0000000..a10f71a --- /dev/null +++ b/scripts/test-run-isolation.py @@ -0,0 +1,48 @@ +"""P0 run-boundary isolation (agent_swarm#8 / Queen anti-grab). + +Verifies the Queen's run-ownership guard: a launcher-minted agent is bound to its own run by the +`{swarm_id}-agent-{N}` id prefix, so cross-run task grab (bid/yield/takeover/self-select) is denied. +Externally-supplied agents (no resolvable prefix) keep the global behavior (fail-soft). +""" +import os +import sys + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) +os.environ.setdefault("REDIS_FAKE", "1") +os.environ.setdefault("ALLOW_MEMORY_STORE", "1") +os.environ.setdefault("OPENAI_API_KEY", "test-key") + +from orchestrator.main import extract_swarm_from_agent, _agent_belongs_to_run + +_failures = [] + + +def check(name, cond): + print(("PASS" if cond else "FAIL") + " - " + name) + if not cond: + _failures.append(name) + + +class _Run: + def __init__(self, swarm_id): + self.swarm_id = swarm_id + + +# extract_swarm_from_agent: strip the -agent-N suffix +check("extract swarm prefix", extract_swarm_from_agent("swarm-abc-agent-1") == "swarm-abc") +check("extract multi-digit N", extract_swarm_from_agent("swarm-xyz123-agent-16") == "swarm-xyz123") +check("extract None for external id", extract_swarm_from_agent("external-worker") is None) +check("extract None for empty", extract_swarm_from_agent("") is None) + +run_a = _Run("swarm-abc") +# same run → allowed +check("own-run agent allowed", _agent_belongs_to_run("swarm-abc-agent-3", run_a) is True) +# cross run → DENIED (the core anti-grab guarantee) +check("cross-run agent denied", _agent_belongs_to_run("swarm-xyz-agent-1", run_a) is False) +# external (unresolvable prefix) → fail-soft allow +check("external agent fail-soft allowed", _agent_belongs_to_run("external-worker", run_a) is True) + +if _failures: + print(f"\nFAILED: {len(_failures)} check(s): {_failures}") + raise SystemExit(1) +print("\nALL PASSED") From c2413b5edde77a7d8c18303a9dbc846ec9896765 Mon Sep 17 00:00:00 2001 From: gongzhiyong Date: Fri, 19 Jun 2026 19:00:15 +0800 Subject: [PATCH 2/6] =?UTF-8?q?feat(orchestrator):=20M2=20=E8=9C=82?= =?UTF-8?q?=E5=90=8E=E7=BB=93=E6=9E=9C=E8=81=9A=E5=90=88=20=E2=80=94=20bes?= =?UTF-8?q?t-of-N=20=E9=80=89=E6=9C=80=E4=BC=98(SC-5/6/7)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 新增 orchestrator/queen.py(蜂后:不执行/不分配,只仲裁最终结果): - collect_candidates:各 completed task 的 impl 产物为候选(复用 quality.collect_generated_files) - score_candidates:用 swarm 共享 test 跑各候选 impl 评分(复用 sandbox.run_tests; fail-closed:未确认隔离则不评分,unscored!=0,组织规则#9) - select_best:纯函数,测试通过率最高者胜(best-of-N — 单模型没有的涌现杠杆) - aggregate_run:终态收集→评分→选最优,best-effort 不破坏终态路径 接入 main.py completed 分支:winner 标到 deliverable.selected,verdict 存 run.metadata[queen]。 测试 scripts/test-queen.py(6 检查全过)。 注:SC-7 的 git push 最优到产物仓 main 待端到端阶段(需 orchestrator git CLI + 凭据持久化); 当前先标记 winner(SWE-bench 语境产物是 patch,选最优即够)。 影响:仅 orchestrator 终态聚合;不涉及 Manager 契约/计费/发布链路。 Co-Authored-By: Claude Opus 4.8 (1M context) --- orchestrator/main.py | 13 ++++ orchestrator/queen.py | 138 ++++++++++++++++++++++++++++++++++++++++++ scripts/test-queen.py | 49 +++++++++++++++ 3 files changed, 200 insertions(+) create mode 100644 orchestrator/queen.py create mode 100644 scripts/test-queen.py diff --git a/orchestrator/main.py b/orchestrator/main.py index ec303b7..d1df4dd 100644 --- a/orchestrator/main.py +++ b/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 diff --git a/orchestrator/queen.py b/orchestrator/queen.py new file mode 100644 index 0000000..6e31551 --- /dev/null +++ b/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}"} diff --git a/scripts/test-queen.py b/scripts/test-queen.py new file mode 100644 index 0000000..295ee45 --- /dev/null +++ b/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") From 50bc8b019fe09b666160f9c0aac9d672e47dfa83 Mon Sep 17 00:00:00 2001 From: gongzhiyong Date: Fri, 19 Jun 2026 19:16:42 +0800 Subject: [PATCH 3/6] =?UTF-8?q?feat(orchestrator):=20M3=20=E8=9C=82?= =?UTF-8?q?=E5=90=8E=E8=B4=A8=E9=87=8F=E9=97=A8=E6=89=93=E5=9B=9E=20?= =?UTF-8?q?=E2=80=94=20=E8=B4=A8=E9=87=8F=E9=A9=B1=E5=8A=A8=E8=BF=AD?= =?UTF-8?q?=E4=BB=A3(SC-9)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 把收敛从"任务完成即停"升级为"质量驱动": - queen.should_bounce(纯函数):最优候选已评分且 < 阈值且未触顶 → 打回;无阈值/未评分/达标/触顶 → 接受(诚实,不在算不出的分上打回,组织规则#9) - queen_quality_gate(接入 refresh_swarm_run_status,在 cross_review 之后/状态提交之前): 评分不达标 → reopen_task 回灌 impl 任务 + review_cycles++ → run 保持 RUNNING 迭代; 达标/触顶 → 继续终态,winner 标到 deliverable - _queen_threshold:QUEEN_ACCEPTANCE_THRESHOLD(env / run.metadata),默认 None=禁用, 保持现有完成语义不变(可由 operator 启用) - 717 复用门已聚合的 run.metadata[queen],不重复聚合 测试 scripts/test-queen.py 扩展 should_bounce(5 检查),共 11 检查全过。 注:SC-8 convergence 全 authoritative(BLOCKED→override run.status)动 Manager 终态语义, 需契约评审,列为后续;质量驱动收敛的核心已由 queen_quality_gate 实现。 SC-10 P_rework 计入评审重开属度量层(E),后续。 影响:仅 orchestrator 终态/重开路径;门默认禁用,不改 Manager 契约/完成语义。 Co-Authored-By: Claude Opus 4.8 (1M context) --- orchestrator/main.py | 60 +++++++++++++++++++++++++++++++++++++------ orchestrator/queen.py | 18 +++++++++++++ scripts/test-queen.py | 17 +++++++++++- 3 files changed, 86 insertions(+), 9 deletions(-) diff --git a/orchestrator/main.py b/orchestrator/main.py index d1df4dd..df7e55b 100644 --- a/orchestrator/main.py +++ b/orchestrator/main.py @@ -692,6 +692,12 @@ async def refresh_swarm_run_status(run): if next_status == "completed": if await run_cross_review(run, tasks): return + # Queen quality gate (M3/SC-9): score the candidates and, if the best fails the acceptance + # bar (and the review-cycle cap isn't hit), send work BACK for another round instead of + # declaring success on substandard output. Disabled by default (no threshold) — keeps + # current completion semantics until an operator sets QUEEN_ACCEPTANCE_THRESHOLD. + if await queen_quality_gate(run, tasks): + return if run.status == next_status: return @@ -726,14 +732,12 @@ async def refresh_swarm_run_status(run): # 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) + # 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 {} + if isinstance(deliverable, dict) and queen_summary.get("winner"): + deliverable["selected"] = queen_summary["winner"] + deliverable["candidate_count"] = queen_summary.get("candidate_count") await swarm_runtime.save_run(run) # Benchmark Group B: grade the run's generated code against its held-out fixture tests in @@ -1152,6 +1156,46 @@ async def run_cross_review(run, tasks) -> bool: return True +def _queen_threshold(run) -> Optional[float]: + """Queen acceptance bar (pass_rate 0-100): run.metadata override → QUEEN_ACCEPTANCE_THRESHOLD + env → None (gate disabled). None keeps the current task-completion completion semantics.""" + raw = (run.metadata or {}).get("queen_acceptance_threshold") + if raw is None: + raw = os.getenv("QUEEN_ACCEPTANCE_THRESHOLD") + try: + return float(raw) if raw not in (None, "") else None + except (TypeError, ValueError): + return None + + +async def queen_quality_gate(run, tasks) -> bool: + """M3/SC-9: the Queen scores the fan-out candidates and, if the best fails the acceptance bar + and the review-cycle cap isn't hit, sends work BACK (reopen impl tasks) for another round + instead of declaring success on substandard output. Stores the verdict on run.metadata['queen'] + (reused by the deliverable). Returns True if it reopened (caller keeps the run RUNNING). + Best-effort: never raises, never bounces on a score it couldn't compute (org rule #9).""" + try: + summary = await queen_mod.aggregate_run(run, tasks) + run.metadata["queen"] = summary + cycles = int(run.metadata.get("review_cycles", 0) or 0) + max_cycles = int(os.getenv("MAX_REVIEW_CYCLES", "2") or 2) + if not queen_mod.should_bounce(summary, _queen_threshold(run), cycles, max_cycles): + await swarm_runtime.save_run(run) + return False + reopened = 0 + for c in summary.get("candidates", []): + if await task_queue.reopen_task(c["task_id"]): + reopened += 1 + run.metadata["review_cycles"] = cycles + 1 + await swarm_runtime.save_run(run) + logger.info("queen: quality gate bounced run %s — reopened %d for rework (cycle %d)", + run.swarm_id, reopened, cycles + 1) + return reopened > 0 + except Exception as exc: + logger.warning("queen quality gate failed for run %s: %s", run.swarm_id, exc) + return False + + async def _historical_success_map(agent_role: str, agent_ids) -> Dict[str, float]: """τ (decision_engine pheromone) per agent, normalized to [0,1] for arbitration.""" out: Dict[str, float] = {} diff --git a/orchestrator/queen.py b/orchestrator/queen.py index 6e31551..85aaeaf 100644 --- a/orchestrator/queen.py +++ b/orchestrator/queen.py @@ -48,6 +48,24 @@ def select_best(candidates: List[Candidate]) -> Optional[Candidate]: 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.""" diff --git a/scripts/test-queen.py b/scripts/test-queen.py index 295ee45..6236ec2 100644 --- a/scripts/test-queen.py +++ b/scripts/test-queen.py @@ -11,7 +11,7 @@ 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 +from orchestrator.queen import Candidate, select_best, should_bounce _failures = [] @@ -43,6 +43,21 @@ 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) + if _failures: print(f"\nFAILED: {len(_failures)} check(s): {_failures}") raise SystemExit(1) From 7314662f975373ce4a0f4750f024013bae2b4cfc Mon Sep 17 00:00:00 2001 From: gongzhiyong Date: Fri, 19 Jun 2026 19:25:43 +0800 Subject: [PATCH 4/6] =?UTF-8?q?feat(orchestrator):=20M4=20=E5=A4=B1?= =?UTF-8?q?=E8=B4=A5=E9=9A=94=E7=A6=BB=20+=20P-guard=20=E7=A1=AE=E8=AE=A4(?= =?UTF-8?q?SC-11/12)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit SC-12 失败隔离:单 task 失败不再拖垮整个 run。 - refresh_swarm_run_status:next_status 仅在"有失败且无任何完成产物"时 failed; 有完成工作则走 completed,交蜂后/convergence 判定(失败仍由 termination_reason 反映)。 - 解决那次 LLM 网关 404 致单 task 失败 → 整 run FAILED 的单点问题。 SC-11 P-guard:guard.diagnose 已由 assess_swarm_health 在派发循环接入(探索确认,零改动); 检测无 Agent/无模型/依赖死锁/预算耗尽/种子不可分解。 全套测试通过:runtime-contract / merge-smoke / workflow-e2e / contract-freeze + task-competition / queen / run-isolation。 影响:仅 orchestrator 终态判定;Manager status 映射不变(failed/completed)。 Co-Authored-By: Claude Opus 4.8 (1M context) --- orchestrator/main.py | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/orchestrator/main.py b/orchestrator/main.py index df7e55b..d3a9eb4 100644 --- a/orchestrator/main.py +++ b/orchestrator/main.py @@ -678,7 +678,13 @@ async def refresh_swarm_run_status(run): await swarm_runtime.save_run(run) return - next_status = "failed" if any(task.status == TaskStatus.FAILED for task in tasks) else "completed" + # Failure isolation (SC-12): a single failed task must NOT sink the whole run (e.g. one task + # tripped a transient LLM-gateway 404). Declare the run failed ONLY when there is no completed + # work to deliver; otherwise take the completed path and let the Queen / convergence judge + # acceptability (the failure is still surfaced in convergence termination_reason). + failed_tasks = [t for t in tasks if t.status == TaskStatus.FAILED] + completed_tasks = [t for t in tasks if t.status == TaskStatus.COMPLETED] + next_status = "failed" if (failed_tasks and not completed_tasks) else "completed" # Master review-and-iterate gate: before declaring success, optionally run the critic and # send rejected work back to the specialists. Off unless ENABLE_REVIEW_LOOP is set, so the From 0c3af0425d6cb42e4613916a934255eaa7e80980 Mon Sep 17 00:00:00 2001 From: gongzhiyong Date: Fri, 19 Jun 2026 19:40:41 +0800 Subject: [PATCH 5/6] =?UTF-8?q?docs:=20=E8=9C=82=E5=90=8E=E6=94=B6?= =?UTF-8?q?=E6=95=9B=E9=97=AD=E7=8E=AF(SC-3/4)=20=E2=80=94=20convergence-p?= =?UTF-8?q?rotocol=20=C2=A77=20+=20rework-plan=20P-=E5=90=8E=E7=BB=AD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 记录已实现的蜂后闭环(M2-M4+P0)现状与诚实差距,标注剩余(SC-7 落main/SC-8 authoritative/M5 北极星)。 Co-Authored-By: Claude Opus 4.8 (1M context) --- docs/swarm/convergence-protocol.md | 17 +++++++++++++++++ docs/swarm/decentralized-rework-plan.md | 1 + 2 files changed, 18 insertions(+) diff --git a/docs/swarm/convergence-protocol.md b/docs/swarm/convergence-protocol.md index d945293..85f656e 100644 --- a/docs/swarm/convergence-protocol.md +++ b/docs/swarm/convergence-protocol.md @@ -127,3 +127,20 @@ consensus = 100 × (未被任何冲突牵连的已完成任务数 / 已完成任 - **质量/预算/风险输入有条件**:`quality` 依赖 Group B fixture 评分(绑定 fixture 时);`budget`/`usage` 需 run 提供;`risks` 需上游注入。缺失时相应原因不触发,回退 `tasks_completed`(不伪造)。 - **消解为保守首版**:不做自动 merge / 自动选胜,硬冲突留待重做或人工。 - **收敛事件不进 Manager 流**:六个 `convergence.*`/`conflict.*`/`consensus.*` 事件**构建器已实现但不经 `emit_event` 外发**(未在 Manager `agent_callback.go` 注册;与 `swarm.health` 同策略,避免向订阅全部的回调投递未登记事件)。登记后方可启用 Manager 侧发送。`termination_reason` 以**新增可选字段**附在 `timeline.updated`,对旧消费方向后兼容。 + +## 7. 蜂后收敛闭环(Queen,agent_swarm#8,`orchestrator/queen.py`) + +fan-out 后的「收口」由**蜂后(Queen)**承担——一个**不执行、不分配任务**的终态仲裁层,把「任务完成即停」升级为「质量驱动收敛」。对齐公开 Swarm 范式的「感知→决策→交互→更新」迭代循环 + 终止「任务完成 ∨ 质量阈值 ∨ 预算/轮次」。 + +**职责(已实现)**: +- **best-of-N 选最优(M2 / SC-5·6·7)**:`queen.aggregate_run` 收集各 agent 候选产物 → `score_candidates`(共享 test 跑各 impl,复用 `sandbox.run_tests`)→ `select_best`(测试通过率最高,纯函数)。winner 标到 `deliverable.selected`,verdict 存 `run.metadata["queen"]`。这是单模型没有的涌现杠杆。 +- **质量门打回(M3 / SC-9)**:`queen_quality_gate` 在 `run_cross_review` 之后、状态提交之前——最优分 < `QUEEN_ACCEPTANCE_THRESHOLD` 且 `review_cycles < MAX_REVIEW_CYCLES` → `reopen_task` 回灌迭代;达标/触顶 → 收敛。`should_bounce` 为纯函数。 +- **失败隔离(M4 / SC-12)**:单 task 失败不拖垮整个 run;仅「全失败且无完成产物」才 failed,否则交蜂后/convergence 判定。 +- **防跨 run 抢夺(P0 / #8)**:`extract_swarm_from_agent` / `_agent_belongs_to_run`——agent 只能竞争/认领自己 run 的 task;`swarm_dispatch` 过滤、`handle_task_bid/yield/takeover` 拒绝跨 run(`cross_run_denied`)。 + +**诚实差距**: +- **落 main 待端到端**:SC-7 的「git push 最优产物到产物仓 `main`」需 orchestrator 加 git CLI + 凭据持久化;当前先标记 winner(SWE-bench 语境产物是 patch,选最优即够)。 +- **质量门默认禁用**:需 operator 设 `QUEEN_ACCEPTANCE_THRESHOLD` 才打回;评分依赖沙箱隔离(`HEICODE_SANDBOX_ISOLATED`),未确认隔离则不评分(unscored ≠ 0,不打回,规则 #9)。 +- **convergence 全 authoritative(SC-8)**:质量驱动收敛已由 `queen_quality_gate` 实现;让 `ConvergenceReport.status` 完全覆盖 `run.status`(BLOCKED 等)动 Manager 终态语义,列为后续。 +- **北极星(M5,后续)**:在 SWE-bench Pro 50 上对比单 Opus 4.8 的 resolved 率,客观验证涌现是否超越——前提是上述闭环 + 接入真实代码执行环境。 +- **测试**:`scripts/test-queen.py`(select_best + should_bounce)、`scripts/test-run-isolation.py`(防抢)。 diff --git a/docs/swarm/decentralized-rework-plan.md b/docs/swarm/decentralized-rework-plan.md index df8136b..b691036 100644 --- a/docs/swarm/decentralized-rework-plan.md +++ b/docs/swarm/decentralized-rework-plan.md @@ -57,6 +57,7 @@ - **P6 去中心化自选**:✅ 已建(`main.py: swarm_dispatch` + `ENABLE_SWARM_DISPATCH`):每个空闲 Agent 感知共享池、按 capability+τ+load+budget **自选**最适任务(统一 #9 可解释打分 + #10 信息素 τ),记录可解释 `dispatch.decision_made`。集成测试 `test-swarm-dispatch.py`。cutover 时成为**唯一**派发,删除 greedy/ACO/scored 与各模式开关。 - **P-cutover 切换为唯一路径**:✅ 已完成。`swarm_dispatch` 为唯一派发(删 greedy/ACO/scored 分支 + `scored_matchmake`);`build_seed_task_specs` 为唯一任务创建(删 planner-fallback `build_planner_task_specs`/`planner_fallback_enabled`);删单 critic Master 评审环(`maybe_run_review_cycle`/`review_loop_enabled`),`run_cross_review` 为唯一评审;收敛/提案/竞争/评审原语全部**无条件**(移除全部 `ENABLE_*` 构建开关);`test-workflow-e2e` 改写为 seed→自选→自主分解→执行→收敛全流程(stub agent 改为感知种子后提案分解),`test-merge-smoke` 的 planner/单评审用例改写为 seeder/cross-review;CI 同步。`master_agent.synthesize` 作为汇总工具保留。 - **P-guard 守卫(最后)**:✅ 已完成。`orchestrator/guard.py: diagnose`(纯函数)检测 `NO_AGENTS_CONNECTED`/`NO_CAPABLE_AGENT`/`DEPENDENCY_DEADLOCK`/`BUDGET_EXHAUSTED`/`SEED_UNDECOMPOSED` 并给出可读原因;`main.py: assess_swarm_health` 在派发环检测到「有待办却本 tick 无任何分派」时诊断受影响 run,存 `run.metadata["health"]` 并在不健康时发内部 `swarm.health` 事件(仅诊断,不改 run)。测试 `test-swarm-guard.py`。**这是唯一保留的非正常路径处理。** +- **P-后续 蜂后收敛闭环(agent_swarm#8,已实现 M2–M4)**:fan-out 后的「收口」由 `orchestrator/queen.py` **蜂后**(不执行/不分配的终态仲裁层)承担——best-of-N 选最优(M2)、质量门打回迭代(M3,`QUEEN_ACCEPTANCE_THRESHOLD` 默认禁用)、失败隔离(M4,单 task 失败不拖垮整个 run)、防跨 run 抢夺(P0,`extract_swarm_from_agent`)。设计与诚实差距见 `convergence-protocol.md §7`。**剩余**:SC-7 git push 最优产物落 `main`(待端到端)、SC-8 convergence 全 authoritative、**北极星 M5**(SWE-bench Pro 50 对比单 Opus 4.8 的 resolved 率,客观验证涌现是否超越)。 --- From 2672a3d4b6207a07a854c4b7dea57d41a3f0fc9b Mon Sep 17 00:00:00 2001 From: gongzhiyong Date: Fri, 19 Jun 2026 21:38:43 +0800 Subject: [PATCH 6/6] =?UTF-8?q?feat(orchestrator):=20SC-7=20=E8=9C=82?= =?UTF-8?q?=E5=90=8E=E8=90=BD=20main=20=E2=80=94=20git=20push=20=E6=9C=80?= =?UTF-8?q?=E4=BC=98=E4=BA=A7=E7=89=A9(#16)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 让蜂后选最优后把产物合并到产物仓 main,run 交付一份连贯产物而非 N 个碎片分支: - create 存 git grant 引用(repo_url + secret_ref,均非明文)到 run.metadata;凭据不存 - queen.promote_to_main:从 secret_ref 现取凭据 → clone base_branch → 写最优产物 → commit → push; _auth_url 嵌入并 URL-encode 凭据(不入日志);best-effort 不破坏终态 - main.py 终态:winner → promote_to_main → deliverable.promoted_to_main{branch,commit_sha} - Dockerfile.orchestrator 加 git CLI - test-queen 扩展(auth_url 编码 + no_git_grant 分支),14 检查全过;全套契约通过 注:真实 git push e2e 需重建 orchestrator 镜像 + 部署 + gitea 产物仓验证(后续); 当前纯代码 + 单元测完成。凭据经 secret_ref 现取、不存 run、不入日志(组织规则#8)。 影响:仅 orchestrator 终态聚合;不涉及 Manager 契约/计费/发布链路。 Co-Authored-By: Claude Opus 4.8 (1M context) --- Dockerfile.orchestrator | 2 +- orchestrator/main.py | 27 +++++++++++++- orchestrator/queen.py | 81 +++++++++++++++++++++++++++++++++++++++++ scripts/test-queen.py | 20 +++++++++- 4 files changed, 126 insertions(+), 4 deletions(-) diff --git a/Dockerfile.orchestrator b/Dockerfile.orchestrator index c10a509..a57f475 100644 --- a/Dockerfile.orchestrator +++ b/Dockerfile.orchestrator @@ -4,7 +4,7 @@ FROM python:3.11-slim # to create/teardown agent Pods + per-swarm key Secrets. Without it the k8s backend fails (0 agents). # Pinned to the cluster minor (AKS 1.34) per kubectl skew policy. (linux/amd64 — AKS default node arch.) RUN apt-get update \ - && apt-get install -y --no-install-recommends curl ca-certificates \ + && apt-get install -y --no-install-recommends curl ca-certificates git \ && KUBECTL_VERSION="$(curl -fsSL https://dl.k8s.io/release/stable-1.34.txt)" \ && curl -fsSL "https://dl.k8s.io/release/${KUBECTL_VERSION}/bin/linux/amd64/kubectl" -o /usr/local/bin/kubectl \ && chmod +x /usr/local/bin/kubectl \ diff --git a/orchestrator/main.py b/orchestrator/main.py index d3a9eb4..b00cec6 100644 --- a/orchestrator/main.py +++ b/orchestrator/main.py @@ -741,9 +741,20 @@ async def refresh_swarm_run_status(run): # 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 {} - if isinstance(deliverable, dict) and queen_summary.get("winner"): - deliverable["selected"] = queen_summary["winner"] + winner = queen_summary.get("winner") if isinstance(queen_summary, dict) else None + if isinstance(deliverable, dict) and winner: + deliverable["selected"] = winner deliverable["candidate_count"] = queen_summary.get("candidate_count") + # SC-7: promote the winning artifact to the repo's main → one coherent deliverable on + # main, not N scattered agent branches. No-op when the run has no git grant. + try: + promo = await queen_mod.promote_to_main(run, tasks, winner.get("task_id")) + run.metadata["queen_promotion"] = promo + if isinstance(deliverable, dict) and promo.get("promoted"): + deliverable["promoted_to_main"] = { + "branch": promo.get("branch"), "commit_sha": promo.get("commit_sha")} + except Exception as exc: + logger.warning("queen promote_to_main failed for run %s: %s", run.swarm_id, exc) await swarm_runtime.save_run(run) # Benchmark Group B: grade the run's generated code against its held-out fixture tests in @@ -1939,6 +1950,18 @@ async def create_swarm_run_from_request( if created and run.status == "running": await create_tasks_for_run(run, body) await launch_swarm_agents(run, body) + # Persist the git grant REFERENCE (repo_url + secret_ref — both non-secret; the actual + # credential is resolved from secret_ref at promote time and never stored) so the Queen can + # push the winning artifact to the repo's main at run terminal (SC-7). Never store creds. + _git_grant = agent_launcher._first_git_grant(body) + if _git_grant: + _meta = _git_grant.get("metadata") or {} + run.metadata["git_grant"] = { + "repo_url": _meta.get("repo_url") or _git_grant.get("repo_url"), + "secret_ref": _git_grant.get("secret_ref"), + "base_branch": _meta.get("base_branch", "main"), + } + await swarm_runtime.save_run(run) return { "success": True, diff --git a/orchestrator/queen.py b/orchestrator/queen.py index 85aaeaf..7689d62 100644 --- a/orchestrator/queen.py +++ b/orchestrator/queen.py @@ -154,3 +154,84 @@ async def aggregate_run(run, tasks) -> Dict[str, Any]: except Exception as exc: # Queen never breaks the run's terminal path logger.warning("queen: aggregate_run failed for %s: %s", getattr(run, "swarm_id", "?"), exc) return {"winner": None, "candidate_count": 0, "reason": f"error:{exc!r}"} + + +def _auth_url(repo_url: str, user: Optional[str], pw: Optional[str]) -> str: + """Embed credentials into an http(s) clone URL. URL-encodes the password (handles '@','/', etc). + The result is NEVER logged.""" + import urllib.parse + if not user or not pw or "://" not in repo_url: + return repo_url + scheme, rest = repo_url.split("://", 1) + return f"{scheme}://{urllib.parse.quote(user, safe='')}:{urllib.parse.quote(pw, safe='')}@{rest}" + + +def _git_promote(env: Dict[str, str], base_branch: str, files: List[Any], swarm_id: str) -> Dict[str, Any]: + """Sync git: clone base_branch → overwrite with winner files → commit → push base_branch. + Blocking (run via asyncio.to_thread). Credentials live only in the clone URL, never logged.""" + import os + import shutil + import subprocess + import tempfile + repo_url = env["GIT_REPO_URL"] + auth_url = _auth_url(repo_url, env.get("GIT_USERNAME"), env.get("GIT_PASSWORD")) + workdir = tempfile.mkdtemp(prefix="queen-promote-") + repo_dir = os.path.join(workdir, "repo") + + def git(*args, cwd=None): + return subprocess.run(["git", *args], cwd=cwd or workdir, + capture_output=True, text=True, timeout=180) + + try: + if git("clone", "--depth", "1", "--branch", base_branch, auth_url, "repo").returncode != 0: + return {"promoted": False, "reason": "clone_failed"} + for f in files: + dest = os.path.join(repo_dir, f.path) + os.makedirs(os.path.dirname(dest) or repo_dir, exist_ok=True) + with open(dest, "w", encoding="utf-8") as fh: + fh.write(f.content) + git("config", "user.email", "queen@heicode.swarm", cwd=repo_dir) + git("config", "user.name", "HeiCode Queen", cwd=repo_dir) + git("add", "-A", cwd=repo_dir) + c = git("commit", "-m", f"queen: promote best swarm artifact ({swarm_id})", cwd=repo_dir) + if c.returncode != 0: + return {"promoted": False, "reason": "no_changes"} + if git("push", "origin", base_branch, cwd=repo_dir).returncode != 0: + return {"promoted": False, "reason": "push_failed"} + sha = git("rev-parse", "HEAD", cwd=repo_dir).stdout.strip() + return {"promoted": True, "commit_sha": sha, "branch": base_branch} + except Exception as exc: + return {"promoted": False, "reason": f"error:{exc!r}"} + finally: + shutil.rmtree(workdir, ignore_errors=True) + + +async def promote_to_main(run, tasks, winner_task_id: str) -> Dict[str, Any]: + """SC-7: push the winning candidate's files to the artifact repo's base branch (main), so the + run delivers ONE coherent artifact, not N scattered agent branches. Credentials are resolved + from the grant's secret_ref at call time (never stored/logged). Best-effort: never raises.""" + try: + grant = (run.metadata or {}).get("git_grant") or {} + repo_url = grant.get("repo_url") + if not repo_url: + return {"promoted": False, "reason": "no_git_grant"} + wtask = next((t for t in tasks if getattr(t, "task_id", None) == winner_task_id), None) + if wtask is None: + return {"promoted": False, "reason": "winner_not_found"} + from .quality import collect_generated_files + cf = collect_generated_files([wtask]) + files = (cf.get("impl") or []) + (cf.get("agent_tests") or []) + if not files: + return {"promoted": False, "reason": "winner_no_files"} + from .agent_launcher import resolve_git_grant + env = resolve_git_grant({"resource_grants": [{ + "resource_type": "git", "secret_ref": grant.get("secret_ref"), + "metadata": {"repo_url": repo_url, "base_branch": grant.get("base_branch", "main")}, + }]}) + if not env or "GIT_REPO_URL" not in env: + return {"promoted": False, "reason": "grant_unresolved"} + return await asyncio.to_thread(_git_promote, env, grant.get("base_branch", "main"), + files, getattr(run, "swarm_id", "?")) + except Exception as exc: + logger.warning("queen: promote_to_main failed for %s: %s", getattr(run, "swarm_id", "?"), exc) + return {"promoted": False, "reason": f"error:{exc!r}"} diff --git a/scripts/test-queen.py b/scripts/test-queen.py index 6236ec2..95bf473 100644 --- a/scripts/test-queen.py +++ b/scripts/test-queen.py @@ -11,7 +11,9 @@ 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, should_bounce +import asyncio + +from orchestrator.queen import Candidate, select_best, should_bounce, _auth_url, promote_to_main _failures = [] @@ -58,6 +60,22 @@ check("no threshold → no bounce", should_bounce(_summary(0.0), None, cycles=0, # 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)