registry

package
v0.5.30-beta Latest Latest
Warning

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

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

Documentation

Overview

Package registry owns live runner registrations and run-lease coordination in the control plane.

Index

Constants

View Source 
const (
	RunStatusOpening   = protocol.RunStatusOpening
	RunStatusRunning   = protocol.RunStatusRunning
	RunStatusSucceeded = protocol.RunStatusSucceeded
	RunStatusFailed    = protocol.RunStatusFailed
	RunStatusCanceled  = protocol.RunStatusCanceled
	RunStatusLost      = protocol.RunStatusLost
)

Variables

View Source 
var (
	// ErrRunnerNotFound indicates that a stable runner registration does not exist.
	ErrRunnerNotFound = errors.New("runner not found")
	// ErrRunnerConnected indicates that a live runner must be stopped before removal.
	ErrRunnerConnected = errors.New("runner is connected")
	// ErrRunnerActiveRun indicates inconsistent state that still references an active run.
	ErrRunnerActiveRun = errors.New("runner has an active run")
)

Functions

This section is empty.

Types

type ConversationAffinity 

type ConversationAffinity struct {
	RunnerID           string
	EnvironmentProfile string
}

ConversationAffinity is the durable environment selection for one conversation. 

type Link interface {
	Call(ctx context.Context, method string, params any, result any) error
	CallTracked(ctx context.Context, method string, params any, result any, onRequestID func(string)) error
	Notify(ctx context.Context, method string, params any) error
	Close() error
	Done() <-chan struct{}
	Err() error
}

Link is the live symmetric RPC connection retained for one runner generation.

type Options 

type Options struct {
	HeartbeatInterval time.Duration
	HeartbeatTimeout  time.Duration
	RunLeaseGrace     time.Duration
	Now               func() time.Time
	NewID             func(prefix string) (string, error)
	Persistence       Persistence
}

Options configures registry liveness and test seams.

type PersistedState 

type PersistedState struct {
	Runners    []Runner
	Runs       []Run
	Affinities map[string]ConversationAffinity
}

PersistedState is the durable registry state restored when the control plane starts.

type Persistence 

type Persistence interface {
	Load(ctx context.Context) (PersistedState, error)
	SaveRunner(ctx context.Context, runner Runner) error
	SaveRun(ctx context.Context, run Run) error
	SaveRunnerAndRun(ctx context.Context, runner Runner, run Run) error
	SaveRunnerAndRuns(ctx context.Context, runner Runner, runs []Run) error
	BindConversation(ctx context.Context, conversationID, runnerID, environmentProfile string, now time.Time) error
	ConversationAffinity(ctx context.Context, conversationID string) (ConversationAffinity, bool, error)
	RemoveRunner(ctx context.Context, runnerID string, force bool) (RemovalResult, error)
	Close() error
}

Persistence stores durable runner identity, run state, and conversation affinity.

type Registry 

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

Registry coordinates stable runner identity, live connection generations, and concurrent runs.

func New 

func New(parent context.Context, options Options) (*Registry, error)

New creates a live registry and restores any configured durable state.

func (*Registry) BindConversation 

func (r *Registry) BindConversation(ctx context.Context, conversationID, runnerID string) error

BindConversation establishes authoritative affinity between a conversation and runner.

func (*Registry) BindConversationWithEnvironmentProfile 

func (r *Registry) BindConversationWithEnvironmentProfile(ctx context.Context, conversationID, runnerID, environmentProfile string) error

BindConversationWithEnvironmentProfile establishes and immediately persists authoritative affinity.

func (*Registry) CallRun 

func (r *Registry) CallRun(ctx context.Context, runID, method string, params any, result any) error

CallRun invokes a run-scoped runner method after generation and active-run validation.

func (*Registry) CancelRun 

func (r *Registry) CancelRun(ctx context.Context, runID, reason string) error

CancelRun cancels active runner operations but leaves run.close responsible for releasing the lease.

func (*Registry) Close 

func (r *Registry) Close() error

Close disconnects all live runner generations and stops liveness monitoring.

func (*Registry) CloseRun 

func (r *Registry) CloseRun(ctx context.Context, runID string, status RunStatus, runErr error) error

CloseRun releases one runner lease and records its terminal status.

func (*Registry) CommitConversationAffinity 

func (r *Registry) CommitConversationAffinity(ctx context.Context, conversationID string) error

CommitConversationAffinity persists a pending first-turn reservation after the conversation record itself has been saved.

func (*Registry) DeliverToolUpdate 

func (r *Registry) DeliverToolUpdate(runnerID, connectionID string, generation int64, params runnerpayload.ToolUpdateParams) error

DeliverToolUpdate forwards a generation-fenced transient tool update.

func (*Registry) Detach 

func (r *Registry) Detach(runnerID, connectionID string, generation int64, cause error)

Detach marks a connection offline only when it is still the current fenced generation.

func (*Registry) EnvironmentError 

