swarm

package
v0.2.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jul 6, 2026 License: MIT Imports: 16 Imported by: 0

Documentation

Overview

Package swarm adds a multi-agent SWARM on top of zero's single sub-agent mechanism (internal/specialist): an orchestrator can spawn and coordinate MULTIPLE specialist members that run concurrently, communicate via per-agent mailboxes, and hand work off. It composes internal/specialist (to launch each member), internal/daemon (pooled workers when a daemon is running) and internal/background. Everything is additive and opt-in — the existing single "task" tool is unchanged; swarm tools are only active when invoked.

Every member runs under the SAME sandbox + risk/autonomy policy as the orchestrator: the swarm never grants a member more authority than its parent.

Index

Constants

View Source 
const (
	SpawnToolName   = "swarm_spawn"
	SendToolName    = "swarm_send"
	InboxToolName   = "swarm_inbox"
	StatusToolName  = "swarm_status"
	HandoffToolName = "swarm_handoff"
	CollectToolName = "swarm_collect"
)

Tool names. All swarm tools are additive and only act when invoked; the existing single "Task" tool is unchanged.

View Source 
const ScheduleToolName = "swarm_schedule"

ScheduleToolName is the recurring-spawn scheduler tool.

Variables

View Source 
var ErrMailboxFull = errors.New("swarm: mailbox is full")

ErrMailboxFull is returned when an inbox is at MaxMessages.

View Source 
var ErrMemberTemporary = errors.New("swarm: temporary member failure")

ErrMemberTemporary marks a member failure as transient (worker died, transient I/O) so the swarm relaunches it within maxMemberRestarts, mirroring daemon EXIT_TEMPFAIL semantics. Permanent failures are not retried.

View Source 
var ErrMessageTooLarge = errors.New("swarm: message exceeds size cap")

ErrMessageTooLarge is returned when a single message exceeds MaxMessageBytes.

View Source 
var ErrTaskExists = errors.New("swarm: task already registered")

ErrTaskExists is returned when registering a duplicate task ID.

View Source 
var ErrUnknownAgentType = errors.New("swarm: unknown agent type")

ErrUnknownAgentType is returned when a definition lookup misses.

View Source 
var ErrUnknownTask = errors.New("swarm: unknown task")

ErrUnknownTask is returned when updating/removing a missing task ID.

Functions

func RegisterTools 

func RegisterTools(registry *tools.Registry, sw *Swarm)

RegisterTools registers the seven swarm tools backed by one shared Swarm. The caller owns the Swarm's lifetime (call Swarm.Close on shutdown).

Types

type Coordinator 

type Coordinator struct {
	// contains filtered or unexported fields
}

Coordinator is the in-memory task registry + team color assigner shared by an orchestrator and its members. It is safe for concurrent use.

func NewCoordinator 

func NewCoordinator() *Coordinator

NewCoordinator returns an empty coordinator using the wall clock.

func (*Coordinator) Color 

func (c *Coordinator) Color(agentID string) string

Color returns the stable color assigned to an agent (assigning one on first use), so status output colors each member consistently.

func (*Coordinator) Complete 

func (c *Coordinator) Complete(id, result string) error

Complete marks a task done with its result.

func (*Coordinator) CompleteWithSession 

func (c *Coordinator) CompleteWithSession(id, result, sessionID string) error

CompleteWithSession marks a task done with its result and the member's durable child session id (so the TUI can drill into the member's conversation).

func (*Coordinator) Fail 

func (c *Coordinator) Fail(id, errMsg string) error

Fail marks a task failed with an error message.

func (*Coordinator) FailWithSession 

func (c *Coordinator) FailWithSession(id, errMsg, sessionID string) error

FailWithSession marks a task failed but keeps its child session id, so a member that ran and failed (e.g. exited non-zero / hit max-turns) is still drillable.

func (*Coordinator) Get 

func (c *Coordinator) Get(id string) (Task, bool)

Get returns a snapshot of one task.

func (*Coordinator) List 

func (c *Coordinator) List() []Task

List returns snapshots of all tasks, ordered by creation time then ID for stable output.

func (*Coordinator) Reassign 

func (c *Coordinator) Reassign(id, newAgentID string) error

Reassign transfers a task to a new agent (handoff/orphan-adoption). The task returns to pending under the new owner unless it is already terminal.

