server

package
v0.1.0 Latest Latest
Warning

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

 Go to latest Published: Aug 9, 2026 License: Apache-2.0 Imports: 28 Imported by: 0

Documentation

Index

Constants

View Source 
const (
	HeaderAcpConnectionID = "Acp-Connection-Id"
	HeaderAcpSessionID    = "Acp-Session-Id"

	// CookieAcpAffinity sticks a client to the same backend for one connection.
	CookieAcpAffinity = "acp_affinity"
)

Header names for the ACP Streamable HTTP / WebSocket transport (RFD).

Variables

View Source 
var (
	ErrInvalidRequest            = errors.New("invalid request")
	ErrMethodNotFound            = errors.New("method not found")
	ErrAgentNotFound             = errors.New("agent not found")
	ErrSessionNotFound           = errors.New("session not found")
	ErrSessionStoreNotConfigured = errors.New("session store is not configured")
	ErrStreamingNotSupported     = errors.New("streaming not supported")
	ErrInternal                  = errors.New("internal server error")
)

Sentinel errors for classification via errors.Is / errors.As.

View Source 
var ErrRequestCancelled = errors.New("request cancelled")

ErrRequestCancelled is returned when a request is aborted via session/cancel or context cancellation.

View Source 
var ErrWireSessionUnsupported = errors.New("wire sessions not supported by this protocol")

ErrWireSessionUnsupported is returned by Protocol session methods when the protocol has no wire-session lifecycle (e.g. simple SSE).

Functions

func ElicitationResultToSelectionPayload 

func ElicitationResultToSelectionPayload(raw json.RawMessage, opts []interrupt.UserChoice) (action string, resolution []byte, err error)

ElicitationResultToSelectionPayload maps an accept response to the harness interrupt resolution payload. Returns action and optional selection JSON.

func IsClientError 

func IsClientError(err error) bool

IsClientError reports whether err is safe to surface to the client (validation, not-found, bad payload). Internal failures return false.

func JSONRPCErrorCode 

func JSONRPCErrorCode(err error) int

JSONRPCErrorCode maps err to a JSON-RPC 2.0 error code.

func ParseToolPermissionFromInterruptData 

func ParseToolPermissionFromInterruptData(data []byte) (interruptID string, perm interrupt.ToolPermissionInterrupt, err error)

ParseToolPermissionFromInterruptData extracts a tool permission interrupt from yield data.

func ParseUserSelectionFromInterruptData 

func ParseUserSelectionFromInterruptData(data []byte) (interruptID string, opts []interrupt.UserChoice, err error)

ParseUserSelectionFromInterruptData extracts options from StreamEventInterrupt Data payload shape {"interruptId":"...","data":<serialized UserSelectionInterrupt>}.

func PermissionToACPParams 

func PermissionToACPParams(sessionID, toolCallID string, perm interrupt.ToolPermissionInterrupt) map[string]any

PermissionToACPParams builds session/request_permission params.

func PublicError 

func PublicError(err error) error

PublicError returns a wire-safe error: client errors pass through unchanged; all other errors become ErrInternal so internal details are not leaked.

func RequestPermissionResultToPayload 

func RequestPermissionResultToPayload(raw json.RawMessage) (resolution []byte, cancelled bool, err error)

RequestPermissionResultToPayload maps a client permission response to the harness resolution payload. cancelled yields a non-nil err suitable for ending the turn.

func SelectionToElicitationParams 

func SelectionToElicitationParams(sessionID, toolCallID, question string, opts []interrupt.UserChoice) (map[string]any, error)

SelectionToElicitationParams builds form-mode elicitation/create params from a user-selection interrupt.

Types

type AgentSpec 

type AgentSpec struct {
	Name       string
	Config     tacklr.Config
	Model      tacklr.InferenceStrategy
	Tools      []*tacklr.Tool
	MCPConfigs []mcp.MCPConfig
	SubAgents  []*tacklr.SubAgent
	WatchDog   tacklr.AgentWatchDog
	Store      stores.BaseStore
	// ExaAPIKey enables built-in web_search and web_fetch (or use process EXA_API_KEY).
	ExaAPIKey string
}

