#!/usr/bin/env python3 """Create a real swarm task, wait for start and half-progress, and render a poster page.""" from __future__ import annotations import argparse import base64 import json import os import subprocess import sys import time from dataclasses import dataclass from datetime import datetime, timezone from pathlib import Path from typing import Any, Optional from urllib.error import HTTPError, URLError from urllib.parse import urlencode from urllib.request import Request, urlopen ROOT = Path(__file__).resolve().parents[1] OUTPUT_DIR = ROOT / "artifacts" / "swarm-posters" @dataclass class ApiConfig: base_url: str token: str def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser( description="Create a swarm task, wait for real progress milestones, and build a poster.", ) parser.add_argument( "--base-url", default=os.getenv("SWARM_BASE_URL", "http://52.139.240.116:8000"), help="Base URL of the running swarm runtime.", ) parser.add_argument( "--runtime-token", default=os.getenv("AGNET_RUNTIME_SERVICE_TOKEN"), help="Runtime bearer token. If omitted, read from the Kubernetes secret.", ) parser.add_argument( "--budget-seconds", type=int, default=40, help="Task budget duration used to compute the half-progress milestone.", ) parser.add_argument( "--poll-interval", type=float, default=3.0, help="Polling interval in seconds.", ) parser.add_argument( "--timeout-seconds", type=int, default=240, help="Maximum time to wait for the half-progress milestone.", ) parser.add_argument( "--objective", default=( "Build a blog system MVP in the configured workspace repository. " "Deliver a runnable implementation with a simple backend and frontend for posts: " "list posts, view a post, create, edit, and delete. Keep setup minimal and include short run instructions." ), help="High-level objective for the swarm.", ) parser.add_argument( "--task-description", default=( "Create a runnable blog system MVP in the target repository with post list, " "post detail, and create/edit/delete flows. Keep the implementation pragmatic and " "coherent with the existing repo. Commit and push the result branch when changes are ready." ), help="Concrete task description sent to the agent.", ) parser.add_argument( "--poster-title", default="Swarm Blog MVP Live Demo", help="Poster title rendered into the HTML.", ) return parser.parse_args() def iso_now() -> str: return datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ") def to_human_time(timestamp: Optional[float]) -> str: if not timestamp: return "-" return datetime.fromtimestamp(timestamp, tz=timezone.utc).strftime("%Y-%m-%d %H:%M:%S UTC") def read_runtime_token(explicit: Optional[str]) -> str: if explicit: return explicit cmd = [ "kubectl", "get", "secret", "-n", "swarm-system", "agnet-runtime-secrets", "-o", "jsonpath={.data.runtime-service-token}", ] encoded = subprocess.check_output(cmd, text=True).strip() if not encoded: raise RuntimeError("Runtime token was not provided and could not be read from Kubernetes.") return base64.b64decode(encoded).decode("utf-8") def api_request( config: ApiConfig, path: str, method: str = "GET", body: Optional[dict[str, Any]] = None, headers: Optional[dict[str, str]] = None, query: Optional[dict[str, Any]] = None, ) -> dict[str, Any]: query_string = f"?{urlencode(query)}" if query else "" url = f"{config.base_url.rstrip('/')}{path}{query_string}" request_headers = { "Authorization": f"Bearer {config.token}", "Content-Type": "application/json", } if headers: request_headers.update(headers) payload = None if body is not None: payload = json.dumps(body).encode("utf-8") req = Request(url, data=payload, headers=request_headers, method=method) try: with urlopen(req, timeout=30) as resp: return json.loads(resp.read().decode("utf-8")) except HTTPError as exc: raw = exc.read().decode("utf-8", errors="replace") raise RuntimeError(f"API request failed with {exc.code}: {raw}") from exc except URLError as exc: raise RuntimeError(f"API request failed: {exc}") from exc def create_swarm(config: ApiConfig, args: argparse.Namespace) -> dict[str, Any]: request_id = iso_now() correlation_id = f"corr_poster_demo_{request_id}" idempotency_key = f"idem_poster_demo_{request_id}" manager_deployment_id = f"dep_poster_demo_{request_id}" agents = [ { "task_id": f"poster-blog-{request_id.lower()}", "role": "fullstack", "title": "Implement blog system MVP", "description": args.task_description, "depends_on": [], } ] body = { "orchestration_plan": { "objective": args.objective, "sub_mode": "code", "risk_level": "low", "budget": { "duration_seconds": args.budget_seconds, "token_limit": 120000, }, "agents": agents, }, "agents": agents, "callback": { "url": f"{config.base_url.rstrip('/')}/health", "subscribed_events": [ "deployment.status_changed", "task.created", "task.running", "task.completed", "task.failed", "timeline.updated", "artifact.created", ], }, "metadata": { "manager_deployment_id": manager_deployment_id, "correlation_id": correlation_id, }, } response = api_request( config, "/api/swarms", method="POST", body=body, headers={ "X-Correlation-Id": correlation_id, "X-Idempotency-Key": idempotency_key, }, ) if not response.get("success"): raise RuntimeError(f"Swarm creation failed: {json.dumps(response, ensure_ascii=False)}") return response["data"] def fetch_tasks(config: ApiConfig, swarm_id: str) -> list[dict[str, Any]]: response = api_request(config, f"/api/swarms/{swarm_id}/tasks") return response["data"]["tasks"] def fetch_logs(config: ApiConfig, swarm_id: str, limit: int = 100) -> list[dict[str, Any]]: response = api_request( config, f"/api/swarms/{swarm_id}/logs", query={"limit": limit}, ) return response["data"]["events"] def fetch_metrics(config: ApiConfig, swarm_id: str) -> dict[str, Any]: response = api_request(config, f"/api/swarms/{swarm_id}/metrics") return response["data"] def fetch_agents(base_url: str) -> list[dict[str, Any]]: req = Request(f"{base_url.rstrip('/')}/agents") with urlopen(req, timeout=15) as resp: return json.loads(resp.read().decode("utf-8"))["agents"] def manual_assign(base_url: str, task_id: str, agent_id: str) -> bool: payload = json.dumps({"task_id": task_id, "agent_id": agent_id}).encode("utf-8") req = Request( f"{base_url.rstrip('/')}/tasks/assign", data=payload, headers={"Content-Type": "application/json"}, method="POST", ) try: with urlopen(req, timeout=15) as resp: data = json.loads(resp.read().decode("utf-8")) return bool(data.get("task_id") or data.get("status")) except Exception: return False def find_event(events: list[dict[str, Any]], event_type: str) -> Optional[dict[str, Any]]: for event in events: if event.get("event_type") == event_type: return event return None def last_event(events: list[dict[str, Any]], event_type: str) -> Optional[dict[str, Any]]: for event in reversed(events): if event.get("event_type") == event_type: return event return None def render_html( poster_path: Path, args: argparse.Namespace, swarm_data: dict[str, Any], task: dict[str, Any], metrics: dict[str, Any], started_at: float, halfway_at: float, halfway_observed_at: float, latest_events: list[dict[str, Any]], ) -> None: latest_status = task.get("status", "unknown") latest_heartbeat = last_event(latest_events, "task.heartbeat") heartbeat_time = latest_heartbeat.get("occurred_at") if latest_heartbeat else "-" runtime_ratio = metrics.get("budget", {}).get("duration_ratio") ratio_display = f"{runtime_ratio * 100:.0f}%" if isinstance(runtime_ratio, (int, float)) else "50%+" html = f""" {args.poster_title}
Live Swarm Poster