func (r *Registry) EnvironmentError(runnerID, connectionID string, generation int64, params protocol.EnvironmentErrorParams) error

EnvironmentError marks an active run failed after an asynchronous runner notification.

func (*Registry) ExecuteTool 

func (r *Registry) ExecuteTool(ctx context.Context, params runnerpayload.ToolExecuteParams, updates func(runnerpayload.ToolUpdateParams)) (runnerpayload.ToolExecuteResult, error)

ExecuteTool invokes tool.execute and routes replaceable transient updates to the supplied sink.

func (*Registry) ForgetConversation 

func (r *Registry) ForgetConversation(conversationID string)

ForgetConversation removes affinity after central conversation deletion commits.

func (*Registry) Heartbeat 

func (r *Registry) Heartbeat(runnerID, connectionID string, generation int64, params protocol.HeartbeatParams) error

Heartbeat applies application health from the current connection generation.

func (*Registry) ManifestChanged 

func (r *Registry) ManifestChanged(runnerID, connectionID string, generation int64, params protocol.ManifestChangedParams) error

ManifestChanged updates the idle manifest digest for the current generation.

func (*Registry) OpenRun 

func (r *Registry) OpenRun(ctx context.Context, runnerID string, params protocol.RunOpenParams) (runnerpayload.Manifest, error)

OpenRun reserves a run lease, opens the remote environment, validates its manifest, and marks the run active.

func (*Registry) Register 

func (r *Registry) Register(params protocol.RegisterParams, link Link) (protocol.RegisterResult, error)

Register upserts one stable workspace identity and atomically replaces its live generation.

func (*Registry) ReleasePendingConversationAffinity 

func (r *Registry) ReleasePendingConversationAffinity(conversationID string) bool

ReleasePendingConversationAffinity forgets an unpublished first-run reservation. Durable bindings and active run leases are never removed by this method.

func (*Registry) RemoveRunner 

func (r *Registry) RemoveRunner(ctx context.Context, runnerID string, force bool) (RemovalResult, error)

RemoveRunner deletes an offline stable registration and its run history. Force explicitly abandons any conversations pinned to the runner.

func (*Registry) ResolveConversationAffinity 

func (r *Registry) ResolveConversationAffinity(ctx context.Context, conversationID string) (ConversationAffinity, bool, error)

ResolveConversationAffinity refreshes one durable affinity while retaining a pending first-turn reservation that has not created a conversation record yet.

func (*Registry) Run 

func (r *Registry) Run(id string) (Run, bool)

Run returns one run snapshot.

func (*Registry) Runner 

func (r *Registry) Runner(id string) (Runner, bool)

Runner returns one stable runner snapshot.

func (*Registry) RunnerForConversation 

func (r *Registry) RunnerForConversation(conversationID string) (string, bool)

RunnerForConversation returns the current durable or pending runner affinity.

func (*Registry) Runners 

func (r *Registry) Runners() []Runner

Runners returns deterministic runner snapshots ordered by stable ID.

func (*Registry) SetEnvironmentErrorHandler 

func (r *Registry) SetEnvironmentErrorHandler(handler func(conversationID string))

SetEnvironmentErrorHandler installs the control-plane cancellation hook for asynchronous runner failures.

type RemovalResult 

type RemovalResult struct {
	RunnerID                      string `json:"runnerId"`
	RemovedRuns                   int    `json:"removedRuns"`
	RemovedConversationAffinities int    `json:"removedConversationAffinities"`
}

RemovalResult describes durable state removed with one runner registration.

type Run 

type Run struct {
	ID             string    `db:"id" json:"id"`
	ConversationID string    `db:"conversation_id" json:"conversationId"`
	RunnerID       string    `db:"runner_id" json:"runnerId"`
	Status         RunStatus `db:"status" json:"status"`
	ManifestDigest string    `db:"manifest_digest" json:"manifestDigest,omitempty"`
	ManifestJSON   string    `db:"manifest_json" json:"-"`
	Error          string    `db:"error" json:"error,omitempty"`
	CreatedAt      time.Time `db:"created_at" json:"createdAt"`
	UpdatedAt      time.Time `db:"updated_at" json:"updatedAt"`
}

Run is a snapshot of one top-level runner environment lease.

type RunStatus 

type RunStatus = protocol.RunStatus

RunStatus is retained as an API alias for the shared runner lease status.

type Runner 