type ClientBridge 

type ClientBridge struct {

	// Caps is protected by mu; use GetCaps/SetCaps from concurrent stdio handlers.
	Caps ClientCapabilities
	// contains filtered or unexported fields
}

ClientBridge sends JSON-RPC requests to the Client and demuxes responses by id. Safe for concurrent Call from tool/turn goroutines; one bridge per connection.

func NewClientBridge 

func NewClientBridge(w MessageWriter) *ClientBridge

NewClientBridge creates a bridge that writes requests through w.

func (*ClientBridge) Call 

func (b *ClientBridge) Call(ctx context.Context, method string, params any) (json.RawMessage, error)

Call sends a JSON-RPC request and waits for the matching response or ctx cancel.

func (*ClientBridge) GetCaps 

func (b *ClientBridge) GetCaps() ClientCapabilities

GetCaps returns a snapshot of client capabilities (safe for concurrent use).

func (*ClientBridge) SetCaps 

func (b *ClientBridge) SetCaps(c ClientCapabilities)

SetCaps stores client capabilities (safe for concurrent use).

func (*ClientBridge) TryCompleteResponse 

func (b *ClientBridge) TryCompleteResponse(body []byte) bool

TryCompleteResponse returns true if body is a JSON-RPC response that completed a waiter.

type ClientCapabilities 

type ClientCapabilities struct {
	ElicitationForm bool
	ElicitationURL  bool
}

ClientCapabilities captures client features from initialize.

func ParseClientCapabilities 

func ParseClientCapabilities(params json.RawMessage) ClientCapabilities

ParseClientCapabilities extracts elicitation mode support from initialize params.

type ConfigOption 

type ConfigOption struct {
	ID           string              `json:"id"`
	Name         string              `json:"name"`
	Description  string              `json:"description,omitempty"`
	Category     string              `json:"category"`
	Type         string              `json:"type"`
	CurrentValue string              `json:"currentValue"`
	Options      []ConfigOptionValue `json:"options"`
}

ConfigOption describes a selectable session configuration option returned by session/new and session/load.

type ConfigOptionValue 

type ConfigOptionValue struct {
	Value       string `json:"value"`
	Name        string `json:"name"`
	Description string `json:"description,omitempty"`
}

ConfigOptionValue is one choice within a select-type ConfigOption.

type Conn 

type Conn struct {
	Writer MessageWriter
	RPC    *ClientBridge
	Caps   ClientCapabilities
}

Conn is one client connection (stdio session, or a logical HTTP request scope).

type Connection 

type Connection struct {
	ID     string
	Bridge *ClientBridge
	Writer MessageWriter
	// contains filtered or unexported fields
}

Connection is one client transport connection (WebSocket or Streamable HTTP). Harness sessions live in Registry; this is ephemeral wire state only.

func (*Connection) Context 

func (c *Connection) Context() context.Context

Context is cancelled when the connection is removed or shut down.

type ConnectionRegistry 

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

ConnectionRegistry tracks active ACP connections by Acp-Connection-Id.

func NewConnectionRegistry 

func NewConnectionRegistry() *ConnectionRegistry

NewConnectionRegistry returns an empty registry.

func (*ConnectionRegistry) Create 

func (r *ConnectionRegistry) Create(bridge *ClientBridge, writer MessageWriter) *Connection

Create registers a new connection with a generated id and cancellable context. bridge/writer may be filled in after Create (WebSocket accept path).

func (*ConnectionRegistry) Get 

func (r *ConnectionRegistry) Get(id string) *Connection

Get returns the connection for id, or nil.

func (*ConnectionRegistry) Remove 

func (r *ConnectionRegistry) Remove(id string)

Remove deletes the connection and cancels its context. Safe if missing or r is nil.

type ElicitationResult 

type ElicitationResult struct {
	Action  string         `json:"action"`
	Content map[string]any `json:"content"`
}

ElicitationResult is the Client response to elicitation/create.

type EventStream 

type EventStream struct {
	Events  <-chan streaming.StreamEvent
	Harness *tacklr.AgentHarness
	// contains filtered or unexported fields
}

