workers

package
v0.3.0 Latest Latest
Warning

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

 Go to latest Published: Jun 26, 2026 License: Apache-2.0 Imports: 11 Imported by: 0

Documentation

Overview

Package workers owns the lifecycle of every sub-agent spawned by the root. See .agents/WORKER_RUNTIME.md and decision #5 for the design.

A Worker wraps an ADK runner running in its own goroutine. The Manager keeps a process-wide registry of live workers, allocates their sandboxes, multiplexes their event streams, and exposes the lifecycle primitives (spawn, query, list, inspect, collect, kill) that the root agent's spawn tools call into.

Index

Constants

This section is empty.

Variables

View Source 
var ErrCollectTimeout = errors.New("collect timed out")

ErrCollectTimeout is returned by Collect when the configured wait elapses before the worker reaches idle/done.

View Source 
var ErrUnknownWorker = errors.New("unknown worker")

ErrUnknownWorker is returned by Manager methods when the worker_id does not exist in the registry.

View Source 
var ErrWorkerNameConflict = errors.New("worker name already in use")

ErrWorkerNameConflict is returned by Spawn when the requested name collides with a live worker.

Functions

This section is empty.

Types

type Driver 

type Driver interface {
	// Send delivers a user message to the worker. Returns once the
	// message has been accepted (NOT once the worker has finished).
	// The driver is responsible for publishing every event it
	// produces onto bus.
	Send(ctx context.Context, message string, bus *EventBus, workerID string) error

	// WaitIdle blocks until the worker reaches idle (the previous
	// Send has finished producing events) or ctx is cancelled.
	WaitIdle(ctx context.Context) error

	// Output returns the worker's last assistant message. Called by
	// Collect after WaitIdle.
	Output() string

	// Close releases any resources the driver owns. Called once on
	// Kill / Collect.
	Close() error
}

Driver is the per-worker engine that the Manager drives. The real production implementation wires ADK's runner + session.Service + agent.Builder; tests provide a fakeDriver that scripts behaviour without any LLM. Keeping the Manager generic over Driver is what lets us test lifecycle semantics with zero networking.

type DriverFactory 

type DriverFactory func(workerID string, spec Spec, sandbox string) (Driver, error)

DriverFactory builds the Driver for a freshly-spawned worker. The Manager calls it once per Spawn, after sandbox allocation. Real implementations close over the agent.Builder, providers and MCPs registries.

type EventBus 

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

EventBus is a fan-out channel for worker events. Each subscriber gets its own buffered channel; slow subscribers see Drops counted but do not block the publisher.

The implementation is intentionally simple: a slice of subscribers guarded by a RWMutex, and Publish drops the event for any subscriber whose buffer is full.

func NewEventBus 

func NewEventBus() *EventBus

NewEventBus returns a ready-to-use bus.

func (*EventBus) Close 

func (b *EventBus) Close()

Close closes every subscriber channel. After Close, Publish becomes a no-op. Used by the Manager on shutdown.

func (*EventBus) Drops 

func (b *EventBus) Drops() uint64

Drops reports how many events were dropped because of full subscriber buffers. Mostly for diagnostics / status bar.

func (*EventBus) Publish 

func (b *EventBus) Publish(evt WorkerEvent)

Publish fans out evt to every subscriber. Index is assigned by the bus so callers do not have to coordinate. Subscribers whose buffer is full lose the event (counted via Drops).

func (*EventBus) Subscribe 

func (b *EventBus) Subscribe() (<-chan WorkerEvent, func())

Subscribe registers a new subscriber and returns the channel plus an unsubscribe function. Callers must drain the channel until the unsubscribe is called.

type EventKind 

type EventKind int

EventKind classifies a WorkerEvent. Mirrors WORKER_RUNTIME.md. 

const (
	// EventToolCall is emitted when the worker invokes a tool.
	EventToolCall EventKind = iota

	// EventToolResult carries the result of a tool call back to the
	// worker (already redacted).
	EventToolResult

	// EventThought is a model "thinking" step, when the provider
	// exposes one.
	EventThought

	// EventAssistantMessage is a chunk of assistant text.
	EventAssistantMessage

	// EventStatusChange signals a Status transition. Useful for the
	// TUI to repaint without polling.
	EventStatusChange
)

func (EventKind) String 

func (k EventKind) String() string

String renders an EventKind as a short token used by the audit log and the inspect_agent tool output.

type Kind 

type Kind int

Kind classifies a worker by how it was spawned. 

const (
	// KindStatic means the worker was instantiated from an
	// AgentTemplate in agents.yaml.
	KindStatic Kind = iota

	// KindDynamic means the worker was built at runtime from a Spec
	// the root produced. Phase 6 work; defined here so the rest of
	// the Manager stays kind-agnostic.
	KindDynamic
)