{args.poster_title}

真实蜂群任务已成功创建，并已记录到“任务开始”和“进行过半”两段里程碑。下面的内容全部来自当前运行中的 swarm 状态，而不是手工拼接。

Demo Goal

在目标仓库里开发一个博客系统 MVP，包括文章列表、详情和基础增删改能力。

Swarm
{swarm_data["swarm_id"]}
Task Status
{latest_status}
Budget Progress
{ratio_display}
Timeline

任务开始

任务已被 agent 领取并进入运行态。开始时间：{to_human_time(started_at)}

进行到一半

脚本按 budget 的 50% 自动确认里程碑。预算过半时间：{to_human_time(halfway_at)}，实际记录时间：{to_human_time(halfway_observed_at)}

当前执行中

任务仍处于 {latest_status}，最近心跳：{heartbeat_time}

Poster Snapshot
Mid-Run
这张海报展示的是蜂群任务已经真正启动，并且已经跑到预算半程时的现场状态。
Swarm Meta
Deployment{swarm_data["deployment_id"]}
Task{task["task_id"]}
Agent{task.get("assigned_agent_id") or "-"}
Attempts{task.get("attempt", 0)}
Budget{args.budget_seconds}s
Output File{poster_path.name}
Latest Events
1{latest_events[-3]["event_type"] if len(latest_events) >= 3 else "-"}
2{latest_events[-2]["event_type"] if len(latest_events) >= 2 else "-"}
3{latest_events[-1]["event_type"] if latest_events else "-"}
注：如果后续继续执行，日志与状态还会变化；这张图锁定的是“开始后已过半”的那一刻。
""" poster_path.write_text(html, encoding="utf-8") def main() -> int: args = parse_args() OUTPUT_DIR.mkdir(parents=True, exist_ok=True) token = read_runtime_token(args.runtime_token) config = ApiConfig(base_url=args.base_url, token=token) swarm_data = create_swarm(config, args) swarm_id = swarm_data["swarm_id"] deadline = time.time() + args.timeout_seconds started_at = None task_snapshot = None latest_events: list[dict[str, Any]] = [] while time.time() < deadline: tasks = fetch_tasks(config, swarm_id) latest_events = fetch_logs(config, swarm_id, limit=100) if tasks: task_snapshot = tasks[0] if task_snapshot.get("status") == "pending" and not task_snapshot.get("assigned_agent_id"): for agent in fetch_agents(config.base_url): if agent.get("status") == "idle": if manual_assign(config.base_url, task_snapshot["task_id"], agent["agent_id"]): break if task_snapshot.get("started_at"): started_at = float(task_snapshot["started_at"]) break running_event = find_event(latest_events, "task.running") if running_event: started_at = datetime.fromisoformat( running_event["occurred_at"].replace("Z", "+00:00") ).timestamp() break time.sleep(args.poll_interval) if started_at is None or task_snapshot is None: raise RuntimeError("Task did not reach a running state before timeout.") halfway_at = started_at + args.budget_seconds / 2 while time.time() < deadline: tasks = fetch_tasks(config, swarm_id) latest_events = fetch_logs(config, swarm_id, limit=100) task_snapshot = tasks[0] if time.time() >= halfway_at: break if task_snapshot.get("status") in {"completed", "failed"}: break time.sleep(args.poll_interval) metrics = fetch_metrics(config, swarm_id) halfway_observed_at = time.time() run_id = swarm_id.replace("swarm-", "") poster_path = OUTPUT_DIR / f"swarm-poster-{run_id}.html" summary_path = OUTPUT_DIR / f"swarm-poster-{run_id}.json" render_html( poster_path=poster_path, args=args, swarm_data=swarm_data, task=task_snapshot, metrics=metrics, started_at=started_at, halfway_at=halfway_at, halfway_observed_at=halfway_observed_at, latest_events=latest_events, ) summary = { "swarm": swarm_data, "task": task_snapshot, "metrics": metrics, "started_at": started_at, "halfway_at": halfway_at, "halfway_observed_at": halfway_observed_at, "poster_html": str(poster_path), } summary_path.write_text(json.dumps(summary, indent=2, ensure_ascii=False), encoding="utf-8") print(json.dumps(summary, ensure_ascii=False, indent=2)) return 0 if __name__ == "__main__": try: raise SystemExit(main()) except KeyboardInterrupt: print("Interrupted.", file=sys.stderr) raise SystemExit(130)