type Runner struct {
	ID                 string             `json:"id"`
	DisplayName        string             `json:"displayName,omitempty"`
	Host               protocol.Host      `json:"host"`
	Workspace          protocol.Workspace `json:"workspace"`
	KodeletVersion     string             `json:"kodeletVersion"`
	ManifestDigest     string             `json:"manifestDigest,omitempty"`
	ManifestChanged    bool               `json:"manifestChanged"`
	CompatibilityError string             `json:"compatibilityError,omitempty"`
	Status             RunnerStatus       `json:"status"`
	Connected          bool               `json:"connected"`
	ConcurrentRuns     bool               `json:"concurrentRuns"`
	ActiveRunID        string             `json:"activeRunId,omitempty"`
	ActiveRunIDs       []string           `json:"activeRunIds,omitempty"`
	ConnectionID       string             `json:"connectionId,omitempty"`
	Generation         int64              `json:"generation"`
	ConnectedAt        time.Time          `json:"connectedAt,omitempty"`
	LastHeartbeatAt    time.Time          `json:"lastHeartbeatAt,omitempty"`
	CreatedAt          time.Time          `json:"createdAt"`
	UpdatedAt          time.Time          `json:"updatedAt"`
}

Runner is a safe snapshot of one stable runner registration.

type RunnerReferencedError 

type RunnerReferencedError struct {
	RunnerID        string
	ConversationIDs []string
}

RunnerReferencedError prevents an ordinary removal from abandoning conversation affinity.

func (*RunnerReferencedError) Error 

func (e *RunnerReferencedError) Error() string

type RunnerStatus 

type RunnerStatus string

RunnerStatus is the scheduler-facing state of a runner registration. 

const (
	RunnerStatusOffline      RunnerStatus = "offline"
	RunnerStatusConnecting   RunnerStatus = "connecting"
	RunnerStatusIdle         RunnerStatus = "idle"
	RunnerStatusBusy         RunnerStatus = "busy"
	RunnerStatusError        RunnerStatus = "error"
	RunnerStatusIncompatible RunnerStatus = "incompatible"
)

type SQLitePersistence 

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

SQLitePersistence stores runner state in Kodelet's shared SQLite database.

func NewSQLitePersistence 

func NewSQLitePersistence(ctx context.Context, dbPath, ownerID string) (*SQLitePersistence, error)

NewSQLitePersistence opens a durable runner store at dbPath. Database migrations must have run before this constructor is used.

func (*SQLitePersistence) BindConversation 

func (s *SQLitePersistence) BindConversation(ctx context.Context, conversationID, runnerID, environmentProfile string, now time.Time) error

BindConversation durably establishes affinity without silently moving an existing conversation.

func (*SQLitePersistence) Close 

func (s *SQLitePersistence) Close() error

Close releases the SQLite connection.

func (*SQLitePersistence) ConversationAffinity 

func (s *SQLitePersistence) ConversationAffinity(ctx context.Context, conversationID string) (ConversationAffinity, bool, error)

ConversationAffinity reads the current durable affinity so a long-running control plane can observe external deletion.

func (*SQLitePersistence) Load 

func (s *SQLitePersistence) Load(ctx context.Context) (PersistedState, error)

Load restores records for this authenticated owner.

func (*SQLitePersistence) RemoveRunner 

func (s *SQLitePersistence) RemoveRunner(ctx context.Context, runnerID string, force bool) (RemovalResult, error)

RemoveRunner deletes one offline registration and its durable run history. Every durable conversation affinity blocks ordinary removal.

func (*SQLitePersistence) SaveRun 

func (s *SQLitePersistence) SaveRun(ctx context.Context, run Run) error

SaveRun inserts or updates one top-level runner run.

func (*SQLitePersistence) SaveRunner 

func (s *SQLitePersistence) SaveRunner(ctx context.Context, runner Runner) error

SaveRunner inserts or updates one stable runner registration.

func (*SQLitePersistence) SaveRunnerAndRun 

func (s *SQLitePersistence) SaveRunnerAndRun(ctx context.Context, runner Runner, run Run) error

SaveRunnerAndRun atomically stores a runner registration and one of its runs.

func (*SQLitePersistence) SaveRunnerAndRuns 

func (s *SQLitePersistence) SaveRunnerAndRuns(ctx context.Context, runner Runner, runs []Run) error

SaveRunnerAndRuns atomically stores a runner registration and its changed runs.

type Session 

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

Session binds one WebSocket peer to a generation-fenced runner registration.

func NewSession 

func NewSession(registry *Registry, ui UIRequestRouter) *Session

NewSession creates a connection handler. Attach must be called before the peer starts.

func (*Session) Attach 

func (s *Session) Attach(peer Link)

Attach binds the symmetric peer used for registration and reverse calls.

func (*Session) Detach 

func (s *Session) Detach(cause error)

Detach removes this session only when it remains the current runner generation.

func (*Session) HandleNotification 

func (s *Session) HandleNotification(ctx context.Context, method string, params json.RawMessage)

HandleNotification implements protocol.NotificationHandler.

func (*Session) HandleRequest 

func (s *Session) HandleRequest(ctx context.Context, method string, params json.RawMessage) (any, *protocol.RPCError)

HandleRequest implements protocol.RequestHandler.

type UIRequestRouter 

type UIRequestRouter interface {
	HandleRunnerUIRequest(ctx context.Context, runnerID string, method string, params json.RawMessage) (any, *protocol.RPCError)
}

UIRequestRouter handles runner-originated extension UI requests in the control plane.

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.