attach

package
v1.8.0 Latest Latest
Warning

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

 Go to latest Published: May 23, 2026 License: Apache-2.0 Imports: 22 Imported by: 0

Documentation

Overview

Package attach implements live-tail + inject over HTTP/SSE for headless core-agent deployments. See docs/attach-mode-design.md.

Server side (agent process): 

reg := attach.NewSessionRegistry()
ag, _ := agent.New(m, agent.WithSessionRegistry(reg), ...)
srv, _ := attach.NewServer(reg, attach.Options{
    Addr:     ":7777",
    TLSCert:  "/etc/certs/server.crt",
    TLSKey:   "/etc/certs/server.key",
    ClientCA: "/etc/certs/ca.crt",  // mTLS
    ReadOnly: false,
})
go srv.ListenAndServe()

Client side (operator on a laptop, or another binary): 

core-agent ls https://pod-ip:7777
core-agent attach https://pod-ip:7777/sessions/<app>/<sid>

The protocol is HTTP + Server-Sent Events. Four endpoints: 

GET  /sessions                                   list active sessions
GET  /sessions/<app>/<sid>/events?since=N        SSE event stream
POST /sessions/<app>/<sid>/inject                queue an inbox message
POST /sessions/<app>/<sid>/wake                  wake a deferred subagent

URL shortcut: /sessions/<sid> works when <sid> is unambiguous across registered apps. Returns 409 with a helpful message on collision.

Index

Constants

View Source 
const (
	ToolSourceBuiltin = "builtin"
	ToolSourceMCP     = "mcp"
	ToolSourceSkill   = "skill"
	ToolSourceOther   = "other"
)

Tool source classifications surfaced via GET /sessions/.../tools. Bare strings (not a typed enum) so JSON clients downstream — the TUI, an eventual WebUI, operator scripts — don't have to know a Go type to reason about them.

View Source 
const (
	AgentStateRunning  = "running"
	AgentStateDeferred = "deferred"
	AgentStatePaused   = "paused"
	AgentStateIdle     = "idle"
)

Agent run-states surfaced via GET /sessions/.../status. "running" covers any active turn; "deferred" means the scheduler is sleeping the agent until NextWakeAt; "paused" means the autonomous loop was explicitly paused (future, via /pause); "idle" means the agent is alive but not currently turning.

View Source 
const (
	DefaultHeartbeatTTL = 60 * time.Second
)

Default TTL bounds. See docs/peer-registration-design.md "TTL + heartbeat policy" for the rationale.

Variables

View Source 
var ErrAmbiguousSession = errors.New("attach: session id is ambiguous across registered apps; use the /sessions/<app>/<sessionID> form")

ErrAmbiguousSession is returned by LookupSingle when more than one registered session shares the same SessionID across different apps — the caller must use the qualified two-segment form.

View Source 
var ErrPeerEndpointRequired = errors.New("attach: peer Endpoint is required")

ErrPeerEndpointRequired is returned when RegisterRequest.Endpoint is empty.

View Source 
var ErrPeerNameRequired = errors.New("attach: peer Name is required")

ErrPeerNameRequired is returned when RegisterRequest.Name is empty.

View Source 
var ErrPeerNotFound = errors.New("attach: peer registration not found")

ErrPeerNotFound is returned when Lookup / Heartbeat / Deregister can't find the registration ID.

View Source 
var ErrSessionExists = errors.New("attach: session already registered")

ErrSessionExists is returned by Register when the (app, user, sid) triple is already registered.

View Source 
var ErrSessionNotFound = errors.New("attach: session not found")

ErrSessionNotFound is returned by Lookup / Unregister when no matching entry exists.

Functions

This section is empty.

Types

type AgentInfo 

type AgentInfo struct {
	ID              string    `json:"id"`
	Name            string    `json:"name"`
	Status          string    `json:"status"` // running | done | failed | paused
	StartedAt       time.Time `json:"started_at"`
	ParentSessionID string    `json:"parent_session_id,omitempty"`
	LastReport      string    `json:"last_report,omitempty"` // most recent report body, truncated
}