type Manager 

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

Manager owns every live worker. It is the single chokepoint for the spawn / supervise / collect / kill verbs the root agent uses.

func NewManager 

func NewManager(cfg ManagerConfig) *Manager

NewManager constructs a Manager. The bus, sandbox allocator and driver factory must be set in cfg; the Manager does not invent sensible defaults for them because the production wiring depends on caller-owned state (the App).

func (*Manager) Collect 

func (m *Manager) Collect(ctx context.Context, id string, timeout time.Duration) (WorkerInfo, error)

Collect waits for the worker to reach idle/done/failed/killed, captures the output, and removes the worker from the registry. The timeout defaults to ManagerConfig.CollectTimeout when zero.

func (*Manager) Get 

func (m *Manager) Get(id string) (WorkerInfo, error)

Get returns one worker's info or ErrUnknownWorker.

func (*Manager) GlobalBus 

func (m *Manager) GlobalBus() *EventBus

GlobalBus returns the multiplex bus used by the TUI sidebar and the audit logger.

func (*Manager) Inspect 

func (m *Manager) Inspect(id string, _ int) (WorkerInfo, error)

Inspect peeks at the recent events of a worker. since is a 1-based cursor; the Manager keeps the events in the per-worker bus, so callers must Subscribe / drain rather than rely on a history buffer. For Phase 5 Inspect returns the current snapshot only (events_since is a TODO that lands when we wire a ring buffer).

func (*Manager) Kill 

func (m *Manager) Kill(id string, reason string) error

Kill cancels the worker's context and marks it killed. The worker stays in the registry briefly (until Collect or Shutdown) so the root can read its last status.

