From 4309eac2ce42df17b8cfdbf97a9cccb160e2ffe4 Mon Sep 17 00:00:00 2001 From: gongzhiyong Date: Tue, 16 Jun 2026 00:19:14 +0800 Subject: [PATCH 1/6] =?UTF-8?q?fix(#66):=20=E9=80=80=E5=BD=B9=20=5Fmanager?= =?UTF-8?q?=5Fprovided=5Fagents=20=E9=80=83=E7=94=9F=E9=97=A8=20+=20pod=20?= =?UTF-8?q?=E6=95=B0[3,16]=E4=B8=8A=E4=B8=8B=E9=99=90=20+=20=E6=B6=88?= =?UTF-8?q?=E8=B4=B9=20metadata.max=5Fagents=5Fper=5Fuser?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 去中心化是唯一行为(对齐 runtime-contract §3.3): 所有蜂群统一播种单一目标任务 + Swarm 自己拉 agent 池; orchestration_plan.agents 不再控制拉起/任务创建, 退化为无害元数据。修复"任务建了但无 agent 认领、永久 pending"。 - main.py: 删除 _manager_provided_agents 两处分支(任务创建改无条件播种、拉起永远执行) + 函数退役 - agent_launcher.py: launch_count clamp 到 [AGENT_LAUNCH_MIN_POOL=3, AGENT_LAUNCH_MAX_POOL=16] - main.py: max_agents_per_user(body) 消费 metadata.max_agents_per_user(HM 下发; >0 优先, 否则 env) - main.py: WS 注册兜底按 agent 所属 run 的 metadata cap 反查(fail-soft 回退 env), 与拉起口径一致 - docs/integration/runtime-contract.md §3.3: 同步架构师裁定口径(2026-06-15) - tests: test-agent-launcher / test-max-agents-per-user / test-merge-smoke 同步断言 Refs #66 Co-Authored-By: Claude Opus 4.8 (1M context) --- docs/integration/runtime-contract.md | 4 +- orchestrator/agent_launcher.py | 46 +++++++++++-- orchestrator/main.py | 96 +++++++++++++++++++--------- scripts/test-agent-launcher.py | 58 +++++++++++++++-- scripts/test-max-agents-per-user.py | 46 +++++++++++++ scripts/test-merge-smoke.py | 10 ++- 6 files changed, 216 insertions(+), 44 deletions(-) diff --git a/docs/integration/runtime-contract.md b/docs/integration/runtime-contract.md index ab2d6da..79d887e 100644 --- a/docs/integration/runtime-contract.md +++ b/docs/integration/runtime-contract.md @@ -61,8 +61,10 @@ create 响应 `data`：`deployment_id`、`runtime_deployment_id`、`manager_depl ### 3.3 蜂群专家 agent 拉起环境契约（agent launch env，回应 agent_swarm#16） > **状态（团队决议）**：**Swarm 运行时负责拉起专家 agent 并执行每用户限额**（不再由 AM 拉起）。下表 env 字段 + 拉起方/限额口径**已定**。实现：`orchestrator/agent_launcher.py` + `main.launch_swarm_agents`，测试 `scripts/test-agent-launcher.py`。 +> +> **架构师裁定（2026-06-15，取代此前逃生门口径）**：**所有蜂群一律走去中心化主路——Swarm 永远播种单一目标任务（`build_seed_task_specs`），并按用户拉起 agent 池。** `orchestration_plan.agents`（HM 现会发 `[{role:general}]`）**不控制任务创建，也不控制 agent 拉起**：既不会让运行时跳过播种、改按 breakdown 建任务，也不会让运行时认为「caller 自行拉起 agent」而不拉池。此前「Manager 显式 agent breakdown 由 caller 拉起 / honored as-is」的逃生门（`_manager_provided_agents`）已**彻底退役删除**，本节其余口径（拉起后端、限额、模型 key 注入 §3.3.1）不变。这与 CLAUDE.md「去中心化蜂群是唯一行为（cutover 已完成）」一致。 -**拓扑**：专家 agent 仍是**独立进程**，主动出站连编排器 WS（`/ws/{agent_id}`，见 §5/security-boundary §6）。**拉起方 = Swarm 运行时**：编排器在 create 播种后由 `agent_launcher` 按 provisioning 策略拉起一个能力多样的 agent 池。去中心化下**无「按 run 分解算出 agent 数/角色」这一步**（种子任务无 required caps、任意 agent 可认领，子任务按能力自路由），故 agent 池**按用户**拉起、为固定能力集。§1 的「派发」= 把任务指派给**已连入**的 agent；**拉起 agent 进程**是本节定义的 Swarm 新职责。 +**拓扑**：专家 agent 仍是**独立进程**，主动出站连编排器 WS（`/ws/{agent_id}`，见 §5/security-boundary §6）。**拉起方 = Swarm 运行时**：编排器在 create **无条件播种**单一目标任务后，由 `agent_launcher` 按 provisioning 策略拉起一个能力多样的 agent 池。去中心化下**无「按 run 分解算出 agent 数/角色」这一步**，且**无「按 caller 提供的 agent breakdown 拉起」这一步**（种子任务无 required caps、任意 agent 可认领，子任务按能力自路由），故 agent 池**按用户**拉起、为固定能力集。§1 的「派发」= 把任务指派给**已连入**的 agent；**拉起 agent 进程**是本节定义的 Swarm 新职责。 **拉起后端**（`AGENT_LAUNCH_BACKEND`，fail-soft——拉起失败不影响 create）： - `none`（默认）：不自动拉起，agent 由外部供给（保留 CI/e2e 与「外部拉起」部署）； diff --git a/orchestrator/agent_launcher.py b/orchestrator/agent_launcher.py index 78616bd..e933b7b 100644 --- a/orchestrator/agent_launcher.py +++ b/orchestrator/agent_launcher.py @@ -53,14 +53,35 @@ def launch_backend() -> str: return (os.getenv("AGENT_LAUNCH_BACKEND", "none") or "none").strip().lower() -def desired_pool_size() -> int: - """How many agents to launch per run (before the per-user cap is applied).""" +def min_pool_size() -> int: + """Hard floor on the launched-pod count, env-tunable (AGENT_LAUNCH_MIN_POOL, default 3).""" try: - return max(0, int(os.getenv("AGENT_LAUNCH_POOL_SIZE", "3") or 3)) + return max(0, int(os.getenv("AGENT_LAUNCH_MIN_POOL", "3") or 3)) except ValueError: return 3 +def max_pool_size() -> int: + """Hard ceiling on the launched-pod count, env-tunable (AGENT_LAUNCH_MAX_POOL, default 16).""" + try: + return max(1, int(os.getenv("AGENT_LAUNCH_MAX_POOL", "16") or 16)) + except ValueError: + return 16 + + +def desired_pool_size() -> int: + """How many agents to launch per run (before the per-user cap is applied), clamped to + [AGENT_LAUNCH_MIN_POOL, AGENT_LAUNCH_MAX_POOL] = [3, 16] by default.""" + try: + raw = max(0, int(os.getenv("AGENT_LAUNCH_POOL_SIZE", "3") or 3)) + except ValueError: + raw = 3 + lo, hi = min_pool_size(), max_pool_size() + if lo > hi: # defensive: keep the window sane if mis-set + lo = hi + return max(lo, min(hi, raw)) + + def orchestrator_ws_url() -> str: """WS base the launched agent connects back to (deployment-set).""" return (os.getenv("ORCHESTRATOR_PUBLIC_URL") @@ -219,8 +240,23 @@ class AgentLaunchSpec: def launch_count(*, pool_size: int, limit: int, connected_user_agents: int) -> int: - """How many to launch: pool size, but never pushing the user over the per-user cap.""" - return max(0, min(pool_size, limit - max(0, connected_user_agents))) + """How many agents to launch, with a HARD pod-count window [MIN, MAX] = [3, 16] by default. + + count = max(MIN, min(MAX, min(pool_size, limit - max(0, connected)))) + + `limit` is the per-user cap (env MAX_AGENTS_PER_USER or the run's metadata.max_agents_per_user; + see main.max_agents_per_user). `pool_size` is desired_pool_size() (already clamped to [MIN, MAX]). + + INTENTIONAL boundary: the MIN floor (3) takes priority over the per-user-cap headroom — when a + user already has many agents connected, `limit - connected` can be < MIN, yet we still floor to + MIN. This is the architect's [3,16] hard-floor-first rule. The WS register-time per-user cap + (main.websocket_endpoint) still hard-rejects connections beyond the SAME cap, so any pods launched + above `limit` simply fail to register (fail-closed) rather than over-provisioning the user.""" + lo, hi = min_pool_size(), max_pool_size() + if lo > hi: # defensive: keep the window sane if mis-set + lo = hi + headroom = min(pool_size, limit - max(0, connected_user_agents)) + return max(lo, min(hi, headroom)) def plan_launch_specs( diff --git a/orchestrator/main.py b/orchestrator/main.py index 4caab88..6333685 100644 --- a/orchestrator/main.py +++ b/orchestrator/main.py @@ -145,14 +145,53 @@ manager = ConnectionManager() AGENT_SLOTS: Dict[str, int] = {} -def max_agents_per_user() -> int: - """Max concurrent agents one user may have connected (per-user swarm cap). Env-tunable; default 10.""" +def _env_max_agents_per_user() -> int: try: return max(1, int(os.getenv("MAX_AGENTS_PER_USER", "10") or 10)) except ValueError: return 10 +def max_agents_per_user(body: Optional[Dict[str, Any]] = None) -> int: + """Max concurrent agents one user may have connected (per-user swarm cap / ceiling). + + HM sends `metadata.max_agents_per_user = min(plan SwarmMaxAgents, user override)` on the swarm + create body, and only when > 0. When `body` carries a positive integer there, it WINS (this is + the per-user CEILING for that run); otherwise we fall back to env `MAX_AGENTS_PER_USER` (default + 10). `body` is OPTIONAL: callers without a create body (e.g. the WS register-time cap on an agent + whose owning run can't be resolved) still get the env value — DO NOT make body required. + """ + if isinstance(body, dict): + cap = ((body.get("metadata") or {}).get("max_agents_per_user")) + if isinstance(cap, int) and not isinstance(cap, bool) and cap > 0: + return cap + return _env_max_agents_per_user() + + +async def _per_user_cap_for_agent(agent_id: str) -> int: + """Per-user cap to enforce at WS registration for `agent_id`. + + Keeps the register-time cap consistent with what launch_swarm_agents planned: launcher-minted + agent ids are `{swarm_id}-agent-{N}`, so we strip the `-agent-N` suffix, resolve the owning run, + and reuse ITS metadata.max_agents_per_user ceiling. If the run can't be resolved (externally + supplied agent, malformed id, or a run not yet/no longer persisted) we fall back to the env cap. + Fail-soft: any lookup error falls back to env (never blocks registration on a lookup hiccup). + """ + try: + marker = "-agent-" + idx = agent_id.rfind(marker) + if idx <= 0: + return _env_max_agents_per_user() + swarm_id = agent_id[:idx] + run = await swarm_runtime.get_run_by_identifier(swarm_id) + if run is None: + return _env_max_agents_per_user() + return max_agents_per_user({"metadata": run.metadata or {}}) + except Exception as exc: # never block registration on a cap-lookup error + logger.debug("per-user cap lookup failed for agent %s: %s; using env cap", agent_id, exc) + return _env_max_agents_per_user() + + def record_agent_slots(agent_id: str, message: dict): """Update the cached free-slot count for an agent from a protocol message.""" slots = message.get("available_slots") @@ -807,13 +846,6 @@ def request_context_headers(request: Request) -> Dict[str, Optional[str]]: } -def _manager_provided_agents(body: Dict[str, Any]) -> bool: - """Whether the Manager supplied an explicit agent breakdown (Manager-first; never overridden).""" - normalized = swarm_runtime.normalize_create_request(body) - plan = normalized.get("orchestration_plan") or {} - return bool(plan.get("agents") or normalized.get("agents")) - - def build_seed_task_specs(run, body: Dict[str, Any], base_specs: List[Dict[str, Any]]) -> List[Dict[str, Any]]: """Decentralized rework P2: seed the run with ONE objective-carrying task instead of a Master-decomposed plan. @@ -1177,12 +1209,12 @@ async def create_tasks_for_run(run, body: Dict[str, Any]) -> int: """Create the runtime task graph and emit task.created events.""" created_count = 0 task_specs = swarm_runtime.build_task_descriptions(body) - # Swarm task creation (the ONLY path): when the Manager supplied an explicit agent breakdown we - # honor it (Manager-first contract); otherwise we SEED a single objective-carrying task and the + # Swarm task creation (the ONLY path): ALWAYS seed a single objective-carrying task and let the # agents grow the graph bottom-up via proposals (handle_task_proposal). No up-front Master - # decomposition — this repo is the swarm runtime (see rework plan §0). - if not _manager_provided_agents(body): - task_specs = build_seed_task_specs(run, body, task_specs) + # decomposition and no escape hatch for a caller-supplied agent breakdown — this repo is the + # decentralized swarm runtime and seeding is unconditional. `orchestration_plan.agents` does NOT + # control task creation (see rework plan §0 / runtime-contract §3.3, architect ruling 2026-06-15). + task_specs = build_seed_task_specs(run, body, task_specs) task_id_map = { task_spec["task_id"]: f"{run.swarm_id}-{task_spec['task_id']}" for task_spec in task_specs @@ -1663,8 +1695,10 @@ def _launch_note(backend: str, planned: int, launched: int, model_key_resolved: "agents externally, or set AGENT_LAUNCH_BACKEND=kubernetes (prod) / subprocess (dev). " "With no agent connected the seeded task is never claimed (see health.blockers).") if planned == 0: - return ("0 agents planned — the per-user cap (MAX_AGENTS_PER_USER) is already met by connected " - "agents, or AGENT_LAUNCH_POOL_SIZE is 0.") + return ("0 agents planned — the launched-pod window is [AGENT_LAUNCH_MIN_POOL, " + "AGENT_LAUNCH_MAX_POOL] (default [3,16]); a 0 here means the window was mis-set " + "(MIN=0 with no headroom). Note the per-user cap = the run's " + "metadata.max_agents_per_user ceiling (HM-sent) or env MAX_AGENTS_PER_USER.") if launched == 0: note = (f"backend={backend}: planned {planned} but launched 0 — the launch backend failed " f"(see orchestrator logs; for kubernetes verify kubectl/RBAC + Pod Workload Identity, " @@ -1685,21 +1719,19 @@ async def launch_swarm_agents(run, body: Dict[str, Any]) -> None: capped at MAX_AGENTS_PER_USER, with the model key resolved server-side from billing_context.secret_ref. No-op unless AGENT_LAUNCH_BACKEND is set (default 'none'). Fail-soft. - Manager-provided explicit agent breakdowns are honored as-is (those agents are provisioned by - the caller), so we only auto-launch the pool for the decentralized seed flow. + Every swarm goes through this unconditionally — the runtime always seeds + launches the pool. + `orchestration_plan.agents` does NOT control launch; there is no caller-provisioned escape hatch + (architect ruling 2026-06-15, runtime-contract §3.3). The launched-pod count self-regulates via + plan_launch_specs / launch_count: max(MIN, min(MAX, min(pool_size, cap − already-connected))), + with a HARD window [AGENT_LAUNCH_MIN_POOL, AGENT_LAUNCH_MAX_POOL] = [3, 16] by default (the MIN + floor takes priority over cap headroom). `cap` = the run's metadata.max_agents_per_user ceiling + (HM-sent, min(plan SwarmMaxAgents, user override)) when present, else env MAX_AGENTS_PER_USER. Records the launch outcome on run.metadata["agent_launch"] (backend / planned / launched / model_key_resolved / note) — surfaced in /result + /diagnostics so a run with 0 expert agents explains itself instead of hanging silently (agent_swarm#56). The note carries no secret — only a bool for whether the model key resolved. """ - if _manager_provided_agents(body): - run.metadata["agent_launch"] = { - "backend": "manager_provided", "planned": 0, "launched": 0, "launched_ids": [], - "note": "Manager provided explicit agents; the runtime did not auto-launch a pool.", - } - await swarm_runtime.save_run(run) - return user_id = ((body.get("metadata") or {}).get("runtime_headers") or {}).get("x_user_id") connected = manager.user_agent_count(user_id) if user_id else 0 backend = agent_launcher.launch_backend() @@ -1712,7 +1744,7 @@ async def launch_swarm_agents(run, body: Dict[str, Any]) -> None: specs = agent_launcher.plan_launch_specs( run, body, connected_user_agents=connected, - limit=max_agents_per_user(), + limit=max_agents_per_user(body), pool_size=agent_launcher.desired_pool_size(), model_key=model_key, orchestrator_url=agent_launcher.orchestrator_ws_url(), @@ -2681,11 +2713,15 @@ async def websocket_endpoint(websocket: WebSocket, agent_id: str): capabilities = data.get("capabilities", []) user_id = data.get("user_id") - # Per-user swarm cap: a user may have at most MAX_AGENTS_PER_USER (default 10) agents - # connected at once. Agents that omit user_id are unbound and not capped. A reconnect by - # an already-counted agent_id is allowed (can_bind_user handles it). + # Per-user swarm cap: a user may have at most `limit` agents connected at once. The + # limit MUST match what launch_swarm_agents planned with, or we'd "plan 12 but reject at + # 10". The launcher uses the owning run's metadata.max_agents_per_user ceiling (HM-sent); + # so here we resolve THAT run by agent_id and reuse its cap, falling back to env when the + # run can't be found (externally-supplied agents, or pre-cap legacy ids). + # Agents that omit user_id are unbound and not capped. A reconnect by an already-counted + # agent_id is allowed (can_bind_user handles it). if user_id: - limit = max_agents_per_user() + limit = await _per_user_cap_for_agent(agent_id) if not manager.can_bind_user(agent_id, user_id, limit): logger.warning( "Rejecting agent %s: user %s already at max agents (%d connected)", diff --git a/scripts/test-agent-launcher.py b/scripts/test-agent-launcher.py index 14bfa8d..be4ea72 100644 --- a/scripts/test-agent-launcher.py +++ b/scripts/test-agent-launcher.py @@ -36,18 +36,65 @@ class FakeRun: def test_launch_count(): - check("count = min(pool, limit-connected)", al.launch_count(pool_size=3, limit=10, connected_user_agents=0) == 3) - check("count respects per-user cap", al.launch_count(pool_size=5, limit=10, connected_user_agents=8) == 2) - check("count never negative (already over cap)", al.launch_count(pool_size=3, limit=10, connected_user_agents=10) == 0) - check("count clamps to pool when cap is high", al.launch_count(pool_size=3, limit=100, connected_user_agents=0) == 3) + # Default hard pod-count window is [MIN, MAX] = [3, 16]. count = max(3, min(16, min(pool, cap-connected))). + check("count = max(MIN, min(pool, limit-connected)) — pool 3, no connected -> 3", + al.launch_count(pool_size=3, limit=10, connected_user_agents=0) == 3) + # MIN floor takes PRIORITY over cap headroom: pool 5, cap 10, 8 connected -> headroom 2, floored to MIN 3. + check("MIN floor (3) wins over cap headroom (headroom 2 -> 3)", + al.launch_count(pool_size=5, limit=10, connected_user_agents=8) == 3) + # Even when the user is already AT/over the cap, the MIN floor still applies (intentional [3,16] + # hard-floor-first rule; WS register-time cap fail-closes any pods that exceed the cap). + check("MIN floor still applies when already at cap (headroom 0 -> 3)", + al.launch_count(pool_size=3, limit=10, connected_user_agents=10) == 3) + check("count clamps to pool when cap is high (pool 3 -> 3)", + al.launch_count(pool_size=3, limit=100, connected_user_agents=0) == 3) + # MAX ceiling (16): a big pool with plenty of headroom is capped at 16. + check("MAX ceiling (16) caps a large pool", + al.launch_count(pool_size=50, limit=100, connected_user_agents=0) == 16) + # A high pool but limited cap headroom: pool 50, cap 20, 12 connected -> headroom 8, within [3,16] -> 8. + check("cap headroom (8) honored when within [MIN,MAX]", + al.launch_count(pool_size=50, limit=20, connected_user_agents=12) == 8) + # env-overridable window: MIN=1, MAX=4 -> headroom 2 honored (not floored), big pool capped at 4. + os.environ["AGENT_LAUNCH_MIN_POOL"] = "1" + os.environ["AGENT_LAUNCH_MAX_POOL"] = "4" + check("env MIN=1 -> small headroom not floored", + al.launch_count(pool_size=5, limit=10, connected_user_agents=8) == 2) + check("env MAX=4 -> large pool capped at 4", + al.launch_count(pool_size=50, limit=100, connected_user_agents=0) == 4) + os.environ.pop("AGENT_LAUNCH_MIN_POOL") + os.environ.pop("AGENT_LAUNCH_MAX_POOL") + + +def test_pool_window(): + # desired_pool_size() is clamped to [MIN, MAX] = [3, 16] by default. + for k in ("AGENT_LAUNCH_POOL_SIZE", "AGENT_LAUNCH_MIN_POOL", "AGENT_LAUNCH_MAX_POOL"): + os.environ.pop(k, None) + check("desired_pool_size default = 3 (== MIN)", al.desired_pool_size() == 3) + os.environ["AGENT_LAUNCH_POOL_SIZE"] = "1" + check("desired_pool_size floored to MIN 3 (requested 1)", al.desired_pool_size() == 3) + os.environ["AGENT_LAUNCH_POOL_SIZE"] = "99" + check("desired_pool_size capped to MAX 16 (requested 99)", al.desired_pool_size() == 16) + os.environ["AGENT_LAUNCH_POOL_SIZE"] = "8" + check("desired_pool_size passes through within window (8)", al.desired_pool_size() == 8) + # env-overridable window. + os.environ["AGENT_LAUNCH_MIN_POOL"] = "5" + os.environ["AGENT_LAUNCH_MAX_POOL"] = "6" + os.environ["AGENT_LAUNCH_POOL_SIZE"] = "1" + check("env MIN=5 floors desired_pool_size", al.desired_pool_size() == 5) + os.environ["AGENT_LAUNCH_POOL_SIZE"] = "20" + check("env MAX=6 caps desired_pool_size", al.desired_pool_size() == 6) + check("min_pool_size()/max_pool_size() read env", al.min_pool_size() == 5 and al.max_pool_size() == 6) + for k in ("AGENT_LAUNCH_POOL_SIZE", "AGENT_LAUNCH_MIN_POOL", "AGENT_LAUNCH_MAX_POOL"): + os.environ.pop(k, None) def test_plan_specs(): run = FakeRun() body = {"orchestration_plan": {"objective": "x"}, "billing_context": {"default_model_id": "gpt-x"}} + # 8 connected, cap 10 -> headroom 2, but MIN floor 3 wins (intentional [3,16] hard-floor-first). specs = al.plan_launch_specs(run, body, connected_user_agents=8, limit=10, pool_size=3, model_key="sk-test", orchestrator_url="ws://orch:8000", user_id="u-1") - check("plan caps at limit (8 connected, cap 10 -> launch 2)", len(specs) == 2) + check("plan applies MIN floor over cap headroom (8 connected, cap 10 -> 3 not 2)", len(specs) == 3) s = specs[0] check("spec env has ORCHESTRATOR_URL", s.env.get("ORCHESTRATOR_URL") == "ws://orch:8000") check("spec env has model key (server-side injected)", s.env.get("OPENAI_API_KEY") == "sk-test") @@ -171,6 +218,7 @@ def test_k8s_manifests(): def main(): import asyncio test_launch_count() + test_pool_window() test_plan_specs() test_resolve_model_key() test_azkv_resolver() diff --git a/scripts/test-max-agents-per-user.py b/scripts/test-max-agents-per-user.py index 12de227..eac0393 100644 --- a/scripts/test-max-agents-per-user.py +++ b/scripts/test-max-agents-per-user.py @@ -64,6 +64,50 @@ def test_unit(): check("unit: env limit default respected (=3 here)", orch.max_agents_per_user() == 3) +def test_metadata_cap(): + # HM sends metadata.max_agents_per_user = min(plan SwarmMaxAgents, user override), only when > 0. + # When present as a positive int it WINS over env; otherwise we fall back to env (=3 here). + check("metadata cap (>0 int) wins over env", + orch.max_agents_per_user({"metadata": {"max_agents_per_user": 7}}) == 7) + check("body=None -> env cap (callers without a body unaffected)", + orch.max_agents_per_user() == 3) + check("body without the field -> env cap", + orch.max_agents_per_user({"metadata": {}}) == 3) + check("metadata cap = 0 ignored -> env cap (HM only sends when > 0)", + orch.max_agents_per_user({"metadata": {"max_agents_per_user": 0}}) == 3) + check("metadata cap negative ignored -> env cap", + orch.max_agents_per_user({"metadata": {"max_agents_per_user": -5}}) == 3) + check("metadata cap bool ignored -> env cap (True is not a valid count)", + orch.max_agents_per_user({"metadata": {"max_agents_per_user": True}}) == 3) + check("metadata cap non-int ignored -> env cap", + orch.max_agents_per_user({"metadata": {"max_agents_per_user": "9"}}) == 3) + + +async def test_ws_cap_per_agent(): + """The register-time per-user cap resolves the owning run's metadata cap by agent_id, so it + stays consistent with what launch_swarm_agents planned (no 'plan 12 but reject at 10').""" + await orch.redis_client.connect() # REDIS_FAKE in-memory store (this test runs before the server boots) + # Seed a run whose metadata carries a HM-sent cap of 5; launcher mints `{swarm_id}-agent-N`. + run = await orch.swarm_runtime.get_or_create_run( + { + "mode": "swarm", + "orchestration_plan": {"objective": "x"}, + "callback": {"url": "https://hm.example/cb"}, + "metadata": {"manager_deployment_id": "mdep-1", "max_agents_per_user": 5}, + }, + idempotency_key=None, + correlation_id="corr-ws-cap", + ) + run_obj = run[0] + swarm_id = run_obj.swarm_id + cap = await orch._per_user_cap_for_agent(f"{swarm_id}-agent-2") + check("WS cap for launcher agent uses run metadata cap (=5, not env 3)", cap == 5) + cap_ext = await orch._per_user_cap_for_agent("externally-supplied-agent") + check("WS cap for non-launcher id falls back to env (=3)", cap_ext == 3) + cap_missing = await orch._per_user_cap_for_agent("swarm-doesnotexist-agent-1") + check("WS cap for unknown run falls back to env (=3)", cap_missing == 3) + + async def _register(agent_id, user_id=None): """Open a WS, send register (optionally with user_id), return (ws, response_dict).""" ws = await websockets.connect(f"{WS}/ws/{agent_id}") @@ -142,6 +186,8 @@ async def test_integration(): async def main(): test_unit() + test_metadata_cap() + await test_ws_cap_per_agent() await test_integration() print() if failures: diff --git a/scripts/test-merge-smoke.py b/scripts/test-merge-smoke.py index 8ff5c09..5c78004 100644 --- a/scripts/test-merge-smoke.py +++ b/scripts/test-merge-smoke.py @@ -75,9 +75,13 @@ async def test_seeder(): check("seed preserves base context", specs[0]["context"].get("orchestration_plan") == {"x": 1}) check("seed marks is_seed + objective", specs[0]["context"].get("is_seed") is True and specs[0]["context"].get("objective") == "Build a calculator") - # Manager-provided agents are honored (Manager-first; seeder does not override). - check("Manager-provided agent breakdown bypasses the seeder", - orch._manager_provided_agents({"orchestration_plan": {"agents": [{"role": "impl"}]}}) is True) + # Seeding is UNCONDITIONAL (architect ruling 2026-06-15): a caller-supplied + # orchestration_plan.agents no longer bypasses the seeder — every swarm still gets exactly one + # seed task and the pool is auto-launched (the old _manager_provided_agents escape hatch is gone). + specs_with_agents = orch.build_seed_task_specs( + run, {"orchestration_plan": {"agents": [{"role": "impl"}]}}, base_specs) + check("orchestration_plan.agents does NOT bypass the seeder (one seed task regardless)", + len(specs_with_agents) == 1 and specs_with_agents[0]["source"] == "seed") async def test_agent_peer_routing(): From f9f1d9a6d5403f8ddaf09afa29ef011cd3d59bd6 Mon Sep 17 00:00:00 2001 From: gongzhiyong Date: Tue, 16 Jun 2026 01:41:06 +0800 Subject: [PATCH 2/6] =?UTF-8?q?fix:=20=E6=B3=A8=E5=85=A5=20GIT=5FREPO=5FUR?= =?UTF-8?q?L=EF=BC=8C=E8=AE=A9=20agent=20=E5=85=8B=E9=9A=86=E7=BB=91?= =?UTF-8?q?=E5=AE=9A=E4=BB=93=E5=BA=93=E5=88=B0=20workspace=EF=BC=88?= =?UTF-8?q?=E4=BF=AE=E5=A4=8D=E4=BB=BB=E5=8A=A1=E5=9B=A0=E7=A9=BA=20worksp?= =?UTF-8?q?ace=20=E5=BF=85=E8=B4=A5=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 根因(实测复现): agent 启动未注入 GIT_REPO_URL → 从不 clone → /workspace 为空 → 任何"读/改仓库代码"的任务被 LLM 以 "no source files to analyze or repair" 判 failed。 对照: 简单建文件任务在空 workspace 下仍 success，证明执行管线/模型/解析/落盘均正常。 公开库只需 GIT_REPO_URL(无凭据)。 - agent_launcher.git_launch_env(body): 从 resource_grants[].metadata.repo_url 提取仓库, 注入 GIT_REPO_URL(+GIT_PROVIDER/GIT_DEFAULT_BRANCH/GIT_BASE_BRANCH)。纯函数,无 I/O。 - plan_launch_specs(git_env=...): 合并进每个 agent 的 env。 - launch_swarm_agents: 解析并传入 git_env; 记 agent_launch.repo_bound 诊断字段。 - 私有库凭据(GIT_TOKEN 经 grant secret_ref + 入 k8s Secret)留作干净后续。 - tests: test-agent-launcher 新增 test_git_launch_env + plan_specs 的 git_env 合并断言。 Refs #66 Co-Authored-By: Claude Opus 4.8 (1M context) --- orchestrator/agent_launcher.py | 47 +++++++++++++++++++++++++++++++++- orchestrator/main.py | 5 +++- scripts/test-agent-launcher.py | 34 ++++++++++++++++++++++++ 3 files changed, 84 insertions(+), 2 deletions(-) diff --git a/orchestrator/agent_launcher.py b/orchestrator/agent_launcher.py index e933b7b..a4f896b 100644 --- a/orchestrator/agent_launcher.py +++ b/orchestrator/agent_launcher.py @@ -232,6 +232,47 @@ def model_id(body: Dict[str, Any]) -> Optional[str]: or plan.get("model_id") or body.get("model_id") or os.getenv("OPENAI_MODEL")) +def git_launch_env(body: Dict[str, Any]) -> Dict[str, str]: + """Translate a bound git resource grant into the agent's workspace env so the agent clones the + repo into WORKSPACE_DIR before working (without it /workspace is empty and any "read/repair the + repo" task fails with "no source files to analyze"). Pure: reads resource_grants only, no I/O. + + Sets `GIT_REPO_URL` (+ `GIT_PROVIDER` / `GIT_DEFAULT_BRANCH` when present) from the grant, mirroring + the HM git-binding → env contract (agent-capability-schema §3 / agent_template_env.go). PUBLIC repos + clone with just `GIT_REPO_URL` (no credentials). Private-repo credential injection (`GIT_TOKEN` via + the grant's azkv `secret_ref`, routed through the per-swarm k8s Secret) is intentionally NOT handled + here yet — tracked as a follow-up. Returns {} when no git grant is bound. + """ + grants: List[Dict[str, Any]] = [] + grants += body.get("resource_grants") or [] + plan = body.get("orchestration_plan") or {} + grants += plan.get("resource_grants") or [] + for agent in (plan.get("agents") or []): + grants += agent.get("resource_grants") or [] + for grant in grants: + if not isinstance(grant, dict): + continue + meta = grant.get("metadata") or {} + rtype = str(grant.get("resource_type") or grant.get("type") or "").lower() + repo = str((meta or {}).get("repo_url") or "").strip() + if not repo and rtype == "git": + scope = str(grant.get("binding_scope") or "").strip() + if scope.startswith(("http://", "https://", "git@", "ssh://")): + repo = scope + if not repo: + continue + env = {"GIT_REPO_URL": repo} + provider = str((meta or {}).get("provider") or "").strip() + if provider: + env["GIT_PROVIDER"] = provider + branch = str((meta or {}).get("default_branch") or "").strip() + if branch: + env["GIT_DEFAULT_BRANCH"] = branch + env["GIT_BASE_BRANCH"] = branch # agent/git_operations reads GIT_BASE_BRANCH for the base ref + return env + return {} + + @dataclass class AgentLaunchSpec: agent_id: str @@ -269,12 +310,14 @@ def plan_launch_specs( model_key: Optional[str], orchestrator_url: str, user_id: Optional[str] = None, + git_env: Optional[Dict[str, str]] = None, ) -> List[AgentLaunchSpec]: """Pure: plan the agent pool for a run (count capped by the per-user limit) + each agent's env. The model key is injected into the launch env (not the create body). Capabilities cycle through 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). + 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. """ count = launch_count(pool_size=pool_size, limit=limit, connected_user_agents=connected_user_agents) caps = pool_capabilities() or ["general"] @@ -295,6 +338,8 @@ def plan_launch_specs( env["OPENAI_MODEL"] = mid if user_id: env["HEICODE_USER_ID"] = user_id + if git_env: + env.update(git_env) specs.append(AgentLaunchSpec(agent_id=env["AGENT_ID"], capabilities=cap_csv, env=env)) return specs diff --git a/orchestrator/main.py b/orchestrator/main.py index 6333685..8ab0e86 100644 --- a/orchestrator/main.py +++ b/orchestrator/main.py @@ -1736,9 +1736,11 @@ async def launch_swarm_agents(run, body: Dict[str, Any]) -> None: connected = manager.user_agent_count(user_id) if user_id else 0 backend = agent_launcher.launch_backend() model_key = agent_launcher.resolve_model_key(body) + git_env = agent_launcher.git_launch_env(body) info: Dict[str, Any] = { "backend": backend, "planned": 0, "launched": 0, "launched_ids": [], - "model_key_resolved": bool(model_key), "note": "", + "model_key_resolved": bool(model_key), "repo_bound": bool(git_env.get("GIT_REPO_URL")), + "note": "", } try: specs = agent_launcher.plan_launch_specs( @@ -1749,6 +1751,7 @@ async def launch_swarm_agents(run, body: Dict[str, Any]) -> None: model_key=model_key, orchestrator_url=agent_launcher.orchestrator_ws_url(), user_id=user_id, + git_env=git_env, ) info["planned"] = len(specs) launched = await agent_launcher.launch(specs, swarm_id=run.swarm_id) diff --git a/scripts/test-agent-launcher.py b/scripts/test-agent-launcher.py index be4ea72..5010579 100644 --- a/scripts/test-agent-launcher.py +++ b/scripts/test-agent-launcher.py @@ -107,6 +107,39 @@ def test_plan_specs(): model_key=None, orchestrator_url="ws://orch", user_id=None) check("no model key -> OPENAI_API_KEY omitted", "OPENAI_API_KEY" not in specs2[0].env) check("no user -> HEICODE_USER_ID omitted", "HEICODE_USER_ID" not in specs2[0].env) + # git_env (GIT_REPO_URL) merged into every agent's env so the agent clones the bound repo. + specs3 = al.plan_launch_specs(run, body, connected_user_agents=0, limit=10, pool_size=3, + model_key="sk-test", orchestrator_url="ws://orch", user_id="u-1", + git_env={"GIT_REPO_URL": "https://github.com/Fasthei/swe", "GIT_PROVIDER": "github"}) + check("git_env merged into every spec (GIT_REPO_URL)", + all(sp.env.get("GIT_REPO_URL") == "https://github.com/Fasthei/swe" for sp in specs3)) + check("git_env GIT_PROVIDER propagated", specs3[0].env.get("GIT_PROVIDER") == "github") + check("no git_env -> GIT_REPO_URL omitted", "GIT_REPO_URL" not in specs2[0].env) + + +def test_git_launch_env(): + # Public repo bound via resource grant (metadata.repo_url) -> GIT_REPO_URL injected, no creds. + body = {"resource_grants": [{ + "resource_type": "git", "binding_scope": "https://github.com/Fasthei/swe", + "metadata": {"repo_url": "https://github.com/Fasthei/swe", "provider": "github", "default_branch": "main"}, + "ref": "azkv://heicode-vault.vault.azure.net/secrets/users-3-bindings-x", # present but NOT resolved (public) + }]} + env = al.git_launch_env(body) + check("git_launch_env: GIT_REPO_URL from grant metadata.repo_url", env.get("GIT_REPO_URL") == "https://github.com/Fasthei/swe") + check("git_launch_env: GIT_PROVIDER from metadata", env.get("GIT_PROVIDER") == "github") + check("git_launch_env: GIT_DEFAULT_BRANCH + GIT_BASE_BRANCH from metadata", + env.get("GIT_DEFAULT_BRANCH") == "main" and env.get("GIT_BASE_BRANCH") == "main") + check("git_launch_env: public repo injects NO credentials (GIT_TOKEN absent)", "GIT_TOKEN" not in env) + # grant found via orchestration_plan.resource_grants too + body_plan = {"orchestration_plan": {"resource_grants": [{"resource_type": "git", + "metadata": {"repo_url": "https://github.com/a/b"}}]}} + check("git_launch_env: reads plan.resource_grants", al.git_launch_env(body_plan).get("GIT_REPO_URL") == "https://github.com/a/b") + # git resource via binding_scope (no metadata.repo_url) when resource_type == git + body_scope = {"resource_grants": [{"resource_type": "git", "binding_scope": "https://gitea.x/o/r"}]} + check("git_launch_env: falls back to binding_scope for git type", al.git_launch_env(body_scope).get("GIT_REPO_URL") == "https://gitea.x/o/r") + # no git grant -> empty (a non-git grant is ignored) + check("git_launch_env: no git grant -> {}", al.git_launch_env({"resource_grants": [{"resource_type": "database", "metadata": {"host": "h"}}]}) == {}) + check("git_launch_env: empty body -> {}", al.git_launch_env({}) == {}) def test_resolve_model_key(): @@ -220,6 +253,7 @@ def main(): test_launch_count() test_pool_window() test_plan_specs() + test_git_launch_env() test_resolve_model_key() test_azkv_resolver() test_command_backend_build() From b394ce99d517885d1c663ba312555b2f389fc4ec Mon Sep 17 00:00:00 2001 From: gongzhiyong Date: Tue, 16 Jun 2026 04:33:01 +0800 Subject: [PATCH 3/6] fix(swarm): reap orphan agents on terminal run + retry backoff + surface task errors MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 孤儿 agent 事故修复(2026-06-15 swarm-69e470561bd7-seed 跨 workspace 执行)。 不改去中心化认领逻辑(swarm_dispatch 不动),仅治理 agent pod 生命周期与可观测性: - B1 终态回收: refresh_swarm_run_status 终态后 best-effort stop_launched 回收 pod+secret (此前仅 Manager stop 才回收,自然完成/失败的 run agent 残留 → 孤儿留在共享池抢别的 swarm 任务) - B2 防驱逐: agent pod 加 karpenter.sh/do-not-disrupt(AGENT_POD_ALLOW_DISRUPTION=1 可关) - stop_launched: backend=kubernetes 时按标签删,不再被内存集合 _k8s_swarms 门控(跨重启可靠) - C 重试退避: Task.next_retry_at + fail_task 指数退避 5→30→180s(cap 300, TASK_RETRY_BACKOFF_*), is_task_ready 门控;TASK_MAX_RETRIES 可配 - D 错误上浮: task_executor 失败时聚合 subtask error 到顶层 error(根治通用 "Task failed"), _execute_subtask 打印 LLM 响应片段 - 配额硬上限 16: _clamp_user_cap 契约测试全过: runtime-contract / merge-smoke / workflow-e2e / contract-freeze / max-agents-per-user / security-boundary。 Co-Authored-By: Claude Opus 4.8 (1M context) --- agent/task_executor.py | 34 +++++++++++++++++++++++++++++-- orchestrator/agent_launcher.py | 28 +++++++++++++++++++++++-- orchestrator/main.py | 37 +++++++++++++++++++++++++++++++--- orchestrator/task_queue.py | 35 +++++++++++++++++++++++++++++++- 4 files changed, 126 insertions(+), 8 deletions(-) diff --git a/agent/task_executor.py b/agent/task_executor.py index 99e4795..1df4b6f 100644 --- a/agent/task_executor.py +++ b/agent/task_executor.py @@ -117,7 +117,7 @@ class TaskExecutor: success = all(r["status"] in ["completed", "handed_off"] for r in results) awaiting_handoff = any(r["status"] == "handed_off" for r in results) - return { + payload = { "success": success, "task_id": task_id, "subtasks": results, @@ -125,6 +125,19 @@ class TaskExecutor: "agent_id": self.agent_id, "usage": self._usage_payload(time.time() - started_at), } + if not success: + # Surface the real per-subtask failure detail as a top-level `error` so the + # orchestrator records WHY (agent/main.py falls back to a generic "Task failed" + # when this key is absent — see incident 2026-06-15 swarm-69e470561bd7-seed). + failed = [r for r in results if r.get("status") not in ("completed", "handed_off")] + detail = "; ".join( + f"{((r.get('subtask') or {}).get('description') or 'subtask')[:80]}: " + f"{r.get('error') or r.get('summary') or 'no detail reported'}" + for r in failed + ) or "subtask(s) failed without detail" + payload["error"] = detail + logger.error(f"Task {task_id} failed: {detail}") + return payload except Exception as e: logger.error(f"Error executing task {task_id}: {e}") return { @@ -166,6 +179,7 @@ Return ONLY the JSON array, no other text.""" }] async def _execute_subtask(self, subtask: dict, task_id: str, context: dict, peer_collaboration_callback: Optional[Callable]) -> dict: + content = "" try: description = subtask["description"] workspace_files = self._summarize_workspace() @@ -270,10 +284,26 @@ Return ONLY the JSON, no other text.""" if apply_result["errors"]: result["status"] = "failed" result["error"] = "; ".join(apply_result["errors"]) + logger.warning( + f"Subtask for task {task_id} failed applying file changes: {result['error']}" + ) + else: + # LLM returned 2xx but declared the subtask not completed: log its reason and a + # bounded snippet of the model output so the failure is diagnosable from agent logs. + logger.warning( + "Subtask for task %s reported status=%r (error=%s summary=%s); llm_response[:500]=%s", + task_id, result.get("status"), result.get("error"), + result.get("summary"), (content or "").strip()[:500], + ) result["subtask"] = subtask return result except Exception as e: - logger.error(f"Error executing subtask: {e}") + # Parse failures / API errors land here; include a bounded model-output snippet + # (model-generated text, no secrets) to explain why parsing/execution failed. + logger.error( + f"Error executing subtask for task {task_id}: {e}; " + f"llm_response[:500]={(content or '').strip()[:500]}" + ) return { "subtask": subtask, "status": "failed", diff --git a/orchestrator/agent_launcher.py b/orchestrator/agent_launcher.py index a4f896b..1c54b38 100644 --- a/orchestrator/agent_launcher.py +++ b/orchestrator/agent_launcher.py @@ -439,6 +439,20 @@ def pod_resources() -> Dict[str, Any]: } +def pod_annotations() -> Dict[str, str]: + """Annotations applied to every agent Pod. + + `karpenter.sh/do-not-disrupt` keeps the node autoscaler from consolidating/evicting an agent + while its run is still live — mid-run eviction is what orphaned an agent on 2026-06-15 (its + siblings were reaped as "Underutilized", leaving a lone idle agent). The run's own teardown + (stop_launched) removes the pods when the run reaches a terminal state, so this only protects + in-flight work. Set AGENT_POD_ALLOW_DISRUPTION=1 to opt out (e.g. cost-sensitive dev clusters). + """ + if os.getenv("AGENT_POD_ALLOW_DISRUPTION", "").lower() in {"1", "true", "yes"}: + return {} + return {"karpenter.sh/do-not-disrupt": "true"} + + def _k8s_labels(spec: AgentLaunchSpec, swarm_id: str) -> Dict[str, str]: labels = {"app": "heicode-swarm-agent", "heicode-swarm-id": swarm_id} uid = spec.env.get("HEICODE_USER_ID") @@ -480,9 +494,15 @@ def build_pod_manifest(spec: AgentLaunchSpec, *, namespace: str, swarm_id: str, } if service_account: pod_spec["serviceAccountName"] = service_account + metadata: Dict[str, Any] = { + "name": spec.agent_id, "namespace": namespace, "labels": _k8s_labels(spec, swarm_id), + } + annotations = pod_annotations() + if annotations: + metadata["annotations"] = annotations return { "apiVersion": "v1", "kind": "Pod", - "metadata": {"name": spec.agent_id, "namespace": namespace, "labels": _k8s_labels(spec, swarm_id)}, + "metadata": metadata, "spec": pod_spec, } @@ -537,7 +557,11 @@ async def stop_launched(swarm_id: str) -> int: stopped += 1 except Exception as exc: logger.warning("failed to stop launched agent proc for %s: %s", swarm_id, exc) - if swarm_id in _k8s_swarms: + # The in-memory `_k8s_swarms` set is lost on orchestrator restart, so don't gate teardown on + # it: when the kubernetes backend is active, delete by label authoritatively. The label + # selector + --ignore-not-found makes this idempotent and safe to call for any swarm_id + # (including runs launched before a restart — which is exactly how agents got orphaned). + if swarm_id in _k8s_swarms or launch_backend() == "kubernetes": _k8s_swarms.discard(swarm_id) try: proc = await asyncio.create_subprocess_exec( diff --git a/orchestrator/main.py b/orchestrator/main.py index 8ab0e86..1c1f709 100644 --- a/orchestrator/main.py +++ b/orchestrator/main.py @@ -145,9 +145,28 @@ manager = ConnectionManager() AGENT_SLOTS: Dict[str, int] = {} +# Absolute ceiling on concurrent agents per user, enforced regardless of what HM/plan/env requests. +# Product rule: a tenant may run at most 16 agents at once (architect ruling 2026-06-15). +MAX_AGENTS_PER_USER_HARD_CAP = 16 + + +def _clamp_user_cap(value: int) -> int: + """Clamp a requested per-user agent cap into [1, MAX_AGENTS_PER_USER_HARD_CAP].""" + return max(1, min(MAX_AGENTS_PER_USER_HARD_CAP, value)) + + +def _default_task_retries() -> int: + """Default retry budget for runtime tasks (env TASK_MAX_RETRIES, default 3). Paired with the + task-queue retry backoff so retries span minutes rather than seconds.""" + try: + return max(1, int(os.getenv("TASK_MAX_RETRIES", "3") or 3)) + except ValueError: + return 3 + + def _env_max_agents_per_user() -> int: try: - return max(1, int(os.getenv("MAX_AGENTS_PER_USER", "10") or 10)) + return _clamp_user_cap(int(os.getenv("MAX_AGENTS_PER_USER", "10") or 10)) except ValueError: return 10 @@ -164,7 +183,8 @@ def max_agents_per_user(body: Optional[Dict[str, Any]] = None) -> int: if isinstance(body, dict): cap = ((body.get("metadata") or {}).get("max_agents_per_user")) if isinstance(cap, int) and not isinstance(cap, bool) and cap > 0: - return cap + # HM's value wins, but never above the hard product ceiling (16). + return _clamp_user_cap(cap) return _env_max_agents_per_user() @@ -731,6 +751,17 @@ async def refresh_swarm_run_status(run): except Exception as exc: logger.warning("benchmark capture failed for run %s: %s", run.swarm_id, exc) + # Reap the run's agent pods/secret now that it has reached a terminal state. Previously teardown + # ran only on a Manager-requested stop (stop_swarm_run), so naturally completed/failed runs left + # their agents Running — those idle, registered agents then lingered in the shared pool and could + # self-select another run's tasks (incident 2026-06-15: a leftover agent ran another swarm's seed + # against the wrong workspace). Best-effort; never fails the terminal transition. Does NOT touch + # task self-selection — only the worker pod lifecycle. + try: + await agent_launcher.stop_launched(run.swarm_id) + except Exception as exc: + logger.warning("agent teardown on terminal run %s failed: %s", run.swarm_id, exc) + # Lifespan context manager @asynccontextmanager @@ -1262,7 +1293,7 @@ async def create_tasks_for_run(run, body: Dict[str, Any]) -> int: root_task_id=runtime_root_task_id, source=task_spec.get("source", "runtime_bridge"), context=task_context, - max_retries=body.get("max_retries", 3), + max_retries=body.get("max_retries") or _default_task_retries(), ) await swarm_runtime.attach_task(run, task.task_id) await swarm_runtime.emit_event( diff --git a/orchestrator/task_queue.py b/orchestrator/task_queue.py index d788532..7579fe4 100644 --- a/orchestrator/task_queue.py +++ b/orchestrator/task_queue.py @@ -1,5 +1,6 @@ """Task queue management with dependency-aware failure recovery.""" import json +import os import time import uuid import logging @@ -12,6 +13,24 @@ from .agent_registry import agent_registry, AgentStatus logger = logging.getLogger(__name__) +def _retry_backoff_seconds(retry_count: int) -> float: + """Delay before a failed task becomes dispatchable again (exponential, capped). + + Defaults: 5s → 30s → 180s … capped at 300s. Spreads retries over minutes instead of burning + the whole budget in seconds, so a swarm's own agents (which may be cold-starting / waiting on + node scale-up) have time to register before the seed exhausts its retries (incident 2026-06-15). + Tunable via TASK_RETRY_BACKOFF_{BASE,FACTOR,CAP}. + """ + try: + base = float(os.getenv("TASK_RETRY_BACKOFF_BASE", "5") or 5) + factor = float(os.getenv("TASK_RETRY_BACKOFF_FACTOR", "6") or 6) + cap = float(os.getenv("TASK_RETRY_BACKOFF_CAP", "300") or 300) + except ValueError: + base, factor, cap = 5.0, 6.0, 300.0 + exponent = max(0, retry_count - 1) + return min(cap, base * (factor ** exponent)) + + class TaskStatus(str, Enum): """Task status enumeration.""" PENDING = "pending" @@ -44,6 +63,7 @@ class Task(BaseModel): child_task_ids: List[str] = Field(default_factory=list) retry_count: int = 0 max_retries: int = 3 + next_retry_at: Optional[float] = None context: Dict = Field(default_factory=dict) @@ -172,6 +192,11 @@ class TaskQueue: if task.status != TaskStatus.PENDING: return False + # Respect retry backoff: a task re-queued after a failure is not dispatchable until its + # next_retry_at has passed (see fail_task / _retry_backoff_seconds). + if task.next_retry_at and time.time() < task.next_retry_at: + return False + for dependency_id in task.depends_on: dependency = await self.get_task(dependency_id) if not dependency or dependency.status != TaskStatus.COMPLETED: @@ -322,11 +347,17 @@ class TaskQueue: task.assigned_agent_id = None task.started_at = None + # Backoff: gate re-dispatch until next_retry_at (is_task_ready enforces it) so retries + # spread over minutes rather than all firing within seconds. + delay = _retry_backoff_seconds(task.retry_count) + task.next_retry_at = time.time() + delay + # Re-add to pending queue await redis_client.lpush(self.PENDING_QUEUE_KEY, task_id) logger.warning( - f"Task {task_id} failed (retry {task.retry_count}/{task.max_retries}): {reason}" + f"Task {task_id} failed (retry {task.retry_count}/{task.max_retries}): {reason} " + f"(next retry in {delay:.0f}s)" ) else: task.status = TaskStatus.FAILED @@ -384,6 +415,7 @@ class TaskQueue: task.assigned_agent_id = None task.started_at = None task.blocked_reason = None + task.next_retry_at = None # a capacity release is not a failure — no backoff await self._save_task(task) await self.requeue_task(task_id) @@ -417,6 +449,7 @@ class TaskQueue: task.started_at = None task.completed_at = None task.blocked_reason = None + task.next_retry_at = None # a review reopen is not a failure — no backoff await self._save_task(task) await self.requeue_task(task_id) logger.info(f"Re-opened task {task_id} for another review cycle") From 6152235c96c70ec64df418dc07b88dad503e098f Mon Sep 17 00:00:00 2001 From: gongzhiyong Date: Tue, 16 Jun 2026 04:52:40 +0800 Subject: [PATCH 4/6] fix(agent): execute against cloned repo root, not empty per-task subdir MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 每个 SWE 任务都因 "Empty workspace: no files detected" 失败:仓库 clone 到 /workspace, 但 TaskExecutor 用的是空的 /workspace/.agent_tasks/ 子目录(从未播种仓库内容), 模型看不到任何源码 → 正确地拒绝执行。 Option B(架构裁定 2026-06-16):TaskExecutor.workspace_dir 改为仓库根 self.workspace_dir, context.workspace_dir 同步。git(workspace_git 同根)即可 commit/push 真实改动。 MAX_CONCURRENT_TASKS 默认 1:单一 /workspace git checkout 非并发安全,蜂群并行靠多 agent。 Co-Authored-By: Claude Opus 4.8 (1M context) --- agent/main.py | 18 +++++++++++++----- 1 file changed, 13 insertions(+), 5 deletions(-) diff --git a/agent/main.py b/agent/main.py index e20b2b7..9ecb370 100644 --- a/agent/main.py +++ b/agent/main.py @@ -72,7 +72,11 @@ class AgentRuntimeDisconnected(RuntimeError): class Agent: """Agent that connects to orchestrator and executes tasks.""" - MAX_CONCURRENT_TASKS = int(os.getenv("MAX_CONCURRENT_TASKS", "4")) + # One task at a time per agent: tasks execute against the shared cloned repo root (see + # _execute_assignment, Option B 2026-06-16) and the per-agent git checkout (workspace_git, + # branch/commit/push) is single-repo and not concurrency-safe. Swarm parallelism comes from + # MULTIPLE agents, not multiple tasks per agent. Override via env only on isolated workspaces. + MAX_CONCURRENT_TASKS = int(os.getenv("MAX_CONCURRENT_TASKS", "1")) TASK_TIMEOUT_SECONDS = int(os.getenv("TASK_TIMEOUT_SECONDS", "60")) PEER_REPLY_TIMEOUT_SECONDS = int(os.getenv("PEER_REPLY_TIMEOUT_SECONDS", "20")) HEARTBEAT_INTERVAL_SECONDS = 15 @@ -382,12 +386,16 @@ class Agent: await self.send_status_update("busy", task_id, "Starting task execution") start_time = time.time() - task_workspace = self.task_workspace(task_id) - task_workspace.mkdir(parents=True, exist_ok=True) git_enabled = False try: - executor = TaskExecutor(agent_id=self.agent_id, workspace_dir=str(task_workspace)) + # Execute against the cloned repository root (Option B, 2026-06-16). The previous + # per-task subdir (/workspace/.agent_tasks/) was created empty and never + # seeded with the repo, so the executor saw an empty workspace and every SWE task + # failed with "Empty workspace: no files detected". Using the repo root means the + # model reads the real source and git (workspace_git, rooted at the same dir) + # commits the actual edits. One-task-per-agent (MAX_CONCURRENT_TASKS) keeps this safe. + executor = TaskExecutor(agent_id=self.agent_id, workspace_dir=str(self.workspace_dir)) git_enabled = await self.workspace_git.is_git_workspace() if git_enabled: branch_created = await self.workspace_git.create_result_branch(task_id) @@ -401,7 +409,7 @@ class Agent: description=description, context={ **context, - "workspace_dir": str(task_workspace), + "workspace_dir": str(self.workspace_dir), "repo_workspace_dir": str(self.workspace_dir), "git_repo_url": self.git_repo_url, "agent_id": self.agent_id, From 62d6fdf78ca3015c201fbfd443374facc65fe95c Mon Sep 17 00:00:00 2001 From: gongzhiyong Date: Tue, 16 Jun 2026 05:01:25 +0800 Subject: [PATCH 5/6] fix(agent): per-task git worktree so concurrent tasks each get an isolated repo copy MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 取代上一版 Option B(执行器共用仓库根 + 串行 1 任务)。每个任务从本 agent 自己的 clone 切出独立 git worktree(完整仓库内容 + 独立分支/索引),执行器在其中读写真实源码, 任务级 GitOperations 在该 worktree 提交/推送结果分支,完成后回收 worktree。 - 单 agent 可并发多任务(MAX_CONCURRENT_TASKS 恢复 4),互不共用 checkout/index - worktree 置于仓外 /tmp/agent-worktrees(AGENT_WORKTREE_BASE 可配),主 checkout 不受污染 - _git_admin_lock 仅串行 worktree add/remove 等共享 .git plumbing,任务执行仍并行 - 跨 agent 隔离不变:每个 agent 仍各自 clone 一份(各自 pod) - 根治 "Empty workspace: no files detected"(worktree 自带仓库文件,已端到端验证) GitOperations 新增 add_task_worktree / remove_task_worktree。 Co-Authored-By: Claude Opus 4.8 (1M context) --- agent/git_operations.py | 40 ++++++++++++++++++++++++++ agent/main.py | 62 +++++++++++++++++++++++++---------------- 2 files changed, 78 insertions(+), 24 deletions(-) diff --git a/agent/git_operations.py b/agent/git_operations.py index a00455e..cffa18d 100644 --- a/agent/git_operations.py +++ b/agent/git_operations.py @@ -164,6 +164,46 @@ class GitOperations: logger.error(f"Error creating result branch: {e}") return False + async def add_task_worktree(self, path: str, task_id: str) -> Optional[str]: + """Create an isolated git worktree at `path` on a fresh per-task result branch. + + Cut from this agent's clone (off origin/ or HEAD), the worktree has the full repo + contents plus its own index/HEAD, so one agent can run several tasks concurrently without + them clobbering a shared checkout. Returns the branch name, or None on failure. + """ + try: + root = await self.repo_root() or self.workspace_dir + safe_task_id = task_id.replace("/", "-")[:12] + branch = f"agent/{self.agent_id}/{safe_task_id}-{int(time.time())}" + + await self._run_git_command(["git", "fetch", "origin"], cwd=root) + base_ref = f"origin/{self.base_branch}" + if (await self._run_git_command(["git", "rev-parse", "--verify", base_ref], cwd=root)).returncode != 0: + base_ref = "HEAD" + logger.warning(f"Base branch origin/{self.base_branch} not found; worktree from HEAD") + + Path(path).parent.mkdir(parents=True, exist_ok=True) + result = await self._run_git_command( + ["git", "worktree", "add", "-b", branch, path, base_ref], cwd=root, + ) + if result.returncode != 0: + logger.error(f"git worktree add failed: {result.stderr}") + return None + logger.info(f"Created task worktree {path} on branch {branch}") + return branch + except Exception as e: + logger.error(f"Error creating task worktree: {e}") + return None + + async def remove_task_worktree(self, path: str) -> None: + """Tear down a per-task worktree (best-effort). The branch ref is kept (already pushed).""" + try: + root = await self.repo_root() or self.workspace_dir + await self._run_git_command(["git", "worktree", "remove", "--force", path], cwd=root) + logger.info(f"Removed task worktree {path}") + except Exception as e: + logger.warning(f"Failed to remove task worktree {path}: {e}") + async def commit_changes(self, message: str) -> Optional[str]: try: cwd = await self.repo_root() or self.workspace_dir diff --git a/agent/main.py b/agent/main.py index 9ecb370..1b20594 100644 --- a/agent/main.py +++ b/agent/main.py @@ -72,11 +72,10 @@ class AgentRuntimeDisconnected(RuntimeError): class Agent: """Agent that connects to orchestrator and executes tasks.""" - # One task at a time per agent: tasks execute against the shared cloned repo root (see - # _execute_assignment, Option B 2026-06-16) and the per-agent git checkout (workspace_git, - # branch/commit/push) is single-repo and not concurrency-safe. Swarm parallelism comes from - # MULTIPLE agents, not multiple tasks per agent. Override via env only on isolated workspaces. - MAX_CONCURRENT_TASKS = int(os.getenv("MAX_CONCURRENT_TASKS", "1")) + # An agent may run several tasks at once; each executes in its OWN git worktree cut from this + # agent's clone (see _execute_assignment / GitOperations.add_task_worktree), so concurrent tasks + # never share a checkout/index. Swarm parallelism = many agents × this per-agent concurrency. + MAX_CONCURRENT_TASKS = int(os.getenv("MAX_CONCURRENT_TASKS", "4")) TASK_TIMEOUT_SECONDS = int(os.getenv("TASK_TIMEOUT_SECONDS", "60")) PEER_REPLY_TIMEOUT_SECONDS = int(os.getenv("PEER_REPLY_TIMEOUT_SECONDS", "20")) HEARTBEAT_INTERVAL_SECONDS = 15 @@ -115,14 +114,20 @@ class Agent: self._peer_executor: Optional[TaskExecutor] = None self.workspace_git = GitOperations(str(self.workspace_dir), self.agent_id) + # Serializes only the brief shared-repo git plumbing (worktree add/remove + fetch) so + # concurrent tasks don't race on .git locks. Task EXECUTION stays parallel. + self._git_admin_lock = asyncio.Lock() def available_slots(self) -> int: # Return how many additional tasks this agent can currently accept. return max(0, self.MAX_CONCURRENT_TASKS - len(self.active_tasks)) def task_workspace(self, task_id: str) -> Path: - # Compute the dedicated per-task working directory inside the agent workspace. - return self.workspace_dir / ".agent_tasks" / task_id + # Per-task git worktree path. Kept OUTSIDE the repo root (default /tmp/agent-worktrees) so + # the main checkout never sees it as untracked, and namespaced by agent_id to avoid + # collisions. `git worktree add` requires the leaf to not pre-exist, so we don't mkdir it. + base = Path(os.getenv("AGENT_WORKTREE_BASE", "/tmp/agent-worktrees")) + return base / self.agent_id / task_id.replace("/", "-") async def safe_send(self, payload: dict): # Serialize and send a websocket message while holding a lock to prevent concurrent writes. @@ -386,22 +391,28 @@ class Agent: await self.send_status_update("busy", task_id, "Starting task execution") start_time = time.time() + task_workspace = self.task_workspace(task_id) git_enabled = False + task_git = None try: - # Execute against the cloned repository root (Option B, 2026-06-16). The previous - # per-task subdir (/workspace/.agent_tasks/) was created empty and never - # seeded with the repo, so the executor saw an empty workspace and every SWE task - # failed with "Empty workspace: no files detected". Using the repo root means the - # model reads the real source and git (workspace_git, rooted at the same dir) - # commits the actual edits. One-task-per-agent (MAX_CONCURRENT_TASKS) keeps this safe. - executor = TaskExecutor(agent_id=self.agent_id, workspace_dir=str(self.workspace_dir)) - git_enabled = await self.workspace_git.is_git_workspace() - if git_enabled: - branch_created = await self.workspace_git.create_result_branch(task_id) - if not branch_created: - logger.warning("Failed to create task result branch; continuing without git push") - git_enabled = False + # 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 + # the full repo contents on a fresh result branch; the executor reads/writes the real + # 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(): + 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: + git_enabled = True + task_git = GitOperations(str(task_workspace), self.agent_id) + task_git.result_branch = result_branch + else: + logger.warning("Failed to create task worktree; executing on repo root without git push") + exec_dir = str(task_workspace) if git_enabled else str(self.workspace_dir) + executor = TaskExecutor(agent_id=self.agent_id, workspace_dir=exec_dir) result = await asyncio.wait_for( executor.execute_task( @@ -409,7 +420,7 @@ class Agent: description=description, context={ **context, - "workspace_dir": str(self.workspace_dir), + "workspace_dir": exec_dir, "repo_workspace_dir": str(self.workspace_dir), "git_repo_url": self.git_repo_url, "agent_id": self.agent_id, @@ -425,12 +436,12 @@ 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: - commit_sha = await self.workspace_git.commit_changes( + if git_enabled and task_git: + commit_sha = await task_git.commit_changes( message=f"Task {task_id}: {description[:50]}" ) if commit_sha: - branch_name = await self.workspace_git.push_results() + branch_name = await task_git.push_results() result["git_branch"] = branch_name result["commit_sha"] = commit_sha else: @@ -473,6 +484,9 @@ class Agent: await self.send_task_result(task_id, False, {"error": str(e), "success": False}) await self.send_status_update("idle", None, f"Task failed: {e}") finally: + if git_enabled: + async with self._git_admin_lock: + await self.workspace_git.remove_task_worktree(str(task_workspace)) self.active_tasks.pop(task_id, None) ACTIVE_TASKS.set(len(self.active_tasks)) self.current_task_id = None From 26a40c7770e8b664d52f2b27e3b7604f9322a47f Mon Sep 17 00:00:00 2001 From: gongzhiyong Date: Tue, 16 Jun 2026 05:23:04 +0800 Subject: [PATCH 6/6] fix(agent): failure reason leads with model's diagnosis, not the task prompt MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 之前聚合用 "<任务描述前80字>: ",失败 reason 被整段 prompt 污染, 真正的"看到什么/缺什么"被挤到后面。改为以模型 summary 为主、附上 distinct error, reason 直接就是诊断本身(例:"No source files found...; only README/jsonl present. (Required files like src/database/redis/main.js are missing.)")。 Co-Authored-By: Claude Opus 4.8 (1M context) --- agent/task_executor.py | 23 +++++++++++++++-------- 1 file changed, 15 insertions(+), 8 deletions(-) diff --git a/agent/task_executor.py b/agent/task_executor.py index 1df4b6f..9271a56 100644 --- a/agent/task_executor.py +++ b/agent/task_executor.py @@ -126,15 +126,22 @@ class TaskExecutor: "usage": self._usage_payload(time.time() - started_at), } if not success: - # Surface the real per-subtask failure detail as a top-level `error` so the - # orchestrator records WHY (agent/main.py falls back to a generic "Task failed" - # when this key is absent — see incident 2026-06-15 swarm-69e470561bd7-seed). + # 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 + # "Task failed" (agent/main.py uses this as the failure reason). Lead with the model's + # summary/error — NOT the task prompt — so the reason reads as the diagnosis, not the + # ask. summary is usually the fuller "saw X, missing Y" narrative; append a distinct + # error for the concise cause. failed = [r for r in results if r.get("status") not in ("completed", "handed_off")] - detail = "; ".join( - f"{((r.get('subtask') or {}).get('description') or 'subtask')[:80]}: " - f"{r.get('error') or r.get('summary') or 'no detail reported'}" - for r in failed - ) or "subtask(s) failed without detail" + explanations = [] + for r in failed: + summary = str(r.get("summary") or "").strip() + error = str(r.get("error") or "").strip() + text = summary or error or "subtask failed without detail" + if summary and error and error not in summary: + text = f"{summary} ({error})" + explanations.append(text) + detail = "; ".join(explanations) or "subtask(s) failed without detail" payload["error"] = detail logger.error(f"Task {task_id} failed: {detail}") return payload