supervisor

package
v1.4.7 Latest Latest
Warning

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

 Go to latest Published: Aug 17, 2026 License: MIT Imports: 16 Imported by: 0

Documentation

Overview

Package supervisor provides Erlang-style process supervision using suture. It manages MCP server processes with automatic restart and graceful termination.

Index

Constants

This section is empty.

Variables

View Source 
var ErrBackendUnavailable = errors.New("managed HTTP backend unavailable")
View Source 
var ErrDrainContextHasDeadline = errors.New("drain context must not carry a deadline")

Functions

This section is empty.

Types

type BackendCoordinator added in v1.3.3 

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

BackendCoordinator serializes backend lifecycle changes with application requests. Cleanup-triggered recycle drains existing requests without cancelling them and rejects new dispatch while draining.

func NewBackendCoordinator added in v1.3.3 

func NewBackendCoordinator() *BackendCoordinator

func (*BackendCoordinator) BeginRequest added in v1.3.3 

func (c *BackendCoordinator) BeginRequest() (func(), error)

func (*BackendCoordinator) DrainAndRecycle added in v1.3.3 

func (c *BackendCoordinator) DrainAndRecycle(ctx context.Context, recycle func() error) error

DrainAndRecycle blocks new dispatch, waits for every admitted application request to finish, then invokes recycle exactly once. Successful recycle transitions to probing; callers must complete readiness before MarkReady.

func (*BackendCoordinator) InFlight added in v1.3.3 

func (c *BackendCoordinator) InFlight() int

func (*BackendCoordinator) MarkProbing added in v1.3.3 

func (c *BackendCoordinator) MarkProbing()

func (*BackendCoordinator) MarkProcessLost added in v1.3.3 

func (c *BackendCoordinator) MarkProcessLost()

func (*BackendCoordinator) MarkReady added in v1.3.3 

func (c *BackendCoordinator) MarkReady()

func (*BackendCoordinator) Ready added in v1.3.3 

func (c *BackendCoordinator) Ready() bool

func (*BackendCoordinator) State added in v1.3.3 

func (c *BackendCoordinator) State() BackendState

type BackendState added in v1.3.3 

type BackendState string
const (
	BackendStarting   BackendState = "starting"
	BackendProbing    BackendState = "probing"
	BackendReady      BackendState = "ready"
	BackendDraining   BackendState = "draining"
	BackendRecycling  BackendState = "recycling"
	BackendRestarting BackendState = "restarting"
)

type BridgeProvider 

type BridgeProvider interface {
	ForwardRequest(ctx context.Context, data []byte) ([]byte, error)
}

BridgeProvider provides access to the stdio bridge for health checks.

type HealthChecker 

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

HealthChecker performs periodic health checks on MCP servers. It probes servers using the MCP tools/list method and tracks consecutive failures.

func NewHealthChecker 

func NewHealthChecker(bridge BridgeProvider, cfg HealthCheckerConfig) *HealthChecker

NewHealthChecker creates a new health checker for an MCP server.

func (*HealthChecker) IsHealthy 

func (hc *HealthChecker) IsHealthy() bool

IsHealthy returns the current health status.

func (*HealthChecker) Start 

func (hc *HealthChecker) Start()

Start begins periodic health checking.

func (*HealthChecker) Status 

func (hc *HealthChecker) Status() HealthStatus

Status returns detailed health status.

func (*HealthChecker) Stop 

func (hc *HealthChecker) Stop()

Stop stops the health checker.

type HealthCheckerConfig 

type HealthCheckerConfig struct {
	CheckInterval    time.Duration // How often to check (default: 30s)
	CheckTimeout     time.Duration // Timeout for each check (default: 10s)
	FailureThreshold int           // Consecutive failures before unhealthy (default: 3)
	Logger           *slog.Logger
	OnUnhealthy      func() // Called when threshold exceeded
}

HealthCheckerConfig configures the health checker.

type HealthStatus 

type HealthStatus struct {
	Healthy          bool      `json:"healthy"`
	ConsecutiveFails int       `json:"consecutive_fails"`
	LastCheck        time.Time `json:"last_check"`
	LastError        string    `json:"last_error,omitempty"`
}

HealthStatus contains the current health check status.

type ManagedProcess 

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

ManagedProcess represents a supervised MCP server process. It implements suture.Service for integration with the supervision tree.

func NewManagedProcess 

