feat: swarm — big tasks decomposed across specialist agents running in parallel

run_team was one shallow round: up to three sequential delegations, no
dependencies, and no way for an agent to consult a peer mid-task. Fine for
"check X then answer", useless for "build me a thing".

aios_swarm.py adds a real swarm:
  PLAN        an agent turns the request into a DAG of subtasks, each assigned
              to the specialist that suits it, with explicit dependencies
  EXECUTE     independent subtasks run CONCURRENTLY (ThreadPoolExecutor, wave by
              wave); a subtask starts only once its deps have produced output,
              and dependents receive those outputs as context
  CONSULT     a running agent can emit `ASK <agent>: <question>`, gets a real
              answer back, and finishes with it — agents talking, not just
              output chaining
  SYNTHESIZE  one agent reads the whole blackboard and writes the final answer

Triggered automatically for big asks on brain/team via a cheap heuristic so
ordinary chat never pays for a planning call, or forced with the 🐝 Swarm chip /
"swarm" target. A plan with one task is returned as a plain single-agent answer
rather than dressed up as a swarm. Live DAG in the dashboard's Swarm view, fed by
the same activity feed, plus GET/POST /api/swarm and `swarm.enabled` config.

The engine is injected with route/ask_llm/act so it never imports the hub back,
which also makes it testable against a stub router. Verified that way: peak
parallelism 2 on independent subtasks, correct dependency ordering, a real
consult round-trip (claudecode -> opencode -> claudecode), narrated-and-fenced
JSON still parsed, and a failing subtask unblocking its dependents instead of
deadlocking.

Two real bugs found while testing against live agents:
- llm_chat's 120s timeout was too short for claude-code, whose every call
  cold-starts the Claude CLI. Planning is a big prompt and timed out, surfacing
  as "no usable plan". Now 300s, overridable via AIOS_LLM_TIMEOUT.
- claude-code is an AGENT, not a chat model: given a planning prompt it replies
  "I'll find and read the README..." or returns an empty body, so it is an
  unreliable planner backend. The planner now puts its instructions in the user
  turn (a long system prompt full of JSON braces reliably returned empty there),
  retries once, and on failure reports it honestly instead of inventing a plan —
  and the hub falls back to run_team so a swarm can never return a blank reply.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Author: ZD Studios

aios.config.example.yaml

@@ -83,6 +83,12 @@ updates:
                            # nothing executes) when an agent says SAFE. Service-tier
                            # projects with dependency trees always wait for you.
 
+swarm:
+  enabled: true            # big requests are decomposed into subtasks, run by
+                           # different specialist agents in parallel; agents can
+                           # consult each other mid-task, then one synthesizes.
+                           # Small asks are untouched (a cheap heuristic gates it).
+
 watchdog:
   enabled: true            # if an agent stops responding, auto-restart it
   interval: 45             # seconds between health sweeps; a healthy agent diagnoses failures

aios.py

@@ -478,6 +478,7 @@ def service_specs(cfg: dict) -> dict:
             "AIOS_WATCHDOG": "1" if cfg_get(cfg, "watchdog.enabled", True) else "0",
             "AIOS_WATCHDOG_INTERVAL": str(cfg_get(cfg, "watchdog.interval", 45)),
             # supervised updates: agents review upstream changes before they land
+            "AIOS_SWARM": "1" if cfg_get(cfg, "swarm.enabled", True) else "0",
             "AIOS_SUPERVISED_UPDATES": "1" if cfg_get(cfg, "updates.supervised", True) else "0",
             "AIOS_UPDATE_INTERVAL": str(cfg_get(cfg, "updates.check_interval", 21600)),
             "AIOS_AUTO_APPLY": "1" if cfg_get(cfg, "updates.auto_apply_safe", True) else "0",

aios_hub.py

@@ -35,6 +35,7 @@
 import aios_tools as tools    # noqa: E402  shell + channel/skill catalogs
 import aios_updates as updates  # noqa: E402  agent-supervised dependency updates
 import aios_web as web          # noqa: E402  YouTube / Reddit / HN / page readers
+import aios_swarm as swarm      # noqa: E402  decompose big tasks across agents
 
 PORT = int(os.environ.get("AIOS_HUB_PORT", "8787"))
 ROOT = Path(os.environ.get("AIOS_ROOT", Path(__file__).resolve().parent))
@@ -115,7 +116,7 @@ def read_system_prompt() -> str:
 
 
 # Chat "targets" the AIOS API exposes as OpenAI-style models.
-TARGETS = ["brain", "team", "opencode", "crewai", "claudecode", "fabric", "all"]
+TARGETS = ["brain", "team", "swarm", "opencode", "crewai", "claudecode", "fabric", "all"]
 
 MEMORY_FILE = ROOT / ".aios" / "memory.json"
 USAGE_FILE = ROOT / ".aios" / "usage.json"
