pool

package
v0.28.0 Latest Latest
Warning

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

Go to latest
Published: Jun 30, 2026 License: MIT Imports: 24 Imported by: 0

Documentation

Overview

Package pool manages the global agent subprocess slot count + FIFO queue. Files:

  • buffer.go — per-session message buffer (drain on slot grant)
  • pool.go — slot allocation, FIFO queue, factory hookup

Buffer rationale: when a session is queued (no slot), incoming user messages must not vanish — they're appended to the on-disk PendingInput list (so a wick restart preserves them) AND held in a transient buffer here. When the slot is granted, the entire buffer is drained as one combined input to the spawned agent. See agents-design.md §5.1.1.

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

This section is empty.

Types

type ActiveEntry

type ActiveEntry struct {
	SessionID      string
	AgentName      string
	ProviderType   string // resolved provider type (claude / codex / gemini)
	ProviderName   string // instance name within that type
	CWD            string // resolved workspace path, used by RouteByCWD
	PID            int
	Queued         int  // messages waiting after the current turn (RespawnQueue)
	Respawns       bool // one process per turn (codex): a dead PID between turns is normal, not a zombie
	Lifecycle      string
	Substate       string
	LastActive     time.Time
	InFlightEvents []store.TurnEvent
	// PartialText is the assistant text accumulated so far for the
	// in-flight turn (everything received via TextDelta but not yet
	// flushed by Done). Empty when no turn is mid-stream. Used by the
	// SSE snapshot so a refresh keeps the partial bubble visible.
	PartialText string
}

ActiveEntry is the public snapshot view of one running agent. Lifecycle / Substate / PID / LastActive are populated when the pool can read them; older callers that only check SessionID + AgentName keep working.

type AgentFactory

type AgentFactory interface {
	Build(opt FactoryOptions) (BuildResult, error)
}

AgentFactory builds an agent ready to Start. The pool wires the OnExit hook itself (so it can free the slot); the factory should not.

BuildResult.OnStarted is called by the pool right after a.Start succeeds — that's when the OS pid is known and the first user message has been drained from the buffer. Factories use it to finish writing the spawn `start` event with both fields. Optional; nil = nothing to record.

type Buffer

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

Buffer is a small per-session message queue. Operations are idempotent w.r.t. the on-disk meta.json so a crash between in-memory append and disk persist doesn't lose user input.

func NewBuffer

func NewBuffer(layout config.Layout, sessionID string) (*Buffer, error)

NewBuffer reads any pending_input persisted on disk so we resume queued messages after a wick restart.

func (*Buffer) Append

func (b *Buffer) Append(text string) error

Append adds a new line and persists it to meta.PendingInput so a crash before drain doesn't drop it.

func (*Buffer) Drain

func (b *Buffer) Drain() (string, error)

Drain returns all buffered lines joined by newline and clears the buffer (both in-memory and on-disk PendingInput). Returns "" if empty.

func (*Buffer) Len

func (b *Buffer) Len() int

Len reports the buffered line count without modifying state.

type BuildResult

type BuildResult struct {
	Agent     *provider.Agent
	State     *state.Machine
	Store     *store.Store
	OnStarted func(meta SpawnStartMeta)
}

BuildResult bundles everything Build returns. New code should pull fields from here; the bare-tuple shape is gone so we don't have to thread one more channel through every callsite when we add another hook later.

type Capacity added in v0.15.2

type Capacity struct {
	Scope     string // "global" or "type/name"
	Used      int    // active + in-flight spawns counted against this scope
	Max       int    // configured cap; 0 = unlimited (provider scope only)
	Remaining int    // slots still grantable right now (Max<=0 → -1 = unlimited)
}

Capacity is a used / max / free snapshot for one scope. Max == 0 means "unlimited" at that scope (only meaningful for the provider scope; the global cap is always a positive number after New() normalises it).

func (Capacity) Unlimited added in v0.15.2

func (c Capacity) Unlimited() bool

Unlimited reports whether this scope imposes no finite cap.

type ClaudeFactory