func NewManagedProcess(name string, cfg *config.ServerConfig, supCfg config.SupervisionConfig, logger *slog.Logger) *ManagedProcess

NewManagedProcess creates a new managed process for the given server config. Managed-http ownership is Linux-only for now: non-Linux proc identity reads fail closed before publication, so the backend is not silently unleased.

func NewManagedProcessWithOwnership added in v1.4.5 

func NewManagedProcessWithOwnership(name string, cfg *config.ServerConfig, supCfg config.SupervisionConfig, logger *slog.Logger, store ownership.LeaseStore, daemonID string, options ...ProcessOption) *ManagedProcess

func (*ManagedProcess) Config 

func (p *ManagedProcess) Config() *config.ServerConfig

Config returns the server configuration.

func (*ManagedProcess) LastError 

func (p *ManagedProcess) LastError() error

LastError returns the last error that caused the process to exit.

func (*ManagedProcess) LifecycleEvents added in v1.3.3 

func (p *ManagedProcess) LifecycleEvents() <-chan struct{}

LifecycleEvents coalesces process lifecycle transitions. Consumers must read LifecycleSnapshot after each signal; the snapshot is the durable truth.

func (*ManagedProcess) LifecycleSnapshot added in v1.3.3 

func (p *ManagedProcess) LifecycleSnapshot() (ServiceState, uint64)

func (*ManagedProcess) Name 

func (p *ManagedProcess) Name() string

Name returns the server name.

func (*ManagedProcess) PID 

func (p *ManagedProcess) PID() int

PID returns the process ID, or 0 if not running.

func (*ManagedProcess) RequestRestart added in v1.3.3 

func (p *ManagedProcess) RequestRestart() error

RequestRestart terminates the current child process. Suture owns the subsequent restart and lifecycle generation.

func (*ManagedProcess) RestartCount 

func (p *ManagedProcess) RestartCount() int

RestartCount returns the number of times this process has been restarted.

func (*ManagedProcess) Serve 

func (p *ManagedProcess) Serve(ctx context.Context) error

Serve implements suture.Service. It spawns the subprocess and blocks until it exits or context is cancelled.

func (*ManagedProcess) State 

func (p *ManagedProcess) State() ServiceState

State returns the current service state.

func (*ManagedProcess) Status 

func (p *ManagedProcess) Status() ServiceStatus

Status returns a snapshot of the current service status.

func (*ManagedProcess) Stdin 

func (p *ManagedProcess) Stdin() io.WriteCloser

Stdin returns the stdin pipe for the process. Used by the bridge to send MCP messages to the subprocess.

func (*ManagedProcess) Stdout 

func (p *ManagedProcess) Stdout() io.ReadCloser

Stdout returns the stdout pipe for the process. Used by the bridge to receive MCP messages from the subprocess.

func (*ManagedProcess) String 

func (p *ManagedProcess) String() string

String implements fmt.Stringer for logging.

func (*ManagedProcess) Uptime 

func (p *ManagedProcess) Uptime() time.Duration

Uptime returns how long the process has been running.

type Metrics 

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

Metrics tracks request statistics for an MCP server. Uses a ring buffer for response times to avoid memory allocations and provide O(1) insertions with bounded memory usage.

func NewMetrics 

func NewMetrics() *Metrics

NewMetrics creates a new metrics tracker.

func (*Metrics) RecordRequest 

func (m *Metrics) RecordRequest(duration time.Duration, isError bool)

RecordRequest records a request completion.

func (*Metrics) Reset 

func (m *Metrics) Reset()

Reset resets all metrics.

func (*Metrics) Snapshot 

func (m *Metrics) Snapshot() MetricsSnapshot

Snapshot returns current metrics.

type MetricsSnapshot 

type MetricsSnapshot struct {
	RequestCount    int64         `json:"request_count"`
	ErrorCount      int64         `json:"error_count"`
	Uptime          time.Duration `json:"uptime"`
	AvgResponseTime time.Duration `json:"avg_response_time"`
	P95ResponseTime time.Duration `json:"p95_response_time"`
}

MetricsSnapshot contains a point-in-time metrics snapshot.

type Option added in v1.4.5 

type Option func(*Supervisor)

Option configures optional managed-backend ownership wiring.

func WithLeaseStore added in v1.4.5 

func WithLeaseStore(store ownership.LeaseStore, daemonID string) Option

func WithProcessOptions added in v1.4.5 

func WithProcessOptions(options ...ProcessOption) Option

type ProbeError 

