Help us shape the future of Go! Take the 2026 Go Developer Survey

acp

package
v1.0.0 Latest Latest
Warning

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

 Go to latest Published: Oct 29, 2026 License: Apache-2.0 Imports: 25 Imported by: 0

Documentation

Index

Constants

View Source 
const (
	ACPChildMarkerEnv   = "QUARTET_ACP_CHILD"
	ACPChildMarkerValue = "1"
)

ACPChildMarkerEnv / ACPChildMarkerValue identify subprocesses that the quartet process itself spawned as ACP agent children. The pair is injected into cmd.Env by NewConn (pkg/acp/conn.go) and inherited across fork/exec, so the whole subprocess tree (including sh/node wrappers that npx creates) carries it.

Orphan cleanup on Linux (pkg/acp/process_unix.go) uses this marker as the authoritative "this tree is ours to kill" check — keyword-based cmdline matching is only a cheap prefilter and cannot, on its own, authorize a kill(-pgid) on a PPID==1 root. Without the marker check, substring collisions (e.g. "openclaw" matching /etc/sysop/mongoosev3-agent/plugin/ openclaw-collector/openclaw-collector, or "kimi" matching an unrelated user command in an ssh session) would escalate to killing system-level service process groups.

Variables

This section is empty.

Functions

func AllowedAgentCommands 

func AllowedAgentCommands() []string

AllowedAgentCommands returns a snapshot of the current allowed command list.

func BeginRuntimeMaintenance 

func BeginRuntimeMaintenance(key string) (func(), error)

BeginRuntimeMaintenance is fail-fast: it never interrupts an active prompt, authentication attempt, probe or constructor. The caller must release it.

func CleanupOrphanedConns 

func CleanupOrphanedConns()

CleanupOrphanedConns finds ACP agent subprocesses orphaned by a previous parent crash and kills their process groups. This should be called once at server startup before any new connections are created.

Matching has three stages:

  1. Cheap filter — allowlist keyword appears as a substring of /proc/<pid>/cmdline. Keywords like "openclaw", "droid", "kimi" are specific enough to look distinctive but can still collide with unrelated processes (e.g. /etc/sysop/mongoosev3-agent/plugin/ openclaw-collector/openclaw-collector or an ssh session running a user command happening to contain "kimi"). That substring match on its own is NOT safe to escalate to kill(-pgid).

  2. Marker guard — every ACP subprocess NewConn launches is started with QUARTET_ACP_CHILD=1 in its env. Children inherit env across fork/ exec, so the whole subtree (including the sh/node wrappers npx creates) carries the marker. Before we kill the PPID==1 root's process group we require that marker on both the matched process AND the root — a non-quartet root (sshd, mongoosev3-agent, any system service) has no way to carry it and is skipped.

  3. System-root blacklist — even if a future bug or env propagation issue lets the marker leak into something it should not, we refuse to kill anything whose root cmdline basename matches a known system service (sshd, systemd, init, mongoose*-agent, ...). Belt-and- suspenders: the marker check is the contract, this is the safety net for when the contract is violated.

Earlier revisions only did stage 1 and then kill(-pgid) on the root, which produced real system-level kills of sshd and sysop agents on machines where unrelated processes happened to share a keyword.

func CloseAllConns 

func CloseAllConns()

CloseAllConns closes all tracked connections. Use during graceful shutdown.

func IsAllowedAgentCommand 

func IsAllowedAgentCommand(agentType string) bool

IsAllowedAgentCommand checks whether the given agentType is in the whitelist.

func IsBenignCloseErr 

func IsBenignCloseErr(err error) bool

IsBenignCloseErr reports whether err signals that the underlying ACP transport or JSON-RPC connection has already been closed. Best-effort notifications such as CancelSession on shutdown / run-switch routinely race the connection close (peer exit, idle reap, prior Conn.Close), and the resulting transport-closed / connection-closed return is part of the expected teardown — not a fault to alert on. Callers should downgrade the log level for these and keep WARN/ERROR for genuine RPC failures.

func IsSessionAlreadyLoaded 

func IsSessionAlreadyLoaded(err error) bool

