Files
Agentswarm/orchestrator/main.py
T
gongzhiyongandClaude Opus 4.8 fb1eeeebfa fix(swebench): 真实仓 patch 捕获 + agent 稳定性
#1 稳定性:HEARTBEAT_TIMEOUT 30→120(可配 AGENT_HEARTBEAT_TIMEOUT)——真实仓 clone/长
模型调用不再误杀 agent。
#2 正确性/忠实度:
- 透传 GIT_BASE_COMMIT;agent clone 改为浅 fetch+checkout 该 commit(在正确基线上读/改,
  且大仓也快);emit_patch 持久化 base_commit + _git_diff 对 SHA 走浅 fetch。
- agent 读文件预算可配并调大(AGENT_CTX_MAX_FILES=40/BYTES=40000),LLM 看到完整文件,
  避免全量重写截断 → diff 失真。launcher 透传这些 env 给 agent pod。

import + runtime-contract/result-aggregator/security-boundary 测试全绿。

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-22 00:25:38 +08:00

3522 lines
159 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""FastAPI orchestrator with WebSocket support for agent coordination."""
import asyncio
import json
import logging
import os
import time
from collections import defaultdict
from contextlib import asynccontextmanager
from typing import Any, Dict, List, Optional
from fastapi import FastAPI, Header, Request, WebSocket, WebSocketDisconnect, HTTPException
from fastapi.responses import JSONResponse, PlainTextResponse, StreamingResponse
from pydantic import BaseModel, Field
from prometheus_client import Counter, Gauge, Histogram, generate_latest, CONTENT_TYPE_LATEST
from dotenv import load_dotenv
# Load a local .env (gitignored) before the planner is imported, so its OpenAI client and the
# review/synthesis steps pick up credentials without them being passed on the command line.
load_dotenv()
from .redis_client import redis_client
from .agent_registry import agent_registry, AgentStatus, AgentMetadata
from .handoff_manager import handoff_manager, HandoffRequest
from .task_queue import task_queue, TaskStatus
from .swarm_runtime import RuntimeValidationError, swarm_runtime, TERMINAL_CLIENT_EVENT_TYPES
from .planner import planner
from .master_agent import master_agent
from .quality import evaluate_run_quality
from .decision_engine import decision_engine
from .dispatch_score import (
DispatchCandidate, rank_candidates, build_dispatch_decision_event,
normalize_capability_match, normalize_tau, normalize_load,
)
from . import convergence as convergence_mod
from . import audit as audit_mod
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 result_aggregator
from . import agent_launcher
# Configure logging
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)
# Prometheus metrics
AGENTS_CREATED = Counter('swarm_agents_created_total', 'Total agents created')
AGENTS_FAILED = Counter('swarm_agents_failed_total', 'Total agents failed')
AGENTS_ACTIVE = Gauge('swarm_agents_active', 'Currently active agents')
TASKS_CREATED = Counter('swarm_tasks_created_total', 'Total tasks created')
TASKS_COMPLETED = Counter('swarm_tasks_completed_total', 'Total tasks completed')
TASKS_FAILED = Counter('swarm_tasks_failed_total', 'Total tasks failed')
HANDOFFS_TOTAL = Counter('swarm_handoffs_total', 'Total handoffs performed')
HANDOFF_DURATION = Histogram('swarm_handoff_duration_seconds', 'Handoff latency in seconds')
TASK_DURATION = Histogram('swarm_task_duration_seconds', 'Task duration in seconds')
AGENT_STATUS = Gauge('swarm_agent_status', 'Agent status by ID', ['agent_id', 'status'])
WEBSOCKET_CONNECTIONS = Gauge('swarm_websocket_connections', 'Active WebSocket connections')
ERRORS_TOTAL = Counter('swarm_errors_total', 'Total errors', ['error_type'])
_AUTH_WARNING_EMITTED = False
# WebSocket connection manager
class ConnectionManager:
"""Manages WebSocket connections for agents."""
def __init__(self):
self.active_connections: Dict[str, WebSocket] = {}
# Per-user concurrent-agent accounting (cap enforced at register; see max_agents_per_user).
# Agents that register without a user_id are "unbound" and not subject to the per-user cap.
self.agent_user: Dict[str, str] = {} # agent_id -> user_id (bound agents)
self.user_agents: Dict[str, set] = defaultdict(set) # user_id -> {agent_id}
async def connect(self, agent_id: str, websocket: WebSocket):
"""Accept and store WebSocket connection."""
await websocket.accept()
self.active_connections[agent_id] = websocket
logger.info(f"Agent {agent_id} connected via WebSocket")
def disconnect(self, agent_id: str):
"""Remove WebSocket connection."""
if agent_id in self.active_connections:
del self.active_connections[agent_id]
logger.info(f"Agent {agent_id} disconnected")
def user_agent_count(self, user_id: str) -> int:
"""How many distinct agents this user currently has connected."""
return len(self.user_agents.get(user_id, ()))
def can_bind_user(self, agent_id: str, user_id: str, limit: int) -> bool:
"""Whether `agent_id` may register under `user_id` without exceeding the per-user cap.
A reconnect by an already-counted agent_id is always allowed (it adds no new agent).
"""
agents = self.user_agents.get(user_id, set())
if agent_id in agents:
return True
return len(agents) < limit
def bind_user(self, agent_id: str, user_id: str):
"""Record that `agent_id` belongs to `user_id` (call only after a passing can_bind_user)."""
self.agent_user[agent_id] = user_id
self.user_agents[user_id].add(agent_id)
def unbind(self, agent_id: str):
"""Drop an agent's user binding on disconnect (no-op for unbound agents)."""
user_id = self.agent_user.pop(agent_id, None)
if user_id is None:
return
agents = self.user_agents.get(user_id)
if agents is not None:
agents.discard(agent_id)
if not agents:
self.user_agents.pop(user_id, None)
async def send_message(self, agent_id: str, message: dict) -> bool:
"""Send message to specific agent. Returns True if it was delivered."""
if agent_id in self.active_connections:
try:
await self.active_connections[agent_id].send_json(message)
return True
except Exception as e:
logger.error(f"Failed to send message to agent {agent_id}: {e}")
return False
return False
async def broadcast(self, message: dict):
"""Broadcast message to all connected agents."""
for agent_id, connection in self.active_connections.items():
try:
await connection.send_json(message)
except Exception as e:
logger.error(f"Failed to broadcast to agent {agent_id}: {e}")
manager = ConnectionManager()
# Last self-reported free capacity per agent, learned from register/heartbeat/accept/reject
# messages sent by the merged agent runtime. Used to avoid dispatching to an agent that
# reports it is momentarily full. Older agents omit these fields and are treated as having
# capacity (default 1), preserving the previous one-task-per-agent behavior.
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 _clamp_user_cap(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:
# HM's value wins, but never above the hard product ceiling (16).
return _clamp_user_cap(cap)
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`.
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")
if isinstance(slots, int):
AGENT_SLOTS[agent_id] = max(0, slots)
def agent_has_capacity(agent_id: str) -> bool:
"""Return whether an agent currently reports free capacity (default: yes)."""
return AGENT_SLOTS.get(agent_id, 1) > 0
async def deposit_pheromone(task, agent_id: str, *, success: bool, result=None, run=None):
"""Group A learning (always on): update the agent's τ trail for this task's role.
Passive observation — never changes dispatch behavior by itself (selection is gated
separately by ENABLE_ACO_DISPATCH) and never fails the caller. cost_ratio is the task's
model cost as a fraction of the run budget when both are known, else 0 (no fabricated
penalty — see decision_engine module docstring).
"""
if not task or not agent_id:
return
try:
cost_ratio = 0.0
if run is not None and isinstance(result, dict):
budget = ((run.request_body or {}).get("orchestration_plan") or {}).get("budget") or {}
max_cost = budget.get("max_cost_usd")
if isinstance(max_cost, (int, float)) and max_cost > 0:
cost = float((result.get("usage") or {}).get("model_cost_usd") or 0.0)
cost_ratio = cost / float(max_cost)
await decision_engine.deposit(
agent_role=task.agent_role,
agent_id=agent_id,
success=success,
cost_ratio=cost_ratio,
)
except Exception as exc:
logger.debug(f"pheromone deposit skipped for {getattr(task, 'task_id', '?')}: {exc}")
# Background task for failure detection
async def failure_detection_loop():
"""Periodically check for failed agents and reassign their tasks."""
while True:
try:
await asyncio.sleep(10) # Check every 10 seconds
failed_agents = await agent_registry.check_failed_agents()
for agent_id in failed_agents:
# Reassign tasks from failed agent
reassigned_tasks = await task_queue.reassign_agent_tasks(agent_id)
if reassigned_tasks:
logger.info(
f"Reassigned {len(reassigned_tasks)} tasks from failed agent {agent_id}"
)
# Disconnect WebSocket
manager.disconnect(agent_id)
recovered_tasks = await task_queue.recover_orphaned_tasks(
set(manager.active_connections.keys())
)
if recovered_tasks:
logger.info(f"Recovered {len(recovered_tasks)} orphaned tasks")
except Exception as e:
logger.error(f"Error in failure detection loop: {e}")
def review_max_cycles() -> int:
try:
return int(os.getenv("MAX_REVIEW_CYCLES", "2") or 2)
except ValueError:
return 2
async def build_dispatch_context(run, task) -> Dict[str, Any]:
"""Enrich a task's context at dispatch with completed dependency artifacts and live peers.
This is what lets specialists see each other's work: dependency_artifacts carries the
results of completed upstream tasks, and peer_agents lists other agents currently connected
on the run so the executor can consult them for overlapping fields. Purely additive — it
does not change the Manager-facing API.
"""
context = dict(task.context or {})
dependency_artifacts = []
for dep_id in task.depends_on or []:
dep = await task_queue.get_task(dep_id)
if dep and dep.status == TaskStatus.COMPLETED:
parsed = parse_task_result(dep) or {}
dependency_artifacts.append({
"task_id": dep_id,
"agent_role": dep.agent_role,
"summary": summarize_task_result(parsed),
"files_modified": parsed.get("files_modified") or [],
"changes": parsed.get("changes"),
})
if dependency_artifacts:
context["dependency_artifacts"] = dependency_artifacts
peer_agents = []
for other_id in run.task_ids:
if other_id == task.task_id:
continue
other = await task_queue.get_task(other_id)
if not other or not other.assigned_agent_id:
continue
if other.assigned_agent_id in manager.active_connections:
peer_agents.append({
"agent_id": other.assigned_agent_id,
"role": other.agent_role,
"capabilities": other.required_capabilities,
})
if peer_agents:
context["peer_agents"] = peer_agents
# Surface the task's role as the executor's specialist role, so the LLM prompt is
# role-aware and the testing/documentation peer-consult path can activate. The Task's
# agent_role is authoritative; fall back to any role already in context, then "general".
context.setdefault(
"specialist_role",
getattr(task, "agent_role", None) or context.get("agent_role") or "general",
)
context.setdefault("run_goal", run.objective)
context.setdefault("user_prompt", run.objective)
return context
async def task_dispatch_loop():
"""Assign pending tasks to idle, connected agents."""
while True:
try:
await asyncio.sleep(2)
pending_count = await task_queue.get_pending_count()
if pending_count == 0:
continue
idle_agents = await agent_registry.get_idle_agents()
connected_idle_agents = [
agent for agent in idle_agents
if agent.agent_id in manager.active_connections
and agent_has_capacity(agent.agent_id)
]
# Swarm dispatch (the ONLY dispatch path): each idle agent perceives the shared task
# pool and SELF-SELECTS the best-fit ready task by capability + pheromone(τ) + load +
# budget, recording an explainable dispatch.decision_made. There is no central greedy
# assignment, no mode toggle — this repo is the swarm runtime (see rework plan §0).
assigned = await swarm_dispatch(connected_idle_agents)
# P-guard: pending work but nothing got dispatched this tick → the swarm may be stuck.
# Diagnose each affected run and record/emit a health report explaining why (no capable
# agent, dependency deadlock, budget, etc.). Advisory — does not change the run.
if not assigned and await task_queue.get_pending_count() > 0:
diagnosed = set()
for task in await task_queue.get_all_tasks():
if task.status != TaskStatus.PENDING or task.task_id in diagnosed:
continue
run = await swarm_runtime.get_run_for_task(task.task_id)
if run and run.swarm_id not in diagnosed and run.status not in ("completed", "failed", "stopped"):
await assess_swarm_health(run)
diagnosed.add(run.swarm_id)
diagnosed.update(run.task_ids)
except Exception as e:
logger.error(f"Error in task dispatch loop: {e}")
async def finalize_dispatch(agent, task, *, decision=None, dispatch_event=None) -> bool:
"""Shared dispatch tail used by all selection modes (greedy / ACO / scored).
Assigns the task, emits task.claimed, records the selection telemetry (Group A decision
and/or #9 dispatch decision — both INTERNAL run state, not Manager events), builds the
dispatch context, and sends the assignment. Returns False (and requeues) if the assign lost
a race. Extracted so the three modes never diverge.
"""
success = await task_queue.assign_task(task.task_id, agent.agent_id)
if not success:
latest_task = await task_queue.get_task(task.task_id)
if latest_task and latest_task.status == TaskStatus.PENDING:
await task_queue.requeue_task(task.task_id)
return False
run = await swarm_runtime.get_run_for_task(task.task_id)
if run:
await swarm_runtime.emit_event(
run,
"task.claimed",
task_id=task.task_id,
agent_instance_id=agent.agent_id,
payload={
"task_id": task.task_id,
"agent_role": task.agent_role,
"agent_id": agent.agent_id,
},
)
if decision is not None:
# Group A telemetry: feeds tau/eta/p_decision in the collector.
await swarm_runtime.record_decision(run, decision.telemetry())
if dispatch_event is not None:
# #9: internal dispatch-decision record (audit/benchmark-replayable). NOT a Manager
# event — a Manager-facing dispatch.decision_made needs event-schema registration.
await swarm_runtime.record_dispatch_decision(run, dispatch_event)
dispatch_context = await build_dispatch_context(run, task) if run else task.context
await manager.send_message(agent.agent_id, {
"type": "task_assignment",
"task_id": task.task_id,
"description": task.description,
"context": dispatch_context,
})
logger.info(f"Dispatched task {task.task_id} to connected idle agent {agent.agent_id}")
return True
async def _maybe_topup_agents(run, backlog: int) -> None:
"""Elastic top-up for one run: if its PENDING backlog exceeds idle connected agents and the
pool is below the cap, launch the deficit (capped by MAX_POOL and the run's per-user limit).
New agents get non-colliding ids (id_start = current pool size). Fail-soft."""
swarm_id = run.swarm_id
prefix = f"{swarm_id}-agent-"
live = sum(1 for aid in manager.active_connections if aid.startswith(prefix))
launched = run.metadata.get("launched_agents") or []
pool = max(live, len(launched)) # avoid re-launching agents still connecting
idle = sum(1 for a in await agent_registry.get_idle_agents()
if a.agent_id.startswith(prefix) and a.agent_id in manager.active_connections
and agent_has_capacity(a.agent_id))
body = run.request_body or {}
cap = min(agent_launcher.max_pool_size(), max_agents_per_user(body))
if backlog <= idle or pool >= cap:
return
delta = min(backlog - idle, cap - pool)
if delta <= 0:
return
user_id = ((body.get("metadata") or {}).get("runtime_headers") or {}).get("x_user_id")
specs = agent_launcher.plan_launch_specs(
run, body, connected_user_agents=0, limit=max_agents_per_user(body),
pool_size=delta, model_key=agent_launcher.resolve_model_key(body),
orchestrator_url=agent_launcher.orchestrator_ws_url(), user_id=user_id,
git_env=agent_launcher.resolve_git_grant(body), count_override=delta, id_start=pool)
launched_ids = await agent_launcher.launch(specs, swarm_id=swarm_id)
if launched_ids:
run.metadata["launched_agents"] = launched + launched_ids
await swarm_runtime.save_run(run)
logger.info("autoscale: run %s +%d agents (backlog=%d idle=%d pool=%d cap=%d)",
swarm_id, len(launched_ids), backlog, idle, pool, cap)
async def agent_autoscale_loop():
"""Elastic agent pool (demand-driven). Initial launch stays small (AGENT_LAUNCH_MIN_POOL); this
loop grows each run's pool only as its PENDING-task backlog demands, up to MAX_POOL / per-user
cap — so trivial runs use few agents and heavy runs scale out. Gated by ENABLE_AGENT_AUTOSCALING;
no-op when the launch backend is 'none'. Fail-soft (never raises into the event loop)."""
if os.getenv("ENABLE_AGENT_AUTOSCALING", "false").lower() not in {"1", "true", "yes"}:
return
if agent_launcher.launch_backend() == "none":
return
interval = float(os.getenv("AUTOSCALE_INTERVAL_SECONDS", "5"))
while True:
try:
await asyncio.sleep(interval)
if await task_queue.get_pending_count() == 0:
continue
backlog: Dict[str, list] = {} # swarm_id -> [run, count]
for task in await task_queue.get_all_tasks():
if task.status != TaskStatus.PENDING:
continue
run = await swarm_runtime.get_run_for_task(task.task_id)
if not run or run.status != "running":
continue
entry = backlog.setdefault(run.swarm_id, [run, 0])
entry[1] += 1
for run, n in backlog.values():
await _maybe_topup_agents(run, n)
except Exception as e:
logger.error(f"Error in agent autoscale loop: {e}")
async def compute_convergence_report(run, tasks, *, terminal: bool):
"""Build a run-state snapshot and evaluate the convergence report (#12, shadow by default).
Pure-input wrapper over convergence.evaluate_convergence: assembles tasks (with parsed
results + depends_on), budget/usage, the Group B quality grade, and the review-cycle counters,
then returns the ConvergenceReport. The caller stores it on the run and may attach the
termination_reason — it does NOT (yet) override run.status (shadow adoption; see the plan doc).
"""
task_dicts = []
used_cost = 0.0
for t in tasks:
result = parse_task_result(t) or {}
try:
used_cost += float((result.get("usage") or {}).get("model_cost_usd") or 0.0)
except Exception:
pass
task_dicts.append({
"task_id": t.task_id,
"status": t.status.value if hasattr(t.status, "value") else t.status,
"depends_on": t.depends_on,
"result": result,
# #70: retry counters let convergence classify a FAILED run as max_retries_exceeded.
"retry_count": t.retry_count,
"max_retries": t.max_retries,
})
plan = (run.request_body or {}).get("orchestration_plan") or {}
budget = plan.get("budget") or {}
run_state = {
"tasks": task_dicts,
"budget": budget,
"usage": {"total_cost_usd": used_cost},
"quality": run.quality or {},
"review_cycles": int(run.metadata.get("review_cycles", 0) or 0),
"max_review_cycles": review_max_cycles(),
}
return convergence_mod.evaluate_convergence(run_state)
async def _run_budget_pressure(task) -> Optional[float]:
"""Fraction of the run budget already consumed (real signal for dispatch scoring), or None.
Run-level (same across agents for a task; it prioritizes tasks on cheaper runs). None when
the run has no budget or no usage yet — never fabricated.
"""
run = await swarm_runtime.get_run_for_task(task.task_id)
if not run:
return None
budget = ((run.request_body or {}).get("orchestration_plan") or {}).get("budget") or {}
max_cost = budget.get("max_cost_usd")
if not isinstance(max_cost, (int, float)) or max_cost <= 0:
return None
consumed = 0.0
for tid in run.task_ids:
t = await task_queue.get_task(tid)
if not t or not t.result:
continue
try:
data = json.loads(t.result) if isinstance(t.result, str) else t.result
consumed += float((data.get("usage") or {}).get("model_cost_usd") or 0.0)
except Exception:
continue
return max(0.0, min(1.0, consumed / float(max_cost)))
async def assess_swarm_health(run, *, connected_agent_ids=None):
"""P-guard: diagnose whether the swarm can make progress on a run, and why not.
Assembles a snapshot (tasks + connected agents' capabilities + budget) and runs the pure
guard. Stores the report on run.metadata["health"], and appends unhealthy reports to a bounded
internal run.metadata["health_log"], so a stall is visible to an operator/Manager pulling the
run. INTERNAL only — does NOT emit a Manager event (see the comment below) and does not change
the run. Returns the HealthReport.
"""
tasks = [await task_queue.get_task(tid) for tid in run.task_ids]
tasks = [t for t in tasks if t]
task_dicts = [{
"task_id": t.task_id,
"status": t.status.value if hasattr(t.status, "value") else t.status,
"required_capabilities": t.required_capabilities,
"depends_on": t.depends_on,
"source": t.source,
} for t in tasks]
# Connected agents' capabilities (those actually reachable over the WS).
agent_ids = connected_agent_ids if connected_agent_ids is not None else list(manager.active_connections.keys())
caps = []
for aid in agent_ids:
a = await agent_registry.get_agent(aid)
if a:
caps.append(list(a.capabilities or []))
budget_state = convergence_mod.derive_budget_state({
"budget": ((run.request_body or {}).get("orchestration_plan") or {}).get("budget") or {},
"usage": {"total_cost_usd": sum(_task_cost_for_health(t) for t in tasks)},
})
report = guard_mod.diagnose({
"tasks": task_dicts,
"connected_agent_caps": caps,
"budget_state": budget_state,
})
# Stored on the run (+ a bounded log) so an operator/Manager can pull it via /diagnostics and
# /result. We deliberately do NOT introduce an unregistered `swarm.health` Manager event; instead
# the stall is surfaced ONCE per distinct blocker-set on the registered `timeline.updated` event
# (below), so a consumer watching /events and the cockpit (#39) sees *why* a run is stuck instead
# of silence — addressing agent_swarm#56.
report_dict = report.to_dict()
run.metadata["health"] = report_dict
if not report.healthy:
run.metadata["health_log"] = (run.metadata.get("health_log", [])[-49:] + [report_dict])
# Dedup by the report summary: emit on the first tick a (new) blocker-set appears; reset when the
# run becomes healthy again so a later re-block re-notifies. No raw content, only blocker reasons.
notify_sig = "" if report.healthy else report.summary
do_emit = bool(notify_sig) and run.metadata.get("health_notified_sig") != notify_sig
run.metadata["health_notified_sig"] = notify_sig
await swarm_runtime.save_run(run)
if do_emit:
await swarm_runtime.emit_event(run, "timeline.updated", payload={
"summary": report.summary,
"title": report.summary,
"blockers": [b["reason"] for b in report.blockers],
})
return report
def _task_cost_for_health(task) -> float:
try:
data = parse_task_result(task) or {}
return float((data.get("usage") or {}).get("model_cost_usd") or 0.0)
except Exception:
return 0.0
async def swarm_dispatch(idle_agents):
"""P6: each idle agent perceives the eligible ready tasks and SELF-SELECTS the best fit.
Decentralized, agent-centric (the OpenAI-Swarm/stigmergy spirit: no central scheduler assigns;
each agent picks from the shared pool). The fit score unifies #9's explainable dimensions with
#10's pheromone τ as `historical_success`: capability_match + τ + load + budget_pressure. The
chosen pairing is recorded as an explainable `dispatch.decision_made` (internal). Returns the
list of (agent_id, task_id) assigned this tick.
"""
assigned = []
for agent in idle_agents:
if not agent_has_capacity(agent.agent_id):
continue
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)
candidates.append(DispatchCandidate(
agent_id=agent.agent_id,
task_id=t.task_id,
agent_role=t.agent_role,
capability_match=normalize_capability_match(t.required_capabilities, agent.capabilities),
historical_success=normalize_tau(tau),
load=normalize_load(AGENT_SLOTS.get(agent.agent_id, 1)),
permission_fit=1.0,
budget_pressure=await _run_budget_pressure(t),
))
ranked = rank_candidates(candidates)
chosen = ranked[0]
chosen_task = next(t for t in ready if t.task_id == chosen.task_id)
excluded = {c.task_id: "lower_fit" for c in ranked[1:]} # other tasks this agent passed over
event = build_dispatch_decision_event(candidates, chosen, excluded, task_id=chosen.task_id)
await task_queue.remove_pending_task(chosen_task.task_id)
if await finalize_dispatch(agent, chosen_task, dispatch_event=event):
assigned.append((agent.agent_id, chosen_task.task_id))
return assigned
async def refresh_swarm_run_status(run):
"""Update a swarm run once all known tasks have reached terminal states."""
if run.status == "stopped":
return
if not run.task_ids:
return
tasks = []
for task_id in run.task_ids:
task = await task_queue.get_task(task_id)
if task:
tasks.append(task)
if not tasks:
return
if any(task.status in {TaskStatus.PENDING, TaskStatus.ASSIGNED, TaskStatus.IN_PROGRESS} for task in tasks):
if run.status != "running":
run.status = "running"
await swarm_runtime.save_run(run)
return
if any(task.status == TaskStatus.BLOCKED for task in tasks):
blocked_with_active_child = False
for task in tasks:
if task.status != TaskStatus.BLOCKED:
continue
for child_task_id in task.child_task_ids:
child_task = await task_queue.get_task(child_task_id)
if child_task and child_task.status in {
TaskStatus.PENDING,
TaskStatus.ASSIGNED,
TaskStatus.IN_PROGRESS,
TaskStatus.BLOCKED,
}:
blocked_with_active_child = True
break
if blocked_with_active_child:
break
if blocked_with_active_child:
if run.status != "running":
run.status = "running"
await swarm_runtime.save_run(run)
return
# 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
# default completion semantics (and Manager contract) are unchanged.
# P5 (#11): peer cross-review gate — when enabled and >=2 peer reviews were submitted, aggregate
# them (replacing the single-critic Master review) and reopen rework targets on rejection. At
# P-cutover this becomes the only review path; until then it precedes the Master loop.
# Swarm review (the ONLY review path): >=2 peers independently validate; disagreement is
# arbitrated and rework targets reopened. No single-critic Master gate. run_cross_review is a
# no-op (returns False) when fewer than 2 reviews were submitted.
if next_status == "completed":
if await run_cross_review(run, tasks):
return
# 协作聚合质量门(①合并 ②沙箱验证 ③不达标打回):聚合节点合并各 agent 互补产出 → 跑测试验证 →
# 不达标且未到轮次上限则 reopen 打回(run 保持 running,下轮重做),达标/达上限则落 main。
# 阈值未设(AGGREGATE_ACCEPTANCE_THRESHOLD)时不打回,保持原完成语义。
agg_cycles = int(run.metadata.get("review_cycles", 0) or 0)
agg_max = int(os.getenv("MAX_REVIEW_CYCLES", "2") or 2)
agg = await result_aggregator.finalize_run(
run, tasks, run.objective, _accept_threshold(run), agg_cycles, agg_max)
run.metadata["aggregation"] = agg
if agg.get("decision") == "bounce":
from .quality import collect_generated_files
reopened = 0
for t in tasks:
if collect_generated_files([t]).get("impl"):
if await task_queue.reopen_task(t.task_id):
reopened += 1
run.metadata["review_cycles"] = agg_cycles + 1
run.metadata["rework_reopens"] = int(run.metadata.get("rework_reopens", 0) or 0) + reopened
await swarm_runtime.save_run(run)
logger.info("aggregation: bounced run %s — reopened %d for rework (cycle %d, pass_rate=%s)",
run.swarm_id, reopened, agg_cycles + 1,
(agg.get("validation") or {}).get("pass_rate"))
return
await swarm_runtime.save_run(run)
if run.status == next_status:
return
run.status = next_status
await swarm_runtime.save_run(run)
# Swarm convergence (#12): always compute a report explaining WHY the run stopped
# (termination_reason + consensus + conflicts), stored on the run and surfaced in the status
# payload. Authoritative status-override is a follow-on; today next_status (task bookkeeping)
# and the report agree on completed/failed.
termination_reason = None
try:
report = await compute_convergence_report(run, tasks, terminal=True)
run.metadata["convergence"] = report.to_dict()
termination_reason = report.to_dict().get("termination_reason")
await swarm_runtime.save_run(run)
except Exception as exc:
logger.warning("convergence eval failed for run %s: %s", run.swarm_id, exc)
deliverable = None
final_summary = None
if next_status == "completed":
deliverable = build_run_deliverable(run, tasks)
# Synthesize the specialist results into one coherent, user-facing answer (a swarm tool,
# not a central controller — it only summarizes the converged outputs).
completed = [t for t in tasks if t.status == TaskStatus.COMPLETED]
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
# 聚合产物已由上面的协作聚合质量门产出并落 main(run.metadata['aggregation']);折进 deliverable。
agg = run.metadata.get("aggregation") or {}
if isinstance(deliverable, dict) and agg.get("finalized"):
deliverable["aggregated_files"] = agg.get("files")
deliverable["merged_file_count"] = agg.get("merged_file_count")
promo = agg.get("promotion") or {}
if promo.get("promoted"):
deliverable["promoted_to_main"] = {
"branch": promo.get("branch"), "commit_sha": promo.get("commit_sha")}
await swarm_runtime.save_run(run)
# Benchmark Group B: grade the run's generated code against its held-out fixture tests in
# the sandbox. Gated (ENABLE_QUALITY_EVAL) + fixture-bound; a no-op otherwise. Never fails
# the run — a grading error just leaves quality unrecorded (reward stays NaN).
try:
quality = await evaluate_run_quality(run, tasks)
if quality:
await swarm_runtime.record_quality(run, quality)
except Exception as exc:
logger.warning("quality eval failed for run %s: %s", run.swarm_id, exc)
await swarm_runtime.emit_event(
run,
"deployment.status_changed",
payload=swarm_runtime.status_payload(
run,
phase="Deliver" if next_status == "completed" else "Execute",
deliverable=deliverable,
),
)
# Frozen terminal swarm event (agent_swarm#15.2): the client cockpit keys its terminal banner
# off swarm.{completed|failed|stopped}. Emitted alongside deployment.status_changed (which HM
# still consumes for AgentDeployment.Status); stopped is emitted in swarm_runtime.stop_run.
swarm_terminal_payload = {
"status": next_status,
"task_count": len(tasks),
}
if termination_reason:
swarm_terminal_payload["termination_reason"] = termination_reason
# Carry the user-facing answer in the terminal event so the client gets the result from the
# one event it already watches (no reconstruction): the synthesized summary + deliverable
# facts (git branch/commit/files + artifact_ids). Full result also at GET …/{id}/result.
if next_status == "completed":
if final_summary:
swarm_terminal_payload["summary"] = final_summary
if deliverable:
swarm_terminal_payload["deliverable"] = deliverable
await swarm_runtime.emit_event(run, f"swarm.{next_status}", payload=swarm_terminal_payload)
timeline_payload = {
"summary": final_summary or f"Swarm run {next_status}",
"status": next_status,
"task_count": len(tasks),
}
if run.metadata.get("review_summary"):
timeline_payload["review_summary"] = run.metadata["review_summary"]
if termination_reason:
# Backward-compatible addition: explains WHY the run ended (#12).
timeline_payload["termination_reason"] = termination_reason
await swarm_runtime.emit_event(run, "timeline.updated", payload=timeline_payload)
await maybe_emit_budget_alert(run)
# Benchmark data capture (阶段2): on a terminal run, compute + persist the run's SwarmMetrics
# so real-user data accrues for later EMPIRICAL calibration (O / coefficients / S_gain — these
# cannot be defined a priori, only fitted from real runs). Read-only over run state; never
# fails the run. Default-on; set BENCHMARK_CAPTURE=0 to disable. Export is separately gated
# (BENCHMARK_EXPORT_TARGET, off by default) — see benchmark/export.
if (os.getenv("BENCHMARK_CAPTURE", "on").strip().lower() not in ("0", "off", "false", "no")):
try:
from benchmark.collectors.capture import capture_run_metrics
await capture_run_metrics(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
async def lifespan(app: FastAPI):
"""Startup and shutdown events."""
# Startup
logger.info("Starting orchestrator...")
# Fail-closed at boot: if code-executing quality eval is enabled, isolation MUST be confirmed.
# Refuse to start otherwise — a platform safety guarantee, not an ops-only convention.
from .quality import assert_quality_eval_safe
assert_quality_eval_safe()
await redis_client.connect()
# Start background tasks
failure_task = asyncio.create_task(failure_detection_loop())
dispatch_task = asyncio.create_task(task_dispatch_loop())
autoscale_task = asyncio.create_task(agent_autoscale_loop())
yield
# Shutdown
logger.info("Shutting down orchestrator...")
failure_task.cancel()
dispatch_task.cancel()
autoscale_task.cancel()
await redis_client.disconnect()
# Create FastAPI app
app = FastAPI(title="Swarm Orchestrator", version="1.0.0", lifespan=lifespan)
# Pydantic models for API
class RegisterRequest(BaseModel):
"""Agent registration request."""
agent_id: str
capabilities: List[str]
class TaskCreateRequest(BaseModel):
"""Task creation request."""
description: str
context: Dict = Field(default_factory=dict)
max_retries: int = 3
class TaskAssignRequest(BaseModel):
"""Task assignment request."""
task_id: str
agent_id: str
class StopDeploymentRequest(BaseModel):
"""Agent Manager stop request."""
reason: str = "Heicode Manager requested stop"
manager_deployment_id: Optional[str] = None
def error_response(
status_code: int,
code: str,
message: str,
request_id: Optional[str] = None,
) -> JSONResponse:
"""Return the Agent Manager compatible error envelope."""
return JSONResponse(
status_code=status_code,
content={
"success": False,
"error": {
"code": code,
"message": message,
"request_id": request_id,
},
},
)
async def require_runtime_auth(request: Request) -> Optional[JSONResponse]:
"""Validate service-to-service runtime auth when configured."""
global _AUTH_WARNING_EMITTED
token = (
os.getenv("AGENT_RUNTIME_SERVICE_TOKEN")
or os.getenv("AGNET_RUNTIME_SERVICE_TOKEN")
)
if not token:
if not _AUTH_WARNING_EMITTED:
logger.warning(
"AGENT_RUNTIME_SERVICE_TOKEN/AGNET_RUNTIME_SERVICE_TOKEN is not configured; runtime API is in insecure dev mode"
)
_AUTH_WARNING_EMITTED = True
return None
authorization = request.headers.get("authorization", "")
expected = f"Bearer {token}"
if authorization != expected:
return error_response(
401,
"UNAUTHORIZED",
"Missing or invalid service token",
request.headers.get("x-correlation-id"),
)
return None
def request_context_headers(request: Request) -> Dict[str, Optional[str]]:
"""Capture common Agent Manager request headers for audit/debug state."""
return {
"authorization_present": "authorization" in request.headers,
"x_user_id": request.headers.get("x-user-id"),
"x_binding_scope": request.headers.get("x-binding-scope"),
"x_agent_request_id": request.headers.get("x-agent-request-id"),
"x_correlation_id": request.headers.get("x-correlation-id"),
"x_idempotency_key": request.headers.get("x-idempotency-key"),
}
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.
This is the demotion of the Master: the orchestrator only *injects the initial task*; it does
NOT call master_agent.plan / decompose up front. Agents perceive the seed + shared state and
grow the task graph themselves (autonomous task generation = P3). The seed has no required
capabilities so any agent may self-select it.
"""
objective = run.objective or "Complete swarm objective"
base_context = (base_specs[0].get("context") if base_specs else {}) or {}
return [{
"task_id": "seed",
"title": "swarm seed",
"description": objective,
"agent_role": "general",
"required_capabilities": [], # any agent may claim the seed (self-selection)
"depends_on": [],
"parent_task_id": None,
"root_task_id": "seed",
"source": "seed",
"workflow_mode": "swarm",
"allow_handoff": True,
"context": {
**base_context,
"agent_role": "general",
"workflow_mode": "swarm",
"is_seed": True,
"objective": objective,
},
}]
async def handle_task_proposal(agent_id: str, message: Dict[str, Any]) -> Dict[str, Any]:
"""Decentralized rework P3 (#7): an executing agent proposes a NEW task from shared state.
This is the bottom-up decomposition that replaces the Master's top-down plan: an agent working
the seed (or any task) perceives the shared run state and proposes follow-up subtasks. The
orchestrator reviews each proposal (confidence floor / dedup-merge / per-run budget) and, on
ACCEPT, enqueues a real PENDING task carrying full lineage. Proposal lifecycle is stored on the
run (internal telemetry, NOT a Manager event); an accepted proposal emits the standard,
registered `task.created`. This is the swarm's only decomposition path (no Master plan).
Returns a small result dict for the agent ack. Never raises into the WS loop.
"""
origin_task_id = message.get("origin_task_id") or message.get("task_id")
run = await swarm_runtime.get_run_for_task(origin_task_id) if origin_task_id else None
if not run:
return {"accepted": False, "decision": "no_run"}
try:
proposal = autonomous_mod.TaskProposal(
proposed_by_agent_id=agent_id,
description=(message.get("description") or "").strip(),
proposal_reason=(message.get("reason") or message.get("proposal_reason") or "").strip(),
proposal_confidence=message.get("confidence", message.get("proposal_confidence", 0.0)),
title=message.get("title"),
agent_role=message.get("agent_role", "general"),
required_capabilities=message.get("required_capabilities") or [],
depends_on=message.get("depends_on") or [],
lineage=autonomous_mod.ProposalLineage(
origin_task_id=origin_task_id,
trigger_event=message.get("trigger_event"),
shared_state_snapshot=message.get("shared_state_snapshot") or {},
),
)
except ValueError as exc:
return {"accepted": False, "decision": "invalid", "reason": str(exc)}
# Per-run proposal budget: a hard cap on how many bottom-up tasks one run may spawn.
cap = int(os.getenv("AGENT_PROPOSAL_BUDGET", "5"))
accepted_so_far = int(run.metadata.get("proposal_accepted_count", 0) or 0)
policy = autonomous_mod.ProposalPolicy(remaining_proposal_budget=max(0, cap - accepted_so_far))
existing = []
for tid in run.task_ids:
t = await task_queue.get_task(tid)
if t:
existing.append(autonomous_mod.ExistingTaskRef(
task_id=t.task_id, description=t.description,
agent_role=t.agent_role,
status=t.status.value if hasattr(t.status, "value") else str(t.status),
))
outcome = autonomous_mod.review_proposal(proposal, policy, existing)
# Internal proposal-lifecycle telemetry on the run (not a Manager event).
run.metadata.setdefault("proposals", [])
run.metadata["proposals"] = (run.metadata["proposals"][-49:] +
[{"event_type": ev["event_type"], **ev}
for ev in autonomous_mod.build_lifecycle_events_for_outcome(proposal, outcome)])
result: Dict[str, Any] = {"accepted": False, "decision": outcome.decision.value,
"reason": outcome.reason, "proposal_id": proposal.proposal_id}
if outcome.decision == autonomous_mod.ProposalDecision.MERGE:
result["merge_target_task_id"] = outcome.merge_target_task_id
if outcome.decision == autonomous_mod.ProposalDecision.ACCEPT:
spec = autonomous_mod.ingest_accepted_proposal(proposal, swarm_id=run.swarm_id)
new_task_id = f"{run.swarm_id}-{proposal.proposal_id}"
task = await task_queue.create_task(
task_id=new_task_id,
title=spec["title"],
description=spec["description"],
agent_role=spec["agent_role"],
required_capabilities=spec["required_capabilities"],
depends_on=spec["depends_on"], # runtime ids (origin task is already runtime-scoped)
source=autonomous_mod.PROPOSED_SOURCE,
context={**spec["context"], "swarm_id": run.swarm_id,
"runtime_deployment_id": run.deployment_id},
enqueue=True,
)
await swarm_runtime.attach_task(run, task.task_id)
run.metadata["proposal_accepted_count"] = accepted_so_far + 1
await swarm_runtime.save_run(run)
await swarm_runtime.emit_event(
run, "task.created", task_id=task.task_id,
payload={"task_id": task.task_id, "agent_role": task.agent_role,
"source": autonomous_mod.PROPOSED_SOURCE,
"proposed_by_agent_id": agent_id},
)
result["accepted"] = True
result["task_id"] = task.task_id
else:
await swarm_runtime.save_run(run)
return result
async def handle_review_decision(agent_id: str, message: Dict[str, Any]) -> Dict[str, Any]:
"""Decentralized rework P5 (#11): a peer agent submits an independent structured review.
Reviews accumulate on the run; `run_cross_review` aggregates >=2 of them (the cross-validation
closure that is the swarm's only review — no single-critic Master gate).
"""
task_id = message.get("task_id") or (message.get("affected_tasks") or [None])[0]
run = await swarm_runtime.get_run_for_task(task_id) if task_id else None
if not run:
return {"recorded": False, "reason": "no_run"}
try:
decision = cross_review_mod.ReviewDecision(
verdict=message.get("verdict", "pass"),
reviewer_agent_id=agent_id,
evidence=message.get("evidence") or [],
failed_criteria=message.get("failed_criteria") or [],
affected_tasks=message.get("affected_tasks") or [],
recommended_rework=message.get("recommended_rework") or [],
confidence=message.get("confidence", 1.0),
weight=message.get("weight", 1.0),
summary=message.get("summary", ""),
)
except ValueError as exc:
return {"recorded": False, "reason": str(exc)}
reviews = run.metadata.setdefault("reviews", [])
# latest review per reviewer wins (an agent may revise its verdict)
reviews = [r for r in reviews if r.get("reviewer_agent_id") != agent_id]
reviews.append(decision.to_dict())
run.metadata["reviews"] = reviews
await swarm_runtime.save_run(run)
return {"recorded": True, "review_count": len(reviews)}
async def run_cross_review(run, tasks) -> bool:
"""P5 (#11): aggregate >=2 collected peer reviews; reopen rework targets if rejected.
Returns True when it reopened tasks (the run stays 'running' for a redo cycle). Returns False
when there aren't >=2 reviews, the cycle budget is spent,
or the cross-review accepted the work — letting the caller proceed to completion. Records the
arbitrated verdict + rework attributions on the run (internal telemetry).
"""
reviews_raw = run.metadata.get("reviews") or []
if len(reviews_raw) < 2:
return False
cycles = int(run.metadata.get("review_cycles", 0) or 0)
if cycles >= review_max_cycles():
return False
decisions = [cross_review_mod.ReviewDecision(**r) for r in reviews_raw]
verdict = cross_review_mod.aggregate_reviews(decisions)
run.metadata["cross_review"] = verdict.to_dict()
# Client-visible review timeline (#34): emit REDACTED projections (allowlist only — no
# evidence/summary/reason free text; see cross_review.*_client_payload). HM registers these 4
# types; the cockpit renders the review/rework timeline from them.
reviewer_ids = [d.reviewer_agent_id for d in decisions]
await swarm_runtime.emit_event(run, "review.started",
payload=cross_review_mod.review_started_client_payload(run.swarm_id, reviewer_ids, cycle=cycles))
await swarm_runtime.emit_event(run, "review.decision_made",
payload=cross_review_mod.review_decision_client_payload(run.swarm_id, verdict, cycle=cycles))
task_owner = {t.task_id: t.assigned_agent_id for t in tasks if t.assigned_agent_id}
valid_targets = [tid for tid in verdict.rework_targets if tid in run.task_ids]
if verdict.accepted or not valid_targets:
# A prior cycle's rework was redone and is now accepted → mark those targets complete.
if cycles > 0:
for a in (run.metadata.get("rework_attributions") or []):
await swarm_runtime.emit_event(run, "rework.completed",
payload=cross_review_mod.rework_completed_client_payload(
run.swarm_id, task_id=a.get("target_task_id"),
root_cause=a.get("root_cause"), cycle=cycles))
run.metadata["reviews"] = [] # consumed
await swarm_runtime.save_run(run)
return False
attributions = cross_review_mod.build_rework_attributions(verdict, task_owner=task_owner)
reopened = [tid for tid in valid_targets if await task_queue.reopen_task(tid)]
run.metadata["rework_attributions"] = (run.metadata.get("rework_attributions", [])[-49:] +
[a.to_dict() for a in attributions])
run.metadata["reviews"] = [] # consumed; reviewers re-review the redone work next cycle
if not reopened:
await swarm_runtime.save_run(run)
return False
run.metadata["review_cycles"] = cycles + 1
# SC-10: tally quality-driven reopens for P_rework (reopen_task doesn't bump retry_count).
run.metadata["rework_reopens"] = int(run.metadata.get("rework_reopens", 0) or 0) + len(reopened)
run.metadata["review_summary"] = verdict.summary
run.status = "running"
await swarm_runtime.save_run(run)
await swarm_runtime.emit_event(run, "timeline.updated", payload={
"summary": f"Cross-review cycle {cycles + 1}: {verdict.summary}",
"status": "running", "retry_tasks": reopened,
"disagreement": verdict.disagreement,
})
# Client-visible rework timeline (#34): one redacted rework.requested per reopened target.
for a in attributions:
if a.target_task_id in reopened:
await swarm_runtime.emit_event(run, "rework.requested",
payload=cross_review_mod.rework_requested_client_payload(run.swarm_id, a, cycle=cycles + 1))
logger.info(f"Cross-review reopened {reopened} on run {run.swarm_id} (cycle {cycles + 1})")
return True
def _accept_threshold(run) -> Optional[float]:
"""协作聚合验收阈值(pass_rate 0-100):run.metadata 覆盖 → AGGREGATE_ACCEPTANCE_THRESHOLD env →
None(门禁用)。None 保持当前完成语义(不打回);设了才启用质量驱动的"不达标打回迭代"。"""
raw = (run.metadata or {}).get("aggregate_acceptance_threshold")
if raw is None:
raw = os.getenv("AGGREGATE_ACCEPTANCE_THRESHOLD")
try:
return float(raw) if raw not in (None, "") else None
except (TypeError, ValueError):
return None
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] = {}
for aid in agent_ids:
tau = await decision_engine.get_tau(agent_role, aid)
out[aid] = normalize_tau(tau) or 0.5
return out
async def handle_task_bid(agent_id: str, message: Dict[str, Any]) -> Dict[str, Any]:
"""Decentralized rework P4 (#8): record an agent's bid for a task (de-dup by agent, latest wins).
Bids accumulate on the run; `arbitrate_and_assign` later picks a winner. Internal telemetry
only (not a Manager event).
"""
task_id = message.get("task_id")
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("run-boundary: rejected cross-run bid — agent %s ∉ run %s (task %s)",
agent_id, run.swarm_id, task_id)
return {"recorded": False, "reason": "cross_run_denied"}
bid = competition_mod.TaskBid(
task_id=task_id, agent_id=agent_id,
confidence=message.get("confidence", 0.5),
estimated_cost=message.get("estimated_cost", 0.0),
estimated_time=message.get("estimated_time", 0.0),
risk_score=message.get("risk_score", 0.0),
reason=message.get("reason", ""),
capabilities=message.get("capabilities") or [],
current_load=message.get("current_load", 0),
)
bids = run.metadata.setdefault("bids", {})
lst = [b for b in bids.get(task_id, []) if b.get("agent_id") != agent_id]
lst.append(bid.model_dump())
bids[task_id] = lst
await swarm_runtime.save_run(run)
return {"recorded": True, "task_id": task_id, "bid_count": len(lst)}
async def arbitrate_and_assign(run, task_id: str) -> Optional[Dict[str, Any]]:
"""Arbitrate the bids collected for a PENDING task and assign the winner (#8).
Deterministic (task_competition.arbitrate) + τ-weighted. Assigns the winner via the shared
finalize_dispatch tail, records the audit trail on the run, and clears the task's bids.
Returns the arbitration result payload, or None if there was nothing to arbitrate.
"""
bids_raw = (run.metadata.get("bids") or {}).get(task_id) or []
task = await task_queue.get_task(task_id)
if not bids_raw or not task:
return None
bids = [competition_mod.TaskBid(**b) for b in bids_raw]
hist = await _historical_success_map(task.agent_role, [b.agent_id for b in bids])
result = competition_mod.arbitrate(
bids, required_capabilities=task.required_capabilities, historical_success=hist,
)
if result.winner_agent_id and task.status == TaskStatus.PENDING:
winner = await agent_registry.get_agent(result.winner_agent_id)
if winner:
await task_queue.remove_pending_task(task_id)
assigned = await finalize_dispatch(winner, task)
if not assigned:
await task_queue.requeue_task(task_id)
_, payload = competition_mod.arbitrated_event(result)
run.metadata.setdefault("arbitrations", [])
run.metadata["arbitrations"] = run.metadata["arbitrations"][-49:] + [payload]
(run.metadata.get("bids") or {}).pop(task_id, None)
await swarm_runtime.save_run(run)
return payload
async def handle_task_yield(agent_id: str, message: Dict[str, Any]) -> Dict[str, Any]:
"""P4 (#8): an agent voluntarily releases a task back for re-competition (reuses release_task)."""
task_id = message.get("task_id")
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("run-boundary: rejected cross-run yield — agent %s ∉ run %s (task %s)",
agent_id, run.swarm_id, task_id)
return {"released": False, "reason": "cross_run_denied"}
yield_msg = competition_mod.TaskYield(
task_id=task_id, agent_id=agent_id,
release_with_reason=message.get("reason", message.get("release_with_reason", "")),
recommend_agent=message.get("recommend_agent"),
)
released = await task_queue.release_task(task_id, agent_id)
_, payload = competition_mod.yielded_event(yield_msg)
run.metadata.setdefault("yields", [])
run.metadata["yields"] = run.metadata["yields"][-49:] + [payload]
await swarm_runtime.save_run(run)
return {"released": bool(released), "task_id": task_id, "recommend_agent": yield_msg.recommend_agent}
async def handle_task_takeover(agent_id: str, message: Dict[str, Any]) -> Dict[str, Any]:
"""P4 (#8): an agent requests takeover of a held task; arbitrate requester vs incumbent.
Requester wins → release from incumbent and assign requester (only when decisive). The
incumbent is weighed on identical terms (a neutral bid carrying its capabilities), so takeover
is never a free steal — it must out-score the holder.
"""
task_id = message.get("task_id")
task = await task_queue.get_task(task_id) if task_id else None
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("run-boundary: rejected cross-run takeover — agent %s ∉ run %s (task %s)",
agent_id, run.swarm_id, task_id)
return {"taken_over": False, "reason": "cross_run_denied"}
incumbent_id = task.assigned_agent_id
requester_bid = competition_mod.TaskBid(
task_id=task_id, agent_id=agent_id,
confidence=message.get("confidence", 0.7),
capabilities=message.get("capabilities") or [],
reason=message.get("reason", "takeover request"),
)
bids = [requester_bid]
if incumbent_id:
incumbent = await agent_registry.get_agent(incumbent_id)
bids.append(competition_mod.TaskBid(
task_id=task_id, agent_id=incumbent_id, confidence=0.5,
capabilities=(incumbent.capabilities if incumbent else []),
))
hist = await _historical_success_map(task.agent_role, [b.agent_id for b in bids])
result = competition_mod.arbitrate(bids, required_capabilities=task.required_capabilities,
historical_success=hist)
taken = False
if result.winner_agent_id == agent_id and result.decisive and incumbent_id != agent_id:
await task_queue.release_task(task_id, incumbent_id)
winner = await agent_registry.get_agent(agent_id)
if winner:
await task_queue.remove_pending_task(task_id)
taken = await finalize_dispatch(winner, task)
_, payload = competition_mod.arbitrated_event(result)
run.metadata.setdefault("arbitrations", [])
run.metadata["arbitrations"] = run.metadata["arbitrations"][-49:] + [payload]
await swarm_runtime.save_run(run)
return {"taken_over": bool(taken), "winner_agent_id": result.winner_agent_id,
"decisive": result.decisive, "task_id": task_id}
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): 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 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
if task_spec.get("task_id")
}
for task_spec in task_specs:
graph_task_id = task_spec.get("task_id")
runtime_task_id = task_id_map.get(graph_task_id, graph_task_id)
runtime_depends_on = [
task_id_map.get(dependency_id, dependency_id)
for dependency_id in (task_spec.get("depends_on") or [])
]
runtime_parent_task_id = task_id_map.get(task_spec.get("parent_task_id"))
runtime_root_task_id = task_id_map.get(
task_spec.get("root_task_id") or graph_task_id,
runtime_task_id,
)
task_context = {
**task_spec.get("context", {}),
"runtime_deployment_id": run.deployment_id,
"swarm_id": run.swarm_id,
"manager_deployment_id": run.manager_deployment_id,
"correlation_id": run.correlation_id,
"task_graph_id": graph_task_id,
"task_title": task_spec.get("title"),
"depends_on": runtime_depends_on,
"agent_role": task_spec.get("agent_role", "general"),
"required_capabilities": task_spec.get("required_capabilities") or [],
"parent_task_id": runtime_parent_task_id,
"root_task_id": runtime_root_task_id,
"source": task_spec.get("source", "runtime_bridge"),
"workflow_mode": task_spec.get("workflow_mode", "single_agent"),
"allow_handoff": task_spec.get("allow_handoff", False),
"model_id": (body.get("orchestration_plan") or {}).get("model_id")
or body.get("model_id"),
}
task = await task_queue.create_task(
task_id=runtime_task_id,
title=task_spec.get("title"),
description=task_spec["description"],
agent_role=task_spec.get("agent_role", "general"),
required_capabilities=task_spec.get("required_capabilities") or [],
depends_on=runtime_depends_on,
parent_task_id=runtime_parent_task_id,
root_task_id=runtime_root_task_id,
source=task_spec.get("source", "runtime_bridge"),
context=task_context,
max_retries=body.get("max_retries") or _default_task_retries(),
)
await swarm_runtime.attach_task(run, task.task_id)
await swarm_runtime.emit_event(
run,
"task.created",
task_id=task.task_id,
payload=swarm_runtime.task_event_payload(task, task_spec),
)
TASKS_CREATED.inc()
created_count += 1
if created_count:
await swarm_runtime.emit_event(run, "timeline.updated", payload={
"summary": f"Created {created_count} task(s)",
"task_count": created_count,
})
return created_count
def task_payload(task) -> Dict[str, Any]:
"""Return the common task payload shape used by task APIs."""
return {
"task_id": task.task_id,
"title": task.title or task.context.get("task_title") or task.description[:80],
"description": task.description,
"status": task.status.value,
"agent_role": task.agent_role,
"required_capabilities": task.required_capabilities,
"depends_on": task.depends_on,
"parent_task_id": task.parent_task_id,
"root_task_id": task.root_task_id,
"source": task.source,
"task_graph_id": task.context.get("task_graph_id"),
"assigned_agent_id": task.assigned_agent_id,
"created_at": task.created_at,
"started_at": task.started_at,
"completed_at": task.completed_at,
"attempt": task.retry_count,
"max_retries": task.max_retries,
"retry_count": task.retry_count,
"blocked_reason": task.blocked_reason,
"context": task.context,
}
def handoff_workflow_enabled(task) -> bool:
"""Return whether a task should use the DAG-style handoff workflow."""
return (
os.getenv("ENABLE_SUBTASK_HANDOFF", "false").lower() in {"1", "true", "yes"}
and task.context.get("workflow_mode") == "multi_agent"
)
def summarize_task_result(result: Any) -> str:
"""Return a user-facing summary for task completion events."""
if isinstance(result, dict):
summary = result.get("summary")
if isinstance(summary, str) and summary.strip():
return summary.strip()
subtasks = result.get("subtasks") or []
for subtask_result in reversed(subtasks):
summary = subtask_result.get("summary")
if isinstance(summary, str) and summary.strip():
return summary.strip()
if isinstance(result, str) and result.strip():
return result.strip()[:500]
return "Task completed"
def parse_task_result(task) -> Optional[Dict[str, Any]]:
"""Parse a task result into a structured dictionary when possible."""
if not getattr(task, "result", None):
return None
if isinstance(task.result, dict):
return task.result
if isinstance(task.result, str):
try:
parsed = json.loads(task.result)
return parsed if isinstance(parsed, dict) else {"summary": str(parsed)}
except json.JSONDecodeError:
return {"summary": task.result}
return None
def build_deliverable_fact(
result: Optional[Dict[str, Any]],
artifact: Optional[Dict[str, Any]] = None,
) -> Dict[str, Any]:
"""Build the structured deliverable fact consumed by Manager."""
result = result or {}
files_modified = list(result.get("files_modified") or [])
files_deleted = list(result.get("files_deleted") or [])
has_branch = bool(result.get("git_branch"))
commit_sha = result.get("commit_sha") or ""
artifact_ids = [artifact["artifact_id"]] if artifact else []
has_deliverable = bool(artifact_ids or has_branch or files_modified or files_deleted)
summary_only = not bool(has_branch or files_modified or files_deleted)
return {
"has_deliverable": has_deliverable,
"summary_only": summary_only,
"artifact_ids": artifact_ids,
"files_modified": files_modified,
"files_deleted": files_deleted,
"has_diff": bool(has_branch or commit_sha or files_modified or files_deleted),
"commit_sha": commit_sha,
}
def build_run_deliverable(run, tasks: List[Any]) -> Dict[str, Any]:
"""Aggregate deliverable facts across all runtime tasks."""
artifact_ids: List[str] = []
files_modified: List[str] = []
files_deleted: List[str] = []
commit_sha = ""
has_deliverable = False
summary_only = True
for task in tasks:
parsed_result = parse_task_result(task)
artifact = build_result_artifact(run, task, parsed_result) if parsed_result else None
fact = build_deliverable_fact(parsed_result, artifact)
artifact_ids.extend(fact["artifact_ids"])
files_modified.extend(fact["files_modified"])
files_deleted.extend(fact["files_deleted"])
has_deliverable = has_deliverable or fact["has_deliverable"]
summary_only = summary_only and fact["summary_only"]
if fact["commit_sha"] and not commit_sha:
commit_sha = fact["commit_sha"]
return {
"has_deliverable": has_deliverable,
"summary_only": summary_only,
"artifact_ids": artifact_ids,
"files_modified": files_modified,
"files_deleted": files_deleted,
"has_diff": bool(commit_sha or files_modified or files_deleted),
"commit_sha": commit_sha,
}
def infer_tool_count(result: Optional[Dict[str, Any]]) -> int:
"""Infer tool usage count from a task result payload."""
if not result:
return 0
if isinstance(result.get("tool_count"), int):
return result["tool_count"]
tool_calls = result.get("tool_calls")
if isinstance(tool_calls, list):
return len(tool_calls)
return 0
def collect_run_artifacts(run, tasks: List[Any]) -> List[Dict[str, Any]]:
"""Collect stable artifact descriptors from runtime tasks."""
artifacts = []
for task in tasks:
parsed_result = parse_task_result(task)
artifact = build_result_artifact(run, task, parsed_result) if parsed_result else None
if artifact:
artifacts.append(artifact)
return artifacts
def build_workflow_phases(run, tasks: List[Any], events: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
"""Build the fixed workflow phase view for the swarm runtime."""
agents_by_role: Dict[str, List[Dict[str, Any]]] = defaultdict(list)
for task in tasks:
parsed_result = parse_task_result(task)
artifact = build_result_artifact(run, task, parsed_result) if parsed_result else None
usage = (parsed_result or {}).get("usage") or {}
elapsed_seconds = 0
if task.started_at and task.completed_at and task.completed_at >= task.started_at:
elapsed_seconds = int(task.completed_at - task.started_at)
agents_by_role[task.agent_role].append({
"agent_id": task.task_id,
"name": task.title or task.task_id,
"role": task.agent_role,
"status": task.status.value,
"tokens": usage.get("model_tokens", 0),
"tools": infer_tool_count(parsed_result),
"elapsed_seconds": elapsed_seconds,
"artifact_ids": [artifact["artifact_id"]] if artifact else [],
})
handoff_seen = any(event.get("event_type", "").startswith("handoff.") for event in events)
approvals = list(run.approvals.values())
task_statuses = {task.status.value for task in tasks}
terminal_statuses = {"completed", "failed", "cancelled"}
def phase_status(name: str) -> str:
if name == "Plan":
return "completed"
if name == "Dispatch":
if not tasks:
return "pending"
if any(status in {"assigned", "in_progress", "blocked", "completed", "failed", "cancelled"} for status in task_statuses):
return "completed"
return "running"
if name == "Execute":
if "failed" in task_statuses or run.status == "failed":
return "failed"
if task_statuses and task_statuses.issubset(terminal_statuses):
return "completed"
if any(status in {"assigned", "in_progress", "blocked"} for status in task_statuses):
return "running"
return "pending"
if name == "Handoff":
if not handoff_seen:
return "pending"
if any(event.get("event_type") == "handoff.completed" for event in events):
return "completed"
return "running"
if name == "Review":
if run.status == "waiting_approval":
return "running"
if run.status == "blocked":
return "failed"
if approvals:
return "completed"
return "pending"
if name == "Deliver":
if run.status == "completed":
return "completed"
if run.status in {"failed", "stopped", "blocked"}:
return "failed" if run.status == "failed" else run.status
if run.status == "running":
return "running"
return "pending"
return "pending"
phase_agents = {
"Plan": [],
"Dispatch": [],
"Execute": [agent for group in agents_by_role.values() for agent in group],
"Handoff": [agent for group in agents_by_role.values() for agent in group if agent["status"] == "blocked"] if handoff_seen else [],
"Review": [],
"Deliver": [],
}
return [
{
"phase_id": phase_name.lower(),
"name": phase_name,
"status": phase_status(phase_name),
"agents": phase_agents.get(phase_name, []),
}
for phase_name in ["Plan", "Dispatch", "Execute", "Handoff", "Review", "Deliver"]
]
def build_result_artifact(run, task, result: Any) -> Optional[Dict[str, Any]]:
"""Build a callback-safe artifact for completed tasks."""
if not isinstance(result, dict):
return None
summary = summarize_task_result(result)
base_metadata = {
"redacted": True,
"agent_role": task.agent_role,
"runtime_deployment_id": run.deployment_id,
"manager_deployment_id": run.manager_deployment_id,
"task_id": task.task_id,
"parent_task_id": task.parent_task_id,
"root_task_id": task.root_task_id,
}
if result.get("git_branch"):
return {
"artifact_id": f"art_{task.task_id}",
"artifact_type": "code_patch",
"title": task.title or "Agent task result branch",
"summary": summary,
"uri": f"git://repo#{result.get('git_branch')}",
"checksum": result.get("commit_sha"),
"metadata": {
**base_metadata,
"commit_sha": result.get("commit_sha"),
"git_branch": result.get("git_branch"),
},
}
files_modified = result.get("files_modified") or []
files_deleted = result.get("files_deleted") or []
artifact_type = "deployment_manifest" if files_modified or files_deleted else "document"
return {
"artifact_id": f"art_{task.task_id}_summary",
"artifact_type": artifact_type,
"title": task.title or task.context.get("task_title") or "Task execution result",
"summary": summary,
"uri": f"runtime://{run.swarm_id}/artifacts/{task.task_id}",
"metadata": {
**base_metadata,
"files_modified": files_modified,
"files_deleted": files_deleted,
"changes": result.get("changes"),
"git_skipped": result.get("git_skipped"),
"git_error": result.get("git_error"),
},
}
async def emit_task_completion_events(run, task, agent_id: str, result: Any):
"""Emit the completion callback set expected by Agent Manager."""
summary = summarize_task_result(result)
artifact = build_result_artifact(run, task, result)
deliverable = build_deliverable_fact(result, artifact)
payload = {
"task_id": task.task_id,
"agent_role": task.agent_role,
"agent_id": agent_id,
"status": "completed",
"summary": summary,
"parent_task_id": task.parent_task_id,
"root_task_id": task.root_task_id,
"deliverable": deliverable,
}
await swarm_runtime.emit_event(
run,
"task.completed",
task_id=task.task_id,
agent_instance_id=agent_id,
payload=payload,
)
if artifact:
await swarm_runtime.emit_event(
run,
"artifact.created",
task_id=task.task_id,
agent_instance_id=agent_id,
artifact=artifact,
)
await swarm_runtime.emit_event(
run,
"timeline.updated",
task_id=task.task_id,
agent_instance_id=agent_id,
payload={
"summary": summary,
"task_id": task.task_id,
"status": "completed",
},
)
async def finalize_parent_after_child(run, child_task, agent_id: str, success: bool, summary: str):
"""Resolve a blocked parent task after a delegated child reaches a terminal state."""
parent_task_id = child_task.parent_task_id
if not parent_task_id:
return
parent_task = await task_queue.get_task(parent_task_id)
if not parent_task or parent_task.status != TaskStatus.BLOCKED:
return
payload = {
"task_id": parent_task.task_id,
"agent_role": parent_task.agent_role,
"delegated_child_task_id": child_task.task_id,
"status": "completed" if success else "failed",
"summary": summary if success else None,
"reason": None if success else summary,
}
if success:
await task_queue.complete_task(
parent_task.task_id,
json.dumps(
{
"delegated_to": child_task.task_id,
"result": child_task.result,
"summary": summary,
}
),
)
await swarm_runtime.emit_event(
run,
"handoff.completed",
task_id=parent_task.task_id,
agent_instance_id=agent_id,
payload={
**payload,
"child_task_id": child_task.task_id,
"parent_task_id": parent_task.task_id,
"from_role": parent_task.agent_role,
"to_role": child_task.agent_role,
},
)
await swarm_runtime.emit_event(
run,
"task.completed",
task_id=parent_task.task_id,
agent_instance_id=agent_id,
payload=payload,
)
else:
parent_task.retry_count = max(parent_task.max_retries - 1, 0)
await task_queue._save_task(parent_task)
await task_queue.fail_task(parent_task.task_id, summary)
await swarm_runtime.emit_event(
run,
"task.failed",
task_id=parent_task.task_id,
agent_instance_id=agent_id,
payload=payload,
)
await swarm_runtime.emit_event(
run,
"timeline.updated",
task_id=parent_task.task_id,
agent_instance_id=agent_id,
payload={
"summary": (
f"Delegated parent task {parent_task.task_id} completed"
if success else f"Delegated parent task {parent_task.task_id} failed"
),
"task_id": parent_task.task_id,
"child_task_id": child_task.task_id,
"status": "completed" if success else "failed",
},
)
def _launch_note(backend: str, planned: int, launched: int, model_key_resolved: bool) -> str:
"""Human-readable explanation of a launch outcome (agent_swarm#56).
Recorded on the run + surfaced in /result and /diagnostics so an operator immediately sees WHY
a run has no expert agents (the #56 black box), instead of a silent run stuck at `running`.
"""
if backend == "none":
return ("AGENT_LAUNCH_BACKEND=none — the runtime did NOT auto-launch agents. Either supply "
"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 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, "
f"agent_swarm#16 / #60 A.3).")
if not model_key_resolved:
note += (" Also: model key did NOT resolve from billing_context.secret_ref — launched "
"agents would start keyless (#60 A.3).")
return note
note = f"backend={backend}: launched {launched}/{planned} agent(s)."
if not model_key_resolved:
note += (" WARNING: model key did NOT resolve from billing_context.secret_ref — agents start "
"keyless and will error on model calls (#60 A.3).")
return note
async def launch_swarm_agents(run, body: Dict[str, Any]) -> None:
"""Swarm-owned agent launch (agent_swarm#16): launch the per-user expert pool for this run,
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.
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.
"""
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()
model_key = agent_launcher.resolve_model_key(body)
git_env = agent_launcher.resolve_git_grant(body)
info: Dict[str, Any] = {
"backend": backend, "planned": 0, "launched": 0, "launched_ids": [],
"model_key_resolved": bool(model_key),
"git_repo_resolved": bool(git_env and git_env.get("GIT_REPO_URL")),
"git_creds_resolved": bool(git_env and git_env.get("GIT_PASSWORD")),
"note": "",
}
try:
specs = agent_launcher.plan_launch_specs(
run, body,
connected_user_agents=connected,
limit=max_agents_per_user(body),
pool_size=agent_launcher.desired_pool_size(),
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)
info["launched"] = len(launched)
info["launched_ids"] = launched
if launched:
run.metadata["launched_agents"] = launched
info["note"] = _launch_note(backend, len(specs), len(launched), bool(model_key))
except Exception as exc: # never fail run creation on launch
logger.warning("launch_swarm_agents failed for run %s: %s", run.swarm_id, exc)
info["note"] = f"launch raised: {exc}"
run.metadata["agent_launch"] = info
await swarm_runtime.save_run(run)
async def create_swarm_run_from_request(
request: Request,
body: Dict[str, Any],
x_correlation_id: Optional[str],
x_idempotency_key: Optional[str],
):
"""Shared implementation for /api/swarms and /api/agnet/deployments."""
auth_error = await require_runtime_auth(request)
if auth_error:
return auth_error
body = swarm_runtime.normalize_create_request(body)
correlation_id = x_correlation_id or (body.get("metadata") or {}).get("correlation_id")
try:
swarm_runtime.validate_create_request(body)
except RuntimeValidationError as exc:
return error_response(422, exc.code, exc.message, correlation_id)
body.setdefault("metadata", {})
body["metadata"]["runtime_headers"] = request_context_headers(request)
run, created = await swarm_runtime.get_or_create_run(
body=body,
idempotency_key=x_idempotency_key,
correlation_id=correlation_id,
)
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"),
# benchmark: emit_patch must diff against the exact base_commit, not default HEAD
"base_commit": _meta.get("base_commit") or _git_grant.get("base_commit"),
}
await swarm_runtime.save_run(run)
return {
"success": True,
"data": {
"deployment_id": run.deployment_id,
"runtime_deployment_id": run.deployment_id,
"manager_deployment_id": run.manager_deployment_id,
"swarm_id": run.swarm_id,
"mode": run.mode,
"status": run.status,
"runtime_execution_status": run.status,
"created": created,
},
}
async def stop_swarm_run(deployment_or_swarm_id: str, reason: str):
"""Stop a run and cancel non-terminal tasks."""
run = await swarm_runtime.stop_run(deployment_or_swarm_id, reason)
if not run:
raise HTTPException(status_code=404, detail="Runtime deployment not found")
for task_id in run.task_ids:
# Ask the owning agent to abort the in-flight task before we mark it cancelled,
# so the merged agent runtime tears down its execution coroutine promptly.
task = await task_queue.get_task(task_id)
if task and task.assigned_agent_id and task.status in {TaskStatus.ASSIGNED, TaskStatus.IN_PROGRESS}:
await manager.send_message(task.assigned_agent_id, {
"type": "cancel_task",
"task_id": task_id,
})
await task_queue.cancel_task(task_id, reason)
await swarm_runtime.emit_event(run, "timeline.updated", payload={
"summary": "Swarm stopped by Manager",
"reason": reason,
})
# Tear down any Swarm-launched agents for this run (best-effort; subprocess backend only —
# command/k8s backends are torn down by the deployment). agent_swarm#16.
try:
await agent_launcher.stop_launched(run.swarm_id)
except Exception as exc:
logger.warning("stop_launched failed for run %s: %s", run.swarm_id, exc)
return run
async def maybe_emit_budget_alert(run):
"""Emit a simple duration-based budget alert once per run."""
budget = ((run.request_body.get("orchestration_plan") or {}).get("budget") or {})
duration_budget = budget.get("duration_seconds") or budget.get("max_duration_seconds")
if not duration_budget or run.metadata.get("budget_duration_alerted"):
return
elapsed = time.time() - run.created_at
try:
ratio = elapsed / float(duration_budget)
except (TypeError, ValueError, ZeroDivisionError):
return
if ratio < 0.8:
return
run.metadata["budget_duration_alerted"] = True
await swarm_runtime.save_run(run)
await swarm_runtime.emit_event(run, "budget.alert", payload={
"budget_type": "duration",
"threshold": 0.8,
"ratio": ratio,
"elapsed_seconds": elapsed,
"limit_seconds": duration_budget,
})
async def emit_usage_event(run, task, agent_id: str, result: Any):
"""Emit model/runtime usage when the agent returned provider usage."""
if not isinstance(result, dict):
return
usage = result.get("usage")
if not isinstance(usage, dict):
return
plan = run.request_body.get("orchestration_plan") or {}
budget = plan.get("budget") or {}
# Review-retry cost attribution (#16): a task's usage is `review_retry` once cross-review has
# reopened it for rework (it appears in run.metadata["rework_attributions"]); otherwise
# `initial`. The first execution emits before any attribution exists → "initial"; each redo
# emits after run_cross_review recorded the attribution → "review_retry". So billing can split
# initial vs review-retry cost by summing usage events per cost_phase (usage-billing §5).
rework_targets = {
a.get("target_task_id") for a in (run.metadata.get("rework_attributions") or [])
}
cost_phase = "review_retry" if task.task_id in rework_targets else "initial"
payload = {
"model_id": usage.get("model_id") or task.context.get("model_id"),
"model_tokens": usage.get("model_tokens", 0),
"prompt_tokens": usage.get("prompt_tokens", 0),
"completion_tokens": usage.get("completion_tokens", 0),
"model_cost_usd": usage.get("model_cost_usd", 0),
"runtime_seconds": usage.get("runtime_seconds", 0),
"billing_source": usage.get("billing_source", "unknown"),
"cost_phase": cost_phase, # "initial" | "review_retry" (#16 attribution)
"attempt": task.retry_count, # redo count for this task
"manager_deployment_id": run.manager_deployment_id,
"swarm_id": run.swarm_id,
"task_id": task.task_id,
"agent_role": task.agent_role,
"correlation_id": run.correlation_id,
"budget": {
"max_tokens": budget.get("max_tokens") or budget.get("token_limit"),
"max_cost_usd": budget.get("max_cost_usd"),
"consumed_usd": usage.get("model_cost_usd", 0),
"remaining_usd": (
budget.get("max_cost_usd") - usage.get("model_cost_usd", 0)
if isinstance(budget.get("max_cost_usd"), (int, float))
else None
),
},
}
await swarm_runtime.emit_event(
run,
"budget.alert",
task_id=task.task_id,
agent_instance_id=agent_id,
payload=payload,
)
await swarm_runtime.emit_event(
run,
"timeline.updated",
task_id=task.task_id,
agent_instance_id=agent_id,
payload={
"summary": "Usage updated",
"task_id": task.task_id,
"model_id": payload["model_id"],
"model_tokens": payload["model_tokens"],
"model_cost_usd": payload["model_cost_usd"],
"runtime_seconds": payload["runtime_seconds"],
},
)
@app.exception_handler(HTTPException)
async def runtime_http_exception_handler(request: Request, exc: HTTPException):
"""Use Manager error envelopes for runtime API errors."""
if request.url.path.startswith("/api/"):
code = "NOT_FOUND" if exc.status_code == 404 else "HTTP_ERROR"
return error_response(
exc.status_code,
code,
str(exc.detail),
request.headers.get("x-correlation-id"),
)
return JSONResponse(status_code=exc.status_code, content={"detail": exc.detail})
# REST API endpoints
@app.get("/")
async def root():
"""Health check endpoint."""
return {"status": "ok", "service": "swarm-orchestrator"}
@app.get("/health")
async def health():
"""Detailed health check."""
try:
await redis_client.client.ping()
redis_status = "connected"
except Exception:
redis_status = "disconnected"
return {
"status": "ok",
"redis": redis_status,
"active_connections": len(manager.active_connections)
}
async def get_runtime_run_or_404(request: Request, deployment_id: str):
"""Load a runtime run after validating service auth."""
auth_error = await require_runtime_auth(request)
if auth_error:
return None, auth_error
run = await swarm_runtime.get_run_by_identifier(deployment_id)
if not run:
raise HTTPException(status_code=404, detail="Swarm run not found")
return run, None
async def load_runtime_tasks(run) -> List[Any]:
"""Load all known task records for a swarm run."""
tasks = [await task_queue.get_task(task_id) for task_id in run.task_ids]
return [task for task in tasks if task]
def runtime_deployment_response(run) -> Dict[str, Any]:
"""Return the standard runtime deployment envelope."""
return {
"deployment_id": run.deployment_id,
"runtime_deployment_id": run.deployment_id,
"manager_deployment_id": run.manager_deployment_id,
"swarm_id": run.swarm_id,
"mode": run.mode,
"objective": run.objective,
"status": run.status,
"runtime_execution_status": run.status,
"created_at": run.created_at,
"updated_at": run.updated_at,
"task_count": len(run.task_ids),
"approvals": list(run.approvals.values()),
}
async def build_runtime_workflow(run) -> Dict[str, Any]:
"""Build the workflow-oriented runtime response consumed by Manager."""
tasks = await load_runtime_tasks(run)
events_data = await swarm_runtime.list_events(run.swarm_id, limit=500)
events = events_data["events"]
artifacts = collect_run_artifacts(run, tasks)
duration_seconds = int(time.time() - run.created_at)
tokens = 0
tools = 0
for task in tasks:
parsed = parse_task_result(task) or {}
usage = parsed.get("usage") or {}
tokens += int(usage.get("model_tokens", 0) or 0)
tools += infer_tool_count(parsed)
summary = run.objective
for event in reversed(events):
if event.get("event_type") == "timeline.updated":
summary = (event.get("payload") or {}).get("summary") or summary
if summary:
break
return {
"workflow_id": run.swarm_id,
"deployment_id": run.deployment_id,
"runtime_deployment_id": run.deployment_id,
"manager_deployment_id": run.manager_deployment_id,
"task_id": run.manager_deployment_id or run.deployment_id,
"title": run.objective,
"mode": run.mode,
"status": run.status,
"runtime_execution_status": run.status,
"summary": summary,
"agent_count": len(tasks),
"tokens": tokens,
"tools": tools,
"elapsed_seconds": duration_seconds,
"phases": build_workflow_phases(run, tasks, events),
"artifacts": artifacts,
}
async def build_runtime_metrics(run, window: str, step: str) -> Dict[str, Any]:
"""Build metrics for the current swarm run."""
tasks = await load_runtime_tasks(run)
agents = await agent_registry.get_all_agents()
status_counts: Dict[str, int] = {}
for task in tasks:
status_counts[task.status.value] = status_counts.get(task.status.value, 0) + 1
completed_durations = [
task.completed_at - task.started_at
for task in tasks
if task.completed_at and task.started_at and task.completed_at >= task.started_at
]
duration_seconds = time.time() - run.created_at
budget = ((run.request_body.get("orchestration_plan") or {}).get("budget") or {})
duration_budget = budget.get("duration_seconds") or budget.get("max_duration_seconds")
budget_ratio = duration_seconds / float(duration_budget) if duration_budget else None
# Run-level cost split by phase (#16/#37): lets HM's detail usage aggregation show
# initial vs review_retry without deriving from the event stream. Same attribution as
# emit_usage_event — a task counts as review_retry once cross-review recorded it in
# run.metadata["rework_attributions"].
rework_targets = {a.get("target_task_id") for a in (run.metadata.get("rework_attributions") or [])}
cost_by_phase = {"initial": {"cost_usd": 0.0, "model_tokens": 0},
"review_retry": {"cost_usd": 0.0, "model_tokens": 0}}
for task in tasks:
usage = (parse_task_result(task) or {}).get("usage") or {}
phase = "review_retry" if task.task_id in rework_targets else "initial"
cost_by_phase[phase]["cost_usd"] += float(usage.get("model_cost_usd") or 0.0)
cost_by_phase[phase]["model_tokens"] += int(usage.get("model_tokens") or 0)
return {
"deployment_id": run.deployment_id,
"swarm_id": run.swarm_id,
"mode": run.mode,
"window": window,
"step": step,
"status": run.status,
"runtime_execution_status": run.status,
"runtime_duration_seconds": duration_seconds,
"tasks_total": len(tasks),
"tasks_by_status": status_counts,
"agents_connected": len(manager.active_connections),
"agents_registered": len(agents),
"average_task_duration_seconds": (
sum(completed_durations) / len(completed_durations)
if completed_durations else 0
),
"cost_by_phase": cost_by_phase, # {initial,review_retry}:{cost_usd,model_tokens} (#16/#37)
"budget": {
"duration_seconds": duration_budget,
"duration_ratio": budget_ratio,
},
}
async def build_runtime_diagnostics(run) -> Dict[str, Any]:
"""Build the diagnostics response for a swarm run."""
tasks = await load_runtime_tasks(run)
callback_attempts = list(run.metadata.get("callback_attempts") or [])
failures = []
for task in tasks:
if task.status.value in {"failed", "blocked"}:
failures.append({
"task_id": task.task_id,
"status": task.status.value,
"blocked_reason": task.blocked_reason,
"result": parse_task_result(task),
})
return {
"deployment_id": run.deployment_id,
"runtime_deployment_id": run.deployment_id,
"manager_deployment_id": run.manager_deployment_id,
"swarm_id": run.swarm_id,
"mode": run.mode,
"status": run.status,
"runtime_execution_status": run.status,
"correlation_id": run.correlation_id,
"request_headers": run.metadata.get("runtime_headers") or {},
"callback": {
"url": run.callback.url,
"subscribed_events": run.callback.subscribed_events,
"attempts": callback_attempts,
},
"approvals": list(run.approvals.values()),
# P-guard health + agent-launch outcome: why a run can't make progress / has no expert agents
# (agent_swarm#56). `agent_launch` carries no secret (only a model_key_resolved bool).
"health": run.metadata.get("health"),
"agent_launch": run.metadata.get("agent_launch"),
"error_context": {
"task_failures": failures,
"callback_failures": [item for item in callback_attempts if item.get("status") == "failed"],
},
"request_body": run.request_body,
}
@app.get("/api/agent/health")
@app.get("/api/agnet/health")
async def agnet_health():
"""Agent Manager compatible health endpoint."""
return await swarm_runtime.health()
@app.post("/api/agent/swarm/deployments")
@app.post("/api/swarms")
@app.post("/api/agnet/deployments")
async def create_swarm(
request: Request,
x_correlation_id: Optional[str] = Header(default=None),
x_idempotency_key: Optional[str] = Header(default=None),
):
"""Create a swarm runtime deployment."""
body = await request.json()
return await create_swarm_run_from_request(
request,
body,
x_correlation_id,
x_idempotency_key,
)
@app.get("/api/agent/swarm/deployments/{deployment_id}")
@app.get("/api/swarms/{deployment_id}")
@app.get("/api/agnet/deployments/{deployment_id}")
async def get_swarm(deployment_id: str, request: Request):
"""Return a runtime deployment summary."""
run, auth_error = await get_runtime_run_or_404(request, deployment_id)
if auth_error:
return auth_error
return {"success": True, "data": runtime_deployment_response(run)}
@app.get("/api/agent/swarm/deployments/{deployment_id}/tasks")
@app.get("/api/swarms/{deployment_id}/tasks")
@app.get("/api/agnet/deployments/{deployment_id}/tasks")
async def get_swarm_tasks(deployment_id: str, request: Request):
"""Return the runtime task graph for a swarm deployment."""
run, auth_error = await get_runtime_run_or_404(request, deployment_id)
if auth_error:
return auth_error
tasks = [task_payload(task) for task in await load_runtime_tasks(run)]
return {
"success": True,
"data": {
"deployment_id": run.deployment_id,
"swarm_id": run.swarm_id,
"tasks": tasks,
},
}
@app.get("/api/agent/swarm/deployments/{deployment_id}/logs")
@app.get("/api/swarms/{deployment_id}/logs")
@app.get("/api/agnet/deployments/{deployment_id}/logs")
async def get_swarm_logs(
deployment_id: str,
request: Request,
limit: int = 100,
cursor: Optional[str] = None,
):
"""Return callback/timeline events as the runtime log stream."""
run, auth_error = await get_runtime_run_or_404(request, deployment_id)
if auth_error:
return auth_error
logs = await swarm_runtime.list_events(run.swarm_id, limit=limit, cursor=cursor)
return {"success": True, "data": logs}
@app.get("/api/agent/swarm/deployments/{deployment_id}/events")
@app.get("/api/swarms/{deployment_id}/events")
@app.get("/api/agnet/deployments/{deployment_id}/events")
async def get_swarm_events(
deployment_id: str,
request: Request,
limit: int = 100,
cursor: Optional[str] = None,
):
"""Return raw runtime events for a swarm deployment."""
run, auth_error = await get_runtime_run_or_404(request, deployment_id)
if auth_error:
return auth_error
events = await swarm_runtime.list_events(run.swarm_id, limit=limit, cursor=cursor)
return {"success": True, "data": events}
@app.get("/api/agent/swarm/deployments/{deployment_id}/events/stream")
@app.get("/api/swarms/{deployment_id}/events/stream")
@app.get("/api/agnet/deployments/{deployment_id}/events/stream")
async def stream_swarm_events(
deployment_id: str,
request: Request,
after: int = 0,
):
"""SSE real-time event stream for a swarm run (agent_swarm#51).
The realtime overlay over `GET /events?after=<sequence>`: on connect it replays the events
after `after`, then holds the connection and pushes new events as they are emitted. Reuses the
existing `swarm_events:{swarm_id}` store and per-swarm `sequence` (no new storage, no schema
change). Each frame's `id:` is the event `sequence`, so SSE and polling share one cursor space
— a dropped SSE connection can fall back to `/events?after=<last id>` with no gap/dup.
Auth is `require_runtime_auth` (service token): the caller is **HM** (which reverse-proxies to
the client EventSource, heicode-mananger#46), never the client directly. Events are already
redacted at emit time, so frames are streamed as-is. The stream closes after a terminal event
(`swarm.completed` / `swarm.failed` / `swarm.stopped`).
"""
run, auth_error = await get_runtime_run_or_404(request, deployment_id)
if auth_error:
return auth_error
# Last-Event-ID (sent by the browser/HM on reconnect) takes precedence over the `after` query
# (#51 point 4): identical cursor semantics — resume from the event after the last one seen.
start_index = after
last_event_id = request.headers.get("Last-Event-ID")
if last_event_id:
try:
start_index = int(last_event_id)
except (TypeError, ValueError):
pass
start_index = max(0, start_index)
swarm_id = run.swarm_id
async def event_generator():
# `cursor` is the Redis list index to read from. Event with sequence S sits at index S-1,
# so events *after* sequence N start at index N — i.e. cursor == "last sequence seen".
cursor = start_index
# Hint the EventSource how long to wait before reconnecting after a drop.
yield "retry: 3000\n\n"
heartbeat_seconds = 15
poll_seconds = 1.0
last_ping = time.monotonic()
while True:
if await request.is_disconnected():
return
page = await swarm_runtime.list_events(swarm_id, limit=500, cursor=str(cursor))
events = page.get("events", [])
if events:
for event in events:
sequence = event.get("sequence")
data = json.dumps(event, ensure_ascii=False, separators=(",", ":"))
# `data:` is byte-identical to the /events envelope; `event: message` matches the
# client EventSource default channel (Mem0ried points 1–2).
yield f"id: {sequence}\nevent: message\ndata: {data}\n\n"
cursor = sequence if isinstance(sequence, int) else cursor + 1
if event.get("event_type") in TERMINAL_CLIENT_EVENT_TYPES:
# Terminal frame is flushed above; close normally so the client stops
# reconnecting (#51 point 6 / Mem0ried point 3).
return
last_ping = time.monotonic()
# Drain any remaining backlog immediately (no sleep) before holding the connection.
continue
# No new events: heartbeat to keep the long-lived connection alive through the nginx
# ingress + HM reverse-proxy (#51 point 5), then poll again. EventSource ignores
# comment frames, so no `data:` is needed.
now = time.monotonic()
if now - last_ping >= heartbeat_seconds:
yield ": ping\n\n"
last_ping = now
await asyncio.sleep(poll_seconds)
return StreamingResponse(
event_generator(),
media_type="text/event-stream",
headers={
"Cache-Control": "no-cache",
"Connection": "keep-alive",
# Disable nginx proxy buffering so frames flush immediately (#51 point 5).
"X-Accel-Buffering": "no",
},
)
@app.get("/api/agent/swarm/deployments/{deployment_id}/metrics")
@app.get("/api/swarms/{deployment_id}/metrics")
@app.get("/api/agnet/deployments/{deployment_id}/metrics")
async def get_swarm_metrics(
deployment_id: str,
request: Request,
window: str = "15m",
step: str = "60s",
):
"""Return lightweight run/task/agent metrics for Manager UI."""
run, auth_error = await get_runtime_run_or_404(request, deployment_id)
if auth_error:
return auth_error
return {"success": True, "data": await build_runtime_metrics(run, window, step)}
@app.get("/api/agent/swarm/deployments/{deployment_id}/workflow")
@app.get("/api/swarms/{deployment_id}/workflow")
@app.get("/api/agnet/deployments/{deployment_id}/workflow")
async def get_swarm_workflow(deployment_id: str, request: Request):
"""Return the workflow-oriented summary of a swarm deployment."""
run, auth_error = await get_runtime_run_or_404(request, deployment_id)
if auth_error:
return auth_error
return {"success": True, "data": await build_runtime_workflow(run)}
@app.get("/api/agent/swarm/deployments/{deployment_id}/diagnostics")
@app.get("/api/swarms/{deployment_id}/diagnostics")
@app.get("/api/agnet/deployments/{deployment_id}/diagnostics")
async def get_swarm_diagnostics(deployment_id: str, request: Request):
"""Return runtime diagnostics for a swarm deployment."""
run, auth_error = await get_runtime_run_or_404(request, deployment_id)
if auth_error:
return auth_error
return {"success": True, "data": await build_runtime_diagnostics(run)}
async def build_audit_trace_for_run(run) -> Dict[str, Any]:
"""Assemble the replayable audit/lineage trail for a run (issue #17).
Pulls the persisted event stream, per-task model/tool facts (from each task's reported
`usage`), and approvals, and hands them to the pure `audit.build_audit_trace`. Returns the
ordered records + a human-readable replay. Carries no prompt content or secrets.
Reads the FULL event stream by following `list_events`' `next_cursor` to exhaustion — the audit
trail must be complete ("每步 trace 可回放", #17), so it must NOT silently stop at the single-page
cap (`list_events` clamps `limit` to 500). A long run (501+ events) would otherwise lose steps.
"""
events: List[Dict[str, Any]] = []
cursor: Optional[str] = None
while True:
page = await swarm_runtime.list_events(run.swarm_id, limit=500, cursor=cursor)
events.extend(page.get("events", []))
cursor = page.get("next_cursor")
if not cursor:
break
task_facts: Dict[str, audit_mod.TaskAuditFacts] = {}
for tid in run.task_ids:
task = await task_queue.get_task(tid)
if not task:
continue
usage = (parse_task_result(task) or {}).get("usage") or {}
task_facts[tid] = audit_mod.TaskAuditFacts(
agent_role=task.agent_role,
assigned_agent_id=getattr(task, "assigned_agent_id", None),
model_id=usage.get("model_id") or (task.context or {}).get("model_id"),
tool_count=infer_tool_count(parse_task_result(task)),
source=task.source,
parent_task_id=task.parent_task_id,
root_task_id=task.root_task_id,
)
lineage = {
"manager_deployment_id": run.manager_deployment_id,
"deployment_id": run.deployment_id,
"swarm_id": run.swarm_id,
"correlation_id": run.correlation_id,
}
records = audit_mod.build_audit_trace(events, task_facts, run.approvals or {}, lineage=lineage)
return {"records": records, "replay": audit_mod.replay(records), "count": len(records)}
@app.get("/api/agent/swarm/deployments/{deployment_id}/audit")
@app.get("/api/swarms/{deployment_id}/audit")
@app.get("/api/agnet/deployments/{deployment_id}/audit")
async def get_swarm_audit(deployment_id: str, request: Request):
"""Return the replayable audit/lineage trace for a swarm deployment (issue #17)."""
run, auth_error = await get_runtime_run_or_404(request, deployment_id)
if auth_error:
return auth_error
return {"success": True, "data": await build_audit_trace_for_run(run)}
async def build_run_result(run) -> Dict[str, Any]:
"""The swarm's user-facing answer for a run: synthesized summary + deliverable + artifacts.
A first-class result object so the client doesn't reconstruct it from /workflow + /artifacts.
Carries only non-content descriptors (summary text + deliverable facts: git branch/commit/
files + artifact uri/checksum) — the actual artifact content lives at its uri (git/runtime).
"""
tasks = await load_runtime_tasks(run)
termination_reason = (run.metadata.get("convergence") or {}).get("termination_reason")
return {
"deployment_id": run.deployment_id,
"swarm_id": run.swarm_id,
"status": run.status,
"runtime_execution_status": run.status,
"summary": run.metadata.get("final_summary"),
"termination_reason": termination_reason,
"deliverable": build_run_deliverable(run, tasks),
"artifacts": collect_run_artifacts(run, tasks),
# Why a run produced nothing, surfaced on the result itself (agent_swarm#56): the P-guard
# health report (e.g. no_agents_connected) + the agent-launch outcome (backend/launched/note).
# Both are non-content descriptors; `agent_launch` carries no secret (only a resolved bool).
"health": run.metadata.get("health"),
"agent_launch": run.metadata.get("agent_launch"),
}
@app.get("/api/agent/swarm/deployments/{deployment_id}/result")
@app.get("/api/swarms/{deployment_id}/result")
@app.get("/api/agnet/deployments/{deployment_id}/result")
async def get_swarm_result(deployment_id: str, request: Request):
"""Return the swarm's result (summary + deliverable + artifacts) for a deployment."""
run, auth_error = await get_runtime_run_or_404(request, deployment_id)
if auth_error:
return auth_error
return {"success": True, "data": await build_run_result(run)}
@app.post("/api/agent/swarm/deployments/{deployment_id}/input")
@app.post("/api/swarms/{deployment_id}/input")
@app.post("/api/agnet/deployments/{deployment_id}/input")
async def append_swarm_input(deployment_id: str, request: Request):
"""Receive a user follow-up prompt and inject it into the running swarm (agent_swarm#40).
The instruction becomes a new `source="user_append"` task in the run's shared pool (agents
self-select it). If the run already finished (completed/failed), it is reopened to `running`
("continue"); a `stopped` run is rejected (start a new run). The instruction text is the task
description (delivered to the claiming agent over WS) but is **redacted from the event stream**
— the `task.created` event carries only a category message, never the raw prompt.
"""
auth_error = await require_runtime_auth(request)
if auth_error:
return auth_error
try:
body = await request.json()
except json.JSONDecodeError:
body = {}
instruction = (body.get("instruction") or body.get("input") or "").strip()
correlation_id = request.headers.get("x-correlation-id")
if not instruction:
return error_response(422, "INVALID_REQUEST", "instruction is required", correlation_id)
run = await swarm_runtime.get_run_by_identifier(deployment_id)
if not run:
raise HTTPException(status_code=404, detail="Swarm run not found")
if run.status == "stopped":
return error_response(409, "RUN_STOPPED",
"run was stopped; start a new run for new input", correlation_id)
reopened = run.status in ("completed", "failed")
n = len([t for t in run.task_ids if "-input-" in t]) + 1
task = await task_queue.create_task(
task_id=f"{run.swarm_id}-input-{n}",
title="用户追加输入",
description=instruction, # delivered to the agent; NOT echoed to events
agent_role="general",
required_capabilities=[],
root_task_id="seed",
source="user_append",
context={
"swarm_id": run.swarm_id,
"runtime_deployment_id": run.deployment_id,
"manager_deployment_id": run.manager_deployment_id,
"correlation_id": run.correlation_id,
"source": "user_append",
"agent_role": "general",
"workflow_mode": "swarm",
"allow_handoff": True,
},
enqueue=True,
)
await swarm_runtime.attach_task(run, task.task_id)
if reopened:
run.status = "running"
await swarm_runtime.save_run(run)
await swarm_runtime.emit_event(run, "deployment.status_changed",
payload=swarm_runtime.status_payload(run, phase="Execute", reason="user appended input"))
# Redacted task.created — the raw instruction is NOT in the event (security-boundary).
await swarm_runtime.emit_event(run, "task.created", task_id=task.task_id, payload={
"task_id": task.task_id,
"title": "用户追加输入",
"source": "user_append",
"agent_role": "general",
"message": "用户追加输入已注入运行中的蜂群",
})
return {"success": True, "data": {
"deployment_id": run.deployment_id,
"swarm_id": run.swarm_id,
"task_id": task.task_id,
"status": run.status,
"reopened": reopened,
}}
@app.post("/api/agent/swarm/deployments/{deployment_id}/stop")
@app.post("/api/swarms/{deployment_id}/stop")
@app.post("/api/agnet/deployments/{deployment_id}/stop")
async def stop_swarm(
deployment_id: str,
request: Request,
stop_request: Optional[StopDeploymentRequest] = None,
):
"""Stop a runtime deployment."""
auth_error = await require_runtime_auth(request)
if auth_error:
return auth_error
try:
body = await request.json() if request.method == "POST" else {}
except json.JSONDecodeError:
body = {}
reason = (
(stop_request.reason if stop_request else None)
or body.get("reason")
or "Heicode Manager requested stop"
)
run = await stop_swarm_run(deployment_id, reason)
return {
"success": True,
"data": {
"deployment_id": run.deployment_id,
"runtime_deployment_id": run.deployment_id,
"manager_deployment_id": run.manager_deployment_id,
"swarm_id": run.swarm_id,
"status": run.status,
"runtime_execution_status": run.status,
},
}
@app.post("/api/agent/swarm/deployments/{deployment_id}/approvals/{approval_id}")
@app.post("/api/swarms/{deployment_id}/approvals/{approval_id}")
@app.post("/api/agnet/deployments/{deployment_id}/approvals/{approval_id}")
async def decide_swarm_approval(
deployment_id: str,
approval_id: str,
request: Request,
):
"""Receive an approval decision from Manager."""
auth_error = await require_runtime_auth(request)
if auth_error:
return auth_error
decision = await request.json()
run = await swarm_runtime.record_approval_decision(deployment_id, approval_id, decision)
if not run:
raise HTTPException(status_code=404, detail="Swarm run not found")
if run.status == "running" and not run.task_ids:
await create_tasks_for_run(run, run.request_body or {
"orchestration_plan": {"objective": run.objective},
})
return {
"success": True,
"data": {
"deployment_id": run.deployment_id,
"runtime_deployment_id": run.deployment_id,
"manager_deployment_id": run.manager_deployment_id,
"swarm_id": run.swarm_id,
"approval_id": approval_id,
"status": run.status,
"runtime_execution_status": run.status,
},
}
@app.get("/agents")
async def list_agents():
"""List all registered agents."""
agents = await agent_registry.get_all_agents()
return {"agents": [agent.model_dump() for agent in agents]}
@app.get("/agents/idle")
async def list_idle_agents():
"""List all idle agents."""
agents = await agent_registry.get_idle_agents()
return {"agents": [agent.model_dump() for agent in agents]}
@app.get("/agents/{agent_id}")
async def get_agent(agent_id: str):
"""Get specific agent details."""
agent = await agent_registry.get_agent(agent_id)
if not agent:
raise HTTPException(status_code=404, detail="Agent not found")
return agent.model_dump()
@app.post("/tasks")
async def create_task(request: TaskCreateRequest):
"""Create a new task."""
task = await task_queue.create_task(
description=request.description,
context=request.context,
max_retries=request.max_retries
)
TASKS_CREATED.inc()
return task_payload(task)
@app.post("/tasks/assign")
async def assign_task(request: TaskAssignRequest):
"""Manually assign a task to an agent."""
task = await task_queue.get_task(request.task_id)
if not task:
raise HTTPException(status_code=404, detail="Task not found")
success = await task_queue.assign_task(request.task_id, request.agent_id)
if not success:
raise HTTPException(status_code=400, detail="Failed to assign task")
await task_queue.remove_pending_task(request.task_id)
run = await swarm_runtime.get_run_for_task(request.task_id)
if run:
await swarm_runtime.emit_event(
run,
"task.claimed",
task_id=request.task_id,
agent_instance_id=request.agent_id,
payload={
"task_id": request.task_id,
"agent_role": task.agent_role,
"agent_id": request.agent_id,
},
)
# Notify agent via WebSocket
await manager.send_message(request.agent_id, {
"type": "task_assignment",
"task_id": request.task_id,
"description": task.description,
"context": task.context,
})
return {"status": "assigned", "task_id": request.task_id, "agent_id": request.agent_id}
@app.get("/tasks")
async def list_tasks(status: Optional[str] = None):
"""List all tasks, optionally filtered by status."""
task_status = TaskStatus(status) if status else None
tasks = await task_queue.get_all_tasks(status=task_status)
return {"tasks": [task_payload(task) for task in tasks]}
@app.get("/tasks/{task_id}")
async def get_task(task_id: str):
"""Get specific task details."""
task = await task_queue.get_task(task_id)
if not task:
raise HTTPException(status_code=404, detail="Task not found")
return task_payload(task)
@app.get("/handoffs")
async def list_handoffs(agent_id: Optional[str] = None, limit: int = 100):
"""List handoff history."""
handoffs = await handoff_manager.get_handoff_history(agent_id=agent_id, limit=limit)
return {"handoffs": [h.model_dump() for h in handoffs]}
@app.get("/metrics")
async def metrics():
"""Prometheus metrics endpoint."""
# Update gauge metrics before returning
agents = await agent_registry.get_all_agents()
AGENTS_ACTIVE.set(len([a for a in agents if a.status == AgentStatus.BUSY]))
WEBSOCKET_CONNECTIONS.set(len(manager.active_connections))
# Update per-agent status
for agent in agents:
AGENT_STATUS.labels(agent_id=agent.agent_id, status=agent.status.value).set(1)
return PlainTextResponse(generate_latest(), media_type=CONTENT_TYPE_LATEST)
# WebSocket endpoint
@app.websocket("/ws/{agent_id}")
async def websocket_endpoint(websocket: WebSocket, agent_id: str):
"""WebSocket endpoint for agent communication."""
await manager.connect(agent_id, websocket)
try:
# Wait for registration message
data = await websocket.receive_json()
if data.get("type") == "register":
capabilities = data.get("capabilities", [])
user_id = data.get("user_id")
# 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 = 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)",
agent_id, user_id, manager.user_agent_count(user_id),
)
await websocket.send_json({
"type": "registration_rejected",
"agent_id": agent_id,
"reason": "max_agents_per_user_exceeded",
"limit": limit,
})
await websocket.close(code=1008, reason="Max agents per user exceeded")
return
await agent_registry.register_agent(agent_id, capabilities)
if user_id:
manager.bind_user(agent_id, user_id)
record_agent_slots(agent_id, data)
await websocket.send_json({
"type": "registered",
"agent_id": agent_id,
"status": "success"
})
# Main message loop
while True:
try:
message = await asyncio.wait_for(
websocket.receive_json(),
timeout=15.0 # Expect heartbeat every 15s
)
message_type = message.get("type")
if message_type == "heartbeat":
record_agent_slots(agent_id, message)
await agent_registry.update_heartbeat(agent_id)
current_agent = await agent_registry.get_agent(agent_id)
if current_agent and current_agent.current_task_id:
current_task = await task_queue.get_task(current_agent.current_task_id)
run = await swarm_runtime.get_run_for_task(current_agent.current_task_id)
if run and current_task:
await swarm_runtime.emit_event(
run,
"task.heartbeat",
task_id=current_agent.current_task_id,
agent_instance_id=agent_id,
payload={
"task_id": current_agent.current_task_id,
"agent_role": current_task.agent_role,
"agent_id": agent_id,
"status": current_task.status.value,
},
)
await maybe_emit_budget_alert(run)
await websocket.send_json({"type": "heartbeat_ack"})
elif message_type == "handoff":
# Handle handoff request
target_agent_id = message.get("target_agent_id")
task_context = message.get("task_context", {})
handoff_id = await handoff_manager.initiate_handoff(
agent_id, target_agent_id, task_context
)
if handoff_id:
# Notify target agent
await manager.send_message(target_agent_id, {
"type": "handoff_request",
"handoff_id": handoff_id,
"source_agent_id": agent_id,
"task_context": task_context
})
await websocket.send_json({
"type": "handoff_initiated",
"handoff_id": handoff_id,
"status": "success"
})
else:
await websocket.send_json({
"type": "handoff_failed",
"reason": "Target agent not available"
})
elif message_type == "handoff_request":
task_id = message.get("task_id")
subtask = message.get("subtask", {})
target_capabilities = message.get("target_capabilities", [])
task = await task_queue.get_task(task_id)
run = await swarm_runtime.get_run_for_task(task_id)
child_task = None
if (
run and task and handoff_workflow_enabled(task)
and task.context.get("allow_handoff", False)
and task.parent_task_id is None
):
child_context = {
**task.context,
"task_title": subtask.get("description", task.title or task.description[:80]),
"depends_on": [],
"agent_role": ",".join(target_capabilities) or "specialist",
"required_capabilities": target_capabilities,
"parent_task_id": task.task_id,
"root_task_id": task.root_task_id or task.task_id,
"source": "dynamic_handoff",
"workflow_mode": "multi_agent",
"allow_handoff": False,
}
child_task = await task_queue.create_task(
title=subtask.get("description", task.title or task.description[:80]),
description=subtask.get("description", task.description),
context=child_context,
max_retries=task.max_retries,
agent_role=",".join(target_capabilities) or "specialist",
required_capabilities=target_capabilities,
depends_on=[],
parent_task_id=task.task_id,
root_task_id=task.root_task_id or task.task_id,
source="dynamic_handoff",
)
await task_queue.add_child_task(task.task_id, child_task.task_id)
await swarm_runtime.attach_task(run, child_task.task_id)
await swarm_runtime.emit_event(
run,
"task.created",
task_id=child_task.task_id,
agent_instance_id=agent_id,
payload={
"task_id": child_task.task_id,
"title": child_task.title,
"description": child_task.description,
"agent_role": child_task.agent_role,
"status": child_task.status.value,
"depends_on": child_task.depends_on,
"required_capabilities": child_task.required_capabilities,
"parent_task_id": child_task.parent_task_id,
"root_task_id": child_task.root_task_id,
"source": child_task.source,
"attempt": child_task.retry_count,
},
)
if run and task:
from_role = task.agent_role
to_role = ",".join(target_capabilities) or "specialist"
await swarm_runtime.emit_event(
run,
"handoff.requested",
task_id=task_id,
agent_instance_id=agent_id,
payload={
"task_id": task_id,
"from_role": from_role,
"to_role": to_role,
"summary": subtask.get("description", "Subtask handoff requested"),
"target_capabilities": target_capabilities,
"child_task_id": child_task.task_id if child_task else None,
"parent_task_id": task.task_id,
},
)
# Frozen handoff event (agent_swarm#15.2): the client's 13-type set uses
# handoff.created (a new handoff/child task exists). Emitted additively
# alongside handoff.requested when a child task was actually created.
if child_task:
await swarm_runtime.emit_event(
run,
"handoff.created",
task_id=task_id,
agent_instance_id=agent_id,
payload={
"task_id": task_id,
"from_role": from_role,
"to_role": to_role,
"child_task_id": child_task.task_id,
"parent_task_id": task.task_id,
},
)
await swarm_runtime.emit_event(
run,
"timeline.updated",
task_id=task_id,
agent_instance_id=agent_id,
payload={
"summary": subtask.get("description", "Subtask handoff requested"),
"task_id": task_id,
"child_task_id": child_task.task_id if child_task else None,
"status": "blocked" if child_task else "recorded",
},
)
await websocket.send_json({
"type": "handoff_response",
"task_id": task_id,
"status": "recorded",
"child_task_id": child_task.task_id if child_task else None,
})
elif message_type == "handoff_accept":
# Target agent accepts handoff
handoff_id = message.get("handoff_id")
handoff = await handoff_manager.get_handoff(handoff_id)
await handoff_manager.complete_handoff(handoff_id)
if handoff:
task_id = handoff.task_context.get("task_id")
run = await swarm_runtime.get_run_for_task(task_id) if task_id else None
if run:
await swarm_runtime.emit_event(
run,
"handoff.completed",
task_id=task_id,
agent_instance_id=agent_id,
payload={
"handoff_id": handoff_id,
"source_agent_id": handoff.source_agent_id,
"target_agent_id": handoff.target_agent_id,
"task_id": task_id,
# HM requires from_role/to_role; this legacy accept path
# lacks role context, so fall back to agent ids.
"from_role": handoff.source_agent_id,
"to_role": handoff.target_agent_id,
},
)
await websocket.send_json({
"type": "handoff_accepted",
"handoff_id": handoff_id
})
elif message_type == "handoff_reject":
# Target agent rejects handoff
handoff_id = message.get("handoff_id")
reason = message.get("reason", "Rejected by target agent")
await handoff_manager.fail_handoff(handoff_id, reason)
await websocket.send_json({
"type": "handoff_rejected",
"handoff_id": handoff_id
})
elif message_type == "task_start":
# Agent starts working on task
task_id = message.get("task_id")
await task_queue.start_task(task_id)
task = await task_queue.get_task(task_id)
run = await swarm_runtime.get_run_for_task(task_id)
if run and task:
await swarm_runtime.emit_event(
run,
"task.running",
task_id=task_id,
agent_instance_id=agent_id,
payload={
"task_id": task_id,
"agent_role": task.agent_role,
"agent_id": agent_id,
},
)
elif message_type == "blocked_on_handoff":
task_id = message.get("task_id")
child_task_id = message.get("child_task_id")
reason = message.get("reason") or "Waiting on delegated child task"
task = await task_queue.block_task(task_id, reason=reason, release_agent=True)
run = await swarm_runtime.get_run_for_task(task_id)
if run and task:
await swarm_runtime.emit_event(
run,
"task.blocked",
task_id=task_id,
agent_instance_id=agent_id,
payload={
"task_id": task_id,
"agent_role": task.agent_role,
"agent_id": agent_id,
"child_task_id": child_task_id,
"reason": reason,
},
)
await swarm_runtime.emit_event(
run,
"timeline.updated",
task_id=task_id,
agent_instance_id=agent_id,
payload={
"summary": reason,
"task_id": task_id,
"child_task_id": child_task_id,
"status": "blocked",
},
)
await refresh_swarm_run_status(run)
await websocket.send_json({
"type": "task_blocked_ack",
"task_id": task_id,
})
elif message_type == "task_complete":
# Agent completes task
task_id = message.get("task_id")
result = message.get("result")
result_text = json.dumps(result) if isinstance(result, dict) else result
logger.info(f"Received task_complete for {task_id}, result length: {len(result_text) if result_text else 0}")
completed = await task_queue.complete_task(task_id, result_text)
task = await task_queue.get_task(task_id)
run = await swarm_runtime.get_run_for_task(task_id)
if completed and task:
await deposit_pheromone(task, agent_id, success=True, result=result, run=run)
if completed and run and task:
await emit_task_completion_events(run, task, agent_id, result)
await emit_usage_event(run, task, agent_id, result)
await finalize_parent_after_child(
run,
task,
agent_id=agent_id,
success=True,
summary="Delegated child task completed",
)
await refresh_swarm_run_status(run)
await websocket.send_json({
"type": "task_completed",
"task_id": task_id
})
elif message_type == "task_result":
# Backward-compatible handler for older/simple agents
task_id = message.get("task_id")
success = message.get("success")
if success is None:
success = message.get("status") == "completed"
if success:
result = message.get("result")
result_text = json.dumps(result) if isinstance(result, dict) else result
completed = await task_queue.complete_task(task_id, result_text)
task = await task_queue.get_task(task_id)
run = await swarm_runtime.get_run_for_task(task_id)
if completed and task:
await deposit_pheromone(task, agent_id, success=True, result=result, run=run)
if completed and run and task:
await emit_task_completion_events(run, task, agent_id, result)
await emit_usage_event(run, task, agent_id, result)
await finalize_parent_after_child(
run,
task,
agent_id=agent_id,
success=True,
summary="Delegated child task completed",
)
await refresh_swarm_run_status(run)
await websocket.send_json({
"type": "task_completed",
"task_id": task_id
})
else:
reason = message.get("error") or message.get("reason") or "Unknown error"
failed = await task_queue.fail_task(task_id, reason)
task = await task_queue.get_task(task_id)
run = await swarm_runtime.get_run_for_task(task_id)
if failed and task:
await deposit_pheromone(task, agent_id, success=False, run=run)
if failed and run and task:
event_type = "task.retried" if task.status == TaskStatus.PENDING else "task.failed"
await swarm_runtime.emit_event(
run,
event_type,
task_id=task_id,
agent_instance_id=agent_id,
payload={
"task_id": task_id,
"agent_role": task.agent_role,
"reason": reason,
"summary": reason,
"attempt": task.retry_count,
"parent_task_id": task.parent_task_id,
"root_task_id": task.root_task_id,
},
)
if task.status == TaskStatus.FAILED:
await finalize_parent_after_child(
run,
task,
agent_id=agent_id,
success=False,
summary=reason,
)
await refresh_swarm_run_status(run)
await websocket.send_json({
"type": "task_failed_ack",
"task_id": task_id
})
elif message_type == "task_failed":
# Agent reports task failure
task_id = message.get("task_id")
reason = message.get("reason", "Unknown error")
failed = await task_queue.fail_task(task_id, reason)
task = await task_queue.get_task(task_id)
run = await swarm_runtime.get_run_for_task(task_id)
if failed and task:
await deposit_pheromone(task, agent_id, success=False, run=run)
if failed and run and task:
event_type = "task.retried" if task.status == TaskStatus.PENDING else "task.failed"
await swarm_runtime.emit_event(
run,
event_type,
task_id=task_id,
agent_instance_id=agent_id,
payload={
"task_id": task_id,
"agent_role": task.agent_role,
"reason": reason,
"summary": reason,
"attempt": task.retry_count,
"parent_task_id": task.parent_task_id,
"root_task_id": task.root_task_id,
},
)
if task.status == TaskStatus.FAILED:
await finalize_parent_after_child(
run,
task,
agent_id=agent_id,
success=False,
summary=reason,
)
await refresh_swarm_run_status(run)
await websocket.send_json({
"type": "task_failed_ack",
"task_id": task_id
})
elif message_type == "status_update":
# Agent updates its status
raw_status = message.get("status")
if raw_status == "handoff":
raw_status = AgentStatus.HANDOFF_PENDING.value
status = AgentStatus(raw_status)
task_id = message.get("task_id")
await agent_registry.update_status(agent_id, status, task_id)
elif message_type == "task_accepted":
# Merged agent confirms it took (or de-duplicated) an assignment.
record_agent_slots(agent_id, message)
if message.get("status") == "duplicate":
logger.info(
f"Agent {agent_id} reported duplicate task {message.get('task_id')}"
)
else:
logger.debug(
f"Agent {agent_id} accepted task {message.get('task_id')}"
)
elif message_type == "task_rejected":
# Agent was at capacity; return the task to the pending queue for another agent.
record_agent_slots(agent_id, message)
rejected_task_id = message.get("task_id")
reject_reason = message.get("reason", "rejected")
logger.info(
f"Agent {agent_id} rejected task {rejected_task_id} ({reject_reason}); requeuing"
)
released = await task_queue.release_task(rejected_task_id, agent_id=agent_id)
if released:
released_task = await task_queue.get_task(rejected_task_id)
released_run = await swarm_runtime.get_run_for_task(rejected_task_id)
if released_task and released_run:
await swarm_runtime.emit_event(
released_run,
"task.released",
task_id=rejected_task_id,
agent_instance_id=agent_id,
payload={
"task_id": rejected_task_id,
"agent_role": released_task.agent_role,
"reason": reject_reason,
},
)
elif message_type == "peer_message":
# Route peer collaboration messages between agents; stamp the sender so the
# recipient knows where to send its reply.
target_agent_id = message.get("target_agent_id")
delivered = False
if target_agent_id:
delivered = await manager.send_message(
target_agent_id,
{**message, "from_agent_id": agent_id},
)
else:
logger.warning(f"peer_message from {agent_id} missing target_agent_id")
# Record internal communication telemetry (feeds benchmark S_communication).
# Not emitted to the Manager event stream — it is not a registered HM event.
peer_task_id = message.get("task_id")
if peer_task_id:
peer_run = await swarm_runtime.get_run_for_task(peer_task_id)
if peer_run:
await swarm_runtime.record_peer_message(
peer_run,
correlation_id=message.get("correlation_id"),
is_reply=bool(message.get("is_reply")),
delivered=delivered,
)
elif message_type == "task_proposal":
# Decentralized rework P3 (#7): an agent proposes a new task from shared
# state (bottom-up decomposition). Reviewed + (maybe) enqueued; ack back.
try:
proposal_result = await handle_task_proposal(agent_id, message)
except Exception as exc:
logger.warning(f"task_proposal from {agent_id} failed: {exc}")
proposal_result = {"accepted": False, "decision": "error", "reason": str(exc)}
await websocket.send_json({"type": "task_proposal_ack", **proposal_result})
elif message_type == "review_decision":
# P5 (#11): a peer agent submits an independent structured review.
try:
review_result = await handle_review_decision(agent_id, message)
except Exception as exc:
logger.warning(f"review_decision from {agent_id} failed: {exc}")
review_result = {"recorded": False, "reason": str(exc)}
await websocket.send_json({"type": "review_decision_ack", **review_result})
elif message_type == "task_bid":
# P4 (#8): record an agent's bid for a task (arbitration picks a winner).
try:
bid_result = await handle_task_bid(agent_id, message)
except Exception as exc:
logger.warning(f"task_bid from {agent_id} failed: {exc}")
bid_result = {"recorded": False, "reason": str(exc)}
await websocket.send_json({"type": "task_bid_ack", **bid_result})
elif message_type == "task_yield":
# P4 (#8): an agent releases a task back for re-competition.
try:
yield_result = await handle_task_yield(agent_id, message)
except Exception as exc:
logger.warning(f"task_yield from {agent_id} failed: {exc}")
yield_result = {"released": False, "reason": str(exc)}
await websocket.send_json({"type": "task_yield_ack", **yield_result})
elif message_type == "task_takeover_request":
# P4 (#8): an agent requests takeover; arbitrate requester vs incumbent.
try:
takeover_result = await handle_task_takeover(agent_id, message)
except Exception as exc:
logger.warning(f"task_takeover from {agent_id} failed: {exc}")
takeover_result = {"taken_over": False, "reason": str(exc)}
await websocket.send_json({"type": "task_takeover_ack", **takeover_result})
else:
logger.warning(f"Unknown message type from {agent_id}: {message_type}")
except asyncio.TimeoutError:
# No message received within timeout - check heartbeat
agent = await agent_registry.get_agent(agent_id)
if agent:
import time
if time.time() - agent.last_heartbeat > 30:
logger.warning(f"Agent {agent_id} heartbeat timeout")
break
continue
else:
logger.error(f"Agent {agent_id} did not send registration message")
await websocket.close(code=1008, reason="Registration required")
except WebSocketDisconnect:
logger.info(f"Agent {agent_id} disconnected")
except Exception as e:
logger.error(f"Error in WebSocket handler for agent {agent_id}: {e}")
finally:
# Cleanup
manager.disconnect(agent_id)
manager.unbind(agent_id) # free the user's per-user agent slot
AGENT_SLOTS.pop(agent_id, None)
await agent_registry.deregister_agent(agent_id)
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)