reason is an optional human-readable explanation that surfaces on the next Collect as WorkerInfo.Err (and thus CollectResult.Err for the root agent's collect_agent tool). Use it to distinguish e.g. "killed by user from TUI" from "killed by shutdown" so the LLM can react accordingly. Empty reason falls back to the legacy "killed" marker.

func (*Manager) List 

func (m *Manager) List() []WorkerInfo

List returns a snapshot of every live worker, sorted by start time (oldest first), with the worker ID as a tie-breaker. Iterating the workers map directly yields a random order on every call, which makes the /worker list view reshuffle on each refresh and impossible to navigate; this stable ordering is what keeps a selected row put.

func (*Manager) Query 

func (m *Manager) Query(ctx context.Context, id string, message string) error

Query delivers a message to an existing worker and transitions it back to running.

func (*Manager) Shutdown 

func (m *Manager) Shutdown(grace time.Duration) error

Shutdown cancels every worker and waits up to grace for them to drain. Sandboxes are cleaned up on the way out. Always returns nil; partial failures are reported via the event bus.

func (*Manager) Spawn 

func (m *Manager) Spawn(ctx context.Context, spec Spec) (*Worker, error)

Spawn creates a new worker from spec, allocates its sandbox, builds its driver, and registers it. If spec.InitialMessage is non-empty the message is delivered before returning so the worker is already running by the time Spawn returns its ID.

func (*Manager) SubscribeWorker 

func (m *Manager) SubscribeWorker(id string) (history []WorkerEvent, stream <-chan WorkerEvent, cancel func(), err error)

SubscribeWorker attaches a live subscriber to the worker's event bus AND returns the recent-history snapshot so the caller can show context immediately. The returned cancel function unsubscribes from the live stream when the caller is done (typical case: the TUI closes the spy view).

Why history first, channel second: race-free playback of "what happened before I subscribed" is impossible with channels alone because events emitted between the snapshot and the subscription would be lost. We accept a tiny window of duplication: an event already in the snapshot may also appear on the channel if it was published in the few microseconds between snapshot and subscribe. Subscribers de-duplicate on (WorkerID, Index).

type ManagerConfig 

type ManagerConfig struct {
	Sandbox       *SandboxAllocator
	DriverFactory DriverFactory

	// CollectTimeout is the default upper bound for Collect when no
	// explicit timeout is provided. Defaults to 5 minutes.
	CollectTimeout time.Duration
}

ManagerConfig bundles construction-time options.

type SandboxAllocator 

type SandboxAllocator struct {
	// DataDir is the absolute path of the .baifo/data directory.
	// Workspaces live under <DataDir>/workspaces/.
	DataDir string
}

SandboxAllocator owns the per-worker filesystem workspace. Every worker gets an isolated directory under <DataDir>/workspaces/<id>/ that the filesystem-MCP can chroot into. The directory is created at Spawn time and removed at Cleanup time (typically right after Collect or on Shutdown).

This is NOT a security boundary — a process running inside the worker can absolutely escape with an exec(2). The point of the per-worker directory is hygiene: a known, predictable cwd that gets garbage-collected after the worker dies so the filesystem doesn't accumulate scratch from every spawn. Real isolation is the user's responsibility (containers, namespaces, VMs).

func (*SandboxAllocator) Allocate 

func (a *SandboxAllocator) Allocate(workerID string) (string, error)

Allocate returns the absolute workspace path for the worker, creating the directory tree if needed.

func (*SandboxAllocator) Cleanup 

func (a *SandboxAllocator) Cleanup(workerID string) error

Cleanup removes the worker's workspace. Idempotent: returns nil for unknown ids and for unconfigured allocators.

type Spec 

type Spec struct {
	Kind Kind
	Name string
	// Description is a one-liner the worker's parent (typically the
	// root) uses to advertise the worker's purpose. Forwarded to
	// llmagent.Config.Description so ADK's agent-transfer routing
	// can describe it. Optional; an empty string is fine.
	Description string
	Prompt      string
	Provider    string
	Model       string
	Skills      []string
	MCPs        []string

	// Reasoning is the optional reasoning-effort knob forwarded to the
	// agent.Builder ("minimal" / "low" / "medium" / "high"; empty =
	// model default). Lets the root dial a worker's thinking up or down
	// to fit the sub-task.
	Reasoning string

	// ReasoningAPI optionally overrides the Anthropic reasoning API the
	// worker's model uses ("enabled" or "adaptive"); empty auto-detects.
	// Forwarded to the agent.Builder. Ignored by non-anthropic providers.
	ReasoningAPI string

	// AllowedSecrets is forwarded to the agent.Builder so the secrets
	// pipeline scopes correctly.
	AllowedSecrets []string

	// InitialMessage, when non-empty, is sent to the worker right
	// after it is created so it starts processing immediately.
	InitialMessage string
}

Spec is the minimal description the Manager needs to spawn a worker. It is intentionally decoupled from agent.Spec so static and dynamic callers can build it from their own sources (agents.yaml templates vs the LLM's spawn_dynamic_agent payload).

type Status 

type Status int

Status enumerates the lifecycle states described in WORKER_RUNTIME.md. 

const (
	// StatusRunning means the worker is processing input.
	StatusRunning Status = iota

	// StatusIdle means the worker is waiting for the next query.
	StatusIdle

	// StatusDone means the worker terminated normally.
	StatusDone

	// StatusFailed means the worker terminated with an error.
	StatusFailed

	// StatusKilled means the worker was cancelled via kill_agent or
	// at shutdown.
	StatusKilled
)

func (Status) String 

func (s Status) String() string

String renders Status as a lowercase word for the spawn tools.

type ToolCallPayload 

type ToolCallPayload struct {
	Name string
	ID   string
	Args map[string]any
}

ToolCallPayload is the Payload shape published with EventToolCall events. It carries the full call envelope so subscribers can render the tool boundary with as much context as the agent itself sees (name, ID for pairing with the response, and the post-expansion args map after the BeforeToolCallback ran).

Subscribers should treat Args as read-only; the same map is shared with ADK's tool-execution path.

type ToolResultPayload 

type ToolResultPayload struct {
	Name   string
	ID     string
	Result map[string]any
}

ToolResultPayload is the Payload shape published with EventToolResult events. ID pairs with the ToolCallPayload.ID of the matching call so the TUI can collapse the pair into a single card. Result is the post-redaction response map.

type Worker 

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

Worker is the manager's internal record. We expose a tiny method surface so callers can drive supervision without reaching into the struct directly.

func (*Worker) Done 

func (w *Worker) Done() <-chan struct{}

Done returns a channel that closes when the worker reaches a terminal state (done / failed / killed). Useful for collect_agent.

func (*Worker) ID 

func (w *Worker) ID() string

ID returns the worker's stable identifier.

func (*Worker) Info 

func (w *Worker) Info() WorkerInfo

Info returns a snapshot of the worker's public state.

type WorkerEvent 

type WorkerEvent struct {
	WorkerID  string
	Index     int
	Timestamp time.Time
	Kind      EventKind
	Payload   any
}

WorkerEvent is the unit pushed onto the event bus. Index is a monotonic per-worker counter; subscribers reading from the global channel use (WorkerID, Index) to detect gaps.

type WorkerInfo 

type WorkerInfo struct {
	ID        string
	Name      string
	Kind      Kind
	Status    Status
	Spec      Spec
	Sandbox   string
	StartedAt time.Time
	UpdatedAt time.Time
	LastEvent string
	Output    string
	Err       string
}

WorkerInfo is a snapshot of a worker's public state. The Manager hands these out for list_agents / inspect_agent / TUI listings so the caller never gets a live pointer to internal state.

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.