IsSessionAlreadyLoaded reports whether err is an ACP agent's rejection of a session/resume|load issued for a session that is still loaded in the live subprocess. Some agents implement resume/load as a cold-restore-only operation (valid on a fresh connection / after a process restart) and answer with a JSON-RPC error whose payload says the session is "already loaded" when the target is still resident in memory. On a live connection that rejection is benign: it confirms the session is present and ready to prompt, so the caller can keep using the existing session id instead of treating it as a restore failure.

func RegisterAgentRuntime 

func RegisterAgentRuntime(key string, definition RuntimeDefinition) error

func SetEnvProvider 

func SetEnvProvider(p EnvProvider)

SetEnvProvider installs a provider that NewConn consults when populating the subprocess environment. Passing nil clears the provider.

func StartIdleReaper 

func StartIdleReaper() (stop func())

StartIdleReaper launches a background goroutine that periodically scans tracked connections and closes ones that have been idle for longer than connIdleTimeout. This prevents unbounded accumulation of subprocess memory when sessions complete but the server keeps running.

Call the returned stop function to terminate the reaper (e.g. during shutdown).

func UnregisterAgentRuntime 

func UnregisterAgentRuntime(key string)

Types

type AuthenticationRequiredError 

type AuthenticationRequiredError struct {
	Err     error
	Methods []acp.AuthMethod
}

AuthenticationRequiredError preserves the original protocol error and all diagnostics while allowing management code to distinguish missing login.

func (*AuthenticationRequiredError) Error 

func (e *AuthenticationRequiredError) Error() string

func (*AuthenticationRequiredError) Unwrap 

func (e *AuthenticationRequiredError) Unwrap() error

type Conn 

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

Conn is a live connection to an ACP agent subprocess. It owns the child process, captures its stderr, and serializes Prompt calls across sessions (the subprocess supports one active Prompt at a time).

func NewAuthenticationConn 

func NewAuthenticationConn(ctx context.Context, agentType, workdir string, output func(string), link func(string, string)) (*Conn, error)

NewAuthenticationConn only initializes the protocol. Authentication itself is an explicit, separately cancellable action. Terminal methods are executed by the user in an interactive terminal on the backend host; they are never incorrectly sent to the authenticate RPC.

func NewConn 

func NewConn(ctx context.Context, agentType, workdir string) (*Conn, error)

NewConn starts an ACP agent subprocess and completes the initialize handshake. Caller is responsible for tracking the returned Conn for idle reaping if desired; most callers should use NewTrackedConn instead.

func NewProbeConn 

func NewProbeConn(ctx context.Context, agentType, workdir string) (*Conn, error)

NewProbeConn creates a short-lived Conn for capability probing.

Unlike NewTrackedConn it does NOT detach from the caller's cancellation signal. A probe deliberately sets a tight deadline (see services/agent/probe.acpProbeTimeout) that must bound BOTH subprocess startup + initialize handshake AND the throwaway session/new that follows, so an agent that is installed but not logged in — and therefore never answers session/new — can't wedge the probe or the /agent/list request that triggered it. NewTrackedConn preserves an earlier deadline but deliberately ignores direct cancellation for shared creation; a probe must honor both.

The returned Conn is not registered for idle reaping: probe callers Close it immediately after reading its session info.

func NewTrackedConn 

func NewTrackedConn(ctx context.Context, agentType, workdir string) (*Conn, error)

NewTrackedConn creates a new Conn with the standard startup timeout and registers it for idle reaping. Every caller path (service agent, probe) should go through this helper.

func (*Conn) AcquirePromptSlot 

func (c *Conn) AcquirePromptSlot(ctx context.Context, sessionID SessionID) (*PromptSlot, error)

AcquirePromptSlot best-effort cancels any active prompt on this connection, waits for the prompt semaphore, records this session as the active prompt, and returns a slot handle. The slot MUST be Released by the caller (typically via defer) before any other prompt on this connection can proceed.

cancelActivePromptTimeout only bounds the SessionCancel RPC itself; waiting for the previous prompt to release promptSem is governed by ctx. In the service path, ACPAgent.runSem ensures the previous Run has already unwound its handler cleanup before the next Run reaches this method.

Splitting slot acquisition from prompt send lets callers install a stream handler in the interval between "old prompt released the slot" and "subprocess receives new prompt text", closing the residue-delivery race described in ACPAgent.Run.

func (*Conn) AuthMethods 