@@ -320,7 +321,11 @@ def llm_chat(messages: list, system: str | None = None) -> str:
         headers={"Content-Type": "application/json", "Authorization": f"Bearer {key}",
                  "HTTP-Referer": "https://github.com/ZDStudios/AIOS", "X-Title": "The AI OS"})
     try:
-        with urllib.request.urlopen(req, timeout=120) as r:
+        # Generous: when the provider is claude-code, every call shells out to the
+        # Claude CLI, which cold-starts slowly. Swarm planning is a big prompt and
+        # was timing out at 120s, which surfaced as "no usable plan".
+        with urllib.request.urlopen(req, timeout=int(
+                os.environ.get("AIOS_LLM_TIMEOUT", "300"))) as r:
             data = json.loads(r.read())
         return data["choices"][0]["message"]["content"]
     except urllib.error.HTTPError as e:
@@ -621,6 +626,42 @@ def maybe_pin_to_canvas(user_msg: str, reply: str) -> str:
     return title
 
 
+# --------------------------------------------------------------------------- #
+# Swarm — a big request becomes a DAG of subtasks run by different specialists  #
+# in parallel, able to consult each other, then synthesized into one answer.    #
+# --------------------------------------------------------------------------- #
+SWARM_FILE = ROOT / ".aios" / "swarm.json"
+
+
+def swarm_enabled() -> bool:
+    return os.environ.get("AIOS_SWARM", "1") == "1"
+
+
+def swarm_state() -> dict:
+    return _load_json(SWARM_FILE, {})
+
+
+def run_swarm(message: str) -> dict:
+    """Bind the hub's router + activity feed into the swarm engine, and publish
+    live progress so the dashboard can watch the DAG fill in."""
+    def _publish(snap):
+        _save_json(SWARM_FILE, {**snap, "message": message[:400], "ts": time.time()})
+
+    r = swarm.run(
+        message,
+        route=lambda agent, prompt: route(agent, prompt),
+        ask_llm=lambda msgs, system=None: llm_chat(msgs, system=system),
+        act=act,
+        on_update=_publish,
+    )
+    _save_json(SWARM_FILE, {"summary": r.get("summary", ""), "tasks": r.get("tasks", []),
+                            "message": message[:400], "ts": time.time(),
+                            "elapsed": r.get("elapsed"), "finished": True})
+    if r.get("swarm"):
+        brain.audit("swarm", "run", f"{len(r.get('tasks', []))} subtasks: {message[:120]}")
+    return r
+
+
 def route(target: str, message: str, history: list | None = None) -> str:
     history = history or []
     target = (target or "brain").lower()
@@ -645,7 +686,20 @@ def route(target: str, message: str, history: list | None = None) -> str:
         if txt.strip():
             return run_fabric(pat.strip(), txt.strip())
         return run_fabric("summarize", message)
+    if target == "swarm":
+        r = run_swarm(message)
+        # A swarm that couldn't plan must not hand back an empty string — fall
+        # back to the team, which is a single round but always produces an answer.
+        return r.get("reply") or run_team(message, history)
     if target in ("team", "auto", "merge"):
+        # A genuinely big ask gets the swarm — parallel specialists that can
+        # consult each other — rather than run_team's single shallow round.
+        if swarm_enabled() and swarm.looks_big(message):
+            act("plan", "Big task detected — deploying a swarm of agents")
+            r = run_swarm(message)
+            if r.get("swarm") and r.get("reply"):
+                return r["reply"]
+            act("note", "Swarm unavailable — using the team instead.")
         return run_team(message, history)
     if target == "all":
         parts = []
@@ -1446,6 +1500,8 @@ def do_GET(self):
         elif self.path == "/api/activity":
             self._send(200, activity(self._query.get("turn", [""])[0],
                                      int(self._query.get("since", ["0"])[0] or 0)))
+        elif self.path == "/api/swarm":
+            self._send(200, {**swarm_state(), "enabled": swarm_enabled()})
         elif self.path == "/api/dashboard":
             self._send(200, dashboard_cfg())
         elif self.path == "/api/updates":
@@ -1692,6 +1748,16 @@ def do_POST(self):
             brain.audit(payload.get("by", "agent"), "web." + op,
                         f"{q[:120]} -> ok={r.get('ok')}")
             self._send(200, r)
+        elif self.path == "/api/swarm":
+            msg = payload.get("message", "")
+            if not msg:
+                self._send(400, {"error": "no message"})
+            else:
+                act_begin(payload.get("turn", ""))
+                try:
+                    self._send(200, run_swarm(msg))
+                finally:
+                    act_end()
         elif self.path == "/api/dashboard":
             self._send(200, dashboard_update(payload))
         elif self.path == "/api/updates":