EventStream is a running agent turn. Events is closed when the current harness run finishes (complete, error, interrupt park) or the turn context is cancelled and the registry forwarder exits.

Cancel cancels the turn context (session/cancel). Close releases harness resources after the turn has ended. Callers typically: 

defer func() { stream.Cancel(); stream.Close() }()

Harness remains usable after an interrupt park so ResumeInterrupts can run before Close.

func (*EventStream) Cancel 

func (s *EventStream) Cancel()

Cancel cancels the turn context so producers stop. Safe to call multiple times. Does not release the harness; call Close after the event pump finishes.

func (*EventStream) Cancelled 

func (s *EventStream) Cancelled() bool

Cancelled reports whether the turn context has been cancelled.

func (*EventStream) Close 

func (s *EventStream) Close()

Close releases harness resources (idempotent). Call after Cancel or stream end.

func (*EventStream) ResumeInterrupts 

func (s *EventStream) ResumeInterrupts(ctx context.Context, responses map[string][]byte) (<-chan streaming.StreamEvent, error)

ResumeInterrupts resolves pending interrupts and returns a new event stream from the same harness (ACP mid-turn elicitation resume).

func (*EventStream) TurnContext 

func (s *EventStream) TurnContext() context.Context

TurnContext is the context for this turn (cancelled by session/cancel or parent).

type HTTPRoute 

type HTTPRoute struct {
	Method  string // e.g. "POST"
	Pattern string // e.g. "/" or "/resume"
	Handler func(env ProtocolEnv, w http.ResponseWriter, r *http.Request)
}

HTTPRoute is one HTTP endpoint owned by a protocol.

type InterruptEventEnvelope 

type InterruptEventEnvelope struct {
	InterruptId string          `json:"interruptId"`
	Type        string          `json:"type"`
	Data        json.RawMessage `json:"data"`
}

InterruptEventEnvelope is the harness StreamEventInterrupt Data shape.

func ParseInterruptEnvelope 

func ParseInterruptEnvelope(data []byte) (InterruptEventEnvelope, error)

ParseInterruptEnvelope extracts interrupt id, type, and raw data from a yield event.

type MemoryWireStore 

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

MemoryWireStore is an in-process ProtocolWireStore.

func NewMemoryWireStore 

func NewMemoryWireStore() *MemoryWireStore

NewMemoryWireStore returns an empty in-memory wire store.

func (*MemoryWireStore) Delete 

func (s *MemoryWireStore) Delete(_ context.Context, sessionID string) error

func (*MemoryWireStore) Get 

func (s *MemoryWireStore) Get(_ context.Context, sessionID string) ([]byte, error)

func (*MemoryWireStore) Put 

func (s *MemoryWireStore) Put(_ context.Context, sessionID string, payload []byte) error

type MessageWriter 

type MessageWriter interface {
	WriteResult(id json.RawMessage, result any) error
	// WriteError writes a failure. Implementations should use PublicError /
	// JSONRPCErrorCode so internal details are not leaked on the wire.
	WriteError(id json.RawMessage, err error) error
	WriteFrame(data []byte) error
}

MessageWriter is the transport-agnostic sink for protocol responses and streamed frames. Transports adapt HTTP, stdio, and WebSocket to this interface.

type PostgresWireStore 

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

PostgresWireStore implements ProtocolWireStore against Postgres. Table: public.protocol_wire_session (see stores/testdata/session_schema.sql). Shares a *pgx.Conn with stores.PostgresStore when desired; schema is separate from harness session checkpoints.

func NewPostgresWireStore 

func NewPostgresWireStore(conn *pgx.Conn, protocolKey string) *PostgresWireStore

NewPostgresWireStore wraps an existing pgx connection. protocolKey labels rows (e.g. "acp"); empty defaults to "acp".

func (*PostgresWireStore) Delete 

func (s *PostgresWireStore) Delete(ctx context.Context, sessionID string) error

func (*PostgresWireStore) Get 

func (s *PostgresWireStore) Get(ctx context.Context, sessionID string) ([]byte, error)

func (*PostgresWireStore) Put 