func (*Coordinator) Register 

func (c *Coordinator) Register(id, agentID, team, description string) (Task, error)

Register adds a new pending task. The ID must be unique and non-empty.

func (*Coordinator) SetStatus 

func (c *Coordinator) SetStatus(id string, status TaskStatus) error

SetStatus transitions a task. Transitions out of a terminal state are rejected (fail closed) so a late member update cannot resurrect a finished task.

func (*Coordinator) Summarize 

func (c *Coordinator) Summarize() Summary

Summarize aggregates current task counts.

func (*Coordinator) WaitSettled 

func (c *Coordinator) WaitSettled(ctx context.Context, team string) []Task

WaitSettled blocks until every task in the scope (one team, or all teams when team=="") has reached a terminal status, then returns a snapshot of the scope. It also returns — with the latest partial state — as soon as ctx is done, so a stuck or long-running member can't hang the caller forever (the caller bounds it with a timeout / user-cancel context). A scope with no tasks is settled immediately, so an empty or unknown team never blocks. This is the primitive that lets swarm_collect deliver final results in one call instead of forcing the orchestrator to poll swarm_status in a loop.

The settled decision and the returned snapshot are taken in the SAME locked pass, so a concurrent Register/Reassign slipping in between cannot make the call return a non-terminal task it never actually waited on.

type Definition 

type Definition struct {
	AgentType      string
	WhenToUse      string
	Model          string // "inherit" => use the orchestrator's model
	PermissionMode string
	// SystemPrompt returns the member's system prompt for the given task context.
	// It is a func so a definition can fold the task briefing in.
	SystemPrompt func(ctx PromptContext) string
}

Definition is one entry in the agent roster: how a member of a given type behaves (agent type, when to use it, model — "inherit" by default —, permission mode, and a system-prompt builder).

type FuncLauncher 

type FuncLauncher struct {
	Run RunFunc
}

FuncLauncher is the single MemberLauncher implementation: it runs a RunFunc in a goroutine and exposes a handle. Production builds one whose RunFunc calls specialist.Executor.Run (or submits to daemon.Pool); tests build one whose RunFunc is a stub.

func (FuncLauncher) Launch 

func (l FuncLauncher) Launch(ctx context.Context, spec MemberSpec) (MemberHandle, error)

Launch starts the RunFunc for spec and returns a handle to await it.

type JobStatus 

type JobStatus struct {
	ID        string
	Team      string
	AgentType string
	Task      string
	Every     time.Duration
	MaxRuns   int
	Runs      int
	// Skipped counts fires that did not spawn — either because the job's previous
	// spawn was still running (non-overlap) or a spawn attempt errored.
	Skipped int
}

JobStatus is a read-only snapshot of a scheduled job for listing.

type Mailbox 

type Mailbox struct {
	// BaseDir is the root under which <team>/inboxes/<agent>.json live. It must
	// be set (NewMailbox enforces this).
	BaseDir string
	// MaxMessageBytes caps a single serialized message. 0 => defaultMaxMessageBytes.
	MaxMessageBytes int
	// MaxMessages caps the number of messages retained in one inbox. 0 =>
	// defaultMaxMessages. Sends past the cap fail closed.
	MaxMessages int
	// LockTimeout bounds how long Send/MarkRead wait for the inbox lock.
	LockTimeout time.Duration
}

Mailbox is a per-agent, per-team message inbox persisted as a JSON array on disk at <baseDir>/<team>/inboxes/<agent>.json, written atomically under a lock file. It is hardened for untrusted input:

  • inbox files and their parent dirs are owner-only (0600/0700);
  • every write takes an exclusive lock file (bounded retry, stale-break);
  • message bodies and whole inboxes are size-capped (fail closed on oversize);
  • team/agent names are sanitized so a name can never escape the base dir;
  • malformed inbox JSON is rejected rather than silently reset (fail closed).

func NewMailbox 

func NewMailbox(baseDir string) (*Mailbox, error)

NewMailbox validates baseDir and returns a Mailbox with defaults applied.

func (*Mailbox) ReadAndConsume 

func (m *Mailbox) ReadAndConsume(team, recipient string) ([]Message, error)