type ClaudeFactory struct {
	Layout    config.Layout
	Spawner   provider.Spawner // optional override; nil = real claude
	RecordRaw bool
	OnEvent   func(sessionID, agentName string, ev event.AgentEvent)
	OnExit    func(sessionID, agentName string, reason provider.ExitReason)

	// Gate (optional) attaches a static command whitelist to every spawn.
	// When non-nil, Build writes a per-session settings.json + spec
	// file to a temp dir, points the spawner at the settings file,
	// and injects WICK_GATE_SPEC into ExtraEnv so wick-gate finds
	// its config. nil = no gate (fail-open, only safe for tests).
	Gate *GateConfig
	// GateLoader (optional) is called on every Build to fetch the
	// current gate config from the live config store. Takes precedence
	// over Gate when non-nil. This lets operators toggle gate_enabled
	// or edit AllowedCmds in the UI without restarting the server.
	GateLoader func() *GateConfig
	// PermissionModeLoader (optional) is called on every Build to read
	// the current GateConfig.PermissionMode value. Return "bypass" to
	// force --permission-mode bypassPermissions on Claude (and the
	// equivalent on codex/gemini) when no gate hook is installed.
	// Any other value (including empty) means "prompt as normal".
	PermissionModeLoader func() string

	// SystemPromptLoader (optional) returns a global system prompt
	// fragment appended to the loaded preset body on every spawn.
	// Empty string = no append. Lets operators set org-wide rules
	// (prompt-injection defenses, shared conventions) without editing
	// every preset. Preset stays the primary; this only adds to it.
	SystemPromptLoader func() string

	// TraceEventMaxKBLoader (optional) returns the current trace_event_max_kb
	// config value. 0 = no cap.
	TraceEventMaxKBLoader func() int

	// TraceInlineKBLoader (optional) returns the current trace_event_inline_kb
	// config value. Called on every Build so operators can change the threshold
	// without restarting the server. 0 or negative = use DefaultTraceInlineBytes.
	TraceInlineKBLoader func() int

	// ConnectorCatalogLoader (optional) returns a "## Available wick
	// connectors" markdown block listing the connectors the spawning
	// agent should prefer over hand-rolled HTTP. Wired in server.go
	// so the loader can call connectorsSvc and filter to instances
	// whose status is "ready" — connectors the operator has finished
	// configuring. Empty string = no append (no connectors ready, or
	// service unavailable). Inserted between the immutable rules and
	// the preset body so the catalog can't override either layer.
	ConnectorCatalogLoader func() string

	// SpawnLogger (optional) writes one jsonl per spawn under
	// `<base>/backends/spawns/`. Each spawn emits `start` on Build +
	// `exit` from the OnExit hook so the Backends UI can list spawn
	// history per backend by `ls`-ing the directory. nil = no logging.
	SpawnLogger *provider.SpawnLogger

	// MCPToken is the per-boot internal MCP secret forwarded to the
	// claude spawner so agents reach the live MCP server over loopback.
	MCPToken string

	// InstanceOverride pins a specific Instance for every Build call,
	// bypassing the provider.Find registry lookup. Tests use this to
	// inject ExtraArgs / Env without touching userconfig files.
	InstanceOverride *provider.Instance
}

ClaudeFactory is the production AgentFactory: wires a ClaudeParser + ClaudeSpawner into a fresh provider.Agent for each Build call.

The factory owns no per-spawn state; the pool calls Build once per session activation.

func (*ClaudeFactory) Build

func (f *ClaudeFactory) Build(opt FactoryOptions) (BuildResult, error)

Build returns a fresh agent + state machine + store wired for one session+agent. Caller (the pool) is responsible for calling agent.Start.

type FactoryOptions

type FactoryOptions struct {
	SessionID     string
	AgentName     string
	ProviderType  string
	ProviderName  string
	Workspace     string
	ResumeID      string
	IdleTimeout   time.Duration
	KillAfterIdle time.Duration
	OnEvent       func(event.AgentEvent)
	// PresetName is the preset name from session meta. Factory resolves
	// the content from disk — pool passes the name so factory avoids a
	// redundant session.Load.
	PresetName string
	// Origin is the session origin (e.g. "slack", "ui", "rest") written
	// into the spawn log so Recent Spawns can show the channel without a
	// registry lookup.
	Origin string
	// Title / TitleCustom are the session's current title state, surfaced
	// in the "This session" system-prompt block so the agent knows
	// whether it still needs to set a title without a wick_session_info
	// round-trip. Snapshot at spawn time.
	Title       string
	TitleCustom bool
	// MaxTurns caps agentic turns on the spawn (--max-turns). Pulled from
	// the agent entry by the pool; 0 = no cap.
	MaxTurns int
	// ThinkingTokens is the resolved MAX_THINKING_TOKENS env value for the
	// spawn (claude). Pulled from the agent entry by the pool; empty = unset
	// (provider default, thinking on); "0" = disabled; "<n>" = budget.
	ThinkingTokens string
}

