daemon

package
v0.14.10 Latest Latest
Warning

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

 Go to latest Published: Aug 4, 2026 License: MIT Imports: 33 Imported by: 0

Documentation

Overview

Package daemon is STABLE — background snapshot refresh and in-memory cache. It powers stable cached reads.

Index

Constants

View Source 
const (
	RefreshTriggerManual       = "manual"
	RefreshTriggerStartup      = "startup"
	RefreshTriggerInterval     = "interval"
	RefreshTriggerConfigChange = "config-change"
	RefreshTriggerStateChange  = "state-change"
	RefreshTriggerSkillsChange = "skills-change"
	RefreshTriggerBeaconChange = "beacon-change"
	RefreshTriggerDiscovery    = "discovery"
)
View Source 
const (
	RunStreamContentType = "application/x-ndjson"

	RunStreamEventReady       = "ready"
	RunStreamEventStateChange = "state_change"
	RunStreamEventStdout      = "stdout"
	RunStreamEventStderr      = "stderr"
	RunStreamEventResult      = "result"
)
View Source 
const (
	DefaultAddr = "127.0.0.1:42425"
)
View Source 
const ShutdownDrainTimeout = 10 * time.Second

ShutdownDrainTimeout is the maximum time to wait for in-flight requests and the background refresh goroutines to finish when the daemon is stopping.

View Source 
const Version = buildinfo.Version

Variables

This section is empty.

Functions

func ApplyReservationView 

func ApplyReservationView(snap *models.ClusterSnapshot, st *state.ClusterState, ledger *reservation.Ledger)

ApplyReservationView overlays locally persisted reservations onto a snapshot so read paths can reason about allocatable RAM without requiring daemon-only semantics.

func CanReserve 

func CanReserve(snap *models.ClusterSnapshot, node string, mb int64) bool

func CloneSnapshot 

func CloneSnapshot(snap *models.ClusterSnapshot) *models.ClusterSnapshot

func DefaultSnapshotPath 

func DefaultSnapshotPath() string

func FetchSnapshot 

func FetchSnapshot(ctx context.Context, addr string) (*models.ClusterSnapshot, string, error)

func HealthPayload 

func HealthPayload(meta *Metadata) map[string]any

func HttpClientForAddr added in v0.3.0 

func HttpClientForAddr(addr string) (*http.Client, string)

func NormalizeAddr 

func NormalizeAddr(addr string) string

func NormalizeRefreshTrigger added in v0.7.0 

func NormalizeRefreshTrigger(trigger string) (string, error)

NormalizeRefreshTrigger validates and canonicalizes daemon refresh trigger labels. Empty input maps to the explicit manual trigger. Supports comma-separated coalesced triggers by validating, sorting, and deduping them.

func NormalizeRunResult added in v0.7.0 

func NormalizeRunResult(resp execution.GuardedExecutionResult, runErr error) execution.GuardedExecutionResult

NormalizeRunResult folds an execution error into the final guarded result so both streamed and buffered HTTP paths report the same terminal payload.

func RegisterRoutes 

func RegisterRoutes(mux *http.ServeMux, cache SnapshotCache)

func RegisterRoutesWithDeps 

func RegisterRoutesWithDeps(mux *http.ServeMux, cache SnapshotCache, deps RouteDeps)

func RunGuarded added in v0.7.0 

func RunGuarded(ctx context.Context, addr string, req execution.GuardedExecutionRequest, origin models.ExecutionOrigin) (execution.GuardedExecutionResult, error)

RunGuarded preserves the simple final-result helper surface for callers that do not need streamed callbacks. It reuses the same streamed /run transport as other local execution callers so long-running executions are bounded by the caller context rather than the short metadata timeout.

func RunGuardedStream added in v0.7.0 

func RunGuardedStream(ctx context.Context, addr string, req execution.GuardedExecutionRequest, origin models.ExecutionOrigin) (execution.GuardedExecutionResult, error)

RunGuardedStream executes a guarded request through the local AXIS HTTP /run surface using the NDJSON streaming contract.

func Serve 

func Serve(addr string, cache SnapshotCache) error

func WantsRunStream added in v0.7.0 

func WantsRunStream(r *http.Request) bool

WantsRunStream reports whether the caller requested the streaming /run contract via query string or Accept header.

func WireRunStreamResponse added in v0.7.0 

func WireRunStreamResponse(w http.ResponseWriter, r *http.Request, req *execution.GuardedExecutionRequest) (func(execution.GuardedExecutionResult) error, bool, error)

WireRunStreamResponse adapts a guarded execution request to the NDJSON /run contract when the caller requested streaming. Existing callbacks and writers are preserved so HTTP handlers can layer streaming on top of their local refresh logic instead of replacing it.