func (s *PostgresWireStore) Put(ctx context.Context, sessionID string, payload []byte) error

type Protocol 

type Protocol interface {
	Name() string

	// HandleInbound processes one inbound message on a connection-oriented
	// transport (stdio NDJSON). Pure-HTTP protocols may return nil without work.
	HandleInbound(ctx context.Context, env ProtocolEnv, body []byte) error

	// HTTPRoutes returns routes to mount for ServeHTTP. Nil/empty is fine.
	HTTPRoutes() []HTTPRoute

	// OnStreamEvent encodes one harness StreamEvent for the client connection.
	OnStreamEvent(ctx context.Context, env ProtocolEnv, threadID string, stream *EventStream, ev streaming.StreamEvent, reqID json.RawMessage) StreamControl

	// OnStreamClosed is called when the event channel closes without Finished.
	OnStreamClosed(ctx context.Context, env ProtocolEnv, threadID string, reqID json.RawMessage, cancelled bool) error

	// CreateSession allocates a wire session. params is protocol-defined JSON.
	// Stateless protocols return ErrWireSessionUnsupported.
	CreateSession(ctx context.Context, env ProtocolEnv, params json.RawMessage) (sessionID string, result any, err error)

	// LoadSession reattaches a wire session (memory or durable wire store).
	LoadSession(ctx context.Context, env ProtocolEnv, sessionID string, params json.RawMessage) (result any, err error)

	// BindTurn maps a wire session + turn body into a Registry TurnRequest.
	BindTurn(ctx context.Context, env ProtocolEnv, sessionID string, turnParams json.RawMessage) (TurnRequest, error)

	// CloseSession drops live wire binding and may cancel an in-flight turn.
	CloseSession(ctx context.Context, env ProtocolEnv, sessionID string) error
}

Protocol is a complete wire façade over Registry. Transports only provide Conn I/O; protocols own methods, routes, stream policy, and wire-session lifecycle (create/load/bind/close).

Multi-protocol model (ACP, SSE, future A2A):

  • Registry.RunTurn produces protocol-agnostic streaming.StreamEvent values.
  • runTurnStream pumps those events through OnStreamEvent / OnStreamClosed.
  • Each Protocol maps StreamEvent → its client wire and owns any wire session state.

Adding a protocol should not require harness streaming changes. 

var (
	ACP Protocol = NewACPProtocol(NewMemoryWireStore())
	SSE Protocol = SSEProtocol()
)

Built-in protocol aliases.

ACP is a process-scoped default for simple apps (NewServer(reg, server.ACP)). Prefer NewACPProtocol(wire) when you need durable/shared wire state or test isolation. Tests that share a *Registry should use protocolForRegistry (via serveACPRaw) or acpTestServer — not this package-level value for multi-step session flows.

func ACPProtocol 

func ACPProtocol() Protocol

ACPProtocol returns a new ACP protocol with an in-memory wire store. Each call is a fresh instance (own live map + wire store).

func NewACPProtocol 

func NewACPProtocol(wire ProtocolWireStore) Protocol

NewACPProtocol returns an ACP protocol with optional durable wire store. Nil wire uses an in-memory ProtocolWireStore.

func NewACPProtocolMemory 

func NewACPProtocolMemory() Protocol

NewACPProtocolMemory is shorthand for NewACPProtocol(NewMemoryWireStore()). Use when mounting ACP alongside other protocols: 

server.NewServer(reg, server.NewACPProtocolMemory(), server.SSE)

func NewACPProtocolPostgres 

func NewACPProtocolPostgres(conn *pgx.Conn) Protocol

NewACPProtocolPostgres is shorthand for ACP with a Postgres wire store on conn. protocolKey is "acp". Requires table public.protocol_wire_session (see stores/testdata/session_schema.sql). 

server.NewServer(reg, server.NewACPProtocolPostgres(conn))

func SSEProtocol 

func SSEProtocol() Protocol

SSEProtocol returns the SSE/WS wire protocol module.

type ProtocolEnv 