ReadAndConsume reads the recipient's inbox and marks every previously-unread message read under one lock, returning the messages with their read flags as they were BEFORE this read (so the caller can tell which were new). The inbox is empty (no error) if it does not yet exist; malformed inbox JSON is reported as an error (fail closed) and nothing is written.

func (*Mailbox) Send 

func (m *Mailbox) Send(team, recipient string, msg Message) error

Send appends msg to the recipient's inbox under the team, taking the inbox lock. It fails closed on oversize messages or a full inbox.

Each send rewrites the whole inbox (read all → append one → write all), so it is O(N) in the inbox size. That is intentional given the small bounds (MaxMessages × MaxMessageBytes); if those caps are raised substantially, switch to an append-only log with periodic compaction.

type Member 

type Member struct {
	ID        string
	AgentType string
	TaskID    string
	// contains filtered or unexported fields
}

Member is one running agent within a team.

type MemberHandle 

type MemberHandle interface {
	ID() string
	Wait() (MemberResult, error)
}

MemberHandle is a running member. Wait blocks until the member exits and returns its result or error; ID identifies the member.

type MemberLauncher 

type MemberLauncher interface {
	Launch(ctx context.Context, spec MemberSpec) (MemberHandle, error)
}

MemberLauncher turns a MemberSpec into a running member. Production wires two implementations behind this seam — one over internal/specialist.Executor (direct) and one over internal/daemon.Pool (when a daemon is running) — while tests inject a fake. Keeping the seam here means the swarm core has no compile-time dependency on either heavy subsystem.

func NewSpecialistLauncher 

func NewSpecialistLauncher(executor specialist.Executor) MemberLauncher

NewSpecialistLauncher adapts internal/specialist.Executor into a MemberLauncher: each swarm member runs as a specialist sub-agent. This is the production "direct" launch path (used when no daemon pool is wired). The member inherits the orchestrator's model via spec.Model and runs in spec.Cwd, so it stays under the same sandbox + policy as its parent — the swarm never grants a member more authority.

agentType maps to the specialist agent name; the specialist's own definition for that name governs the member's system prompt and tools. RunInBackground is deliberately left false: the swarm provides concurrency itself (one goroutine per member), so each Run executes to completion inside its goroutine.

