xiaohei/Agentswarm
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(swarm): reap orphan agents on terminal run + retry backoff + surface task errors

Browse Source 
孤儿 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) <noreply@anthropic.com>
This commit is contained in:
gongzhiyong
2026-06-16 04:33:01 +08:00
co-authored by Claude Opus 4.8
parent f9f1d9a6d5
commit b394ce99d5
4 changed files with 126 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files
agent/task_executor.py
+32 -2
View File
@@ -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",
orchestrator/agent_launcher.py
+26 -2
View File
@@ -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(
orchestrator/main.py
+34 -3
View File
@@ -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(
orchestrator/task_queue.py
+34 -1
View File
@@ -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")