type ProtocolEnv struct {
	Registry *Registry
	Conn     *Conn
	// Connections is the server-wide connection registry (WebSocket / Streamable HTTP).
	// Nil for pure stdio or tests that only use HandleInbound.
	Connections *ConnectionRegistry
}

ProtocolEnv is the domain + connection context passed into protocol handlers.

type ProtocolWireStore 

type ProtocolWireStore interface {
	Put(ctx context.Context, sessionID string, payload []byte) error
	Get(ctx context.Context, sessionID string) ([]byte, error)
	Delete(ctx context.Context, sessionID string) error
}

ProtocolWireStore persists protocol-owned session envelopes (not harness checkpoints). Payload is opaque JSON defined by each protocol. May share a database connection with BaseStore without sharing schema.

type Registry 

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

Registry serves agents over wire protocols (ACP, SSE, and others).

func NewRegistry 

func NewRegistry(store stores.BaseStore, defaultAgent string, opts ...RegistryOption) *Registry

NewRegistry builds a registry. opts may set telemetry providers.

func (*Registry) AgentModel 

func (r *Registry) AgentModel(agentID string) tacklr.InferenceStrategy

AgentModel returns the inference strategy for a registered agent, or nil.

func (*Registry) CancelSession 

func (r *Registry) CancelSession(sessionID string)

CancelSession cancels the in-flight turn context for the session (if any). Session state is preserved. The turn context is the single cancel signal for harness work, the registry forwarder, and runTurnStream. This is abort-only (ACP session/cancel). It does not start a new turn.

func (*Registry) ConfigOptions 

func (r *Registry) ConfigOptions(currentAgent string) []ConfigOption

ConfigOptions returns selectable agent config options for wire session responses.

func (*Registry) DefaultAgent 

func (r *Registry) DefaultAgent() string

DefaultAgent returns the registry default agent id.

func (*Registry) DropLiveHarness 

func (r *Registry) DropLiveHarness(sessionID string)

DropLiveHarness removes a cached harness (e.g. session/close). Next prompt reloads from store.

func (*Registry) HasAgent 

func (r *Registry) HasAgent(agentID string) bool

HasAgent reports whether agentID is registered.

func (*Registry) RecordSessionCreated 

func (r *Registry) RecordSessionCreated(ctx context.Context)

RecordSessionCreated records a session-created metric (called by protocols).

func (*Registry) Register 

func (r *Registry) Register(agentID string, spec AgentSpec)

func (*Registry) RunTurn 

func (r *Registry) RunTurn(ctx context.Context, req TurnRequest) (*EventStream, error)

RunTurn starts a prompt or resume turn and returns a stream of events. Setup errors (unknown session/agent, validation) are returned synchronously. Runtime errors are delivered as StreamEventError on the channel.

If a turn is already in flight for the session, it is cancelled and allowed to finalize before this turn starts (mid-turn steer via session/prompt).

type RegistryOption 

type RegistryOption func(*Registry)

RegistryOption configures NewRegistry.

func WithMeterProvider 

func WithMeterProvider(mp metric.MeterProvider) RegistryOption

WithMeterProvider sets the MeterProvider for turn and tool metrics. Nil uses the process global meter.

func WithTracer 

func WithTracer(t trace.Tracer) RegistryOption

WithTracer sets an explicit Tracer for turn telemetry. WithTracerProvider is the usual choice for consistent instrumentation names.

func WithTracerProvider 

func WithTracerProvider(tp trace.TracerProvider) RegistryOption

WithTracerProvider sets the TracerProvider for turn telemetry (tracer name telemetry.InstrumentationName). Nil uses the process global.

type RequestPermissionResult 

type RequestPermissionResult struct {
	Outcome struct {
		Outcome  string `json:"outcome"`
		OptionID string `json:"optionId,omitempty"`
	} `json:"outcome"`
}

RequestPermissionResult is the Client response to session/request_permission.

type Server 

type Server struct {
	Registry  *Registry
	Protocols []Protocol
	// Client is set for the active stdio connection (outbound Agent→Client RPC).
	// Prefer Conn.RPC inside protocol handlers; this field supports demux on stdio.
	Client *ClientBridge
	// Connections tracks ACP WebSocket (and future Streamable HTTP) connections.
	Connections *ConnectionRegistry
}