func (c *Conn) AuthMethods() []acp.AuthMethod

AuthMethods is connection-scoped and immutable after initialize.

func (*Conn) Authenticate 

func (c *Conn) Authenticate(ctx context.Context, methodID string) error

func (*Conn) CancelActivePrompt 

func (c *Conn) CancelActivePrompt(ctx context.Context)

CancelActivePrompt sends a Cancel notification for whichever session is currently holding the prompt slot on this connection, if any. The caller supplies a context so this RPC cannot block Stop / run-switch on a stuck subprocess; pass a context.WithTimeout(WithoutCancel(parent), ...) when you want bounded waits even if the parent has already been cancelled.

func (*Conn) CancelSession 

func (c *Conn) CancelSession(ctx context.Context, sessionID SessionID) error

CancelSession sends a Cancel notification for the specified session.

func (*Conn) ClearStreamHandlerIfGen 

func (c *Conn) ClearStreamHandlerIfGen(sid acp.SessionID, gen uint64)

ClearStreamHandlerIfGen clears the handler only if the generation still matches, avoiding a race where a finishing Run clears a newer Run's handler.

func (*Conn) Close 

func (c *Conn) Close()

Close tears down the subprocess. It does not untrack the Conn from the idle-reaper list — dead connections are pruned on the next reap cycle.

func (*Conn) ClosedByIdleReap 

func (c *Conn) ClosedByIdleReap() bool

ClosedByIdleReap reports whether this connection's Close was triggered by the idle reaper. Returns false for any other cause (explicit shutdown, subprocess crash, OOM, handshake error).

func (*Conn) IsAlive 

func (c *Conn) IsAlive() bool

IsAlive reports whether the connection is still usable. A connection that has already been selected by the idle reaper is treated as not alive even if its subprocess has not finished terminating yet, so callers won't race into a Conn that is about to be closed outside the pool lock.

func (*Conn) LastUsedAt 

func (c *Conn) LastUsedAt() time.Time

LastUsedAt returns the timestamp of the last activity on this connection.

func (*Conn) LoadSession 

func (c *Conn) LoadSession(ctx context.Context, sessionID, workdir string) (*SessionResponse, error)

LoadSession restores an existing ACP session on this connection.

func (*Conn) MarkClosedByIdleReap 

func (c *Conn) MarkClosedByIdleReap()

MarkClosedByIdleReap records that this connection is being closed by the idle reaper. The next reconnect attempt uses this to downgrade its "subprocess dead" log from WARN to INFO, so an expected idle recycle doesn't look like a crash in log aggregates. Must be called BEFORE Close so the flag is visible if the reap and reconnect race.

func (*Conn) NewSession 

func (c *Conn) NewSession(ctx context.Context, workdir string) (*SessionResponse, error)

NewSession creates a fresh ACP session on this connection.

func (*Conn) Pid 

func (c *Conn) Pid() int

Pid returns the subprocess PID, or 0 if the process is not running.

func (*Conn) ProbeSession 

func (c *Conn) ProbeSession(ctx context.Context, workdir string) (*SessionResponse, error)

ProbeSession refuses interactive login. Some servers start OAuth from session/new instead of returning -32000; stop that request once they report a login prompt, with the full stderr preserved for the user.

func (*Conn) ReleaseUse 

func (c *Conn) ReleaseUse()

ReleaseUse releases a prior successful TryAcquireUse call.

func (*Conn) RemoveSession 

func (c *Conn) RemoveSession(sid acp.SessionID)

RemoveSession drops the session's registry entry entirely. Call when the session is closed or the owning agent is torn down.

func (*Conn) ResumeSession 

func (c *Conn) ResumeSession(ctx context.Context, sessionID, workdir string) (*SessionResponse, error)

ResumeSession restores an existing ACP session on this connection WITHOUT replaying conversation history. Unlike LoadSession (which the protocol requires to re-stream every prior turn via session/update notifications before responding), session/resume restores the subprocess-side context and returns once the session is ready — no replay events are emitted, so the caller's stream handler never sees stale historical output.

Only valid when SupportsResume reports true. The SDK method is marked UNSTABLE; callers MUST gate on the advertised capability and fall back to LoadSession when resume is unavailable.

func (*Conn) SetSessionConfigOption 