type ProbeError struct {
	Message string
}

ProbeError indicates a health probe failure.

func (*ProbeError) Error 

func (e *ProbeError) Error() string

type ProcessOption added in v1.4.5 

type ProcessOption func(*ManagedProcess)

func WithProcReader added in v1.4.5 

func WithProcReader(reader ownership.ProcReader) ProcessOption

func WithRestartClock added in v1.4.5 

func WithRestartClock(now func() time.Time, sleep func(context.Context, time.Duration) error) ProcessOption

WithRestartClock injects the restart clock and sleep implementation. It is useful for deterministic policy tests and preserves cancellation semantics.

func WithSignaler added in v1.4.5 

func WithSignaler(signaler ownership.Signaler) ProcessOption

func WithTokenGenerator added in v1.4.5 

func WithTokenGenerator(generator func() (string, error)) ProcessOption

type RestartLimitError added in v1.4.5 

type RestartLimitError struct {
	Limit    int
	Attempts int
}

RestartLimitError is terminal for a suture supervisor, while retaining the typed exhaustion details for status and callers.

func (*RestartLimitError) Error added in v1.4.5 

func (e *RestartLimitError) Error() string

func (*RestartLimitError) Unwrap added in v1.4.5 

func (e *RestartLimitError) Unwrap() error

type ServiceState 

type ServiceState string

ServiceState represents the current state of a managed service. 

const (
	StateStopped  ServiceState = "stopped"
	StateStarting ServiceState = "starting"
	StateRunning  ServiceState = "running"
	StateCrashed  ServiceState = "crashed"
	StateFailed   ServiceState = "failed" // Max restarts exceeded
)

type ServiceStatus 

type ServiceStatus struct {
	Name         string
	State        ServiceState
	PID          int
	Port         int
	Uptime       time.Duration
	RestartCount int
	LastError    error
	StartedAt    time.Time
}

ServiceStatus contains runtime information about a managed service.

type StartupProbe 

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

StartupProbe verifies that an MCP server has started correctly. It sends an initialize request and waits for a successful response.

func NewStartupProbe 

func NewStartupProbe(bridge BridgeProvider, cfg StartupProbeConfig) *StartupProbe

NewStartupProbe creates a new startup probe.

func (*StartupProbe) Wait 

func (sp *StartupProbe) Wait(ctx context.Context) error

Wait blocks until the server is ready or timeout expires. Returns nil if server is ready, error if timeout or probe failure.

type StartupProbeConfig 

type StartupProbeConfig struct {
	Timeout      time.Duration // Total time to wait for startup (default: 30s)
	PollInterval time.Duration // How often to retry (default: 1s)
	Logger       *slog.Logger
}

StartupProbeConfig configures the startup probe.

type StartupTimeoutError 

type StartupTimeoutError struct {
	Timeout time.Duration
}

StartupTimeoutError indicates the server didn't start in time.

func (*StartupTimeoutError) Error 

func (e *StartupTimeoutError) Error() string

type Supervisor 

type Supervisor struct {
	*suture.Supervisor
	// contains filtered or unexported fields
}

Supervisor wraps suture.Supervisor with Vision-specific functionality.

func New 

func New(cfg config.SupervisionConfig, logger *slog.Logger) *Supervisor

New creates a new Supervisor with the given configuration.

func NewWithOptions added in v1.4.5 

func NewWithOptions(cfg config.SupervisionConfig, logger *slog.Logger, options ...Option) *Supervisor

func (*Supervisor) AddServer 

func (s *Supervisor) AddServer(name string, serverCfg *config.ServerConfig) (*ManagedProcess, error)

AddServer registers a server to be supervised. The server won't start until Start() is called on the supervisor.

func (*Supervisor) GetServer 

func (s *Supervisor) GetServer(name string) *ManagedProcess

GetServer returns the managed process for a server, or nil if not found.

func (*Supervisor) RemoveServer 

func (s *Supervisor) RemoveServer(name string) error

RemoveServer unregisters and stops a server.

func (*Supervisor) RemoveServerAndWait added in v1.4.5 

func (s *Supervisor) RemoveServerAndWait(ctx context.Context, name string) error

RemoveServerAndWait unregisters a service and waits for its Serve method to finish. This keeps registry shutdown truthful: callers must not clear a process reference or its ownership lease before process-group cleanup ends.

func (*Supervisor) Servers 

func (s *Supervisor) Servers() []string

Servers returns all registered server names.

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.