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 ¶
- type ActiveEntry
- type AgentFactory
- type Buffer
- type BuildResult
- type Capacity
- type ClaudeFactory
- type FactoryOptions
- type GateConfig
- type LifecycleEvent
- type Pool
- func (p *Pool) Active() int
- func (p *Pool) ActiveSnapshot() []ActiveEntry
- func (p *Pool) AutoReplyOn(sessionID string) bool
- func (p *Pool) Capacity() Capacity
- func (p *Pool) Dequeue(sessionID, agentName string) int
- func (p *Pool) DequeueSession(sessionID string) int
- func (p *Pool) EnsureSession(ctx context.Context, sessionID, source, projectID string) error
- func (p *Pool) EnsureSessionOwner(ctx context.Context, sessionID, userID string)
- func (p *Pool) HandleExit(sessionID, agentName string, reason provider.ExitReason)
- func (p *Pool) IdleTimeout() time.Duration
- func (p *Pool) Kill(sessionID, agentName string) error
- func (p *Pool) MaxConcurrent() int
- func (p *Pool) ProviderCapacity(pType, pName string) Capacity
- func (p *Pool) QueueLen() int
- func (p *Pool) QueueSnapshot() []QueueEntry
- func (p *Pool) ReconcileDead()
- func (p *Pool) Send(ctx context.Context, sessionID, agentName, source, role, text string) error
- func (p *Pool) SendWithAttachments(ctx context.Context, ...) error
- func (p *Pool) SendWithProject(ctx context.Context, ...) error
- func (p *Pool) SessionExists(sessionID string) bool
- func (p *Pool) SetAutoReply(sessionID string, on bool)
- func (p *Pool) SetMaxTurns(sessionID, agentName string, maxTurns int) error
- func (p *Pool) SetThinkingTokens(sessionID, agentName, v string) error
- func (p *Pool) Stop()
- type PoolConfig
- type QueueEntry
- type SpawnStartMeta
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 ¶
NewBuffer reads any pending_input persisted on disk so we resume queued messages after a wick restart.
func (*Buffer) Append ¶
Append adds a new line and persists it to meta.PendingInput so a crash before drain doesn't drop it.
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).
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) 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
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
Capacity returns the global slot usage. Caller need not hold p.mu.
func (*Pool) Dequeue ¶
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
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
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
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 ¶
IdleTimeout returns the configured idle timeout. UI consumers use it to render the auto-kill countdown alongside LastActive.
func (*Pool) Kill ¶
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 ¶
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
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) 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) 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
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
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
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
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.
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 ¶
QueueEntry is the public snapshot view of one queued request.
type SpawnStartMeta ¶
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.