Server serves a Registry over one or more wire Protocols.

func NewACPServer 

func NewACPServer(reg *Registry) *Server

NewACPServer returns a Server with ACP and an in-memory wire session store. Suitable for demos, single-process apps, and tests that do not need durable session/load across process restarts. 

reg := server.NewRegistry(stores.NewInMemoryStore(), "agent")
reg.Register("agent", server.AgentSpec{...})
srv := server.NewACPServer(reg)
_ = srv.ServeStdio(ctx, os.Stdin, os.Stdout)
// or: _ = srv.ServeHTTP(ctx, ":8080")  // /acp WebSocket + Streamable HTTP

func NewACPServerPostgres 

func NewACPServerPostgres(reg *Registry, conn *pgx.Conn) *Server

NewACPServerPostgres returns a Server with ACP backed by a Postgres wire store. The registry should already use a harness store (e.g. stores.NewPostgresStore(conn) or InMemoryStore). Wire and harness schemas are separate; sharing *pgx.Conn is fine. 

harness := stores.NewPostgresStore(conn)
reg := server.NewRegistry(harness, "agent")
// reg.Register(...)
srv := server.NewACPServerPostgres(reg, conn)

func NewACPServerWithWire 

func NewACPServerWithWire(reg *Registry, wire ProtocolWireStore) *Server

NewACPServerWithWire returns a Server with ACP using the given wire store for protocol session envelopes (session/new, session/load). The registry's BaseStore remains the harness checkpoint store (agent conversation state). 

wire := server.NewPostgresWireStore(conn, "acp")
srv := server.NewACPServerWithWire(reg, wire)

Or any custom ProtocolWireStore (Redis, SQLite, …).

func NewServer 

func NewServer(r *Registry, protocols ...Protocol) *Server

NewServer wraps a Registry and one or more protocols. The first protocol is used for connection-oriented transports (stdio).

func (*Server) HTTPMux 

func (s *Server) HTTPMux() *http.ServeMux

HTTPMux mounts all protocol HTTP routes. Used by ServeHTTP and tests.

func (*Server) HandleMessage 

func (s *Server) HandleMessage(ctx context.Context, body []byte, w MessageWriter)

HandleMessage dispatches one inbound body on the primary protocol. Used by tests and unary HTTP adapters.

func (*Server) ServeHTTP 

func (s *Server) ServeHTTP(ctx context.Context, addr string) error

ServeHTTP starts an HTTP server mounting all protocol routes.

func (*Server) ServeStdio 

func (s *Server) ServeStdio(ctx context.Context, in io.Reader, out io.Writer) error

ServeStdio serves line-delimited JSON messages over in/out.

type SessionView 

type SessionView struct {
	SessionID     string
	ConfigOptions []ConfigOption
}

SessionView is the domain view of a wire session (returned by protocols).

type StreamControl 

type StreamControl struct {
	Frames        [][]byte
	ReplaceEvents <-chan streaming.StreamEvent
	Finished      bool
	Err           error
}

StreamControl is the protocol's decision after observing one harness event.

type TurnRequest 

type TurnRequest struct {
	SessionID string
	AgentID   string
	ThreadID  string
	Prompt    string
	// UserMessage is multimodal user content (ACP). When set, preferred over Prompt.
	UserMessage *tacklr.Message
	Responses   map[string]json.RawMessage
	Load        bool

	// AllowMissingCheckpoint: when Load is true and the harness store has no
	// row, start a fresh agent instead of failing. Set by wire BindTurn for
	// sessions that may never have been checkpointed yet.
	AllowMissingCheckpoint bool

	// CWD is optional turn context (protocol may set from wire session).
	CWD string

	// MCPServers are session-scoped MCP configs for this turn.
	MCPServers []mcp.MCPConfig
}

TurnRequest describes a prompt or resume turn.

Session mode (ACP): protocol BindTurn fills SessionID, AgentID, MCPServers, Load, and AllowMissingCheckpoint. Registry does not own wire envelopes.

Direct mode (SSE): set AgentID and ThreadID. Load restores from harness store.

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.