func (c *Conn) SetSessionConfigOption(ctx context.Context, sessionID SessionID, configID, value string) (*SessionResponse, error)

SetSessionConfigOption sets an arbitrary session config option by id. It is the single generic path for options that have no dedicated RPC — currently thought_level (e.g. "reasoning_effort") and mode on backends that reject session/set_mode (see SetSessionMode). The response carries the full, freshly-linked ConfigOptions list, returned so callers can refresh selectors that changed as a side effect.

func (*Conn) SetSessionMode 

func (c *Conn) SetSessionMode(ctx context.Context, sessionID SessionID, mode, modeConfigID string) (*SessionResponse, error)

SetSessionMode switches the active mode for the session.

The preferred path is the dedicated session/set_mode RPC, whose response carries no ConfigOptions — so on that path the returned SessionResponse has an empty ConfigOptions list (confirmation only, no refreshed option lists).

Some ACP backends (notably antigravity-acp) do not implement session/set_mode and instead expose the mode as a generic config option. When session/set_mode is rejected as an unknown method (JSON-RPC -32601) and modeConfigID is non-empty, the switch is retried through SetSessionConfigOption keyed by that id; that path returns the full, freshly-linked ConfigOptions list. Pass modeConfigID="" to disable the fallback (any set_mode failure surfaces as-is).

func (*Conn) SetSessionModel 

func (c *Conn) SetSessionModel(ctx context.Context, sessionID SessionID, modelID string) (*SessionResponse, error)

SetSessionModel switches the active model for the session. The dedicated session/set_model RPC was removed from the ACP v1 schema, so model selection now always goes through the generic SetSessionConfigOption API keyed by the "model" config option. The response carries the full, freshly-linked ConfigOptions list, which is returned so callers can refresh model / thought_level selectors that changed as a side effect.

func (*Conn) SetSessionThoughtLevel 

func (c *Conn) SetSessionThoughtLevel(ctx context.Context, sessionID SessionID, configID, value string) (*SessionResponse, error)

SetSessionThoughtLevel switches the active thought_level for the session. thought_level has no dedicated RPC, so it always goes through the generic SetSessionConfigOption API keyed by the config option id (e.g. "reasoning_effort") discovered from the session's ConfigOptions.

func (*Conn) SetStreamHandler 

func (c *Conn) SetStreamHandler(sid acp.SessionID, h agentstream.StreamHandler) uint64

SetStreamHandler installs the handler for the given session and returns a generation number. Pair with ClearStreamHandlerIfGen so a finishing Run cannot clear a newer Run's handler.

func (*Conn) Stderr 

func (c *Conn) Stderr() string

Stderr returns the captured subprocess stderr.

func (*Conn) SupportsResume 

func (c *Conn) SupportsResume() bool

SupportsResume reports whether the agent advertised the sessionCapabilities.resume capability during initialize. When true, the reconnect path can use ResumeSession instead of LoadSession to restore the session without replaying conversation history.

func (*Conn) TouchLastUsed 

func (c *Conn) TouchLastUsed()

TouchLastUsed refreshes lastUsedAt to prevent the idle reaper from closing this connection. Called by the service layer on each user message and by the internal update dispatcher on every session update.

func (*Conn) TryAcquireUse 

func (c *Conn) TryAcquireUse() bool

TryAcquireUse claims this connection for a caller that is about to issue RPCs to the subprocess. It closes the race where the idle reaper selects an idle-but-live Conn for teardown while a new Run is starting on that same Conn. The reaper checks the same useMu before marking/closing candidates.

type EnvProvider 

type EnvProvider func(agentType string) []EnvVar

EnvProvider resolves extra environment variables for a given agent command. Set once at startup from the service layer (e.g. reading user settings); pkg/acp consults it inside NewConn but does not own the lookup logic.

type EnvVar 

type EnvVar struct {
	Key   string
	Value string
}

EnvVar represents a KEY=VALUE environment variable pair injected into the ACP agent subprocess at startup.

type PromptResult 

type PromptResult struct {
	StopReason PromptStopReason
	Usage      *PromptUsage
}

PromptResult is the stable prompt completion view exposed by pkg/acp. Usage may be populated even when SendPrompt returns a non-nil error.

type PromptSlot 

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