spec.PermissionMode (resolved + clamped by buildSpec, never above the orchestrator's) is propagated so the child only runs unsafe/high autonomy when the orchestrator is itself unsafe; otherwise it runs non-unsafe. An empty mode is substituted with the non-unsafe "auto" default so a swarm member can never fall into the historical empty-means-high path — that is the Task tool's behavior, not the swarm's.

type MemberResult 

type MemberResult struct {
	Result    string
	SessionID string
}

MemberResult is what a finished member returns.

type MemberSpec 

type MemberSpec struct {
	// ID identifies the running member (its agent id). For a freshly spawned
	// member this equals TaskID; after an orphan adoption a new agent (new ID)
	// takes over the original TaskID.
	ID             string
	TaskID         string
	AgentType      string
	Team           string
	Task           string
	Cwd            string
	Model          string // resolved (never the "inherit" sentinel)
	PermissionMode string
	SystemPrompt   string
}

MemberSpec is the fully-resolved instruction to launch one member. The swarm builds it from a Definition + task; a MemberLauncher turns it into a running agent. PermissionMode/Model carry the orchestrator's policy so the launcher can never grant a member more authority than its parent.

type Message 

type Message struct {
	From    string `json:"from"`
	To      string `json:"to"`
	Subject string `json:"subject,omitempty"`
	Body    string `json:"body"`
	Type    string `json:"type"`
	Read    bool   `json:"read"`
	Time    string `json:"time,omitempty"`
}

Message is one inbox entry. Time is RFC3339 and set on Send if empty.

type Options 

type Options struct {
	// BaseDir is the mailbox root (required).
	BaseDir string
	// Launcher turns specs into running members (required).
	Launcher MemberLauncher
	// Registry is the agent roster; nil => a fresh built-in registry.
	Registry *Registry
	// Coordinator tracks tasks; nil => a fresh coordinator.
	Coordinator *Coordinator
	// MaxTeamSize caps concurrent members per team; 0 => defaultMaxTeamSize.
	MaxTeamSize int
	// Definitions are user-defined agents to add to the roster (extending or
	// overriding the built-ins). Empty agent types are rejected.
	Definitions []Definition
	// Context is the parent context for all members; members outlive the
	// per-call tool context (a spawn returns immediately). nil => Background.
	// Close cancels the derived context, stopping every member.
	Context context.Context
}

Options configures a Swarm.

type Policy 

type Policy struct {
	Model          string
	PermissionMode string
}

Policy is the orchestrator's live model + permission mode at spawn time. Every member inherits it: a member never runs on a different model than requested, nor with a more permissive mode than its parent. It is passed per call (not stored on the Swarm) because the orchestrator's model/mode can change mid-session (e.g. an escalate-model).

type PromptContext 

type PromptContext struct {
	Team string
	Task string
}

PromptContext is the minimal context handed to a definition's SystemPrompt.

type Registry 

type Registry struct {
	// contains filtered or unexported fields
}

Registry is the agent roster: definitions looked up by agent type. Built-ins are seeded at construction; user-defined definitions extend or override them.

func NewRegistry 

func NewRegistry() *Registry

NewRegistry builds a registry seeded with the built-in roster.

func (*Registry) AgentTypes 

func (r *Registry) AgentTypes() []string

AgentTypes returns the registered agent types, sorted, for status/help.

func (*Registry) Lookup 

func (r *Registry) Lookup(agentType string) (Definition, error)

Lookup returns the definition for agentType, or ErrUnknownAgentType.

func (*Registry) Register 

func (r *Registry) Register(def Definition) error

Register adds or overrides a definition (user-defined agents extend the built-ins). An empty AgentType is rejected.

type RunFunc 

type RunFunc func(ctx context.Context, spec MemberSpec) (MemberResult, error)

RunFunc executes a member to completion. Both production launchers and tests supply one; FuncLauncher adapts it to the MemberLauncher interface by running it in a goroutine.

type Schedule 

type Schedule struct {
	// Every is the interval between fires. Required, must be >= minScheduleInterval.
	Every time.Duration
	// FirstDelay delays the first fire. Zero => the first fire happens after Every.
	FirstDelay time.Duration
	// MaxRuns bounds successful spawns. Zero => unbounded (until cancelled).
	MaxRuns int
	// Daily, when set, recomputes each fire as the next local Hour:Minute rather
	// than adding a fixed Every, so a wall-clock daily time does not drift across
	// DST transitions.
	Daily  bool
	Hour   int
	Minute int
}

Schedule describes when a scheduled job fires. Scheduling is interval-based ("wakeup"): the job first fires after FirstDelay (or Every when FirstDelay is zero), then every Every interval, until MaxRuns successful spawns is reached or the job/scheduler is stopped. A daily "cron" time sets Daily with Hour/Minute (and Every=24h for display/validation); the run loop then recomputes the delay to the next local HH:MM each cycle so it holds across DST (see the swarm_schedule tool's daily_at handling).

type Scheduler 

type Scheduler struct {
	// contains filtered or unexported fields
}

Scheduler fires recurring swarm spawns. It is opt-in: nothing runs unless a job is explicitly added via Add, and Close stops every job. Each job spawns a fresh member per interval through the same Swarm.Spawn path, so members inherit the recorded policy and are bounded by the team's slot cap and queue.

func (*Scheduler) Add 

func (s *Scheduler) Add(pol Policy, teamName, agentType, task, cwd string, sch Schedule) (string, error)

Add registers a recurring spawn and starts its timing loop. It validates the schedule and the agent type up front (fail fast) so a bad job never starts. The returned id identifies the job for List/Cancel.

func (*Scheduler) Cancel 

func (s *Scheduler) Cancel(id string) bool

Cancel stops a scheduled job by id. It reports whether a job was found.

func (*Scheduler) Close 

func (s *Scheduler) Close()

Close cancels every job and waits for their loops to exit. Safe to call more than once.

func (*Scheduler) List 

func (s *Scheduler) List() []JobStatus

List returns a snapshot of every active scheduled job.

type Summary 

type Summary struct {
	Total     int
	Pending   int
	Running   int
	Done      int
	Failed    int
	HandedOff int
}

Summary is an aggregate count of tasks by status for the status tool.

type Swarm 

type Swarm struct {
	// contains filtered or unexported fields
}

Swarm is the façade the swarm tools call. It owns the agent roster, the task coordinator, the mailbox, and the live teams, and enforces that every member inherits the orchestrator's model + permission mode.

func New 

func New(opts Options) (*Swarm, error)

New validates options and returns a Swarm.

func (*Swarm) AdoptOrphans 

func (s *Swarm) AdoptOrphans(pol Policy, teamName, toAgentType string) ([]string, error)

AdoptOrphans re-parents tasks in a team whose owning member is no longer live (e.g. a crashed worker) onto fresh members of toAgentType, returning the adopted task ids. Terminal tasks and tasks with a live owner are left alone.

func (*Swarm) Close 

func (s *Swarm) Close()

Close cancels every running member's context and releases resources. It is safe to call more than once. The scheduler is closed first so no new spawn fires after shutdown begins.

func (*Swarm) Collect 

func (s *Swarm) Collect(teamName string) []Task

Collect returns snapshots of every task in a team, for swarm_collect/status.

func (*Swarm) CollectWait 

func (s *Swarm) CollectWait(ctx context.Context, teamName string) []Task

CollectWait blocks until the team's tasks all reach a terminal state (or ctx is done), then returns their snapshots. This is what swarm_collect uses so the orchestrator gets final results in a single call instead of polling swarm_status repeatedly while members are still running.

func (*Swarm) Coordinator 

func (s *Swarm) Coordinator() *Coordinator

Coordinator exposes the task registry (for status/collect).

func (*Swarm) Handoff 

func (s *Swarm) Handoff(pol Policy, teamName, taskID, toAgentType, note string) (string, error)

Handoff transfers a task to a fresh member of toAgentType, delivering a note to the new member's inbox and marking the original task handed-off. It returns the new task id. A handoff of an already-terminal task is rejected (fail closed).

func (*Swarm) Mailbox 

func (s *Swarm) Mailbox() *Mailbox

Mailbox exposes the message store (for send/inbox tools).

func (*Swarm) Registry 

func (s *Swarm) Registry() *Registry

Registry exposes the roster (for tool listing / user-defined registration).

func (*Swarm) Scheduler 

func (s *Swarm) Scheduler() *Scheduler

Scheduler returns the swarm's recurring-spawn scheduler, creating it on first use. Scheduling is opt-in: until a job is added the scheduler does nothing.

func (*Swarm) Spawn 

func (s *Swarm) Spawn(pol Policy, teamName, agentType, task, cwd string) (string, error)

Spawn registers a task and launches a member of agentType to run it under the given team, inheriting the orchestrator's policy. It returns the task id immediately; the member runs concurrently (under the Swarm's base context, so it outlives this call) and reports its result through the coordinator. If the team is at its slot cap the member is queued and launches when a slot frees (it stays pending in the coordinator until then).

type Task 

type Task struct {
	ID          string
	AgentID     string
	Team        string
	Description string
	Status      TaskStatus
	Result      string
	Err         string
	// SessionID is the member's durable child session id, recorded on completion
	// so the TUI can drill into the member's conversation. Empty until done.
	SessionID string
	CreatedAt time.Time
	UpdatedAt time.Time
}

Task is one unit of work tracked by the coordinator. It is returned by value from snapshots so callers cannot mutate coordinator state without the lock.

type TaskStatus 

type TaskStatus string

TaskStatus is the lifecycle state of a swarm task. 

const (
	// StatusPending: registered but the member has not started.
	StatusPending TaskStatus = "pending"
	// StatusRunning: the member is executing.
	StatusRunning TaskStatus = "running"
	// StatusDone: the member finished successfully.
	StatusDone TaskStatus = "done"
	// StatusFailed: the member exited with an error.
	StatusFailed TaskStatus = "failed"
	// StatusHandedOff: ownership was transferred to another agent.
	StatusHandedOff TaskStatus = "handed-off"
)

type Team 

type Team struct {
	Name string
	// contains filtered or unexported fields
}

Team is a named set of concurrently-running members with a bounded slot count and a FIFO queue for overflow spawns.

func (*Team) QueueDepth 

func (t *Team) QueueDepth() int

QueueDepth reports how many specs are waiting for a slot (status/tests).

func (*Team) Running 

func (t *Team) Running() int

Running reports how many members are currently occupying a slot.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.