AgentInfo is one background subagent the parent agent knows about, surfaced via GET /sessions/.../agents. Populated from the BackgroundAgentManager when one is wired; empty list otherwise.

type AgentRegistrarAdapter 

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

AgentRegistrarAdapter satisfies agent.attachRegistrar (an interface defined in package agent with method set `Register(agent.Registrant) (agent.RegisterEntry, error)` + Unregister) using a *SessionRegistry from this package. We keep the indirection because agent/ cannot import attach/ (would create a cycle — attach/ depends on agent's *Agent via the Registrant shape).

agent.Registrant and attach.Registrant have the same method set, so any *agent.Agent already satisfies the attach.Registrant we need internally. The adapter just relabels the registry's typed return for the agent package's RegisterEntry alias.

Typical wiring in a consumer binary: 

reg := attach.NewSessionRegistry()
ag, err := agent.New(m,
    agent.WithSessionRegistry(attach.NewAgentRegistrarAdapter(reg)),
    // ...
)

func NewAgentRegistrarAdapter 

func NewAgentRegistrarAdapter(reg *SessionRegistry) *AgentRegistrarAdapter

NewAgentRegistrarAdapter wraps reg so it satisfies agent.attachRegistrar.

func (*AgentRegistrarAdapter) Register 

func (a *AgentRegistrarAdapter) Register(ag any) (any, error)

Register accepts any (called from agent/, which can't import attach/ for Registrant), type-asserts to Registrant, and forwards to the wrapped registry. Returns ErrNotRegistrant if the type assertion fails.

func (*AgentRegistrarAdapter) Unregister 

func (a *AgentRegistrarAdapter) Unregister(appName, userID, sessionID string)

Unregister forwards to the registry.

type AgentsProvider 

type AgentsProvider interface {
	AttachAgents() []AgentInfo
}

AgentsProvider is the optional capability for GET /sessions/.../agents. Returns the background subagents tracked by the registrant's BackgroundAgentManager (if any).

type AuthConfig 

type AuthConfig struct {
	// TLSCertFile / TLSKeyFile enable HTTPS on the listener. Required
	// together. Files are loaded once at server start.
	TLSCertFile string
	TLSKeyFile  string

	// ClientCAFile, when set alongside TLSCertFile/TLSKeyFile,
	// enables mTLS: clients must present a cert signed by this CA.
	// If empty, the server is HTTPS-only (server auth, no client
	// auth) — clients then rely on BearerToken (or no auth) for
	// authorization.
	ClientCAFile string

	// BearerToken, when non-empty, requires every request to carry
	// Authorization: Bearer <token>. Compared in constant time.
	// Works alongside mTLS; both must pass if both are configured.
	BearerToken string

	// ReadOnly disables every write endpoint (POST /inject, POST
	// /wake) regardless of auth. Read endpoints (GET /sessions,
	// GET /events) stay open. Useful for read-only mirrors.
	ReadOnly bool
}

AuthConfig describes how the attach server authenticates clients. Zero value (all fields empty) accepts everything — only safe over a Unix socket or other already-trusted transport. Per the design doc: mTLS is the primary auth; bearer-token is the fallback for when you don't have cert infrastructure handy.

func (AuthConfig) LoadTLSConfig 

func (a AuthConfig) LoadTLSConfig() (*tls.Config, error)

LoadTLSConfig builds a *tls.Config from the AuthConfig's TLS material. Returns nil (no TLS) when neither TLSCertFile nor TLSKeyFile is set. Returns an error if exactly one is set, or if the files can't be read.

func (AuthConfig) Middleware 

func (a AuthConfig) Middleware(next http.Handler) http.Handler

Middleware returns an http.Handler that wraps next with bearer-token validation + ReadOnly enforcement. mTLS, if configured, is enforced by the TLS handshake itself (ClientAuth = RequireAndVerifyClientCert) so by the time a request reaches this middleware, the cert has already been validated.

Returns 401 Unauthorized on missing/wrong token; 403 Forbidden when ReadOnly + the request is a write.

type Broadcaster 

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

Broadcaster owns one goroutine per session that pumps events from eventlog.Stream.Watch into N subscriber channels. Subscribers can join any time; replay-then-tail is handled via the since parameter.

One Broadcaster per Entry. Lazily created on first Subscribe and torn down when the last subscriber leaves (refcount).

func NewBroadcaster 

func NewBroadcaster(entry *Entry) (*Broadcaster, error)

NewBroadcaster constructs a broadcaster for one registered session. The pump goroutine is NOT started until the first Subscribe — we don't want background goroutines for sessions nobody's watching.

func (*Broadcaster) Close 

func (b *Broadcaster) Close()

Close cancels the pump goroutine and closes every subscriber channel. Idempotent. Called from Server.Close.

func (*Broadcaster) Subscribe 

func (b *Broadcaster) Subscribe(ctx context.Context, since int64) <-chan Frame

Subscribe adds a new client and returns its frame channel. Replays every frame with seq > since before switching to live-tail (which is invisible to the caller; same channel).

The returned channel is closed when:

  • ctx is cancelled (typical: HTTP request ends), OR
  • the subscriber falls behind subscriberBufferSize frames (the drop-the-subscriber decision; better than stalling everyone).

Caller MUST drain the channel until close to release goroutine resources.

type BroadcasterPool 

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

BroadcasterPool lazily constructs and tracks one Broadcaster per Entry. Server uses this so multiple SSE clients for the same session share one pump goroutine.

func NewBroadcasterPool 

func NewBroadcasterPool() *BroadcasterPool

NewBroadcasterPool returns an empty pool.

func (*BroadcasterPool) Close 

func (p *BroadcasterPool) Close()

Close stops every broadcaster in the pool. Used by Server.Close.

func (*BroadcasterPool) For 

func (p *BroadcasterPool) For(entry *Entry) (*Broadcaster, error)

For returns a Broadcaster for entry, constructing it on first use. Returns an error when the entry's agent has no eventlog (attach requires it).

type Entry 

type Entry struct {
	AppName   string
	UserID    string
	SessionID string

	// Agent is the live registrant. The registry holds it by
	// reference; lifetime is the registrant's, not the registry's.
	Agent Registrant
}

Entry is one registered session as the registry sees it.

type ErrNotRegistrant 

type ErrNotRegistrant struct{ Got any }

ErrNotRegistrant is returned by Register when the supplied agent doesn't satisfy attach.Registrant. Practically this can't happen when *agent.Agent is the consumer (it implements all the methods); the check is defensive against future agent constructor refactors.

func (*ErrNotRegistrant) Error 

func (e *ErrNotRegistrant) Error() string

type Frame 

type Frame struct {
	Seq   int64          `json:"seq"`
	Event *session.Event `json:"event"`
}

Frame is one item the SSE stream emits. Carries the eventlog seq (so reconnecting clients can resume via ?since=N) plus the underlying ADK event. Marshaled to JSON before going out the wire.

type Options 

type Options struct {
	// Registry is the SessionRegistry the server consults to resolve
	// URL session IDs to live agents. Required.
	Registry *SessionRegistry

	// PeerRegistry, when non-nil, enables peer-registration endpoints
	// (POST /peers, GET /peers, etc.) — turning this listener into a
	// discovery hub for other agents. Nil means the peer endpoints
	// are not registered and POST /peers returns 404. See
	// docs/peer-registration-design.md.
	PeerRegistry *PeerRegistry

	// Auth controls TLS + bearer + read-only enforcement. Zero value
	// accepts everything (safe only over a Unix socket or other
	// already-trusted transport).
	Auth AuthConfig

	// Addr is the TCP listen address (e.g. ":7777"). Mutually
	// exclusive with UnixSocket — set exactly one.
	Addr string

	// UnixSocket is the Unix domain socket path (e.g.
	// "/var/run/core-agent.sock"). Mutually exclusive with Addr.
	// When set, Auth.TLS* and Auth.BearerToken are usually omitted
	// — filesystem permissions on the socket file are the auth.
	UnixSocket string

	// ShutdownTimeout caps how long Server.Close waits for in-flight
	// SSE clients to drain. Default 5 seconds.
	ShutdownTimeout time.Duration
}

Options configures NewServer. Zero value is invalid — Registry is required at minimum.

type Peer 

type Peer struct {
	RegistrationID string            `json:"registration_id"`
	Name           string            `json:"name"`
	Endpoint       string            `json:"endpoint"`
	Labels         map[string]string `json:"labels,omitempty"`
	RegisteredAt   time.Time         `json:"registered_at"`
	LastHeartbeat  time.Time         `json:"last_heartbeat"`
	LeaseExpiresAt time.Time         `json:"lease_expires_at"`
}

Peer is one entry in the hub's PeerRegistry. Carries the registration identity, the peer's reachable endpoint, opaque labels for filtering, and the lease state for liveness tracking.

type PeerClient 

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

PeerClient is a thin wrapper around http.Client for talking to a hub agent's peer-registration endpoints. Used by peer binaries (typically wired via cmd/core-agent's --attach-register-to flag).

Single-use lifecycle: construct, RegisterAndHeartbeat, defer Deregister. The goroutine spawned by RegisterAndHeartbeat handles the heartbeat cadence; the caller doesn't have to track it.

func NewPeerClient 

func NewPeerClient(hubURL string, opts ...PeerClientOption) *PeerClient

NewPeerClient builds a client targeting hubURL (e.g. "https://hub.default.svc:7777").

func (*PeerClient) Deregister 

func (c *PeerClient) Deregister(ctx context.Context, registrationID string) error

Deregister removes the registration from the hub. Idempotent on the server side; this client also treats network errors as best-effort (graceful shutdown shouldn't fail because the hub is momentarily unreachable).

func (*PeerClient) Heartbeat 

func (c *PeerClient) Heartbeat(ctx context.Context, registrationID string) (*Peer, error)

Heartbeat extends the lease on the named registration.

func (*PeerClient) Register 

func (c *PeerClient) Register(ctx context.Context, req RegisterRequest) (*Peer, error)

Register POSTs the registration request to the hub. Returns the assigned RegistrationID + lease expiry. Doesn't start a heartbeat goroutine — use RegisterAndHeartbeat for the standard lifecycle.

func (*PeerClient) RegisterAndHeartbeat 

func (c *PeerClient) RegisterAndHeartbeat(ctx context.Context, req RegisterRequest) (cancel func(), err error)

RegisterAndHeartbeat registers the peer and starts a background heartbeat goroutine that runs until ctx is cancelled. Returns a cancel function that triggers Deregister + goroutine shutdown.

Heartbeat cadence is the registered TTL divided by c.heartbeatFraction (default 1/3 of TTL). If the registered TTL is 0 (peer wanted default), the hub-supplied lease expiry is used to derive cadence.

Heartbeat failures are logged but don't fatal — a peer that loses its registration via expired lease just re-registers on the next attempt (the hub's name-based upsert makes this clean).

type PeerClientOption 

type PeerClientOption func(*PeerClient)

PeerClientOption configures NewPeerClient.

func WithPeerBearerToken 

func WithPeerBearerToken(token string) PeerClientOption

WithPeerBearerToken sets the Authorization: Bearer header used on every request to the hub. Required when the hub has Auth.BearerToken set.

func WithPeerHTTPClient 

func WithPeerHTTPClient(client *http.Client) PeerClientOption

WithPeerHTTPClient overrides the default http.Client. Useful for mTLS (caller supplies a Transport with a configured TLSConfig) or for tests with a custom round-tripper.

type PeerRegistry 

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

PeerRegistry is the hub-side state. Independent from SessionRegistry — sessions and peers are orthogonal: a peer's endpoint may itself host its own sessions.

func NewPeerRegistry 

func NewPeerRegistry(opts ...PeerRegistryOption) *PeerRegistry

NewPeerRegistry returns an empty hub registry plus a started prune goroutine that drops expired leases every pruneTick. Call Close to stop the goroutine.

func (*PeerRegistry) Close 

func (r *PeerRegistry) Close() error

Close stops the prune goroutine. Idempotent.

func (*PeerRegistry) Deregister 

func (r *PeerRegistry) Deregister(id string)

Deregister removes the peer by ID. No-op on unknown id — keeps graceful shutdown paths idempotent.

func (*PeerRegistry) Heartbeat 

func (r *PeerRegistry) Heartbeat(id string) (*Peer, error)

Heartbeat extends the lease on the named registration. Returns the new lease expiry. ErrPeerNotFound when id is unknown (peer should re-Register on this error).

func (*PeerRegistry) Len 

func (r *PeerRegistry) Len() int

Len returns the count of live peers (post-prune).

func (*PeerRegistry) List 

func (r *PeerRegistry) List(labelMatch map[string]string) []*Peer

List returns a sorted snapshot of every live peer. labelMatch, if non-empty, filters to peers whose Labels contain every k=v in the match map. Returns a defensive copy of each Peer so callers can't mutate registry state.

func (*PeerRegistry) Prune 

func (r *PeerRegistry) Prune() int

Prune drops every peer whose lease has expired. Returns the count pruned. Called from the background goroutine on a tick; exposed for tests + manual triggering.

func (*PeerRegistry) Register 

func (r *PeerRegistry) Register(req RegisterRequest) (*Peer, error)

Register adds (or upserts on Name match) a peer. Returns the assigned RegistrationID + lease expiry. Name-based upsert avoids orphaned entries when a peer restarts.

type PeerRegistryOption 

type PeerRegistryOption func(*PeerRegistry)

PeerRegistryOption configures NewPeerRegistry.

func WithMaxTTL 

func WithMaxTTL(d time.Duration) PeerRegistryOption

WithMaxTTL caps how long a peer-requested heartbeat TTL can run. Defaults to 5 minutes. Peers asking for longer get clamped.

type RegisterRequest 

type RegisterRequest struct {
	Name            string            `json:"name"`
	Endpoint        string            `json:"endpoint"`
	Labels          map[string]string `json:"labels,omitempty"`
	HeartbeatTTLSec int               `json:"heartbeat_ttl_sec,omitempty"`
}

RegisterRequest is the body the peer POSTs to /peers. Validated inside PeerRegistry.Register; bad values surface as errors the handler maps to 400.

type Registrant 

type Registrant interface {
	// AppName / UserID / SessionID together form the ADK session
	// key. The registry uses (AppName, SessionID) for URL lookup
	// (userID is per-process today; see attach-mode-design.md) but
	// stores all three so /sessions can expose full triples.
	AppName() string
	UserID() string
	SessionID() string

	// EventLog returns the agent's event log handle, or nil if the
	// agent was constructed without WithEventLog. Attach requires a
	// non-nil eventlog for live-tail (broadcaster pumps from
	// Stream.Watch); the server returns a clear error for sessions
	// without one.
	EventLog() *eventlog.Handle

	// Inject queues an inbox message that the next turn drains.
	// Also fires the wake signal internally (see Agent.Inject).
	Inject(message string) error

	// RequestWake fires the agent's wake signal without queuing a
	// message. Used by POST /wake.
	RequestWake()
}

Registrant is the subset of *agent.Agent the registry needs. Defined as an interface so attach/ doesn't import agent/ (avoids an import cycle — agent/ needs to call into attach to register itself).

type Server 

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

Server hosts the attach-mode HTTP endpoints. Construct via NewServer; start via ListenAndServe; stop via Close.

func NewServer 

func NewServer(opts Options) (*Server, error)

NewServer builds a Server. Validates Options; returns an error for invalid combinations (both/neither of Addr/UnixSocket; missing registry; TLS material mismatch).

func (*Server) Addr 

func (s *Server) Addr() string

Addr returns the actual listener address the server bound to. Useful for tests that use Addr=":0" to get an ephemeral port. Returns empty before ListenAndServe runs.

func (*Server) Close 

func (s *Server) Close() error

Close stops the server, waits up to Options.ShutdownTimeout for in-flight SSE clients to disconnect, then tears down the broadcaster pool. Idempotent.

func (*Server) ListenAndServe 

func (s *Server) ListenAndServe() error

ListenAndServe binds the listener and serves until Close or a fatal error. Blocks until the server stops. Returns nil on clean shutdown, the underlying error otherwise.

func (*Server) TLSEnabled 

func (s *Server) TLSEnabled() bool

TLSEnabled reports whether the server is running with TLS (HTTPS endpoints) or plaintext (HTTP). Helpful for log lines / debugging.

type SessionRegistry 

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

SessionRegistry holds every Agent that opted into attach-mode by calling agent.WithSessionRegistry at construction. Keys by the full (AppName, UserID, SessionID) triple but exposes a single-segment SessionID shortcut for the unambiguous case.

func NewSessionRegistry 

func NewSessionRegistry() *SessionRegistry

NewSessionRegistry returns an empty registry.

func (*SessionRegistry) Len 

func (r *SessionRegistry) Len() int

Len returns the number of registered sessions. Useful for tests + the listener's startup log.

func (*SessionRegistry) List 

func (r *SessionRegistry) List() []*Entry

List returns a snapshot of every registered entry, sorted by (AppName, UserID, SessionID) for stable output. Used by GET /sessions so the operator sees a deterministic ordering across requests.

func (*SessionRegistry) Lookup 

func (r *SessionRegistry) Lookup(appName, sessionID string) (*Entry, error)

Lookup returns the entry for the qualified (appName, sessionID) form. userID is not required for lookup — the registry searches across all registered userIDs for the (app, sid) pair. Returns ErrSessionNotFound when there's no match. The full-triple form is what's stored internally; this just searches.

func (*SessionRegistry) LookupSingle 

func (r *SessionRegistry) LookupSingle(sessionID string) (*Entry, error)

LookupSingle resolves the /sessions/<sessionID> shortcut. Returns ErrAmbiguousSession if the SessionID is registered against multiple apps — the caller should then use the qualified form and surface the helpful error to the client.

func (*SessionRegistry) Register 

func (r *SessionRegistry) Register(ag Registrant) (*Entry, error)

Register adds a session. Returns ErrSessionExists if the triple is already present — the caller should not silently overwrite, since a double-register usually means an Agent construction race.

func (*SessionRegistry) Unregister 

func (r *SessionRegistry) Unregister(appName, userID, sessionID string)

Unregister removes a session by its full triple. No-op (returns nil) when the entry doesn't exist — keeps shutdown paths idempotent.

type StatusInfo 

type StatusInfo struct {
	State       string    `json:"state"` // running | deferred | paused | idle
	ModelName   string    `json:"model_name,omitempty"`
	NextWakeAt  time.Time `json:"next_wake_at,omitempty"` // populated when State=deferred
	CurrentTool string    `json:"current_tool,omitempty"` // populated when State=running and a tool is in flight
}

StatusInfo is the response shape of GET /sessions/.../status. ModelName is what the TUI's usage panel labels with; the rest is agent-loop introspection useful for the /status slash command.

type StatusProvider 

type StatusProvider interface {
	AttachStatus() StatusInfo
}

StatusProvider is the optional capability for GET /sessions/.../status. Returns the agent's current run-state + model identity.

type ToolInfo 

type ToolInfo struct {
	Name        string `json:"name"`
	Description string `json:"description,omitempty"`
	Source      string `json:"source"`           // builtin | mcp | skill | other
	Server      string `json:"server,omitempty"` // MCP server attribution when Source=mcp
	GateState   string `json:"gate_state,omitempty"`
}

ToolInfo is one entry in the GET /sessions/.../tools response.

GateState carries the pre-flight projection from permissions.Gate.ToolGateState — empty when no gate is wired (library callers with no permission policy). The TUI v1 fetches the field but doesn't surface it; v1.1 adds the column in the /tools modal.

type ToolsProvider 

type ToolsProvider interface {
	AttachTools() []ToolInfo
}

ToolsProvider is the optional capability a Registrant can implement to surface its tool catalog over GET /sessions/.../tools. The handler type-asserts at request time; absence reports an empty list rather than 501 so old Registrant impls keep working.

Method is named AttachTools (not Tools) to avoid colliding with *agent.Agent.Tools() which already returns []tool.Tool. Agents that implement the attach surface define a distinct method with this shape that does the conversion internally.

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.