swarm-dispatch

Scheduling and dispatch for multi-agent systems. Polls task sources, routes work to idle agents or spawns new ones, tracks lifecycle, retries failures, reconciles external state. Runtime-agnostic — bring your own task source, agent runtime, message transport, and roster.

Install

npm install swarm-dispatch

Quick Start

import { createOrchestrator, createOpenTasksSource } from "swarm-dispatch";

const orchestrator = createOrchestrator(
  // Task source — where to find work
  createOpenTasksSource(opentasksClient),

  // Agent runtime — how to spawn and terminate agents
  {
    spawn: async ({ prompt, taskId, role }) => {
      const agent = await myAgentSystem.spawn({ task: prompt, role });
      return { id: agent.id };
    },
    terminate: async (agentId, reason) => {
      await myAgentSystem.terminate(agentId, reason);
    },
    onStopped: (callback) => {
      return myAgentSystem.onLifecycleEvent((event) => {
        if (event.type === "stopped") callback(event.agentId, event.reason);
      });
    },
  },

  // Config
  {
    claimantId: `${hostname}:${pid}`,
    pollIntervalMs: 15_000,
    defaultRole: "worker",
    concurrency: { global: 5 },
    retry: { maxRetries: 3, baseDelayMs: 10_000, maxDelayMs: 300_000 },
    reconcile: { enabled: true, intervalMs: 60_000, stallTimeoutMs: 300_000 },
    continuation: { delayMs: 1_000, maxTurns: 20 },
  }
);

await orchestrator.start();

// Observe events
orchestrator.onEvent((event) => console.log(event.type, event));

// Inspect state
const snap = orchestrator.snapshot();

// Manual triggers
await orchestrator.dispatchNow();
await orchestrator.reconcileNow();

// Shutdown
await orchestrator.stop();

Architecture

                              ┌───────────────────────┐
                              │     Observers          │
                              │  (events, snapshots)   │
                              └──────────┬────────────┘
                                         │
┌────────────────────────────────────────┼─────────────────────────────────┐
│                              Orchestrator                                │
│                                                                          │
│   State:  claimed · running · retryQueue · completed · totals            │
│                                                                          │
│   Tick:   reconcile → preflight → drain retries → fetch → dispatch       │
│                                                                          │
└─────┬────────────┬────────────┬────────────┬────────────┬───────────────┘
      │            │            │            │            │
      ▼            ▼            ▼            ▼            ▼
  TaskSource  MessagePort   AgentRoster  AgentRuntime   Journal
  (poll+claim) (mail I/O)   (presence)   (spawn/kill)   (recovery)
      │            │            │            │            │
      ▼            ▼            ▼            ▼            ▼
  opentasks   agent-inbox   MAP registry  macro-agent    (future)

Five adapter ports. The orchestrator never imports any specific subsystem directly.

Ports

TaskSource

Where tasks come from. Implements polling, claiming, and state transitions.

interface DispatchTaskSource {
  queryReady(opts?): Promise<DispatchTask[]>;
  claim(taskId, claimantId): Promise<ClaimResult>;
  release(taskId, claimantId, fence?): Promise<void>;
  transition(taskId, action, fence?): Promise<void>;
  getTask(taskId): Promise<DispatchTask>;
  isStillActive?(taskId): Promise<boolean>;
  listInProgress?(): Promise<DispatchTask[]>;
  renewClaim?(taskId, fence): Promise<RenewResult>;
}

Built-in adapter: createOpenTasksSource(client).

AgentRuntime

How agents are spawned and terminated.

interface DispatchAgentRuntime {
  spawn(opts: { prompt, taskId, role }): Promise<SpawnedAgent>;
  terminate(agentId, reason): Promise<void>;
  onStopped(callback): () => void;
  onUsage?(callback): () => void;
}

MessagePort (optional)

Push-based work delivery and result envelopes via agent-inbox or other transports.

interface MessagePort {
  onIncoming(handler): () => void;
  deliver(to, payload): Promise<DeliveryReceipt>;
  cancel(to, correlationId, reason, opts?): Promise<CancelAck>;
  dedupeKey(msg): string | null;
  deliverResult?(to, envelope): Promise<DeliveryReceipt>;
}

Built-in adapter: createAgentInboxPort(router, events, config).

AgentRoster (optional)

Presence-aware view of running agents for route-first dispatch.

interface AgentRoster {
  findAvailable(criteria: { role, tags?, notBusy? }): Promise<AgentRef[]>;
  onAgentStateChanged?(handler): () => void;
}

Built-in adapter: createMapRosterAdapter(registry, eventBus).

Dispatch Modes

| Mode | Behavior | |---|---| | spawn-only | Always spawn a new agent. Default when no roster is configured. | | route-only | Route to idle agents only. If none available, queue for retry. | | prefer-route | Route if possible, spawn as fallback. Default when roster is present. | | prefer-spawn | Spawn if capacity allows, route as fallback. |