PromptSlot is a held prompt slot on a Conn. While holding it, the caller has exclusive right to call SendPrompt on the underlying connection and is safely recorded as the connection's active prompt session. Callers install a stream handler AFTER acquiring the slot so that residue updates from the previous (cancelled) prompt cannot reach the new handler — see the race discussion in services/agent/acp/agent.go.Run.

func (*PromptSlot) Release 

func (s *PromptSlot) Release()

Release returns the slot to the connection. Idempotent — subsequent calls are no-ops — so callers can use both defer and explicit release.

func (*PromptSlot) SendPrompt 

func (s *PromptSlot) SendPrompt(ctx context.Context, text string) (PromptResult, error)

SendPrompt sends the prompt text on the held slot and blocks until the agent finishes the turn. The terminal response is returned to the caller: a successful JSON-RPC exchange can still represent an incomplete turn (cancelled, max_tokens, refusal, and so on). Must be called after AcquirePromptSlot and before Release.

type PromptStopReason 

type PromptStopReason string

PromptStopReason is the terminal state returned by an ACP agent for one prompt turn. Keep the SDK value behind this package boundary so service layers do not need to import the protocol SDK directly. 

const (
	PromptStopReasonEndTurn         PromptStopReason = "end_turn"
	PromptStopReasonMaxTokens       PromptStopReason = "max_tokens"
	PromptStopReasonMaxTurnRequests PromptStopReason = "max_turn_requests"
	PromptStopReasonRefusal         PromptStopReason = "refusal"
	PromptStopReasonCancelled       PromptStopReason = "cancelled"
)

type PromptUsage 

type PromptUsage struct {
	InputTokens       int64
	OutputTokens      int64
	TotalTokens       int64
	CachedReadTokens  *int64
	CachedWriteTokens *int64
	ThoughtTokens     *int64
}

PromptUsage is the optional session-cumulative usage carried by PromptResponse or, for an eino-cli failed turn, its vendor error.data extension. The ACP schema defines these counters across the session; service callers must convert them to a per-turn delta before additive accounting.

type RuntimeDefinition 

type RuntimeDefinition struct {
	Program     string
	Args        []string
	EnvKey      string
	Env         []EnvVar
	OverrideEnv bool
}

func AgentRuntime 

func AgentRuntime(key string) (RuntimeDefinition, bool)

type SessionConfigSelect 

type SessionConfigSelect struct {
	// ConfigID is the ACP config option id (e.g. "reasoning_effort") used to
	// set this value through SetSessionConfigOption. May be empty for agents
	// that omit it.
	ConfigID     string
	CurrentValue string
	Options      []SessionConfigSelectItem
}

SessionConfigSelect is a stable view of a select option in ConfigOptions, avoiding SDK types leaking into higher service layers.

type SessionConfigSelectItem 

type SessionConfigSelectItem struct {
	Description string
	Name        string
	Value       string
}

type SessionID 

type SessionID = acp.SessionID

SessionID aliases the SDK session id so callers can store it without importing the SDK directly.

type SessionResponse 

type SessionResponse struct {
	SessionID     string
	ConfigOptions []acp.SessionConfigOption
	Modes         *acp.SessionModeState
}

SessionResponse is the stable session view exposed by pkg/acp. Keep it independent from the SDK response shape so LoadSession does not need to synthesize a fake NewSessionResponse and callers do not have to choose between an embedded SDK SessionID and an outer shadowing field.

func (*SessionResponse) ModeConfigSelect 

func (r *SessionResponse) ModeConfigSelect() *SessionConfigSelect

ModeConfigSelect returns the mode selector carried in ConfigOptions. Some ACP agents expose mode choices there instead of filling Modes.

func (*SessionResponse) ModelConfigSelect 

func (r *SessionResponse) ModelConfigSelect() *SessionConfigSelect

ModelConfigSelect returns the model selector carried in ConfigOptions. Some ACP agents expose model choices there instead of filling Models.

func (*SessionResponse) ThoughtLevelConfigSelect 

func (r *SessionResponse) ThoughtLevelConfigSelect() *SessionConfigSelect

ThoughtLevelConfigSelect returns the thought_level selector carried in ConfigOptions. Unlike mode, thought_level has no dedicated RPC, so callers drive it through SetSessionConfigOption using the select's ConfigID.

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.