From f26091f2887abc9b923766a83dfae243837be2f7 Mon Sep 17 00:00:00 2001 From: daaximus Date: Wed, 11 Feb 2026 16:12:42 -0600 Subject: [PATCH] feat(coding-agent): add swarm extension for multi-agent pipeline orchestration Adds a standalone extension that orchestrates multi-agent workflows defined in YAML. Supports pipeline (iterative), parallel (fan-in/out), sequential, and arbitrary DAG execution patterns. Each agent runs as a full oh-my-pi subagent with complete tool access. Agents communicate through the shared workspace filesystem. The orchestrator handles lifecycle, dependency ordering, and state tracking. Key components: - YAML schema parser with validation and cycle detection - DAG builder with topological sort into execution waves - Pipeline controller with iteration loop and wave execution - Filesystem state persistence for monitoring and resumability - Standalone runner (run-pipeline.ts) for long-running unattended work - TUI integration via /swarm command Designed for any task type: research, coding, data processing, content creation, analysis workflows, or any multi-step objective that benefits from specialized agents working in coordination. --- .../coding-agent/extensions/swarm/.gitignore | 1 + .../coding-agent/extensions/swarm/README.md | 457 ++++++++++++++++++ .../coding-agent/extensions/swarm/bun.lock | 15 + .../extensions/swarm/extension.ts | 276 +++++++++++ .../extensions/swarm/package.json | 11 + .../extensions/swarm/run-pipeline.ts | 105 ++++ .../extensions/swarm/swarm/dag.ts | 146 ++++++ .../extensions/swarm/swarm/executor.ts | 108 +++++ .../extensions/swarm/swarm/pipeline.ts | 219 +++++++++ .../extensions/swarm/swarm/render.ts | 75 +++ .../extensions/swarm/swarm/schema.ts | 146 ++++++ .../extensions/swarm/swarm/state.ts | 127 +++++ .../extensions/swarm/tsconfig.json | 16 + 13 files changed, 1702 insertions(+) create mode 100644 packages/coding-agent/extensions/swarm/.gitignore create mode 100644 packages/coding-agent/extensions/swarm/README.md create mode 100644 packages/coding-agent/extensions/swarm/bun.lock create mode 100644 packages/coding-agent/extensions/swarm/extension.ts create mode 100644 packages/coding-agent/extensions/swarm/package.json create mode 100644 packages/coding-agent/extensions/swarm/run-pipeline.ts create mode 100644 packages/coding-agent/extensions/swarm/swarm/dag.ts create mode 100644 packages/coding-agent/extensions/swarm/swarm/executor.ts create mode 100644 packages/coding-agent/extensions/swarm/swarm/pipeline.ts create mode 100644 packages/coding-agent/extensions/swarm/swarm/render.ts create mode 100644 packages/coding-agent/extensions/swarm/swarm/schema.ts create mode 100644 packages/coding-agent/extensions/swarm/swarm/state.ts create mode 100644 packages/coding-agent/extensions/swarm/tsconfig.json diff --git a/packages/coding-agent/extensions/swarm/.gitignore b/packages/coding-agent/extensions/swarm/.gitignore new file mode 100644 index 000000000..c2658d7d1 --- /dev/null +++ b/packages/coding-agent/extensions/swarm/.gitignore @@ -0,0 +1 @@ +node_modules/ diff --git a/packages/coding-agent/extensions/swarm/README.md b/packages/coding-agent/extensions/swarm/README.md new file mode 100644 index 000000000..5f4af6c96 --- /dev/null +++ b/packages/coding-agent/extensions/swarm/README.md @@ -0,0 +1,457 @@ +# Swarm Extension + +Multi-agent orchestration for oh-my-pi. Define agent workflows in YAML — pipelines, parallel fan-outs, sequential chains, or any DAG — and run them unattended until completion. + +Each agent is a full oh-my-pi subagent with access to every tool: bash, python, read, write, edit, grep, find, fetch, web_search, browser. The orchestrator manages lifecycle and ordering; agents communicate through the shared workspace filesystem. + +Use it for anything: research pipelines, code generation, data processing, content creation, analysis workflows, CI-like automation — any multi-step task that benefits from specialized agents working in coordination. + +## Setup + +```bash +cd packages/coding-agent/extensions/swarm +bun install +``` + +## Running + +### Standalone (recommended for long-running work) + +```bash +# Foreground — runs until complete, no timeout: +bun packages/coding-agent/extensions/swarm/run-pipeline.ts path/to/swarm.yaml + +# Background — survives terminal close: +nohup bun packages/coding-agent/extensions/swarm/run-pipeline.ts path/to/swarm.yaml \ + > pipeline.log 2>&1 & disown +``` + +The standalone runner has no timeout. It runs iteration after iteration until the pipeline finishes or you kill it. + +### Inside oh-my-pi (TUI) + +Register the extension in your config (`~/.omp/config.json` or `.omp/config.json`): + +```json +{ + "extensions": ["packages/coding-agent/extensions/swarm"] +} +``` + +Then: + +``` +/swarm run path/to/swarm.yaml +/swarm status +/swarm help +``` + +## Monitoring + +State persists to `/.swarm_/` while the pipeline runs: + +``` +.swarm_/ + state/pipeline.json # Live pipeline + per-agent status + logs/orchestrator.log # Wave transitions, iteration progress + logs/.log # Per-agent timestamps and errors + context/ # Agent session artifacts +``` + +Check on a running pipeline: + +```bash +# Quick status +cat workspace/.swarm_mypipeline/state/pipeline.json | python -m json.tool + +# Watch the orchestrator log +tail -f workspace/.swarm_mypipeline/logs/orchestrator.log +``` + +--- + +## YAML Reference + +Every swarm is a single YAML file with a top-level `swarm` key: + +```yaml +swarm: + name: my-pipeline # Identifier (state stored in .swarm_/) + workspace: ./workspace # Working directory (relative to YAML file location) + mode: pipeline # pipeline | parallel | sequential + target_count: 10 # Iterations (pipeline mode only, default: 1) + model: claude-opus-4-6 # Model for all agents (optional) + + agents: + first_agent: + role: short-role-name + task: | + Full instructions for this agent. + extra_context: | + Optional additional system prompt text. + reports_to: + - downstream_agent + waits_for: + - upstream_agent +``` + +### Top-Level Fields + +| Field | Required | Default | Description | +|-------|----------|---------|-------------| +| `name` | yes | — | Pipeline identifier. State directory is `.swarm_/` | +| `workspace` | yes | — | Shared working directory. Relative paths resolve from YAML file location | +| `mode` | no | `sequential` | Execution mode (see below) | +| `target_count` | no | `1` | How many times to repeat the full pipeline. Only meaningful in `pipeline` mode | +| `model` | no | session default | Model ID for all agents. Any omp-configured model works | + +### Agent Fields + +| Field | Required | Description | +|-------|----------|-------------| +| `role` | yes | Short role identifier — becomes the agent's system prompt | +| `task` | yes | Complete instructions sent as user prompt. Use YAML `\|` for multi-line | +| `extra_context` | no | Additional text appended to system prompt | +| `reports_to` | no | List of agent names that depend on this agent | +| `waits_for` | no | List of agent names this agent depends on | + +### Execution Modes + +**`pipeline`** — Repeat the full agent graph `target_count` times. Each iteration runs all waves in order. Use for accumulative work: "find 50 things, one per iteration." + +**`sequential`** — Run agents once, chained by declaration order (unless explicit dependencies override). The default mode. + +**`parallel`** — Run all agents simultaneously (unless explicit dependencies impose ordering). + +### Dependency Resolution + +The orchestrator builds a DAG from `waits_for` and `reports_to`, then groups agents into **waves** using topological sort. Agents in the same wave run in parallel; waves execute in sequence. + +- `waits_for: [a, b]` — this agent won't start until both `a` and `b` finish +- `reports_to: [x]` — equivalent to `x` having `waits_for: [this_agent]` +- No explicit deps + `pipeline`/`sequential` mode — agents chain by YAML declaration order +- No explicit deps + `parallel` mode — all agents run in one wave +- Cycles are detected and rejected before execution + +--- + +## Patterns + +### Pipeline: Iterative Accumulation + +Run the same agent chain N times. Each iteration builds on the previous one's output. Good for: research collection, data gathering, batch processing, iterative refinement. + +```yaml +swarm: + name: research-collector + workspace: ./workspace + mode: pipeline + target_count: 25 + model: claude-opus-4-6 + + agents: + finder: + role: researcher + task: | + Find ONE new source on the topic defined in workspace/topic.md. + + 1. Read processed.txt to see what's already been found + 2. Use web_search to find a new, high-quality source + 3. Append the URL to processed.txt + 4. Write the URL to signals/finder_out.txt: FOUND: + + analyzer: + role: analyst + task: | + Read signals/finder_out.txt for the URL. + Fetch the page and extract key findings. + Read tracking/count.txt, increment it, write back. + Write analysis to analyzed/item_.md + Write to signals/analyzer_out.txt: DONE: + + compiler: + role: technical-writer + task: | + Read signals/analyzer_out.txt for the item number. + Read analyzed/item_.md. + Append a summary to output/report.md under a new section. +``` + +After 25 iterations: 25 sources found, analyzed, and compiled into a single report. + +### Fan-In: Parallel Specialists + +Multiple agents work independently, one synthesizer combines results. Good for: multi-perspective analysis, parallel code review, comprehensive audits. + +```yaml +swarm: + name: codebase-audit + workspace: ./workspace + + agents: + security: + role: security-auditor + task: | + Audit all code in src/ for security vulnerabilities. + Write findings to reports/security.md with severity ratings. + reports_to: + - lead + + performance: + role: performance-analyst + task: | + Profile and analyze src/ for performance bottlenecks. + Write findings to reports/performance.md with benchmarks. + reports_to: + - lead + + architecture: + role: architecture-reviewer + task: | + Review src/ for architectural issues, coupling, and tech debt. + Write findings to reports/architecture.md with refactoring suggestions. + reports_to: + - lead + + lead: + role: engineering-lead + task: | + Read all reports in reports/. + Create a prioritized action plan in output/action_plan.md. + Rank issues by impact and effort. + waits_for: + - security + - performance + - architecture +``` + +Execution: security + performance + architecture run in parallel (wave 1), lead starts after all three complete (wave 2). + +### Sequential Chain: Staged Handoff + +Linear progression through distinct phases. Good for: content pipelines, multi-stage processing, review chains. + +```yaml +swarm: + name: blog-post + workspace: ./workspace + mode: sequential + + agents: + researcher: + role: researcher + task: | + Research the topic in topic.md using web_search. + Write raw findings and source links to research/notes.md + + writer: + role: technical-writer + task: | + Read research/notes.md. + Write a complete blog post draft to drafts/post.md. + Include code examples where relevant. + + editor: + role: editor + task: | + Read drafts/post.md. + Fix grammar, improve flow, tighten prose. + Rewrite to drafts/post.md. + + reviewer: + role: senior-reviewer + task: | + Read drafts/post.md. + Check technical accuracy against research/notes.md. + Add an editorial note at top if issues found, otherwise + copy to output/final.md. +``` + +Execution: researcher -> writer -> editor -> reviewer, one after another. + +### Diamond: Fan-Out Then Fan-In + +One planner, parallel workers, one integrator. Good for: divide-and-conquer, modular code generation, multi-file refactors. + +```yaml +swarm: + name: feature-implementation + workspace: ./workspace + + agents: + planner: + role: architect + task: | + Read the feature spec in spec.md. + Break it into independent implementation tasks. + Write the plan to plan.md with file assignments. + reports_to: + - api + - ui + - tests + + api: + role: backend-developer + task: | + Read plan.md for your assigned files. + Implement the API layer. Write to src/api/. + reports_to: + - integrator + + ui: + role: frontend-developer + task: | + Read plan.md for your assigned files. + Implement the UI components. Write to src/ui/. + reports_to: + - integrator + + tests: + role: test-engineer + task: | + Read plan.md for the full feature scope. + Write integration tests to tests/. + reports_to: + - integrator + + integrator: + role: tech-lead + task: | + Read plan.md and review all code in src/ and tests/. + Wire everything together. Fix any integration issues. + Run the tests and fix failures. + Write status to output/done.md. +``` + +Execution: planner (wave 1) -> api + ui + tests in parallel (wave 2) -> integrator (wave 3). + +### Hybrid: Mixed Dependencies + +Any DAG is valid. Combine patterns freely. + +```yaml +swarm: + name: data-pipeline + workspace: ./workspace + mode: pipeline + target_count: 10 + + agents: + scraper_a: + role: web-scraper + task: | + Scrape data source A. Write to raw/source_a.json + reports_to: + - transformer + + scraper_b: + role: web-scraper + task: | + Scrape data source B. Write to raw/source_b.json + reports_to: + - transformer + + transformer: + role: data-engineer + task: | + Read raw/source_a.json and raw/source_b.json. + Clean, normalize, merge. Write to processed/merged.json + reports_to: + - loader + - validator + + validator: + role: qa-analyst + task: | + Read processed/merged.json. + Validate schema, check for anomalies. + Write report to qa/validation.md + + loader: + role: data-engineer + task: | + Read processed/merged.json. + Append to output/dataset.jsonl +``` + +Execution per iteration: scraper_a + scraper_b (wave 1) -> transformer (wave 2) -> loader + validator (wave 3). + +--- + +## Writing Agent Tasks + +### What Agents Can Do + +Each agent is a full oh-my-pi session. It can: + +- **bash/python**: Run commands, scripts, install packages, process data +- **read/write/edit**: Create and modify files in the workspace +- **grep/find**: Search the workspace (or anywhere on disk) +- **web_search**: Search the internet (via configured provider) +- **fetch**: Download web pages, APIs, documents +- **browser**: Navigate websites, scrape dynamic content, take screenshots + +### Inter-Agent Communication + +The orchestrator starts and stops agents in the right order. It does **not** pass data between them. Agents communicate through files in the shared workspace. + +Design your own protocol. Common patterns: + +**Signal files** — lightweight status flags an agent writes when done: +``` +signals/finder_out.txt -> "FOUND:https://example.com" +signals/analyzer_out.txt -> "DONE:42" +signals/reviewer_out.txt -> "APPROVED" or "REJECTED:reason" +``` + +**Structured output** — detailed results other agents read: +``` +analyzed/item_1.md -> Full analysis document +results/report.json -> Machine-readable data +output/final.docx -> Accumulated deliverable +``` + +**Tracking files** — prevent duplicate work across pipeline iterations: +``` +processed.txt -> Items already handled (one per line) +tracking/count.txt -> Current item counter +tracking/status.json -> Cumulative state +``` + +### Tips for Reliable Agents + +- **Be explicit about paths.** Agents start fresh each iteration — they don't remember previous runs. Tell them exactly where to read input and write output. +- **Check existing state.** In pipeline mode, tell agents to read tracking files before doing work: "Read processed.txt to avoid duplicates." +- **Use numbered outputs.** `item_1.md`, `item_2.md` etc. so iterations don't clobber each other. +- **Handle failure.** Tell agents what to do when things go wrong: "If the source lacks depth, write SKIP to signals/out.txt and explain why." +- **Keep signal files simple.** One line, parseable format. Complex data goes in structured output files. +- **Scope the task tightly.** An agent that tries to do five things will do zero well. One clear objective per agent. + +--- + +## Models + +Any model configured in omp works. Set it in the YAML: + +```yaml +swarm: + model: claude-opus-4-6 +``` + +Or omit `model` to use your session's default. Check `packages/ai/src/models.json` for available model IDs. + +--- + +## Architecture + +``` +extension.ts TUI entry point (registers /swarm command) +run-pipeline.ts Standalone runner (no TUI, no timeout) +swarm/ + schema.ts YAML parsing + validation + dag.ts Dependency graph, cycle detection, topological sort + executor.ts Spawns agents via oh-my-pi's runSubprocess + pipeline.ts Iteration loop + wave controller + state.ts Filesystem state persistence + render.ts Progress display formatting +``` diff --git a/packages/coding-agent/extensions/swarm/bun.lock b/packages/coding-agent/extensions/swarm/bun.lock new file mode 100644 index 000000000..5bd33faa1 --- /dev/null +++ b/packages/coding-agent/extensions/swarm/bun.lock @@ -0,0 +1,15 @@ +{ + "lockfileVersion": 1, + "configVersion": 1, + "workspaces": { + "": { + "name": "omp-extension-swarm", + "dependencies": { + "yaml": "^2.7.0", + }, + }, + }, + "packages": { + "yaml": ["yaml@2.8.2", "", { "bin": { "yaml": "bin.mjs" } }, "sha512-mplynKqc1C2hTVYxd0PU2xQAc22TI1vShAYGksCCfxbn/dFwnHTNi1bvYsBTkhdUNtGIf5xNOg938rrSSYvS9A=="], + } +} diff --git a/packages/coding-agent/extensions/swarm/extension.ts b/packages/coding-agent/extensions/swarm/extension.ts new file mode 100644 index 000000000..d4b7189f5 --- /dev/null +++ b/packages/coding-agent/extensions/swarm/extension.ts @@ -0,0 +1,276 @@ +/** + * Swarm Extension — Multi-agent pipeline orchestration from YAML definitions. + * + * Registers: + * - /swarm run — Execute a swarm pipeline + * - /swarm status — Show current pipeline status + * + * Usage: Add this extension's directory to your extensions config, + * then use /swarm in any oh-my-pi session. + */ +import * as path from "node:path"; +import * as fs from "node:fs/promises"; +import type { ExtensionAPI, ExtensionCommandContext } from "@oh-my-pi/pi-coding-agent"; +import { parseSwarmYaml, validateSwarmDefinition, type SwarmDefinition } from "./swarm/schema"; +import { buildDependencyGraph, detectCycles, buildExecutionWaves } from "./swarm/dag"; +import { StateTracker } from "./swarm/state"; +import { PipelineController } from "./swarm/pipeline"; +import { renderSwarmProgress } from "./swarm/render"; + +export default function swarmExtension(pi: ExtensionAPI): void { + pi.setLabel("Swarm Orchestrator"); + + pi.registerCommand("swarm", { + description: "Run a multi-agent swarm pipeline from YAML", + getArgumentCompletions: (prefix) => { + const subcommands = ["run", "status", "help"]; + if (!prefix) return subcommands.map((s) => ({ label: s, value: s })); + return subcommands + .filter((s) => s.startsWith(prefix)) + .map((s) => ({ label: s, value: s })); + }, + handler: async (args: string, ctx: ExtensionCommandContext) => { + const parts = args.trim().split(/\s+/); + const subcommand = parts[0] ?? "help"; + + switch (subcommand) { + case "run": { + const yamlPath = parts[1]; + if (!yamlPath) { + ctx.ui.notify("Usage: /swarm run ", "error"); + return; + } + await handleRun(yamlPath, ctx, pi); + return; + } + case "status": { + await handleStatus(parts[1], ctx); + return; + } + case "help": + default: + ctx.ui.notify( + [ + "Swarm — multi-agent pipeline orchestrator", + "", + " /swarm run Run a pipeline", + " /swarm status [name] Show pipeline status", + " /swarm help Show this help", + ].join("\n"), + "info", + ); + return; + } + }, + }); +} + +// ============================================================================ +// /swarm run +// ============================================================================ + +async function handleRun(yamlPath: string, ctx: ExtensionCommandContext, pi: ExtensionAPI): Promise { + // 1. Resolve and read YAML + const resolvedPath = path.isAbsolute(yamlPath) ? yamlPath : path.resolve(ctx.cwd, yamlPath); + + let content: string; + try { + content = await Bun.file(resolvedPath).text(); + } catch { + ctx.ui.notify(`Cannot read file: ${resolvedPath}`, "error"); + return; + } + + // 2. Parse YAML + let def: SwarmDefinition; + try { + def = parseSwarmYaml(content); + } catch (err) { + ctx.ui.notify(`YAML error: ${err instanceof Error ? err.message : String(err)}`, "error"); + return; + } + + // 3. Validate + const validationErrors = validateSwarmDefinition(def); + if (validationErrors.length > 0) { + ctx.ui.notify(`Validation errors:\n${validationErrors.map((e) => ` - ${e}`).join("\n")}`, "error"); + return; + } + + // 4. Build DAG + const deps = buildDependencyGraph(def); + const cycleNodes = detectCycles(deps); + if (cycleNodes) { + ctx.ui.notify(`Cycle detected in agent dependencies: [${cycleNodes.join(", ")}]`, "error"); + return; + } + const waves = buildExecutionWaves(deps); + + // 5. Resolve workspace (relative to YAML file location) + const workspace = path.isAbsolute(def.workspace) + ? def.workspace + : path.resolve(path.dirname(resolvedPath), def.workspace); + + // Ensure workspace exists + await fs.mkdir(workspace, { recursive: true }); + + // 6. Initialize state tracker + const stateTracker = new StateTracker(workspace, def.name); + await stateTracker.init([...def.agents.keys()], def.targetCount, def.mode); + + // 7. Log start + const agentList = [...def.agents.keys()].join(", "); + const waveDesc = waves.map((w, i) => `wave ${i + 1}: [${w.join(", ")}]`).join("; "); + pi.logger.debug("Swarm starting", { + name: def.name, + mode: def.mode, + agents: agentList, + waves: waveDesc, + workspace, + }); + + ctx.ui.notify( + `Starting swarm '${def.name}': ${def.agents.size} agents, ${waves.length} waves, ${def.targetCount} iteration(s)`, + "info", + ); + + // 8. Set up progress widget + const widgetKey = `swarm-${def.name}`; + const updateWidget = () => { + const lines = renderSwarmProgress(stateTracker.state); + ctx.ui.setWidget(widgetKey, lines); + }; + updateWidget(); + + // 9. Resolve infrastructure for agent execution + let authStorage: Awaited> | undefined; + try { + authStorage = await pi.pi.discoverAuthStorage(); + } catch { + // Let runSubprocess discover auth per-agent as fallback + } + + // 10. Run pipeline + const controller = new PipelineController(def, waves, stateTracker); + + const result = await controller.run({ + workspace, + onProgress: () => updateWidget(), + authStorage, + modelRegistry: ctx.modelRegistry, + settings: pi.pi.settings, + }); + + // 11. Clear widget and show summary + ctx.ui.setWidget(widgetKey, undefined); + + const elapsed = stateTracker.state.completedAt + ? formatDuration(stateTracker.state.completedAt - stateTracker.state.startedAt) + : "unknown"; + + const summaryParts = [ + `Swarm '${def.name}' ${result.status}`, + `${result.iterations}/${def.targetCount} iterations`, + `elapsed: ${elapsed}`, + ]; + + if (result.errors.length > 0) { + summaryParts.push(`${result.errors.length} error(s)`); + } + + const summaryType = result.status === "completed" ? "info" : "error"; + ctx.ui.notify(summaryParts.join(" | "), summaryType); + + // Log errors + if (result.errors.length > 0) { + pi.logger.warn("Swarm completed with errors", { errors: result.errors }); + } + + // 12. Send summary to the conversation so the LLM knows what happened + const summaryMessage = buildSummaryMessage(def, result, stateTracker, workspace); + pi.sendMessage( + { + customType: "swarm-result", + content: [{ type: "text", text: summaryMessage }], + display: true, + details: { + swarmName: def.name, + status: result.status, + iterations: result.iterations, + errorCount: result.errors.length, + }, + }, + { triggerTurn: false }, + ); +} + +// ============================================================================ +// /swarm status +// ============================================================================ + +async function handleStatus(name: string | undefined, ctx: ExtensionCommandContext): Promise { + if (!name) { + ctx.ui.notify("Usage: /swarm status (reads .swarm_/state/pipeline.json from cwd)", "info"); + return; + } + + const stateTracker = new StateTracker(ctx.cwd, name); + const state = await stateTracker.load(); + if (!state) { + ctx.ui.notify(`No state found for swarm '${name}' in ${ctx.cwd}`, "error"); + return; + } + + const lines = renderSwarmProgress(state); + ctx.ui.notify(lines.join("\n"), "info"); +} + +// ============================================================================ +// Helpers +// ============================================================================ + +function buildSummaryMessage( + def: SwarmDefinition, + result: { status: string; iterations: number; errors: string[] }, + stateTracker: StateTracker, + workspace: string, +): string { + const lines: string[] = []; + lines.push(`## Swarm Pipeline: ${def.name}`); + lines.push(""); + lines.push(`- **Status**: ${result.status}`); + lines.push(`- **Mode**: ${def.mode}`); + lines.push(`- **Iterations**: ${result.iterations}/${def.targetCount}`); + lines.push(`- **Workspace**: ${workspace}`); + lines.push(`- **State dir**: ${stateTracker.swarmDir}`); + lines.push(""); + + lines.push("### Agent Results"); + lines.push(""); + for (const [name, agent] of Object.entries(stateTracker.state.agents)) { + const duration = + agent.startedAt && agent.completedAt + ? formatDuration(agent.completedAt - agent.startedAt) + : "n/a"; + lines.push(`- **${name}**: ${agent.status} (${duration})${agent.error ? ` — ${agent.error}` : ""}`); + } + + if (result.errors.length > 0) { + lines.push(""); + lines.push("### Errors"); + lines.push(""); + for (const error of result.errors) { + lines.push(`- ${error}`); + } + } + + return lines.join("\n"); +} + +function formatDuration(ms: number): string { + if (ms < 1000) return `${ms}ms`; + if (ms < 60_000) return `${(ms / 1000).toFixed(1)}s`; + const mins = Math.floor(ms / 60_000); + const secs = Math.floor((ms % 60_000) / 1000); + return `${mins}m${secs}s`; +} diff --git a/packages/coding-agent/extensions/swarm/package.json b/packages/coding-agent/extensions/swarm/package.json new file mode 100644 index 000000000..f16052a83 --- /dev/null +++ b/packages/coding-agent/extensions/swarm/package.json @@ -0,0 +1,11 @@ +{ + "name": "omp-extension-swarm", + "version": "1.0.0", + "type": "module", + "pi": { + "extensions": ["./extension.ts"] + }, + "dependencies": { + "yaml": "^2.7.0" + } +} diff --git a/packages/coding-agent/extensions/swarm/run-pipeline.ts b/packages/coding-agent/extensions/swarm/run-pipeline.ts new file mode 100644 index 000000000..904160235 --- /dev/null +++ b/packages/coding-agent/extensions/swarm/run-pipeline.ts @@ -0,0 +1,105 @@ +/** + * Direct pipeline runner — executes a swarm pipeline outside of the TUI. + * + * Usage: bun run-pipeline.ts + */ +import * as path from "node:path"; +import * as fs from "node:fs/promises"; +import { parseSwarmYaml, validateSwarmDefinition } from "./swarm/schema"; +import { buildDependencyGraph, detectCycles, buildExecutionWaves } from "./swarm/dag"; +import { StateTracker } from "./swarm/state"; +import { PipelineController } from "./swarm/pipeline"; +import { renderSwarmProgress } from "./swarm/render"; +import { discoverAuthStorage } from "@oh-my-pi/pi-coding-agent"; +import { ModelRegistry } from "@oh-my-pi/pi-coding-agent/config/model-registry"; +import { Settings } from "@oh-my-pi/pi-coding-agent/config/settings"; + +const yamlPath = process.argv[2]; +if (!yamlPath) { + console.error("Usage: bun run-pipeline.ts "); + process.exit(1); +} + +const resolvedPath = path.resolve(yamlPath); +console.log(`Reading: ${resolvedPath}`); + +const content = await Bun.file(resolvedPath).text(); +const def = parseSwarmYaml(content); + +console.log(`Swarm: ${def.name}`); +console.log(`Mode: ${def.mode}`); +console.log(`Target count: ${def.targetCount}`); +console.log(`Agents: ${[...def.agents.keys()].join(", ")}`); + +// Validate +const errors = validateSwarmDefinition(def); +if (errors.length > 0) { + console.error("Validation errors:", errors); + process.exit(1); +} + +// Build DAG +const deps = buildDependencyGraph(def); +const cycles = detectCycles(deps); +if (cycles) { + console.error("Cycle detected:", cycles); + process.exit(1); +} +const waves = buildExecutionWaves(deps); +console.log(`Waves: ${waves.map((w, i) => `W${i + 1}:[${w.join(",")}]`).join(" -> ")}`); + +// Resolve workspace +const workspace = path.isAbsolute(def.workspace) + ? def.workspace + : path.resolve(path.dirname(resolvedPath), def.workspace); + +await fs.mkdir(workspace, { recursive: true }); +console.log(`Workspace: ${workspace}`); + +// Initialize +const stateTracker = new StateTracker(workspace, def.name); +await stateTracker.init([...def.agents.keys()], def.targetCount, def.mode); + +// Auth + settings +const authStorage = await discoverAuthStorage(); +const modelRegistry = new ModelRegistry(authStorage); +const settings = Settings.isolated(); + +// Progress display +let lastProgressDump = 0; +const PROGRESS_INTERVAL_MS = 5000; + +// Run +console.log("\n--- Pipeline starting ---\n"); + +const controller = new PipelineController(def, waves, stateTracker); +const result = await controller.run({ + workspace, + onProgress: () => { + const now = Date.now(); + if (now - lastProgressDump > PROGRESS_INTERVAL_MS) { + lastProgressDump = now; + const lines = renderSwarmProgress(stateTracker.state); + console.log(lines.join("\n")); + console.log(); + } + }, + authStorage, + modelRegistry, + settings, +}); + +console.log("\n--- Pipeline finished ---\n"); +console.log(`Status: ${result.status}`); +console.log(`Iterations completed: ${result.iterations}/${def.targetCount}`); +if (result.errors.length > 0) { + console.log(`Errors (${result.errors.length}):`); + for (const err of result.errors) { + console.log(` - ${err}`); + } +} +console.log(`\nState saved to: ${stateTracker.swarmDir}`); + +// Final state dump +const lines = renderSwarmProgress(stateTracker.state); +console.log(lines.join("\n")); diff --git a/packages/coding-agent/extensions/swarm/swarm/dag.ts b/packages/coding-agent/extensions/swarm/swarm/dag.ts new file mode 100644 index 000000000..7efc1cbba --- /dev/null +++ b/packages/coding-agent/extensions/swarm/swarm/dag.ts @@ -0,0 +1,146 @@ +/** + * Directed Acyclic Graph operations for swarm agent dependencies. + * + * Builds a dependency graph from waits_for / reports_to relationships, + * detects cycles, and produces execution waves via topological sort. + */ +import type { SwarmDefinition } from "./schema"; + +/** + * Build a dependency map: agent name → set of agents it depends on. + * + * Dependencies come from: + * 1. Explicit `waits_for` declarations + * 2. Implicit from `reports_to` (if A reports_to B, then B depends on A) + * 3. For pipeline/sequential mode with no explicit deps: chain by YAML declaration order + */ +export function buildDependencyGraph(def: SwarmDefinition): Map> { + const deps = new Map>(); + + for (const name of def.agents.keys()) { + deps.set(name, new Set()); + } + + // Explicit waits_for + for (const [name, agent] of def.agents) { + for (const dep of agent.waitsFor) { + if (deps.has(dep)) { + deps.get(name)!.add(dep); + } + } + } + + // reports_to implies the target waits for the reporter + for (const [name, agent] of def.agents) { + for (const target of agent.reportsTo) { + if (deps.has(target)) { + deps.get(target)!.add(name); + } + } + } + + // For pipeline/sequential with no explicit deps, chain by declaration order + if ((def.mode === "pipeline" || def.mode === "sequential") && !hasExplicitDeps(deps)) { + for (let i = 1; i < def.agentOrder.length; i++) { + deps.get(def.agentOrder[i])!.add(def.agentOrder[i - 1]); + } + } + + return deps; +} + +function hasExplicitDeps(deps: Map>): boolean { + for (const s of deps.values()) { + if (s.size > 0) return true; + } + return false; +} + +/** + * Detect cycles in the dependency graph. + * Returns the names of agents involved in cycles, or null if acyclic. + */ +export function detectCycles(deps: Map>): string[] | null { + // Kahn's algorithm: if topological sort doesn't include all nodes, cycles exist + const inDegree = new Map(); + const forward = new Map(); // dependency → its dependents + + for (const [node, nodeDeps] of deps) { + inDegree.set(node, nodeDeps.size); + for (const dep of nodeDeps) { + const list = forward.get(dep) ?? []; + list.push(node); + forward.set(dep, list); + } + } + + const queue: string[] = []; + for (const [node, degree] of inDegree) { + if (degree === 0) queue.push(node); + } + + const sorted: string[] = []; + while (queue.length > 0) { + const node = queue.shift()!; + sorted.push(node); + for (const dependent of forward.get(node) ?? []) { + const newDegree = inDegree.get(dependent)! - 1; + inDegree.set(dependent, newDegree); + if (newDegree === 0) queue.push(dependent); + } + } + + if (sorted.length < deps.size) { + return [...deps.keys()].filter(k => !sorted.includes(k)); + } + + return null; +} + +/** + * Build execution waves from dependency graph via topological sort. + * + * Each wave contains agents whose dependencies are all in earlier waves. + * Agents within a wave can execute in parallel. + */ +export function buildExecutionWaves(deps: Map>): string[][] { + const waves: string[][] = []; + const completed = new Set(); + const remaining = new Set(deps.keys()); + + while (remaining.size > 0) { + const wave: string[] = []; + + for (const node of remaining) { + const nodeDeps = deps.get(node)!; + let ready = true; + for (const dep of nodeDeps) { + if (!completed.has(dep)) { + ready = false; + break; + } + } + if (ready) { + wave.push(node); + } + } + + if (wave.length === 0) { + throw new Error( + `Deadlock: agents [${[...remaining].join(", ")}] cannot make progress. This indicates a bug in cycle detection.`, + ); + } + + // Sort for deterministic execution order + wave.sort(); + + for (const node of wave) { + remaining.delete(node); + completed.add(node); + } + + waves.push(wave); + } + + return waves; +} diff --git a/packages/coding-agent/extensions/swarm/swarm/executor.ts b/packages/coding-agent/extensions/swarm/swarm/executor.ts new file mode 100644 index 000000000..cc9eb416e --- /dev/null +++ b/packages/coding-agent/extensions/swarm/swarm/executor.ts @@ -0,0 +1,108 @@ +/** + * Swarm agent execution via oh-my-pi's subagent infrastructure. + * + * Wraps `runSubprocess` to spawn individual swarm agents with full tool access. + * Each agent runs in the swarm workspace with its task instructions as the user prompt. + */ +import * as path from "node:path"; +import type { AuthStorage } from "@oh-my-pi/pi-coding-agent/session/auth-storage"; +import type { ModelRegistry } from "@oh-my-pi/pi-coding-agent/config/model-registry"; +import type { Settings } from "@oh-my-pi/pi-coding-agent/config/settings"; +import { runSubprocess } from "@oh-my-pi/pi-coding-agent/task/executor"; +import type { AgentDefinition, AgentProgress, AgentSource, SingleResult } from "@oh-my-pi/pi-coding-agent/task/types"; +import type { SwarmAgent } from "./schema"; +import type { StateTracker } from "./state"; + +export interface SwarmExecutorOptions { + workspace: string; + swarmName: string; + iteration: number; + modelOverride?: string; + signal?: AbortSignal; + onProgress?: (agentName: string, progress: AgentProgress) => void; + authStorage?: AuthStorage; + modelRegistry?: ModelRegistry; + settings?: Settings; + stateTracker: StateTracker; +} + +/** + * Execute a single swarm agent as an oh-my-pi subagent. + * + * The agent receives: + * - System prompt: built from role + extra_context + * - User prompt (task): the full task instructions from the YAML + * - Working directory: the swarm workspace + * - Full tool access (bash, python, read, write, edit, grep, find, fetch, web_search, browser) + */ +export async function executeSwarmAgent( + agent: SwarmAgent, + index: number, + options: SwarmExecutorOptions, +): Promise { + const { workspace, swarmName, iteration, modelOverride, signal, onProgress, authStorage, modelRegistry, settings, stateTracker } = options; + + const agentId = `swarm-${swarmName}-${agent.name}-${iteration}`; + + const agentDef: AgentDefinition = { + name: agent.name, + description: `Swarm agent: ${agent.role}`, + systemPrompt: buildSystemPrompt(agent), + source: "project" as AgentSource, + }; + + await stateTracker.updateAgent(agent.name, { + status: "running", + iteration, + startedAt: Date.now(), + }); + await stateTracker.appendLog(agent.name, `Starting iteration ${iteration}`); + + try { + const result = await runSubprocess({ + cwd: workspace, + agent: agentDef, + task: agent.task, + index, + id: agentId, + modelOverride, + signal, + onProgress: (progress) => onProgress?.(agent.name, progress), + authStorage, + modelRegistry, + settings, + enableLsp: false, + artifactsDir: path.join(stateTracker.swarmDir, "context"), + }); + + const status = result.exitCode === 0 ? "completed" as const : "failed" as const; + await stateTracker.updateAgent(agent.name, { + status, + completedAt: Date.now(), + error: result.error, + }); + await stateTracker.appendLog( + agent.name, + `Iteration ${iteration} ${status}${result.error ? `: ${result.error}` : ""}`, + ); + + return result; + } catch (err) { + const error = err instanceof Error ? err.message : String(err); + await stateTracker.updateAgent(agent.name, { + status: "failed", + completedAt: Date.now(), + error, + }); + await stateTracker.appendLog(agent.name, `Iteration ${iteration} error: ${error}`); + throw err; + } +} + +function buildSystemPrompt(agent: SwarmAgent): string { + const parts = [`You are a ${agent.role}.`]; + if (agent.extraContext) { + parts.push(agent.extraContext); + } + return parts.join("\n\n"); +} diff --git a/packages/coding-agent/extensions/swarm/swarm/pipeline.ts b/packages/coding-agent/extensions/swarm/swarm/pipeline.ts new file mode 100644 index 000000000..6573ac814 --- /dev/null +++ b/packages/coding-agent/extensions/swarm/swarm/pipeline.ts @@ -0,0 +1,219 @@ +/** + * Pipeline controller for swarm execution. + * + * Orchestrates execution waves within each iteration: + * - Agents in the same wave execute in parallel + * - Waves execute sequentially (wave N+1 starts after wave N completes) + * - For pipeline mode, iterations repeat the full DAG execution + */ +import type { AuthStorage } from "@oh-my-pi/pi-coding-agent/session/auth-storage"; +import type { ModelRegistry } from "@oh-my-pi/pi-coding-agent/config/model-registry"; +import type { Settings } from "@oh-my-pi/pi-coding-agent/config/settings"; +import type { AgentProgress, AgentSource, SingleResult } from "@oh-my-pi/pi-coding-agent/task/types"; +import type { SwarmDefinition } from "./schema"; +import type { StateTracker } from "./state"; +import { executeSwarmAgent } from "./executor"; + +// ============================================================================ +// Types +// ============================================================================ + +export interface PipelineOptions { + workspace: string; + signal?: AbortSignal; + onProgress?: (state: PipelineProgress) => void; + authStorage?: AuthStorage; + modelRegistry?: ModelRegistry; + settings?: Settings; +} + +export interface PipelineProgress { + iteration: number; + targetCount: number; + currentWave: number; + totalWaves: number; + agents: Record; +} + +export interface PipelineResult { + status: "completed" | "failed" | "aborted"; + iterations: number; + agentResults: Map; + errors: string[]; +} + +// ============================================================================ +// Controller +// ============================================================================ + +export class PipelineController { + #def: SwarmDefinition; + #waves: string[][]; + #stateTracker: StateTracker; + + constructor(def: SwarmDefinition, waves: string[][], stateTracker: StateTracker) { + this.#def = def; + this.#waves = waves; + this.#stateTracker = stateTracker; + } + + async run(options: PipelineOptions): Promise { + const { workspace, signal, onProgress, authStorage, modelRegistry, settings } = options; + const allResults = new Map(); + const errors: string[] = []; + + for (const name of this.#def.agents.keys()) { + allResults.set(name, []); + } + + const targetCount = this.#def.targetCount; + + await this.#stateTracker.appendOrchestratorLog( + `Pipeline '${this.#def.name}' starting: mode=${this.#def.mode} iterations=${targetCount} waves=${this.#waves.length} agents=${this.#def.agents.size}`, + ); + + try { + for (let iteration = 0; iteration < targetCount; iteration++) { + if (signal?.aborted) { + await this.#stateTracker.updatePipeline({ status: "aborted" }); + return { status: "aborted", iterations: iteration, agentResults: allResults, errors }; + } + + await this.#stateTracker.updatePipeline({ iteration }); + await this.#stateTracker.appendOrchestratorLog( + `--- Iteration ${iteration + 1}/${targetCount} ---`, + ); + + const emitProgress = (currentWave: number) => { + onProgress?.({ + iteration, + targetCount, + currentWave, + totalWaves: this.#waves.length, + agents: this.#buildProgressSnapshot(), + }); + }; + + const iterationResults = await this.#runIteration(iteration, { + workspace, + signal, + emitProgress, + authStorage, + modelRegistry, + settings, + }); + + for (const [agentName, result] of iterationResults) { + allResults.get(agentName)!.push(result); + if (result.exitCode !== 0) { + errors.push(`${agentName} (iteration ${iteration + 1}): ${result.error || "exit code " + result.exitCode}`); + } + } + } + + const status = errors.length > 0 ? "failed" as const : "completed" as const; + await this.#stateTracker.updatePipeline({ status, completedAt: Date.now() }); + await this.#stateTracker.appendOrchestratorLog(`Pipeline ${status} (${errors.length} errors)`); + return { status, iterations: targetCount, agentResults: allResults, errors }; + } catch (err) { + const error = err instanceof Error ? err.message : String(err); + await this.#stateTracker.updatePipeline({ status: "failed", completedAt: Date.now() }); + await this.#stateTracker.appendOrchestratorLog(`Pipeline fatal error: ${error}`); + errors.push(error); + return { status: "failed", iterations: 0, agentResults: allResults, errors }; + } + } + + async #runIteration( + iteration: number, + options: { + workspace: string; + signal?: AbortSignal; + emitProgress: (currentWave: number) => void; + authStorage?: AuthStorage; + modelRegistry?: ModelRegistry; + settings?: Settings; + }, + ): Promise> { + const results = new Map(); + let agentIndex = 0; + + for (let waveIdx = 0; waveIdx < this.#waves.length; waveIdx++) { + const wave = this.#waves[waveIdx]; + + if (options.signal?.aborted) break; + + await this.#stateTracker.appendOrchestratorLog( + `Wave ${waveIdx + 1}/${this.#waves.length}: [${wave.join(", ")}]`, + ); + + // Mark agents in this wave as waiting + for (const agentName of wave) { + await this.#stateTracker.updateAgent(agentName, { + status: "waiting", + iteration, + wave: waveIdx, + }); + } + options.emitProgress(waveIdx); + + // Execute all agents in wave in parallel, catching per-agent errors + const waveResults = await Promise.all( + wave.map(async (agentName) => { + const agent = this.#def.agents.get(agentName)!; + const currentIndex = agentIndex++; + try { + const result = await executeSwarmAgent(agent, currentIndex, { + workspace: options.workspace, + swarmName: this.#def.name, + iteration, + modelOverride: this.#def.model, + signal: options.signal, + onProgress: (_name, _progress) => { + options.emitProgress(waveIdx); + }, + authStorage: options.authStorage, + modelRegistry: options.modelRegistry, + settings: options.settings, + stateTracker: this.#stateTracker, + }); + return { agentName, result }; + } catch (err) { + const error = err instanceof Error ? err.message : String(err); + const failResult: SingleResult = { + index: currentIndex, + id: `swarm-${this.#def.name}-${agentName}-${iteration}`, + agent: agentName, + agentSource: "project" as AgentSource, + task: agent.task, + exitCode: 1, + output: "", + stderr: error, + truncated: false, + durationMs: 0, + tokens: 0, + error, + }; + return { agentName, result: failResult }; + } + }), + ); + + for (const { agentName, result } of waveResults) { + results.set(agentName, result); + } + + options.emitProgress(waveIdx); + } + + return results; + } + + #buildProgressSnapshot(): Record { + const snapshot: Record = {}; + for (const [name, agent] of Object.entries(this.#stateTracker.state.agents)) { + snapshot[name] = { status: agent.status, iteration: agent.iteration }; + } + return snapshot; + } +} diff --git a/packages/coding-agent/extensions/swarm/swarm/render.ts b/packages/coding-agent/extensions/swarm/swarm/render.ts new file mode 100644 index 000000000..f0995ff2f --- /dev/null +++ b/packages/coding-agent/extensions/swarm/swarm/render.ts @@ -0,0 +1,75 @@ +/** + * TUI progress rendering for swarm pipeline status. + */ +import type { SwarmState } from "./state"; + +const STATUS_LABELS: Record = { + completed: "[done]", + running: "[....]", + failed: "[FAIL]", + pending: "[ ]", + waiting: "[wait]", + idle: "[idle]", + aborted: "[stop]", +}; + +export function renderSwarmProgress(state: SwarmState): string[] { + const lines: string[] = []; + + const statusLabel = state.status.toUpperCase(); + lines.push(`Swarm: ${state.name} [${statusLabel}]`); + lines.push(`Mode: ${state.mode} | Iteration: ${state.iteration + 1}/${state.targetCount}`); + lines.push(""); + + const agents = Object.values(state.agents); + if (agents.length === 0) { + lines.push(" (no agents)"); + return lines; + } + + for (const agent of agents) { + const icon = STATUS_LABELS[agent.status] ?? "[????]"; + const duration = formatAgentDuration(agent); + const errorSuffix = agent.error ? ` - ${truncate(agent.error, 60)}` : ""; + lines.push(` ${icon} ${agent.name}: ${agent.status}${duration}${errorSuffix}`); + } + + // Summary line + const completed = agents.filter((a) => a.status === "completed").length; + const failed = agents.filter((a) => a.status === "failed").length; + const running = agents.filter((a) => a.status === "running").length; + + lines.push(""); + const parts = [`${completed}/${agents.length} done`]; + if (running > 0) parts.push(`${running} running`); + if (failed > 0) parts.push(`${failed} failed`); + if (state.startedAt) { + parts.push(`elapsed: ${formatDuration(Date.now() - state.startedAt)}`); + } + lines.push(` ${parts.join(" | ")}`); + + return lines; +} + +function formatAgentDuration(agent: { startedAt?: number; completedAt?: number; status: string }): string { + if (agent.startedAt && agent.completedAt) { + return ` (${formatDuration(agent.completedAt - agent.startedAt)})`; + } + if (agent.startedAt && (agent.status === "running" || agent.status === "waiting")) { + return ` (${formatDuration(Date.now() - agent.startedAt)}...)`; + } + return ""; +} + +function formatDuration(ms: number): string { + if (ms < 1000) return `${ms}ms`; + if (ms < 60_000) return `${(ms / 1000).toFixed(1)}s`; + const mins = Math.floor(ms / 60_000); + const secs = Math.floor((ms % 60_000) / 1000); + return `${mins}m${secs}s`; +} + +function truncate(str: string, maxLen: number): string { + if (str.length <= maxLen) return str; + return `${str.slice(0, maxLen - 1)}…`; +} diff --git a/packages/coding-agent/extensions/swarm/swarm/schema.ts b/packages/coding-agent/extensions/swarm/swarm/schema.ts new file mode 100644 index 000000000..83958b63a --- /dev/null +++ b/packages/coding-agent/extensions/swarm/swarm/schema.ts @@ -0,0 +1,146 @@ +/** + * YAML schema parsing, validation, and normalized types for swarm definitions. + */ +import YAML from "yaml"; + +// ============================================================================ +// Raw YAML shape (snake_case, optional fields) +// ============================================================================ + +interface RawSwarmAgentConfig { + role: string; + task: string; + extra_context?: string; + reports_to?: string[]; + waits_for?: string[]; +} + +interface RawSwarmConfig { + name: string; + workspace: string; + mode?: string; + target_count?: number; + model?: string; + agents: Record; +} + +// ============================================================================ +// Normalized types (camelCase, defaults applied) +// ============================================================================ + +export type SwarmMode = "pipeline" | "parallel" | "sequential"; + +export interface SwarmAgent { + name: string; + role: string; + task: string; + extraContext?: string; + reportsTo: string[]; + waitsFor: string[]; +} + +export interface SwarmDefinition { + name: string; + workspace: string; + mode: SwarmMode; + targetCount: number; + model?: string; + agents: Map; + /** Preserves YAML declaration order for implicit pipeline sequencing. */ + agentOrder: string[]; +} + +// ============================================================================ +// Parsing +// ============================================================================ + +const VALID_MODES = new Set(["pipeline", "parallel", "sequential"]); + +export function parseSwarmYaml(content: string): SwarmDefinition { + const raw = YAML.parse(content) as { swarm?: RawSwarmConfig } | null; + if (!raw?.swarm) { + throw new Error("YAML must have a top-level 'swarm' key"); + } + const swarm = raw.swarm; + + if (!swarm.name || typeof swarm.name !== "string") { + throw new Error("swarm.name is required and must be a string"); + } + if (!swarm.workspace || typeof swarm.workspace !== "string") { + throw new Error("swarm.workspace is required and must be a string"); + } + if (!swarm.agents || typeof swarm.agents !== "object" || Object.keys(swarm.agents).length === 0) { + throw new Error("swarm.agents must contain at least one agent"); + } + + const mode = swarm.mode ?? "sequential"; + if (!VALID_MODES.has(mode)) { + throw new Error(`Invalid mode '${mode}'. Must be one of: ${[...VALID_MODES].join(", ")}`); + } + + const agentOrder: string[] = []; + const agents = new Map(); + + for (const [name, config] of Object.entries(swarm.agents)) { + if (!config.role || typeof config.role !== "string") { + throw new Error(`Agent '${name}': 'role' is required`); + } + if (!config.task || typeof config.task !== "string") { + throw new Error(`Agent '${name}': 'task' is required`); + } + + agentOrder.push(name); + agents.set(name, { + name, + role: config.role, + task: config.task.trim(), + extraContext: config.extra_context?.trim(), + reportsTo: Array.isArray(config.reports_to) ? config.reports_to : [], + waitsFor: Array.isArray(config.waits_for) ? config.waits_for : [], + }); + } + + return { + name: swarm.name, + workspace: swarm.workspace, + mode: mode as SwarmMode, + targetCount: swarm.target_count ?? 1, + model: swarm.model, + agents, + agentOrder, + }; +} + +// ============================================================================ +// Validation (semantic — references, constraints) +// ============================================================================ + +export function validateSwarmDefinition(def: SwarmDefinition): string[] { + const errors: string[] = []; + const agentNames = new Set(def.agents.keys()); + + for (const [name, agent] of def.agents) { + for (const dep of agent.waitsFor) { + if (!agentNames.has(dep)) { + errors.push(`Agent '${name}' waits_for unknown agent '${dep}'`); + } + if (dep === name) { + errors.push(`Agent '${name}' cannot wait for itself`); + } + } + for (const target of agent.reportsTo) { + if (!agentNames.has(target)) { + errors.push(`Agent '${name}' reports_to unknown agent '${target}'`); + } + if (target === name) { + errors.push(`Agent '${name}' cannot report to itself`); + } + } + } + + if (def.targetCount < 1) { + errors.push("target_count must be at least 1"); + } + + return errors; +} diff --git a/packages/coding-agent/extensions/swarm/swarm/state.ts b/packages/coding-agent/extensions/swarm/swarm/state.ts new file mode 100644 index 000000000..4d818991d --- /dev/null +++ b/packages/coding-agent/extensions/swarm/swarm/state.ts @@ -0,0 +1,127 @@ +/** + * Filesystem state tracker for swarm pipeline execution. + * + * Persists pipeline and per-agent state to `.swarm_/` in the workspace. + * Supports resumability by loading state from disk. + */ +import * as fs from "node:fs/promises"; +import * as path from "node:path"; + +// ============================================================================ +// State types +// ============================================================================ + +export type PipelineStatus = "idle" | "running" | "completed" | "failed" | "aborted"; +export type AgentStatus = "pending" | "waiting" | "running" | "completed" | "failed"; + +export interface AgentState { + name: string; + status: AgentStatus; + iteration: number; + wave: number; + startedAt?: number; + completedAt?: number; + error?: string; +} + +export interface SwarmState { + name: string; + status: PipelineStatus; + mode: string; + iteration: number; + targetCount: number; + agents: Record; + startedAt: number; + completedAt?: number; +} + +// ============================================================================ +// State tracker +// ============================================================================ + +export class StateTracker { + #swarmDir: string; + #state: SwarmState; + + constructor(workspaceDir: string, name: string) { + this.#swarmDir = path.join(workspaceDir, `.swarm_${name}`); + this.#state = { + name, + status: "idle", + mode: "sequential", + iteration: 0, + targetCount: 1, + agents: {}, + startedAt: Date.now(), + }; + } + + get swarmDir(): string { + return this.#swarmDir; + } + + get state(): Readonly { + return this.#state; + } + + async init(agentNames: string[], targetCount: number, mode: string): Promise { + await fs.mkdir(path.join(this.#swarmDir, "state"), { recursive: true }); + await fs.mkdir(path.join(this.#swarmDir, "logs"), { recursive: true }); + await fs.mkdir(path.join(this.#swarmDir, "context"), { recursive: true }); + + this.#state.targetCount = targetCount; + this.#state.mode = mode; + this.#state.status = "running"; + this.#state.startedAt = Date.now(); + + for (const name of agentNames) { + this.#state.agents[name] = { + name, + status: "pending", + iteration: 0, + wave: 0, + }; + } + + await this.#persist(); + } + + async updateAgent(name: string, update: Partial): Promise { + const agent = this.#state.agents[name]; + if (!agent) return; + Object.assign(agent, update); + await this.#persist(); + } + + async updatePipeline(update: Partial): Promise { + Object.assign(this.#state, update); + await this.#persist(); + } + + async appendLog(agentName: string, message: string): Promise { + const logPath = path.join(this.#swarmDir, "logs", `${agentName}.log`); + const timestamp = new Date().toISOString(); + await fs.appendFile(logPath, `[${timestamp}] ${message}\n`); + } + + async appendOrchestratorLog(message: string): Promise { + const logPath = path.join(this.#swarmDir, "logs", "orchestrator.log"); + const timestamp = new Date().toISOString(); + await fs.appendFile(logPath, `[${timestamp}] ${message}\n`); + } + + async load(): Promise { + const statePath = path.join(this.#swarmDir, "state", "pipeline.json"); + try { + const content = await Bun.file(statePath).text(); + this.#state = JSON.parse(content) as SwarmState; + return this.#state; + } catch { + return null; + } + } + + async #persist(): Promise { + await Bun.write(path.join(this.#swarmDir, "state", "pipeline.json"), JSON.stringify(this.#state, null, 2)); + } +} diff --git a/packages/coding-agent/extensions/swarm/tsconfig.json b/packages/coding-agent/extensions/swarm/tsconfig.json new file mode 100644 index 000000000..6dec4c6cd --- /dev/null +++ b/packages/coding-agent/extensions/swarm/tsconfig.json @@ -0,0 +1,16 @@ +{ + "compilerOptions": { + "target": "ESNext", + "module": "ESNext", + "moduleResolution": "bundler", + "strict": true, + "noEmit": true, + "skipLibCheck": true, + "types": ["bun-types"], + "paths": { + "@oh-my-pi/pi-coding-agent": ["../../src/index.ts"], + "@oh-my-pi/pi-coding-agent/*": ["../../src/*.ts"] + } + }, + "include": ["extension.ts", "swarm/**/*.ts"] +}