FactoryOptions is what the pool hands to the factory. ResumeID is pulled from the session's agents.json by the pool. ProviderType / ProviderName identify which provider runtime instance to spawn against — empty ProviderName resolves to the per-type default whose name equals the type itself ("claude" / "codex" / "gemini"). Both are forwarded to the spawn logger so /tools/agents/providers can surface per-provider history without re-parsing files.

type GateConfig

type GateConfig struct {
	// GateBinary is the absolute path to the wick-gate binary. Required.
	GateBinary string
	// Rules is the whitelist enforced for every spawn under this factory.
	Rules []gate.CommandRule
	// AppName drives the shared spec path (~/.<app>/agents/gate/spec.json).
	// Falls back to "wick" when empty.
	AppName string
	// DefaultScope is written into spec.json as the fallback scope for
	// rules that have an empty Scope field. Typically the default
	// workspace directory so no-scope rules are still path-restricted.
	DefaultScope string
	// TempDirRoot is where per-spawn gate artifacts live. If empty,
	// `<Layout.SessionDir(id)>/gate` is used.
	TempDirRoot string
}

GateConfig describes the gate plumbing: where the wick-gate binary lives + what rules it enforces. The factory writes the shared spec.json from Rules on every spawn so UI changes propagate immediately without restarting the server.

type LifecycleEvent

type LifecycleEvent struct {
	SessionID    string
	AgentName    string
	Lifecycle    string // "spawning" | "killed"
	PID          int
	At           time.Time
	ProviderType string
	ProviderName string
	// Ctx is the spawn-time context from the originating Send. Carries
	// the zerolog logger the HTTP middleware attached, so callbacks
	// can `log.Ctx(ev.Ctx)` and recover the request_id. Never nil —
	// pool sets it to context.Background() when no spawn ctx applies
	// (e.g. exit fired from an already-released runEntry).
	Ctx context.Context
}

LifecycleEvent is emitted for the two transitions the pool drives directly (no parser event triggers them): a fresh spawn coming online, or a subprocess dying. PID is populated for spawning → working; 0 for killed.

type Pool

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

Pool is the global slot manager. It tracks how many agent subprocesses are alive across all sessions, FIFO-queues sessions that arrive while full, and grants slots when one frees up.

Pool deliberately knows nothing about CLI specifics — it asks an AgentFactory to build an *provider.Agent for a given session+agent name. Tests inject a factory that returns agents wired to the fakeSpawner; production wires ClaudeSpawner.

func New

func New(cfg PoolConfig) *Pool

New returns an empty pool.

MaxConcurrent <= 0 means UNLIMITED at the global scope (bounded only by per-provider caps and host resources). The config UI seeds a sane default of 2, but an operator may set 0 deliberately to lift the global ceiling — capacity math treats 0 as "no global cap".

func (*Pool) Active

func (p *Pool) Active() int

Active returns the number of running agents.

func (*Pool) ActiveSnapshot

func (p *Pool) ActiveSnapshot() []ActiveEntry

ActiveSnapshot returns a defensive copy of every running agent in the pool. Used by the Backends UI to show what's eating each slot.

func (*Pool) AutoReplyOn added in v0.28.0

func (p *Pool) AutoReplyOn(sessionID string) bool

AutoReplyOn reports the persisted Slack auto-reply flag (meta.json). It reads the session meta fresh so a restart picks up the last saved state. Missing session or read error → false (fail closed).

func (*Pool) Capacity added in v0.15.2

func (p *Pool) Capacity() Capacity

Capacity returns the global slot usage. Caller need not hold p.mu.

func (*Pool) Dequeue

func (p *Pool) Dequeue(sessionID, agentName string) int

Dequeue drops every queued request matching sessionID+agentName. Returns the number of removed entries — operators use this to cancel a session that has been waiting too long without ever getting a slot. Active spawns are NOT touched; use Kill for that.

func (*Pool) DequeueSession added in v0.14.21

func (p *Pool) DequeueSession(sessionID string) int

