Songhaoz666andClaude Opus 4.8
f281370209
把本仓从「Master 分解→派发→单评审」重构为去中心化自组织蜂群，并设为唯一行为
（本仓即 swarm 运行时，模式选择在仓外，无 ENABLE_swarm 开关）。依据：heicodeDocs 弱中心
定义 + 蜂群文章（stigmergy）+ OpenAI Swarm（handoff）。
流程（全部无条件生效）：
- 播种 build_seed_task_specs（不主控分解）→ 自选 swarm_dispatch（能力+τ+负载+预算，唯一派发）
→ 自主分解 task_proposal（autonomous_tasks）→ 竞争/接管 task_bid/yield/takeover（task_competition）
→ 同伴交叉评审 ≥2 评审者（cross_review，取代单 critic）→ 收敛 ConvergenceReport+termination_reason
（convergence）→ 守卫 guard.diagnose（检测无法运作并列原因）。
删除：planner 主控分解、贪心/ACO/scored 派发与 scored_matchmake、单 critic 评审环、
全部 ENABLE_* swarm 构建开关。master_agent.synthesize 作为汇总工具保留。
测试：test-workflow-e2e 改写为真实 swarm 全流程；test-merge-smoke 改为 seeder/cross-review；
新增 test-swarm-{seed,dispatch,autonomous,competition,cross-review,convergence,guard} + 模块单测；
CI 同步。本地 18 套全绿。
影响范围：agent_swarm（orchestrator + 新模块 + 测试 + docs + CI）。不改 Manager↔Swarm 事件契约
（新事件均为运行时内部状态/遥测，不进 HM 注册表）；不影响 Client/计费/密钥/审计/发布链路。
无新增长期开关。
诚实边界：不动 benchmark 验收（G_E 仍需真实 run，#13）；convergence 暂为解释性（不强制覆盖
run.status）；前端可见状态为跨端（#18）；「去中心化是否更优」未证（需 Group C）。
#6 的 DoD（Path B 决策 + 定位文档）本 PR 已满足并实现 Path B → 关闭 #6。
#7/#8/#11/#12 的运行时机制已落地（事件 + e2e），但其 DoD 含前端可见状态（跨端 #18）与
多 Agent 回放，故以 Refs 链接、不自动关闭，留维护者在前端/验收就绪后关闭。
Closes #6
Refs #7
Refs #8
Refs #11
Refs #12
Refs #18
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2751 lines
118 KiB
Python
2751 lines
118 KiB
Python
"""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
|
|
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
|
|
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 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
|
|
|
|
# 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] = {}
|
|
|
|
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")
|
|
|
|
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] = {}
|
|
|
|
|
|
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 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,
|
|
})
|
|
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"]; emits an internal `swarm.health` event when
|
|
unhealthy so the stall is visible. Advisory only — it 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,
|
|
})
|
|
run.metadata["health"] = report.to_dict()
|
|
await swarm_runtime.save_run(run)
|
|
if not report.healthy:
|
|
await swarm_runtime.emit_event(run, "swarm.health", payload={
|
|
"summary": report.summary, "blockers": 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
|
|
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
|
|
|
|
next_status = "failed" if any(task.status == TaskStatus.FAILED for task in 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
|
|
|
|
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
|
|
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,
|
|
),
|
|
)
|
|
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)
|
|
|
|
|
|
# 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())
|
|
|
|
yield
|
|
|
|
# Shutdown
|
|
logger.info("Shutting down orchestrator...")
|
|
failure_task.cancel()
|
|
dispatch_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 _manager_provided_agents(body: Dict[str, Any]) -> bool:
|
|
"""Whether the Manager supplied an explicit agent breakdown (Manager-first; never overridden)."""
|
|
normalized = swarm_runtime.normalize_create_request(body)
|
|
plan = normalized.get("orchestration_plan") or {}
|
|
return bool(plan.get("agents") or normalized.get("agents"))
|
|
|
|
|
|
def build_seed_task_specs(run, body: Dict[str, Any], base_specs: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
|
|
"""Decentralized rework P2: seed the run with ONE objective-carrying task instead of a
|
|
Master-decomposed plan.
|
|
|
|
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()
|
|
|
|
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:
|
|
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
|
|
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,
|
|
})
|
|
logger.info(f"Cross-review reopened {reopened} on run {run.swarm_id} (cycle {cycles + 1})")
|
|
return True
|
|
|
|
|
|
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"}
|
|
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"}
|
|
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"}
|
|
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): when the Manager supplied an explicit agent breakdown we
|
|
# honor it (Manager-first contract); otherwise we SEED a single objective-carrying task and the
|
|
# agents grow the graph bottom-up via proposals (handle_task_proposal). No up-front Master
|
|
# decomposition — this repo is the swarm runtime (see rework plan §0).
|
|
if not _manager_provided_agents(body):
|
|
task_specs = build_seed_task_specs(run, body, task_specs)
|
|
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", 3),
|
|
)
|
|
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",
|
|
},
|
|
)
|
|
|
|
|
|
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)
|
|
|
|
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,
|
|
})
|
|
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 {}
|
|
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"),
|
|
"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
|
|
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
|
|
),
|
|
"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()),
|
|
"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}/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)}
|
|
|
|
|
|
@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", [])
|
|
await agent_registry.register_agent(agent_id, capabilities)
|
|
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,
|
|
},
|
|
)
|
|
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)
|
|
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)
|