Swarm Executor

intent

swarm executor coordinates multi-agent task execution using standardized SwarmCommand messages, pub/sub orchestration, token-budget governance, and explicit state tracking. use this skill when your task requires multiple agents to work in parallel with clear ownership, bounded cost control, and verifiable completion without duplicating work or burning uncontrolled tokens. avoid it for single-agent tasks. swarm overhead only pays off when parallelism improves throughput, latency, or quality while maintaining strict control over cost, state, and ownership.

inputs

external connections

• redis pub/sub (optional): set env var SWARM_REDIS_URL (e.g., redis://localhost:6379). if not set, skill falls back to in-memory queue. requires read/write permissions on channels named tasks, results, negotiation.
• agent registry: local or remote mapping of agent types to agent ids. required format: json dict with keys like {"developer": ["id_001", "id_002"], "reviewer": ["id_003"]}. pass as env var SWARM_AGENT_REGISTRY or inline dict.
• token oracle: external service or local function to check remaining token budget per agent/tier. if not available, manual budget tracking is required (less safe).

parameters

• coordinator_id: string, identifies the coordinator agent sending commands. default: "coordinator_001".
• use_redis: boolean, use redis or in-memory pub/sub. default: true if SWARM_REDIS_URL set, else false.
• command_schema_path: string, path to json schema for swarmcommand validation. default: schemas/swarm_command.json (bundled).
• negotiation_timeout_sec: int, max seconds to wait for agent negotiation before coordinator decides. default: 300.
• heartbeat_timeout_sec: int, max seconds an agent can stay in in_progress before triggering fallback. default: 600.

context

• agent roles in current workflow (coordinator, executor, reviewer, monitor, fallback owner)
• task dependencies and hierarchy (if any subtasks block others)
• per-agent token budget and tier (e.g., gpt-4 vs gpt-3.5)
• completion criteria and required artifacts for each task

procedure

step 1: assess swarm necessity

inputs: task description, estimated complexity, available agents, latency requirement.

action: evaluate whether swarm adds value. swarm is justified if at least one is true:

• task naturally splits into independent subtasks
• different agents have distinct roles (e.g., coder vs reviewer)
• review must be separated from implementation
• parallel work reduces latency meaningfully
• constraints conflict and require negotiation

output: boolean decision (swarm_justified: true/false) and reasoning string.

edge case: if task is already complex, adding swarm coordination can reduce clarity. document why swarm improves control before proceeding.

---

step 2: define roles and ownership model

inputs: agent registry (mapping of role to agent ids), task scope.

action: assign roles explicitly:

• coordinator: publishes commands, resolves conflicts, gates high-risk actions
• executor(s): performs work on assigned tasks (one per task unless explicitly coordinated)
• reviewer: validates output before completion
• monitor: tracks budget, deadline, and heartbeat; triggers fallback if needed
• fallback_owner: escalation target if primary executor fails

document current owner of each task (initially null until assigned).

output: role assignment dict in json format:

{
  "coordinator": {"type": "coordinator", "id": "001"},
  "executors": [
    {"type": "developer", "id": "003"},
    {"type": "data_analyst", "id": "004"}
  ],
  "reviewer": {"type": "auditor", "id": "005"},
  "monitor": {"type": "monitor", "id": "006"},
  "fallback_owner": {"type": "developer", "id": "002"}
}

edge case: if an agent must play multiple roles (e.g., coder + reviewer in small teams), explicitly note the conflict in completion gate (e.g., "coder implements; separate human review required").

---

step 3: create standardized SwarmCommand for each task

inputs: task description, assigned target agent, token budget, deadline, dependencies, completion gate.

action: construct a SwarmCommand json object with required fields:

{
  "command_id": "cmd_<8-char-uuid>",
  "timestamp": "<ISO 8601 UTC>",
  "sender": {"type": "coordinator", "id": "<coordinator_id>"},
  "target": {"type": "<agent_type>", "id": "<agent_id>"},
  "command": {
    "action": "<verb>",
    "module": "<scope>",
    "requirements": ["<req1>", "<req2>"],
    "output_format": "<format>"
  },
  "metadata": {
    "priority": "<low|medium|high|critical>",
    "token_budget": <int>,
    "deadline": "<ISO 8601 UTC>",
    "dependencies": ["cmd_<id>", ...]
  },
  "negotiation": {
    "allowed": <boolean>,
    "timeout": <int seconds>
  }
}

use bundled coordinator/swarm_protocol.py for deterministic creation and validation; do not hardcode role names in prompt text.

output: validated SwarmCommand json object.

edge cases:

• token_budget must be positive and within agent tier quota (validate against token oracle).
• deadline must be realistic relative to task complexity and available agents.
• if dependencies list is empty, explicitly set to [] (not null).
• if negotiation is not needed, set allowed: false.

---

step 4: validate command before publishing

inputs: SwarmCommand object from step 3.

action: run validation against schema:

• schema format matches schemas/swarm_command.json
• sender and target types exist in agent registry
• priority is one of low | medium | high | critical
• token_budget > 0 and <= agent tier allowance
• dependencies all reference existing command_ids or are empty
• deadline is >= now and reasonable for task scope
• completion gate (in command metadata or separate) is explicit and testable

use bundled coordinator/swarm_protocol.py method validate_command().

output: tuple (is_valid: bool, errors: list[str]).

if not valid, return errors to user and request correction or auto-fix safe fields. do not publish invalid commands.

edge case: if token oracle is unavailable, skip budget tier check and warn in logs; manual review required.

---

step 5: publish to pub/sub and initialize state tracking

inputs: validated SwarmCommand, pub/sub backend (redis or in-memory), agent registry.

action:

• publish command to pub/sub channel tasks (topic key = agent type, e.g., tasks:developer)
• initialize state record in tracking store: {command_id: "cmd_...", state: "pending", assigned_to: null, created_at: "...", started_at: null, completed_at: null, result: null}
• subscribe coordinator to result channel results:<command_id> to watch for completion
• start heartbeat monitor: if command is in in_progress and no heartbeat received within heartbeat_timeout_sec, trigger fallback (step 8)

use coordinator/pubsub.py class PubSubCoordinator for safe publish/subscribe with error handling.

output: command published successfully (no exceptions), state record created, monitor started.

edge cases:

• if redis connection fails and fallback is disabled, raise exception immediately.
• if in-memory queue is used, state tracking is local to process; swarm cannot span multiple processes.
• rate limiting: if coordinator publishes > 10 commands/sec, buffer and throttle to prevent queue overload.

---

step 6: assign task and transition state

inputs: published command_id, target agent_id from registry.

action:

• set command state to assigned
• set assigned_to field to target agent_id
• publish assignment notification to tasks:<target_agent_id> channel
• agent receives assignment, reads command details, confirms readiness (or negotiates in step 7)

output: state transitions from pending to assigned, assigned_to field populated.

edge cases:

• if target agent is unavailable (not in registry or offline), do not assign. escalate to fallback owner (step 8).
• if assignment notification times out (no ack from agent after 30 sec), consider agent offline and reroute.

---

step 7: negotiate if needed (optional)

inputs: command with negotiation.allowed = true, agent response indicating conflict (e.g., insufficient token budget, deadline unrealistic, missing capability).

action:

• agent publishes negotiation message to negotiation:<command_id> channel with reason and counter-proposal (e.g., request +500 token budget, extend deadline by 30 min, or decline if capability missing)
• coordinator receives negotiation within negotiation_timeout_sec window
• coordinator evaluates counter-proposal against budget constraints and available agents
• coordinator publishes decision: approve counter-proposal, reject and assign to different agent, or escalate to human

negotiation output must be a decision, not endless loop:

{
  "decision": "approve_counter_proposal | reject_reroute | escalate",
  "reason": "<explanation>",
  "budget_adjustment": <int or 0>,
  "deadline_adjustment_sec": <int or 0>,
  "new_assigned_to": "<agent_id or null>",
  "blocked": false
}

output: negotiation decision published, state updated accordingly.

edge cases:

• if negotiation timeout expires without decision, coordinator chooses automatically (usually reroute or escalate) rather than hanging.
• if agent negotiates > 2 times for same command, require human confirmation before retry.

---

step 8: monitor execution and trigger fallback

inputs: command in in_progress state, heartbeat_timeout_sec parameter, fallback_owner from role assignment.

action: run continuous monitor:

• poll command state at 30-sec intervals
• if heartbeat (agent publishes progress update) received within heartbeat_timeout_sec, reset timer
• if heartbeat not received within timeout, agent is presumed stuck or dead
• transition command state to failed
• publish command to fallback_owner assignment (agent_id from step 2 role definition)
• fallback owner receives command in pending state with original requirements plus note: "previous executor failed after ; requires fresh start"
• log failure reason as "heartbeat_timeout" for post-mortem

output: command reassigned to fallback owner, state transitions from in_progress to failed to pending (new assignment).

edge cases:

• if fallback_owner is also unresponsive, escalate to human with full command history and logs.
• if command has been reassigned > 2 times, require coordinator (human) review before further reroute.
• if budget is exhausted during fallback cycle, pause and request confirmation.

---

step 9: collect result and validate completion

inputs: executor publishes result message to results:<command_id> channel with fields:

{
  "command_id": "cmd_...",
  "success": true|false,
  "output": "<artifact or summary>",
  "artifact_path": "<file path or url>",
  "token_usage": <int>,
  "tests_passed": true|false,
  "review_status": "approved|pending|rejected",
  "failure_reason": "<string if success=false>",
  "next_action": "<recommendation>"
}

action:

• receive result message
• check for required fields based on completion gate defined in step 3 command metadata
• if tests required but tests_passed missing, mark as incomplete and reopen with explicit artifact list
• if review required but review_status missing, send to reviewer (step 10)
• if all gates passed, transition state to completed
• deduct token_usage from per-agent budget tracker

output: state transitions to completed (if all gates passed) or incomplete (if artifacts missing). result record stored.

edge cases:

• if artifact_path points to missing file, do not mark completed. ask executor to retry or escalate.
• if token_usage exceeds budget, flag overage in logs and alert monitor. do not proceed to next parallel task until budget restored or downgrade applied.
• if success=false, transition to failed and trigger fallback (step 8).

---

step 10: review and sign off (if required by gate)

inputs: result with review_status = pending, reviewer agent_id from role assignment.

action:

• send result summary to reviewer on tasks:reviewer channel
• reviewer reads output/artifact and tests/verification results
• reviewer publishes review decision: approved | rejected | needs_rework
• if approved, transition command state to completed
• if rejected, add reason and send back to executor with rework instructions (reassign as new command_id with dependencies=[original_cmd])
• if needs_rework, same as rejected

output: command state transitions to completed or new rework command created.

edge cases:

• if reviewer is unavailable or unresponsive (no review within 1 hour), escalate to coordinator with command in in_progress state.
• if reviewer rejects without clear reason, require rework instructions before executor retry.

---

step 11: close loop and update budget

inputs: completed command, token usage recorded, result artifacts verified.

action:

• set command state to completed and populate completed_at timestamp
• record final token usage in per-agent budget tracker
• check budget remaining: if < 20%, trigger downgrade rules (step 12)
• publish completion notification to all subscribers (coordinator, monitor, original requester)
• archive command record and result for audit

output: command marked complete, budget updated, completion event published.

edge case: if command required dependent tasks, trigger those pending assignments now (dependencies satisfied).

---

step 12: apply token budget downgrade rules

inputs: per-agent token budget, current usage, remaining budget percentage.

action: evaluate budget remaining:

budget remaining	action
> 40%	continue normal execution, no change
20-40%	compress context in next commands; reduce parallel agent count; warn coordinator
< 20%	require coordinator approval before publishing new subtasks; suggest model downgrade (gpt-3.5 instead of gpt-4)
exceeded	stop publishing new subtasks immediately; mark monitor state as paused; request budget increase or human decision
repeated failure with budget pressure	lower concurrency; route next attempt to reviewer/monitor for manual execution

use bundled coordinator/token_budget.py for deterministic downgrade logic.

output: downgrade action published, monitor state updated, coordinator alerted.

edge cases:

• if downgrade requires model change, validate that target agent type (e.g., gpt-3.5) exists in registry before assignment.
• if coordinator does not respond to approval request within 5 minutes, default to paused (fail-safe).

---

step 13: handle explicit failures and log for audit

inputs: command state, failure reason from executor or monitor.

action: route failures explicitly:

failure type	handler	action
invalid command	coordinator	reject before publish; return schema errors; ask for correction
target unavailable	monitor	reroute to fallback_owner immediately
dependency blocked	monitor	keep command in pending; notify coordinator with blocked dependency list
budget exceeded	monitor	pause all new subtasks; request confirmation
deadline missed	monitor	mark command failed; log deadline violation; trigger fallback
duplicate owner	coordinator	choose owner; cancel duplicate assignment; investigate cause
incomplete result	reviewer	reopen with explicit artifact checklist; send back to executor
repeated failure (> 2 attempts)	monitor	reduce concurrency; route to reviewer/auditor for manual review
network/pub-sub error	monitor	retry publish up to 3 times; if persistent, escalate to human

publish failure record:

{
  "command_id": "cmd_...",
  "failure_type": "<type>",
  "reason": "<explanation>",
  "timestamp": "<ISO 8601>",
  "handler": "<coordinator|monitor|reviewer>",
  "action_taken": "<description>",
  "escalation_to": "<agent_id or null>"
}

output: failure logged, action taken published, escalation routed.

edge cases:

• if executor repeatedly fails same task, do not reroute indefinitely. after 2 reroutes, require human confirmation before proceeding.
• if monitor or coordinator itself fails, manual intervention required (failure of the coordinator is terminal).

---

decision points

• swarm vs single-agent (step 1): if task is simple and one agent can complete it, do not use swarm. only use if parallelism improves throughput, latency, or quality while maintaining control. if decision unclear, document both options and ask user to choose.

• negotiation needed (step 7): if executor indicates conflict (budget, deadline, capability), allow negotiation only if negotiation.allowed = true in command. if false or timeout expires, coordinator decides (approve, reject+reroute, or escalate). no open-ended loops.

• target unavailable (step 6, 8): if assigned agent is offline or not in registry, immediately reroute to fallback_owner. do not wait or retry assignment to same agent.

• budget downgrade (step 12): if remaining budget < 20%, require explicit coordinator approval before publishing new tasks. if no approval within 5 min, pause all pending work (fail-safe). do not continue silently.

• **failure escalation (step 13)