Types

type Collector 

type Collector func(context.Context) (*models.ClusterSnapshot, error)

type Daemon 

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

func New 

func New(interval time.Duration, collector Collector) *Daemon

func NewDefault 

func NewDefault(interval time.Duration) *Daemon

func (*Daemon) AddOnSnapshotChanged added in v0.11.0 

func (d *Daemon) AddOnSnapshotChanged(fn SnapshotChangedFunc) (remove func())

AddOnSnapshotChanged registers a subscriber that will be invoked after each successful refresh whose content hash differs from the previous one seen by that subscriber. The returned function unregisters the hook.

The daemon's interval refresh can produce a byte-identical snapshot when the cluster is idle; hooks are debounced per-subscriber so an idle cluster does not generate redundant notifications (OQ-5).

The hook runs on the refresh goroutine after d.mu has been released. It must not call back into d.Snapshot() with a long-held read lock; clone the snapshot first if persistence is required beyond the callback.

This API is intended for in-process consumers (MCP server, HTTP cache watcher). It is not part of the public HTTP surface.

func (*Daemon) Invalidate 

func (d *Daemon) Invalidate()

func (*Daemon) Ledger added in v0.10.2 

func (d *Daemon) Ledger() *reservation.Ledger

func (*Daemon) MaxRefreshLatency added in v0.10.2 

func (d *Daemon) MaxRefreshLatency() time.Duration

func (*Daemon) Mesh added in v0.10.2 

func (d *Daemon) Mesh() *mesh.Mesh

func (*Daemon) Meta 

func (d *Daemon) Meta() Metadata

func (*Daemon) Refresh 

func (d *Daemon) Refresh(ctx context.Context) error

func (*Daemon) RefreshNow 

func (d *Daemon) RefreshNow(ctx context.Context) error

func (*Daemon) RefreshWithTrigger added in v0.7.0 

func (d *Daemon) RefreshWithTrigger(ctx context.Context, trigger string) error

func (*Daemon) SetSnapshotPath 

func (d *Daemon) SetSnapshotPath(path string)

func (*Daemon) SetStaleThreshold added in v0.10.0 

func (d *Daemon) SetStaleThreshold(threshold time.Duration)

SetStaleThreshold configures how old the cache must be before it is considered stale. The default is 5 minutes. A value <= 0 resets to default.

func (*Daemon) Snapshot 

func (d *Daemon) Snapshot() (*models.ClusterSnapshot, bool)

func (*Daemon) Start 

func (d *Daemon) Start(ctx context.Context)

func (*Daemon) WaitStopped added in v0.4.0 

func (d *Daemon) WaitStopped(ctx context.Context)

WaitStopped blocks until all background goroutines have finished or ctx expires. Call after cancelling the context passed to Start/WatchConfig to ensure in-flight refreshes complete before the process exits.

func (*Daemon) WatchConfig added in v0.4.0 

func (d *Daemon) WatchConfig(ctx context.Context, configPath string)

WatchConfig polls configPath and triggers Invalidate+Refresh whenever the file contents appear or change, or when the file disappears. Runs until ctx is cancelled. This is the primary config-driven cache trigger.

func (*Daemon) WatchDiscovery added in v0.7.0 

func (d *Daemon) WatchDiscovery(ctx context.Context, configPath string)

WatchDiscovery keeps a long-lived UDP discovery watcher aligned with the current config file and refreshes the daemon cache when beacon-derived nodes appear, change, or age out.

func (*Daemon) WatchMesh added in v0.10.2 

func (d *Daemon) WatchMesh(ctx context.Context, self mesh.Peer)

WatchMesh starts the mesh gossip layer and refreshes cache on peer events

func (*Daemon) WatchSkills added in v0.7.0 

func (d *Daemon) WatchSkills(ctx context.Context, skillsPath string)

WatchSkills polls skillsPath and triggers Invalidate+Refresh whenever the learned skills/failures store changes on disk.

func (*Daemon) WatchState added in v0.7.0 

func (d *Daemon) WatchState(ctx context.Context, statePath string)

WatchState polls statePath and triggers Invalidate+Refresh whenever local reservation/failure memory changes on disk. This keeps cached snapshots in step with guarded execution and local placement memory.

type KnowledgeResponse 

type KnowledgeResponse struct {
	Knowledge *knowledge.ClusterKnowledge `json:"knowledge"`
	Skills    []skills.LearnedSkill       `json:"skills"`
	Failures  []skills.LearnedFailure     `json:"failures"`
}

type Metadata 