DequeueSession drops every queued request for the session regardless of agent name, and clears its buffered/pending input so the session won't execute later (even across a restart). Returns the number of queue entries removed. Use this from the operator UI where the caller only knows the session id — Dequeue requires an exact agent match, which the UI often can't supply.

func (*Pool) EnsureSession added in v0.13.0

func (p *Pool) EnsureSession(ctx context.Context, sessionID, source, projectID string) error

EnsureSession is the public wrapper for ensureSession. Workflow's session_init executor calls this to materialize the registry entry + sidebar row up-front, before any agent node actually dispatches a message. Idempotent — a second call for the same sessionID is a no-op (or backfills project binding).

func (*Pool) EnsureSessionOwner added in v0.17.0

func (p *Pool) EnsureSessionOwner(ctx context.Context, sessionID, userID string)

EnsureSessionOwner stamps UserID on an existing session when the session currently has no owner. No-op when the session does not exist or already has an owner.

func (*Pool) HandleExit

func (p *Pool) HandleExit(sessionID, agentName string, reason provider.ExitReason)

HandleExit is the public hook the factory wires into agent.OnExit. It frees the pool slot when a process exit means the AGENT is done — but for respawn-mode providers (codex) one agent spans many short-lived processes, so a turn-boundary exit (ExitClean) or an internal respawn kill (ExitRespawn) must NOT release the slot. Only a terminal reason (idle TTL, Stop/Kill, crash) tears the runEntry down. Without this gate every codex turn-end deletes the entry, and the next Send sees "no live subprocess" and spawns a SECOND concurrent agent (the double-spawn bug).

claude (append mode) keeps the 1-agent-1-process model: every exit is the agent dying, so all reasons release.

func (*Pool) IdleTimeout

func (p *Pool) IdleTimeout() time.Duration

IdleTimeout returns the configured idle timeout. UI consumers use it to render the auto-kill countdown alongside LastActive.

func (*Pool) Kill

func (p *Pool) Kill(sessionID, agentName string) error

Kill stops the running agent for sessionID+agentName. Idempotent if the agent is not currently active — returns nil in that case. The normal onAgentExit hook still fires, releasing the slot and draining the queue.

func (*Pool) MaxConcurrent

func (p *Pool) MaxConcurrent() int

MaxConcurrent surfaces the configured slot cap for the Backends UI. Read-only — change via PoolConfig at construction time.

func (*Pool) ProviderCapacity added in v0.15.2

func (p *Pool) ProviderCapacity(pType, pName string) Capacity

ProviderCapacity returns the EFFECTIVE capacity for one provider instance: bounded by both its own per-instance cap and the global remaining. Used == active+spawning entries for that provider. Remaining is what can actually be granted right now (the min of the two scopes). Caller need not hold p.mu.

func (*Pool) QueueLen

func (p *Pool) QueueLen() int

QueueLen returns the number of queued requests.

func (*Pool) QueueSnapshot

func (p *Pool) QueueSnapshot() []QueueEntry

QueueSnapshot returns a defensive copy of the current FIFO queue (oldest first). Used by the Backends UI to show what's waiting.

func (*Pool) ReconcileDead added in v0.15.2

func (p *Pool) ReconcileDead()

ReconcileDead scans active entries and Stops any whose subprocess is no longer alive at the OS level — a crash or external kill that the reader loop never saw as stdout EOF (common on Windows, where killing a process does not reliably close its pipe). Stop() fires the exit hook, releasing the slot and draining the queue, so a zombie entry can't wedge the pool. Cheap signal-0 probe per entry; safe to call from request handlers (panel open) and a periodic ticker.

func (*Pool) Send

func (p *Pool) Send(ctx context.Context, sessionID, agentName, source, role, text string) error

func (*Pool) SendWithAttachments added in v0.14.18

func (p *Pool) SendWithAttachments(ctx context.Context, sessionID, agentName, source, role, text, projectID string, atts []store.Attachment) error

SendWithAttachments is Send with a list of user-uploaded files. The caller is responsible for materializing the files on disk under SessionDir/uploads/ — the pool only persists the metadata into conversation.jsonl and appends a small `[Attached files]` block to the text sent to the CLI subprocess so it can Read the paths.

func (*Pool) SendWithProject added in v0.14.21

func (p *Pool) SendWithProject(ctx context.Context, sessionID, agentName, source, role, text, projectID string) error

SendWithProject is like Send but binds sessionID to the given project id when auto-creating the session. Pass an empty string for the default.