State Machine

Tasks transition through internal orchestration states:

[Unclaimed] → [Claimed] → [Running] → [Continuing] → [Running] → ...
                              │              │
                              │ (abnormal)   │ (source inactive)
                              ▼              ▼
                         [RetryQueued]    [Released]
                              │
                              │ (exhausted)
                              ▼
                           [Dead]

Continuation vs retry: normal agent exit + still-active task = continuation (short delay, turn counter, same logical run). Abnormal exit = retry (exponential backoff, attempt counter, capped by maxRetries).

Features

| Feature | Description | |---|---| | Poll loop | Configurable interval, queries task source for ready work | | Mail inbound | Work arrives via MessagePort, joins the same eligibility pipeline | | Roster routing | Route to idle agents before spawning; LRU candidate selection | | Eligibility | Static filters (tags, priority, age) + pluggable scoring | | Concurrency | Global + per-dimension + per-host limits, most-restrictive-wins | | Continuation | Back-to-back turns on same logical run, capped by maxTurns or a policy function | | Retry | Exponential backoff, configurable max retries | | Stall detection | Terminates agents with no activity past stallTimeoutMs | | Reconciliation | Detects external state changes (closed, blocked, reassigned), terminates stale agents | | Fence tokens | Claim fencing for multi-dispatcher coordination; optional heartbeat via renewClaim | | Affinity | Continuation prefers prior agent; retry ignores by default; configurable per-mode | | Cancel propagation | Spawned agents: runtime.terminate. Routed agents: messagePort.cancel with x-dispatch/cancel | | Result envelopes | Mail-inbound work gets a typed ResultEnvelope on terminal state | | Events | 17 typed events covering the full dispatch lifecycle | | Snapshot | Real-time orchestrator state: counts, running entries, retry queue, token totals | | Usage tracking | Optional AgentRuntime.onUsage aggregates token totals into snapshot | | Async safety | Mutex serializes all dispatch paths against concurrent over-commit |

Configuration

interface DispatchConfig {
  claimantId: string;
  pollIntervalMs: number;          // default 15_000
  defaultRole: string;             // default "worker"
  concurrency: {
    global: number;                // default 3
    perDimension?: Record<string, Record<string, number>>;
    perHost?: Record<string, number>;
  };
  retry: {
    maxRetries: number;            // default 3
    baseDelayMs: number;           // default 10_000
    maxDelayMs: number;            // default 300_000
  };
  continuation?: {
    delayMs: number;               // default 1_000
    maxTurns: number;              // default 20
  };
  continuationPolicy?: (task, turnCount, lastExit) => 'continue' | 'release';
  eligibility?: EligibilityConfig;
  reconcile?: {
    enabled: boolean;              // default true
    intervalMs: number;            // default 60_000
    stallTimeoutMs?: number;       // default 300_000
  };
  promptBuilder?: PromptBuilder;
  tags?: string[];

  // Optional ports
  messagePort?: MessagePort;
  roster?: AgentRoster;
  dispatchMode?: DispatchMode;     // default: prefer-route if roster, else spawn-only
  continuationAffinity?: Affinity; // default: prefer
  retryAffinity?: Affinity;        // default: ignore
  deliverRetries?: number;         // default: 2
  heartbeatIntervalMs?: number;    // default: 30_000
}

Built-in Adapters

OpenTasks

import { createOrchestrator, createOpenTasksSource } from "swarm-dispatch";

const source = createOpenTasksSource(opentasksClient);

Agent-Inbox (MessagePort)

import { createAgentInboxPort } from "swarm-dispatch";

const messagePort = createAgentInboxPort(inbox.router, inbox.events, {
  dispatcherAgentId: "dispatcher:host:pid",
  classifyMessage: (msg) => {
    if (msg.content?.schema !== "x-dispatch/work") return null;
    return { messageId: msg.id, correlationId: msg.thread_tag, task: msg.content.data, replyTo: { agentId: msg.sender_id } };
  },
});

MAP Roster (AgentRoster)

import { createMapRosterAdapter } from "swarm-dispatch";

const roster = createMapRosterAdapter(mapRegistry, mapEventBus);

Putting it together

const orchestrator = createOrchestrator(source, runtime, {
  claimantId: "host:pid:instance",
  pollIntervalMs: 15_000,
  concurrency: { global: 5 },
  retry: { maxRetries: 3, baseDelayMs: 10_000, maxDelayMs: 300_000 },
  messagePort,
  roster,
  dispatchMode: "prefer-route",
});

Backward Compatibility

createTaskDispatcher is a deprecated alias for createOrchestrator. All Phase 1 code works unchanged. Phase 2 features (MessagePort, AgentRoster, dispatch modes, fencing) are opt-in via config fields.

License

MIT