type Metadata struct {
	Source             string    `json:"source"`
	Ready              bool      `json:"ready"`
	RefreshIntervalSec int       `json:"refresh_interval_sec"`
	LastRefreshTrigger string    `json:"last_refresh_trigger,omitempty"`
	LastConfigEventAt  time.Time `json:"last_config_event_at,omitempty"`
	CollectedAt        time.Time `json:"collected_at,omitempty"`
	NextRefreshAt      time.Time `json:"next_refresh_at,omitempty"`
	LastError          string    `json:"last_error,omitempty"`
	SnapshotPath       string    `json:"snapshot_path,omitempty"`
	ReservedMB         int64     `json:"reserved_mb,omitempty"`
	Version            string    `json:"version,omitempty"`
	CacheAgeSec        int       `json:"cache_age_sec,omitempty"`
	Stale              bool      `json:"stale,omitempty"`
	MeshPeers          int       `json:"mesh_peers,omitempty"`
	StaleThresholdSec  int       `json:"stale_threshold_sec,omitempty"`
	// Phase 3: refresh metrics
	RefreshCount        int64                      `json:"refresh_count"`
	LastRefreshMs       int64                      `json:"last_refresh_duration_ms,omitempty"`
	MaxRefreshLatencyMs int64                      `json:"max_refresh_latency_ms,omitempty"`
	StaleNodes          []string                   `json:"stale_nodes,omitempty"`
	Freshness           *models.DiscoveryFreshness `json:"freshness,omitempty"`
}

func FetchMeta 

func FetchMeta(ctx context.Context, addr string) (Metadata, error)

type RouteDeps 

type RouteDeps struct {
	LoadRuntime          func(context.Context) (*runtimectx.Context, error)
	RunGuarded           func(context.Context, *runtimectx.Context, execution.GuardedExecutionRequest) (execution.GuardedExecutionResult, error)
	ForwardedOriginToken string
}

type RunStreamEmitter added in v0.7.0 

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

RunStreamEmitter emits NDJSON run events and exposes stdout/stderr writers that preserve event framing.

func NewRunStreamEmitter added in v0.7.0 

func NewRunStreamEmitter(w http.ResponseWriter) (*RunStreamEmitter, error)

func (*RunStreamEmitter) Emit added in v0.7.0 

func (e *RunStreamEmitter) Emit(event RunStreamEvent) error

func (*RunStreamEmitter) EmitReady added in v0.7.0 

func (e *RunStreamEmitter) EmitReady(resp execution.GuardedExecutionResult) error

func (*RunStreamEmitter) EmitResult added in v0.7.0 

func (e *RunStreamEmitter) EmitResult(resp execution.GuardedExecutionResult) error

func (*RunStreamEmitter) EmitStateChange added in v0.7.0 

func (e *RunStreamEmitter) EmitStateChange(trigger string, resp execution.GuardedExecutionResult) error

func (*RunStreamEmitter) StderrWriter added in v0.7.0 

func (e *RunStreamEmitter) StderrWriter() io.Writer

func (*RunStreamEmitter) StdoutWriter added in v0.7.0 

func (e *RunStreamEmitter) StdoutWriter() io.Writer

type RunStreamEvent added in v0.7.0 

type RunStreamEvent struct {
	Type    string                            `json:"type"`
	Trigger string                            `json:"trigger,omitempty"`
	Text    string                            `json:"text,omitempty"`
	Result  *execution.GuardedExecutionResult `json:"result,omitempty"`
}

RunStreamEvent is one NDJSON event in the streaming /run contract.

type SnapshotCache 

type SnapshotCache interface {
	Snapshot() (*models.ClusterSnapshot, bool)
	Meta() Metadata
	Invalidate()
	RefreshNow(context.Context) error
	Mesh() *mesh.Mesh
}

type SnapshotChangedFunc added in v0.11.0 

type SnapshotChangedFunc func(snap *models.ClusterSnapshot, trigger string)

SnapshotChangedFunc is invoked by the daemon after a successful refresh whose content hash differs from the previous one for this subscriber. The hook runs synchronously on the refresh goroutine but the daemon intentionally does not hold its write lock during dispatch (callers must be safe to call without holding d.mu).

Implementations should be non-blocking; a slow handler will stall the next refresh tick. The trigger label is one of the RefreshTrigger* constants.

type ToolDef 

type ToolDef struct {
	Name        string         `json:"name"`
	Description string         `json:"description"`
	InputSchema map[string]any `json:"input_schema"`
}

func ToolDefinitions 

func ToolDefinitions() []ToolDef

type ToolsResponse 

type ToolsResponse struct {
	Tools []ToolDef `json:"tools"`
}

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.