func (*Pool) SessionExists added in v0.9.4

func (p *Pool) SessionExists(sessionID string) bool

Send routes a user message into the right session. If a slot is free the agent is spawned and the message sent immediately; else the message is appended to the session's buffer and the request is queued. The on-disk session meta status is updated to reflect running/queued so UI listings stay correct. SessionExists reports whether sessionID already has on-disk state. Cheap stat — no JSON parse. Used by channels (Slack, Telegram) to decide whether the next inbound message starts a brand-new session and needs a one-time origin-context turn injected before the user message.

Implements channels.SessionChecker.

func (*Pool) SetAutoReply added in v0.28.0

func (p *Pool) SetAutoReply(sessionID string, on bool)

SetAutoReply persists the Slack auto-reply flag on the session meta. A missing session or save error is logged and swallowed — the in-memory fallback in the channel keeps the turn working even if persistence fails.

func (*Pool) SetMaxTurns added in v0.15.5

func (p *Pool) SetMaxTurns(sessionID, agentName string, maxTurns int) error

SetMaxTurns persists the per-spawn turn cap on the session's agent entry (creating it if missing) so the next spawn passes --max-turns.

func (*Pool) SetThinkingTokens added in v0.18.7

func (p *Pool) SetThinkingTokens(sessionID, agentName, v string) error

SetThinkingTokens persists the resolved MAX_THINKING_TOKENS env value on the session's agent entry (creating it if missing) so the next spawn applies it. Empty = unset (provider default); "0" = disabled; "<n>" = token budget.

func (*Pool) Stop

func (p *Pool) Stop()

Stop tears down all active agents and waits for trailing post-exit work (markStatus, queue drain). Used on graceful shutdown and by tests to flush goroutines before TempDir cleanup.

type PoolConfig

type PoolConfig struct {
	MaxConcurrent int
	IdleTimeout   time.Duration
	KillAfterIdle time.Duration
	// PreemptIdle, when true, lets a queued send kick out the longest-idle
	// active subprocess (Lifecycle == Idle) so the new session doesn't have
	// to wait for the idle TTL. The preempted session keeps its CLI session
	// ID in agents.json and resumes via --resume on its next message.
	PreemptIdle      bool
	Layout           config.Layout
	Factory          AgentFactory
	DefaultProjectID string
	// OnSessionCreated is called after the pool auto-creates a session for a
	// channel message (e.g. Slack thread_ts). Wire this to
	// manager.Register so the dashboard sees the session immediately.
	OnSessionCreated func(s session.Session)
	// OnAgentAdded is called after the pool auto-adds an agent entry to
	// agents.json (channel sessions bypass the UI AddAgent flow). Wire
	// this to manager.RefreshSession so the in-memory registry reflects
	// the new agent before sendMessage resolves agentName.
	OnAgentAdded func(sessionID string)
	// OnSessionMeta is called after the pool mutates a session's meta on
	// disk (e.g. setLabelIfEmpty derives the first-message title). Wire
	// this to syncSessionMeta so the in-memory registry refreshes and the
	// new title broadcasts over SSE — otherwise the sidebar/list would
	// only catch up on the next page load. Optional; nil = no callback.
	OnSessionMeta func(sessionID string)
	// OnLifecycle fires when the pool transitions a session+agent's
	// lifecycle (Spawning, Killed). Idle/Working transitions are
	// implicit from event flow and are NOT routed here — UIs that
	// want every transition should subscribe to AgentEvent via
	// the factory's OnEvent. Optional; nil = no callback.
	OnLifecycle func(LifecycleEvent)
}

PoolConfig knobs.

DefaultProjectID is the project id used when a session has no project bound. Empty = no default; the pool falls back to a per-session temp dir so claude still has a stable cwd. See agents-design.md §0.2 D4.

type QueueEntry

type QueueEntry struct {
	SessionID string
	AgentName string
	Enqueued  time.Time
}

QueueEntry is the public snapshot view of one queued request.

type SpawnStartMeta

type SpawnStartMeta struct {
	PID              int
	Binary           string
	Argv             []string
	FirstUserMessage string
}

SpawnStartMeta is the post-Start snapshot the pool feeds back to the factory so the spawn log gets a complete `start` record. PID, argv, and binary path are only knowable after Spawner.Spawn returns; FirstUserMessage comes from the buffer drain.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL