runner

package
v0.12.0 Latest Latest
Warning

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

 Go to latest Published: Oct 16, 2026 License: AGPL-3.0 Imports: 20 Imported by: 0

Documentation

Overview

Package runner owns the controller side of the experimental remote execution protocol. An agent thread remains on the controller; one attempt leases one remote sandbox for all of its environment tools.

Index

Constants

View Source 
const (
	CommandOpen  = "open"
	CommandCall  = "call"
	CommandClose = "close"
)
View Source 
const ProtocolVersion = 1

Variables

View Source 
var (
	ErrStoreConflict  = errors.New("runner state revision conflict")
	ErrThreadBusy     = errors.New("agent thread already has an active attempt")
	ErrSessionsActive = errors.New("runner has reserved sessions; wait for cleanup or lease expiry before disabling")
	ErrFenced         = errors.New("remote attempt is fenced")
	ErrClosed         = errors.New("runner broker is closed")
	ErrUnauthorized   = errors.New("runner authentication failed")
	ErrUnknownOutcome = errors.New("remote execution outcome is unknown; do not automatically replay")
	ErrCallConflict   = errors.New("tool call ID was reused with different input")
)

Functions

func ValidateEnvironmentPrompt added in v0.6.0 

func ValidateEnvironmentPrompt(prompt string) error

Environment notes are set by the administrator, never by registration or heartbeat payloads. Existing sessions retain their original prompt snapshot.

Types

type Capabilities 

type Capabilities struct {
	Executor     string            `json:"executor,omitempty"`
	Slots        int               `json:"slots"`
	Labels       map[string]string `json:"labels,omitempty"`
	Environments []string          `json:"environments,omitempty"`
	CPUMillis    int64             `json:"cpu_millis,omitempty"`
	MemoryMiB    int64             `json:"memory_mib,omitempty"`
	GPUs         []GPU             `json:"gpus,omitempty"` // whole-device exclusive reservations
}

type Command 

type Command struct {
	ID         string    `json:"id"`
	Kind       string    `json:"kind"`
	SessionID  string    `json:"session_id"`
	AttemptID  string    `json:"attempt_id"`
	Epoch      string    `json:"epoch"`
	LeaseUntil time.Time `json:"lease_until"`
	Spec       *RunSpec  `json:"spec,omitempty"`
	GPUIds     []string  `json:"gpu_ids,omitempty"`
	Call       *ToolCall `json:"call,omitempty"`
	Reason     string    `json:"reason,omitempty"`
	Abort      bool      `json:"abort,omitempty"` // true: stop now, no final file writes; false: flush then close
}

type DiagnosticStore added in v0.12.0 

type DiagnosticStore interface {
	SaveDiagnostic(context.Context, uint64, json.RawMessage, []SessionStatus) (uint64, error)
}

DiagnosticStore commits a small content-free projection alongside the opaque checkpoint. Operators never need to expand tool transcripts to inspect leases.

type GPU 

type GPU struct {
	ID        string `json:"id"`
	Model     string `json:"model,omitempty"`
	MemoryMiB int64  `json:"memory_mib,omitempty"`
}

type HeartbeatRequest added in v0.12.0 

type HeartbeatRequest struct {
	NodeID   string           `json:"node_id"`
	BootID   string           `json:"boot_id"`
	Protocol int              `json:"protocol"`
	Stopped  []StoppedSession `json:"stopped,omitempty"`
}

Heartbeats carry lifecycle metadata only, independently of command delivery. Nodes negotiate support through PollResponse before using this endpoint.

type Hub 

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

func New 

func New(opts Options) (*Hub, error)

func (*Hub) Acquire 

func (h *Hub) Acquire(ctx context.Context, spec RunSpec) (*Session, error)

Acquire reserves a complete remote execution session. Waiting for capacity does not claim an intent on the node or rotate individual tools between nodes. Cancellation of ctx fences the session even after Acquire has returned.

func (*Hub) Authenticate 

func (h *Hub) Authenticate(r *http.Request) (string, error)

func (*Hub) AuthorizeSession 

func (h *Hub) AuthorizeSession(r *http.Request, sessionID string, write bool) (RunSpec, error)

func (*Hub) Close 

func (h *Hub) Close() error

func (*Hub) CreateRegistrationToken 

func (h *Hub) CreateRegistrationToken(ctx context.Context, ttl time.Duration) (string, time.Time, error)

func (*Hub) DisableIfIdle 

func (h *Hub) DisableIfIdle(commit func() error) error

DisableIfIdle commits the feature switch while excluding new admissions. Fenced attempts still hold reservations until their processes are confirmed stopped or their lease expires; switching these to local execution is unsafe. commit should only persist the central configuration, without calling Hub APIs.

func (*Hub) Handler 

func (h *Hub) Handler() http.Handler

Handler serves protocol-relative paths. Mount with StripPrefix("/runner/v1", hub.Handler()); registration uses its one-time token and every other request uses Authorization: Bearer <node credential>.

func (*Hub) IsClosed 

func (h *Hub) IsClosed() bool

func (*Hub) ListNodes 

func (h *Hub) ListNodes() []NodeView

func (*Hub) SessionStatuses added in v0.12.0 

func (h *Hub) SessionStatuses() []SessionStatus

func (*Hub) SetNodeConcurrencyLimit 

func (h *Hub) SetNodeConcurrencyLimit(id string, limit int) error

A smaller limit drains naturally; it never interrupts running attempts. Zero restores the node's advertised hard ceiling.

func (*Hub) SetNodeEnabled 

func (h *Hub) SetNodeEnabled(id string, enabled bool) error

func (*Hub) SetNodeEnvironmentPrompt added in v0.6.0 

func (h *Hub) SetNodeEnvironmentPrompt(id, prompt string) error

func (*Hub) SetNodeLimits 

func (h *Hub) SetNodeLimits(id string, slots int, cpu, memory int64) error

SetNodeLimits atomically updates admission budgets without resizing or killing running containers. Zero inherits the corresponding advertised hard capacity.

func (*Hub) SetWarmProfiles added in v0.7.0 

func (h *Hub) SetWarmProfiles(profiles []WarmProfile)

Profiles are reconstructed from durable application configuration on startup. MCP/tool initialization belongs to the assigned session, not an idle Pod.

func (*Hub) WithSessionWrite 

func (h *Hub) WithSessionWrite(r *http.Request, sessionID string, commit func() error) error

WithSessionWrite closes the authorization-to-rename race. The callback must only perform its short atomic commit; read/hash/upload before entering here.

type Lease 

type Lease struct {
	SessionID string    `json:"session_id"`
	AttemptID string    `json:"attempt_id"`
	Epoch     string    `json:"epoch"`
	Until     time.Time `json:"until"`
}

type MCPConfig 

type MCPConfig struct {
	ID         int64             `json:"id,omitempty"`
	Name       string            `json:"name"`
	Transport  string            `json:"transport"` // stdio, http, sse
	Command    string            `json:"command,omitempty"`
	Args       []string          `json:"args,omitempty"`
	Env        map[string]string `json:"env,omitempty"`
	URL        string            `json:"url,omitempty"`
	Headers    map[string]string `json:"headers,omitempty"`
	WorkingDir string            `json:"working_dir,omitempty"`
	Insecure   bool              `json:"insecure,omitempty"`
}

type MemoryStore 

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

MemoryStore is useful for isolated tests and explicitly ephemeral deployments. Production controllers should supply a persistent CAS implementation.

func (*MemoryStore) Guard 

func (s *MemoryStore) Guard(ctx context.Context, expected uint64, commit func() error) error

func (*MemoryStore) Load 

func (s *MemoryStore) Load(ctx context.Context) (Snapshot, error)

func (*MemoryStore) Save 

func (s *MemoryStore) Save(ctx context.Context, expected uint64, data json.RawMessage) (uint64, error)

type NodeView 

type NodeView struct {
	PoolStatus         *PoolStatus  `json:"pool_status,omitempty"`
	EnvironmentPrompt  string       `json:"environment_prompt"`
	CPULimitMillis     int64        `json:"cpu_limit_millis"`
	MemoryLimitMiB     int64        `json:"memory_limit_mib"`
	EffectiveCPUMillis int64        `json:"effective_cpu_millis"`
	EffectiveMemoryMiB int64        `json:"effective_memory_mib"`
	ReservedCPUMillis  int64        `json:"reserved_cpu_millis"`
	ReservedMemoryMiB  int64        `json:"reserved_memory_mib"`
	ConcurrencyLimit   int          `json:"concurrency_limit"`
	EffectiveSlots     int          `json:"effective_slots"`
	ID                 string       `json:"id"`
	Name               string       `json:"name"`
	BootID             string       `json:"boot_id"`
	Enabled            bool         `json:"enabled"`
	Online             bool         `json:"online"`
	LastSeen           time.Time    `json:"last_seen"`
	Capabilities       Capabilities `json:"capabilities"`
	ActiveSessions     int          `json:"active_sessions"`
	ReservedGPUs       []string     `json:"reserved_gpus,omitempty"`
}

type NotificationEnvelope 

type NotificationEnvelope struct {
	ID        string              `json:"id"`
	SessionID string              `json:"session_id"`
	AttemptID string              `json:"attempt_id"`
	Items     []tool.Notification `json:"items"`
}

type Options 

type Options struct {
	Store           StateStore
	LeaseDuration   time.Duration    // default 30 seconds
	RegistrationTTL time.Duration    // default 10 minutes, maximum one hour
	PollWait        time.Duration    // default 10 seconds; less than lease duration
	Clock           func() time.Time // injectable for deterministic lease tests
}

type OutcomeError 

type OutcomeError struct{ Cause error }

OutcomeError deliberately remains distinguishable from a normal tool error. The caller must reconcile this attempt, not automatically replay the command.

func (*OutcomeError) Error 

func (e *OutcomeError) Error() string

func (*OutcomeError) FatalExecution 

func (e *OutcomeError) FatalExecution() bool

func (*OutcomeError) Is 

func (e *OutcomeError) Is(target error) bool

func (*OutcomeError) Unwrap 

func (e *OutcomeError) Unwrap() error

type PollRequest 

type PollRequest struct {
	PoolStatus    *PoolStatus            `json:"pool_status,omitempty"`
	NodeID        string                 `json:"node_id"`
	BootID        string                 `json:"boot_id"`
	Protocol      int                    `json:"protocol"`
	Capabilities  *Capabilities          `json:"capabilities,omitempty"` // adopted on a new boot; same-boot changes require restart
	Notifications []NotificationEnvelope `json:"notifications,omitempty"`
}

type PollResponse 

type PollResponse struct {
	Heartbeat                 bool        `json:"heartbeat,omitempty"`
	Pool                      *PoolPolicy `json:"pool,omitempty"`
	Epoch                     string      `json:"epoch"`
	ServerTime                time.Time   `json:"server_time"`
	Leases                    []Lease     `json:"leases"`   // complete live set for this boot; absent/expired sessions must stop
	Commands                  []Command   `json:"commands"` // redelivered until acknowledged; node must deduplicate ID
	AcknowledgedNotifications []string    `json:"acknowledged_notifications,omitempty"`
}

type PoolPolicy added in v0.7.0 

type PoolPolicy struct {
	Slots     int           `json:"slots"`
	CPUMillis int64         `json:"cpu_millis"`
	MemoryMiB int64         `json:"memory_mib"`
	Profiles  []WarmProfile `json:"profiles"`
}

type PoolStatus added in v0.7.0 

type PoolStatus struct {
	Idle      int    `json:"idle"`
	Preparing int    `json:"preparing"`
	Pods      int    `json:"pods"`
	CPUMillis int64  `json:"cpu_millis"`
	MemoryMiB int64  `json:"memory_mib"`
	Error     string `json:"error,omitempty"`
}

type RegisterRequest 

type RegisterRequest struct {
	Token        string       `json:"token"`
	Name         string       `json:"name"`
	BootID       string       `json:"boot_id"`
	Capabilities Capabilities `json:"capabilities"`
	Protocol     int          `json:"protocol"`
}

type RegisterResponse 

type RegisterResponse struct {
	NodeID       string `json:"node_id"`
	Credential   string `json:"credential"` // shown once; controller persists only its hash
	Epoch        string `json:"epoch"`
	Protocol     int    `json:"protocol"`
	LeaseSeconds int    `json:"lease_seconds"`
}

type Resources 

type Resources struct {
	CPUMillis    int64  `json:"cpu_millis,omitempty"`
	MemoryMiB    int64  `json:"memory_mib,omitempty"`
	PIDs         int64  `json:"pids,omitempty"`
	GPUCount     int    `json:"gpu_count,omitempty"`
	GPUModel     string `json:"gpu_model,omitempty"`
	GPUMemoryMiB int64  `json:"gpu_memory_mib,omitempty"` // minimum advertised memory per device
}

type ResultRequest 

type ResultRequest struct {
	EnvironmentTools toolenv.Inventory   `json:"environment_tools,omitempty"`
	NodeID           string              `json:"node_id"`
	BootID           string              `json:"boot_id"`
	CommandID        string              `json:"command_id"`
	SessionID        string              `json:"session_id"`
	AttemptID        string              `json:"attempt_id"`
	Epoch            string              `json:"epoch"`
	Result           tool.Result         `json:"result"`
	MCPTools         []ToolDefinition    `json:"mcp_tools,omitempty"` // Open acknowledgment only
	Notifications    []tool.Notification `json:"notifications,omitempty"`
	Error            string              `json:"error,omitempty"` // execution infrastructure failure, not ordinary tool.IsError
	UnknownOutcome   bool                `json:"unknown_outcome,omitempty"`
}

type ResultResponse 

type ResultResponse struct {
	Accepted bool `json:"accepted"`
}

type RunSpec 

type RunSpec struct {
	PoolCPUMillis    int64             `json:"pool_cpu_millis,omitempty"`
	PoolMemoryMiB    int64             `json:"pool_memory_mib,omitempty"`
	ConcurrencyLimit int               `json:"concurrency_limit,omitempty"` // set by the broker, never by the agent
	NodeID           string            `json:"node_id,omitempty"`
	ThreadID         string            `json:"thread_id"`
	TaskID           string            `json:"task_id,omitempty"`
	AgentKey         string            `json:"agent_key,omitempty"`
	Environment      string            `json:"environment,omitempty"`
	Image            string            `json:"image,omitempty"`
	Labels           map[string]string `json:"labels,omitempty"`
	Resources        Resources         `json:"resources,omitempty"`
	Env              map[string]string `json:"env,omitempty"`
	WorkingDir       string            `json:"working_dir"`
	MaxOutputChars   int               `json:"max_output_chars,omitempty"`
	Tools            []ToolDefinition  `json:"tools"` // allowlist; MCPs additionally allow their discovered schemas
	MCPs             []MCPConfig       `json:"mcps,omitempty"`
	Files            json.RawMessage   `json:"files,omitempty"` // versioned runnerfiles manifest
	ConfigVersion    string            `json:"config_version,omitempty"`
}

type Session 

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

func (*Session) AttemptID 

func (s *Session) AttemptID() string

func (*Session) Call 

func (s *Session) Call(ctx context.Context, name string, input json.RawMessage, tc *tool.ToolContext) (tool.Result, error)

func (*Session) Close 

func (s *Session) Close() error

Close drains final filesystem updates before acknowledging closure. Cancellation uses abort instead and immediately fences writes. Both retain reservations until the node confirms process cleanup or the old lease expires.

func (*Session) DrainNotifications 

func (s *Session) DrainNotifications() []tool.Notification

func (*Session) EnvironmentPrompt added in v0.6.0 

func (s *Session) EnvironmentPrompt() string

func (*Session) EnvironmentTools added in v0.6.0 

func (s *Session) EnvironmentTools() toolenv.Inventory

func (*Session) Handles 

func (s *Session) Handles(name string) bool

func (*Session) ID 

func (s *Session) ID() string

func (*Session) MCPTools 

func (s *Session) MCPTools(ctx context.Context) ([]ToolDefinition, error)

func (*Session) NodeID 

func (s *Session) NodeID() string

func (*Session) WaitFailure added in v0.12.0 

func (s *Session) WaitFailure(ctx context.Context) error

WaitFailure lets the owner interrupt model generation when its sandbox dies, even if no further execution tool is called. Normal close is not a failure.

type SessionLostError added in v0.12.0 

type SessionLostError struct{ Cause error }

SessionLostError means this rejected call did not start. Earlier in-flight calls may still have unknown outcomes; replacing the session never replays them.

func (*SessionLostError) Error added in v0.12.0 

func (e *SessionLostError) Error() string

func (*SessionLostError) FatalExecution added in v0.12.0 

func (e *SessionLostError) FatalExecution() bool

func (*SessionLostError) Unwrap added in v0.12.0 

func (e *SessionLostError) Unwrap() error

type SessionStatus added in v0.12.0 

type SessionStatus struct {
	ID           string    `json:"id"`
	TaskID       string    `json:"task_id"`
	ThreadID     string    `json:"thread_id"`
	NodeID       string    `json:"node_id"`
	State        string    `json:"state"`
	ReasonCode   string    `json:"reason_code,omitempty"`
	LeaseUntil   time.Time `json:"lease_until"`
	FinishedAt   time.Time `json:"finished_at,omitempty"`
	CommandCount int       `json:"command_count"`
}

type Snapshot 

type Snapshot struct {
	Revision uint64
	Data     json.RawMessage
}

type StateGuard 

type StateGuard interface {
	Guard(context.Context, uint64, func() error) error
}

StateGuard serializes a short artifact commit against a controller takeover. Persistent stores should implement this in the same transactional lock domain as Save. Without it, file commit fencing is limited to this Hub instance.

type StateStore 

type StateStore interface {
	Load(context.Context) (Snapshot, error)
	Save(context.Context, uint64, json.RawMessage) (uint64, error)
}

StateStore must atomically compare and replace its revision. Only one broker owns a revision; this prevents a superseded controller from renewing leases. A missing record is Snapshot{}; Save(expected=0) creates it atomically.

type StoppedSession added in v0.12.0 

type StoppedSession struct {
	SessionID string `json:"session_id"`
	AttemptID string `json:"attempt_id"`
	Epoch     string `json:"epoch"`
	Reason    string `json:"reason"`
}

type ToolCall 

type ToolCall struct {
	ID             string          `json:"id"`
	Name           string          `json:"name"`
	Input          json.RawMessage `json:"input"`
	WorkingDir     string          `json:"working_dir,omitempty"`
	OutputDir      string          `json:"output_dir,omitempty"`
	MaxOutputChars int             `json:"max_output_chars,omitempty"`
	Env            []string        `json:"env,omitempty"`
	Deadline       time.Time       `json:"deadline,omitempty"`
}

type ToolDefinition 

type ToolDefinition struct {
	Name        string          `json:"name"`
	Description string          `json:"description,omitempty"`
	InputSchema map[string]any  `json:"input_schema,omitempty"`
	Kind        string          `json:"kind,omitempty"` // builtin, command, script, http, mcp
	Exec        json.RawMessage `json:"exec,omitempty"`
	ReadOnly    bool            `json:"read_only,omitempty"`
	Concurrent  bool            `json:"concurrent,omitempty"`
}

type WarmProfile added in v0.7.0 

type WarmProfile struct {
	Environment string            `json:"environment"`
	Idle        int               `json:"idle"`
	Image       string            `json:"image"`
	Resources   Resources         `json:"resources"`
	Env         map[string]string `json:"env,omitempty"`
	Labels      map[string]string `json:"labels,omitempty"`
	WorkingDir  string            `json:"working_dir,omitempty"`
	Files       json.RawMessage   `json:"files,omitempty"`
	NodeID      string            `json:"node_id,omitempty"`
}

Pool policy is supplied by the controller; status is observational only and never used to release reservations or grant execution authority.

func WarmProfileFromSpec added in v0.7.0 

func WarmProfileFromSpec(s RunSpec) WarmProfile

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.