Documentation
¶
Index ¶
- Variables
- func InitEnvProvider()
- type ACPAgent
- func (a *ACPAgent) Cancel()
- func (a *ACPAgent) Close()
- func (a *ACPAgent) FlushPendingMessages()
- func (a *ACPAgent) IsRunning() bool
- func (a *ACPAgent) RequiresRebuild(agentType, workdir string) bool
- func (a *ACPAgent) Run(ctx context.Context, userMessages []*schema.Message, handler agui.EventHandler, ...) (retErr error)
- func (a *ACPAgent) SessionID() pkgacp.SessionID
- func (a *ACPAgent) SetMode(ctx context.Context, mode string) (*model.ACPConfigState, error)
- func (a *ACPAgent) SetModel(ctx context.Context, modelID string) (*model.ACPConfigState, error)
- func (a *ACPAgent) SetThoughtLevel(ctx context.Context, thoughtLevel string) (*model.ACPConfigState, error)
- func (a *ACPAgent) StopAndFlush()
- func (a *ACPAgent) UpdateACPMode(ctx context.Context, mode string) error
- func (a *ACPAgent) UpdateACPModelID(ctx context.Context, modelID string) error
- func (a *ACPAgent) UpdateACPThoughtLevel(ctx context.Context, thoughtLevel string) error
- type ACPService
- type Lease
- type PhaseReporter
- type SessionStore
Constants ¶
This section is empty.
Variables ¶
var ErrACPAgentBusyRebuildRequired = errors.New("acp agent rebuild required but a Run is in flight; retry after the previous Run completes")
ErrACPAgentBusyRebuildRequired is returned when a fresh request asks for an ACP agent whose construction inputs (agentType / workdir) differ from a cached agent that is currently mid-Run. Those inputs are baked into the subprocess + tracked-conn at NewACPAgent() time and cannot be hot-swapped, so silently returning the stale instance would dispatch to the wrong subprocess. The session-level Run gate normally prevents concurrent Runs in the happy path; surface a typed error here for the rare collision instead of silently using a stale agent.
var ErrACPAgentCapacityExceeded = fmt.Errorf("acp %w", sessioncache.ErrCapacityExceeded)
ErrACPAgentCapacityExceeded is returned when the in-memory ACP agent cache is at capacity and there is no idle agent to evict. Wraps the shared sentinel so existing `errors.Is(err, ErrACPAgentCapacityExceeded)` sites continue to match.
Functions ¶
func InitEnvProvider ¶
func InitEnvProvider()
InitEnvProvider wires the service-layer settings lookup into pkg/acp so NewConn can inject per-agent environment variables without the pkg/acp package owning settings repository access. Must be called once at startup (cmd/web/main.go) before any ACP subprocess is launched.
Made explicit (rather than an init()) so startup ordering is visible in the entrypoint and tests can wire a different provider by calling pkgacp.SetEnvProvider directly without import-ordering subtleties.
Types ¶
type ACPAgent ¶
type ACPAgent struct {
// contains filtered or unexported fields
}
ACPAgent wraps a pkg/acp connection + session with the quartet business concerns: chat context persistence, message round aggregation, and transparent reconnect when the subprocess dies.
Concurrency model: conn/acpSession/currentModelID/currentMode can mutate mid-lifetime (reconnectIfNeeded, restoreACPSession, UpdateACP*), while Cancel/Close may fire from an external goroutine (user hits stop, session cleanup). All reads and writes of those fields go through a.mu; long I/O runs against a snapshot taken under the lock so we never block the subprocess on a pipe write while holding it.
func NewACPAgent ¶
func NewACPAgent(ctx context.Context, store SessionStore, sessionID, agentType, workdir, jobID, wsID string) (_ *ACPAgent, retErr error)
NewACPAgent starts an ACP agent subprocess (via pkg/acp), creates or reloads the ACP session for this quartet session, and prepares the chat context manager so Run() can stream events.
func (*ACPAgent) Cancel ¶
func (a *ACPAgent) Cancel()
Cancel stops the current run.
Order matters: cancel the local runCtx FIRST so the in-flight Run starts unwinding immediately, then fire-and-forget the subprocess cancel notification. The notification can stall up to cancelACPSessionTimeout (2s) on a slow stdio peer; running it synchronously would delay the user-visible Stop / StopAndFlush / Close path by that much without affecting how fast the local Run actually exits.
func (*ACPAgent) Close ¶
func (a *ACPAgent) Close()
Close releases the ACPAgent resources. The underlying subprocess is managed by the pool and not killed here.
Cancel() unwinds the in-flight Run, but Run's deferred cleanup (EmitPendingEnds / CollectMessages / token-usage fallback / post-run sync count persist — see services/agent/acp/agent.go:703) still needs to finish before we drop the stream handler. Without the wait, a concurrent Close from cleanupSessions can race the deferred block: late terminal events arrive after RemoveSession and get dropped, the final round's Placeholder ends never reach the UI, and the persisted sync count baseline can lag the on-disk messages.jsonl. All agents share this shutdown ordering.
func (*ACPAgent) FlushPendingMessages ¶
func (a *ACPAgent) FlushPendingMessages()
FlushPendingMessages flushes any accumulated in-memory messages from the current LLM round to persistent storage. This should be called before Cancel() to ensure in-flight messages are not lost. Delegates to round.FlushPending, the shared cleanup path. The persist call is wrapped in round.PersistContext so the canonical persist deadline applies — see round.PersistTimeout for the scope of what that deadline currently covers (lock-wait yes, blocking file I/O no).
func (*ACPAgent) IsRunning ¶
IsRunning reports whether a Run() invocation is currently in flight. Used by the service layer to skip candidates during LRU eviction.
func (*ACPAgent) RequiresRebuild ¶
RequiresRebuild reports whether a cached ACPAgent must be discarded and recreated to honour the next call's parameters. agentType / workdir are baked into the subprocess + tracked-conn at construction (NewTrackedConn, env loading) and cannot be hot-swapped. The model is hot-swapped on the live session via SetSessionModel (from applyPersistedConfig at Run time or an explicit SetModel switch), so it does not factor into rebuild.
func (*ACPAgent) Run ¶
func (a *ACPAgent) Run(ctx context.Context, userMessages []*schema.Message, handler agui.EventHandler, jobID string, report PhaseReporter) (retErr error)
Run sends user messages to the ACP agent and streams events via handler. jobID is threaded into the round builder's log label so WARN lines (round eager flush superseded / drop terminal ...) can be traced back to the concrete job, not just the session pair.
func (*ACPAgent) SessionID ¶
SessionID returns the subprocess-level ACP session id. Exposed for observability (debug logging in external callers); callers MUST treat it as read-only — it can change under the hood when restoreACPSession / reconnectIfNeeded re-restores the subprocess session.
func (*ACPAgent) SetMode ¶
SetMode switches the mode on the live session. The mode RPC carries no ConfigOptions, so the returned state has all lists nil — the caller keeps its current selector lists and just reflects the newly-active mode.
func (*ACPAgent) SetModel ¶
SetModel switches the model on the live session and returns the refreshed selector lists carried in the ACP response (model + thought_level may both change as a linked side effect). Unlike UpdateACPModelID it does not short-circuit on an unchanged value: the caller is an explicit user switch that wants the freshly-linked ConfigOptions back.
func (*ACPAgent) SetThoughtLevel ¶
func (a *ACPAgent) SetThoughtLevel(ctx context.Context, thoughtLevel string) (*model.ACPConfigState, error)
SetThoughtLevel switches the thought_level on the live session and returns the refreshed selector lists carried in the ACP response (model + thought_level may both change as a linked side effect).
func (*ACPAgent) StopAndFlush ¶
func (a *ACPAgent) StopAndFlush()
StopAndFlush atomically cancels any in-flight Run and ensures the builder's accumulated state reaches disk. Replaces the prior FlushPendingMessages -> Cancel idiom, which had a real race window: FlushPendingMessages emitted pending-end events and cleared the builder's accumulators, but the Run was still active and could push new stream events into the just-cleared builder. Late tool terminals arriving after the eager superseded flush would never reach disk — only the canceled placeholder would.
New order:
- Cancel runCtx (and notify the ACP subprocess of cancel) so the Run loop exits ASAP.
- Wait (bounded by stopAndFlushTimeout) for Run to finish its deferred cleanup chain — that chain finalizes the current round while onFlush is still installed, so any in-flight round is persisted atomically with the cancel.
- As a safety net, run FlushPendingMessages. If Run already drained the builder, this is a no-op. If Run hung past the timeout, this gets the builder's accumulators onto disk via the inline persist callback that runs because onFlush has been ClearOnFlush'd.
func (*ACPAgent) UpdateACPMode ¶
UpdateACPMode is the mode counterpart to UpdateACPModelID with the same fail-fast contract: a SetSessionMode failure is returned so the caller does not run under a mode different from what the user requested. Mode affects tool availability and assistant behavior, so "log-and-continue" would produce a silent semantic mismatch. Exception: antigravity-acp does not implement session/set_mode. setMode already falls back to the mode config option for it (see setMode / pkgacp.SetSessionMode); only if that fallback ALSO fails do we log and skip for that agent type rather than failing the whole Run.
func (*ACPAgent) UpdateACPModelID ¶
UpdateACPModelID pushes a model selection to the subprocess session. Returns nil on success or when the selection already matches the last-pushed value (no-op). A failed SetSessionModel surfaces as an error — model identity is part of the Run's semantic contract, so silently continuing on the previously-active model would cause the caller to respond with a model the user did not choose. currentModelID stays unchanged on failure so a retry (next Run with the same id) will re-attempt instead of being short-circuited by the equality check.
func (*ACPAgent) UpdateACPThoughtLevel ¶
UpdateACPThoughtLevel is the thought_level counterpart to UpdateACPMode. thought_level has no dedicated RPC, so it is pushed through the generic SetSessionConfigOption keyed by the config id discovered from the session (e.g. "reasoning_effort"). Same fail-fast contract: a failure is returned so the caller does not run under a thought_level different from what the user requested. currentThoughtLevel stays unchanged on failure so a retry re-attempts instead of being short-circuited by the equality check.
type ACPService ¶
type ACPService interface {
GetOrCreate(ctx context.Context, store SessionStore, wsID, jobID, sessionID, agentType, workdir string) (*Lease, error)
Get(wsID, jobID, sessionID string) (*Lease, bool)
Delete(wsID, jobID, sessionID string)
DeleteByAgent(agentID string) int
DeleteByAgentIdentifiers(identifiers []string) int
// PersistPendingMessages records user input when agent construction fails
// before ACPAgent.Run gets a chance to call ChatContextManager.BeginRun.
// Messages already tagged as pre-persisted are not appended twice.
PersistPendingMessages(ctx context.Context, store SessionStore, wsID, jobID, sessionID string, messages []*schema.Message) error
// SetModel / SetMode / SetThoughtLevel apply a live config switch on the
// session's ACP agent and return the refreshed selector lists. They
// get-or-create the agent, apply the switch, and release the lease.
SetModel(ctx context.Context, store SessionStore, wsID, jobID, sessionID, agentType, workdir, modelID string) (*model.ACPConfigState, error)
SetMode(ctx context.Context, store SessionStore, wsID, jobID, sessionID, agentType, workdir, mode string) (*model.ACPConfigState, error)
SetThoughtLevel(ctx context.Context, store SessionStore, wsID, jobID, sessionID, agentType, workdir, thoughtLevel string) (*model.ACPConfigState, error)
}
ACPService manages ACPAgent instances per session.
func NewACPService ¶
func NewACPService() ACPService
type Lease ¶
type Lease = sessioncache.Lease[*ACPAgent]
Lease is a borrowed reference to a cached *ACPAgent. Callers obtain one from GetOrCreate / Get and MUST Release it (via `defer lease.Release()`) when the operation that needs the agent is done. While outstanding, the cache will not Close() the agent under the caller, even if eviction or Delete is triggered concurrently.
type PhaseReporter ¶
type PhaseReporter func(phase, detail string)
PhaseReporter is an optional progress callback invoked during the "silent" preparation window of a Run (subprocess launch, reconnect, session restore, waiting for the first token). The handler layer wires it to a transient agent_phase event so the UI can show a specific loading label. Always non-nil inside Run (nil callers are replaced with a no-op), so call sites need not nil-check.
type SessionStore ¶
type SessionStore interface {
Get(sessionID string) (*model.Session, bool)
UpdateACPState(sessionID, acpSessionID string, fingerprint repository.MessagesFingerprint) error
UpdateACPSyncFingerprint(sessionID string, fingerprint repository.MessagesFingerprint) error
Touch(sessionID string) error
}
SessionStore is the narrow session.Service surface the ACP agent needs for loading/writing session metadata. Declared here (rather than importing services/session directly) to keep the ACP package decoupled from the session package wiring.
Source Files