pulse

package
v0.31.27 Latest Latest
Warning

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

 Go to latest Published: Apr 30, 2026 License: MIT Imports: 14 Imported by: 0

Documentation

Overview

Package pulse implements the TARS system-surface watchdog.

Pulse is one half of the system surface (the other being reflection). Every tick (1 minute by default) it deterministically collects signals from cron, agent runtime, ops, and telegram delivery, and — only when some threshold is exceeded — asks an LLM to classify the situation into one of three actions: ignore, notify the user, or run a whitelisted autofix.

Pulse is strictly separated from the user surface: its LLM calls may only invoke the pulse_decide tool, and its Go runtime directly calls internal/ops and related packages rather than going through user-facing tool wrappers. Cross-surface leakage is enforced at Registry.Register time — see internal/tool.RegistryScope.

Index

Constants

View Source 
const PulseDecideToolName = "pulse_decide"

PulseDecideToolName is the fixed tool name the decider LLM must call. It must match the tool registered on the pulse tool Registry in internal/tool. Keep this constant in sync with that registration.

Variables

This section is empty.

Functions

func PulseDecideToolSchema 

func PulseDecideToolSchema() llm.ToolSchema

PulseDecideToolSchema returns the llm.ToolSchema the decider passes to the model. Callers that wire pulse into an HTTP handler reuse this to register the same schema on their pulse-scoped tool Registry.

Types

type Action 

type Action string

Action is the category of response pulse may take for a tick. 

const (
	ActionIgnore  Action = "ignore"
	ActionNotify  Action = "notify"
	ActionAutofix Action = "autofix"
)

func ParseAction 

func ParseAction(s string) (Action, error)

ParseAction validates an action string coming from LLM output. Unknown values return an error so the decider can reject malformed responses rather than silently ignoring them.

type AgentRuntimeRunLister added in v0.31.5 

type AgentRuntimeRunLister interface {
	List(limit int) []agentruntime.Run
}

AgentRuntimeRunLister is the narrow interface pulse requires from the agent runtime to find stuck runs. The real *agentruntime.Runtime satisfies it.

type Config 

type Config struct {
	Enabled     bool
	Interval    time.Duration // default 1m
	Timeout     time.Duration // decider LLM call timeout, default 2m
	ActiveHours string        // "HH:MM-HH:MM" in Timezone, default "00:00-24:00"
	Timezone    string        // IANA name or "Local"; default "Local"
}

Config is the runtime configuration for pulse. It is populated from the main server config and passed once at startup. Changes require a restart — pulse does not watch its config file.

type CronJobLister 

type CronJobLister interface {
	List() ([]cron.Job, error)
}

CronJobLister is the narrow interface pulse requires from a cron store to count failing jobs. The real *cron.Store satisfies it.

type Decider 

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

Decider turns a set of signals into a Decision by calling the LLM. It does not perform any side-effects; notification and autofix execution happen downstream.

Decider resolves its client from the router via RolePulseDecider on every call, so operators can retarget pulse's model at runtime by editing llm_role_pulse_decider in config.

func NewDecider 

func NewDecider(router llm.Router, policy DeciderPolicy) *Decider

NewDecider constructs a Decider bound to an llm.Router and a policy. The router's RolePulseDecider mapping decides which tier (typically light) serves the classification calls.

func (*Decider) Decide 

func (d *Decider) Decide(ctx context.Context, signals []Signal) (Decision, error)

Decide calls the LLM with a signal summary and returns the parsed Decision. Errors from the LLM or malformed tool calls propagate up so the caller can record them on the TickOutcome; the caller is expected to treat decider errors as non-fatal (the tick becomes an ignore).

type DeciderPolicy 

type DeciderPolicy struct {
	AllowedAutofixes []string
	MinSeverity      Severity
}

DeciderPolicy carries the runtime-configured knobs the decider shows to the LLM so it can reason about what autofixes are allowed and what the minimum severity floor is.

type Decision 

type Decision struct {
	Action      Action         `json:"action"`
	Severity    Severity       `json:"severity"`
	Title       string         `json:"title,omitempty"`
	Summary     string         `json:"summary,omitempty"`
	Details     map[string]any `json:"details,omitempty"`
	AutofixName string         `json:"autofix_name,omitempty"`
}

Decision is the LLM's classification for a tick that exceeded signal thresholds. When Action is ActionAutofix, AutofixName must name an autofix in the configured whitelist; otherwise the runtime rejects it.

type DeliveryFailureCounter 

type DeliveryFailureCounter interface {
	FailuresWithin(window time.Duration) int
}

DeliveryFailureCounter is the narrow interface pulse requires from the telegram delivery counter. The real counter in internal/tarsserver satisfies it.

type Dependencies 

type Dependencies struct {
	Scanner   *Scanner
	Decider   *Decider
	Router    *NotifyRouter
	Autofixes *autofix.Registry
	State     *State
}

Dependencies bundles the wired-up collaborators that a Runtime needs. Any field may be nil; a Runtime constructed with nil scanner or nil decider is a no-op runtime, which is convenient for wiring code that hasn't finished building its dependencies yet.

type DiskStatProvider 

type DiskStatProvider interface {
	Status(ctx context.Context) (ops.Status, error)
}

DiskStatProvider is the narrow interface pulse requires from the ops manager to read disk usage. The real *ops.Manager satisfies it.

type Notifier 

type Notifier interface {
	Notify(ctx context.Context, event NotifyEvent) error
}

Notifier is the sink pulse emits NotifyEvents to. The real implementation lives in internal/tarsserver where it has access to the event broker and telegram sender. Tests use a tiny fake.

Notify must not block for long and must not return an error that would abort the tick — the runtime records failure in state but proceeds.

type NotifierFunc 

type NotifierFunc func(ctx context.Context, event NotifyEvent) error

NotifierFunc adapts a plain function to the Notifier interface.

func (NotifierFunc) Notify 

func (f NotifierFunc) Notify(ctx context.Context, event NotifyEvent) error

type NotifyConfig 

type NotifyConfig struct {
	MinSeverity Severity
}

NotifyConfig controls the pulse→notifier filter. Decisions whose severity falls below MinSeverity are dropped entirely so operators can run pulse with a high threshold without spamming the UI.

type NotifyEvent 

type NotifyEvent struct {
	Category  string         `json:"category"`
	Severity  Severity       `json:"severity"`
	Title     string         `json:"title"`
	Message   string         `json:"message"`
	Details   map[string]any `json:"details,omitempty"`
	Timestamp time.Time      `json:"timestamp"`
}

NotifyEvent is a transport-neutral notification payload emitted by pulse. The Notifier implementation decides where to fan it out (session event stream, telegram, etc.).

Keep this struct intentionally small: pulse does not know about delivery channels, user sessions, or chat IDs. It simply asks "please tell the user about this".

type NotifyRouter 

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

NotifyRouter converts Decisions to NotifyEvents and forwards them to a Notifier, honoring the minimum-severity floor. It is stateless aside from its configured dependencies.

func NewNotifyRouter 

func NewNotifyRouter(notifier Notifier, cfg NotifyConfig) *NotifyRouter

NewNotifyRouter constructs a router. A nil notifier is allowed — all calls become no-ops, which is useful for tests and for configurations where the user has disabled pulse notifications.

func (*NotifyRouter) Route 

func (r *NotifyRouter) Route(ctx context.Context, decision Decision) (bool, error)

Route delivers a decision to the notifier if its severity meets the configured floor and its action is ActionNotify. Returns (delivered, err). err is non-nil only when the notifier itself failed.

type ReflectionHealthSource 

type ReflectionHealthSource interface {
	ConsecutiveFailures() int
	LastRunAt() time.Time
}

ReflectionHealthSource is the narrow interface pulse requires from the reflection runtime to observe nightly-run health. The real *reflection.State satisfies it without importing pulse in reverse.

type Runtime 

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

Runtime owns the pulse tick loop. It is single-instance per server; constructing more than one is allowed but uncommon.

func NewRuntime 

func NewRuntime(cfg Config, deps Dependencies) *Runtime

NewRuntime constructs a runtime. Call Start to begin the tick loop.

func (*Runtime) RunOnce 

func (r *Runtime) RunOnce(ctx context.Context) TickOutcome

RunOnce executes a single tick synchronously and returns its outcome. It is safe to call even when the loop is not running; tests use it to exercise the orchestration without waiting for a ticker.

func (*Runtime) SetTickHook 

func (r *Runtime) SetTickHook(fn func(outcome TickOutcome))

SetTickHook installs a callback invoked after every tick with the tick's outcome. It is intended for tests and observability; the hook must not block the loop.

func (*Runtime) Snapshot 

func (r *Runtime) Snapshot() Snapshot

Snapshot returns the current runtime state. Safe to call from any goroutine, including while the loop is running.

func (*Runtime) Start 

func (r *Runtime) Start(ctx context.Context)

Start begins the tick loop in a goroutine. Start is idempotent; calling it twice has no effect. The runtime stops when Stop is called or when the provided parent context is canceled.

A disabled runtime (cfg.Enabled == false) is a no-op: Start returns immediately without launching a goroutine.

func (*Runtime) Stop 

func (r *Runtime) Stop()

Stop signals the tick loop to exit and waits for it to drain. Stop is safe to call multiple times.

type Scanner 

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

Scanner collects Signals from the configured sources. It is stateless and safe to call concurrently from multiple ticks, though in practice the runtime serializes ticks.

func NewScanner 

func NewScanner(sources ScannerSources, thresholds Thresholds) *Scanner

NewScanner constructs a Scanner. Callers typically build one at server startup and reuse it across ticks.

func (*Scanner) Scan 

func (s *Scanner) Scan(ctx context.Context) []Signal

Scan runs each enabled signal source once and returns the resulting signals. Sources that fail are surfaced as a SignalKindInfo with severity Warn — we do not propagate errors, because a single broken source should not block the whole tick.

type ScannerSources 

type ScannerSources struct {
	Cron         CronJobLister
	AgentRuntime AgentRuntimeRunLister
	Ops          DiskStatProvider
	Delivery     DeliveryFailureCounter
	Reflection   ReflectionHealthSource
}

ScannerSources bundles the data sources a Scanner reads from. Any field may be nil; nil sources yield no signals for that domain.

type Severity 

type Severity int

Severity categorizes the urgency of a pulse signal or decision.

The severity ladder mirrors log levels so it can flow into existing notification routing without translation. 

const (
	SeverityInfo Severity = iota
	SeverityWarn
	SeverityError
	SeverityCritical
)

func ParseSeverity 

func ParseSeverity(s string) (Severity, error)

ParseSeverity parses a lowercase severity string. Unknown strings return SeverityInfo and an error — callers should treat unknown values as safe defaults rather than failing hard.

func (Severity) AtLeast 

func (s Severity) AtLeast(min Severity) bool

AtLeast returns true if s meets or exceeds the given minimum severity. Used by notify routing to filter decisions below the configured floor.

func (Severity) String 

func (s Severity) String() string

String returns the lowercase severity name used in config, logs, and API responses.

type Signal 

type Signal struct {
	Kind     SignalKind     `json:"kind"`
	Severity Severity       `json:"severity"`
	Summary  string         `json:"summary"`
	Details  map[string]any `json:"details,omitempty"`
	At       time.Time      `json:"at"`
}

Signal is a single observation made by the signal scanner. A pulse tick collects zero or more Signals; if none exceed their threshold the LLM decider is never invoked.

type SignalKind 

type SignalKind string

SignalKind identifies the domain a signal came from. It is used as a tag in the LLM prompt so the decider can reason about signal origin. 

const (
	SignalKindCronFailures         SignalKind = "cron_failures"
	SignalKindStuckAgentRuntimeRun SignalKind = "stuck_agentruntime_run"
	SignalKindDiskUsage            SignalKind = "disk_usage"
	SignalKindDeliveryFailures     SignalKind = "delivery_failures"
	SignalKindReflectionFailure    SignalKind = "reflection_failure"
)

type Snapshot 

type Snapshot struct {
	LastTickAt     time.Time     `json:"last_tick_at"`
	LastDecision   *Decision     `json:"last_decision,omitempty"`
	LastErr        string        `json:"last_err,omitempty"`
	TotalTicks     int           `json:"total_ticks"`
	TotalSkipped   int           `json:"total_skipped"`
	TotalDecisions int           `json:"total_decisions"`
	TotalAutofixes int           `json:"total_autofixes"`
	TotalNotifies  int           `json:"total_notifies"`
	Recent         []TickOutcome `json:"recent"`
}

Snapshot is a point-in-time view of state safe to serialize to JSON.

type State 

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

State is the in-memory runtime state of a Pulse Runtime. It tracks the last tick, the last decision, and a ring buffer of recent tick outcomes for observability.

State is safe for concurrent use. The API surface is intentionally narrow: writers (the runtime) call RecordTick with a complete outcome, and readers (the HTTP handler and tests) retrieve snapshots.

func NewState 

func NewState(capacity int) *State

NewState creates a state with the given ring buffer capacity. A non- positive capacity falls back to 50 entries.

func (*State) RecordTick 

func (s *State) RecordTick(outcome TickOutcome)

RecordTick appends an outcome, updating counters and last-* fields.

func (*State) Snapshot 

func (s *State) Snapshot() Snapshot

Snapshot returns a copy of the current state.

type Thresholds 

type Thresholds struct {
	// CronConsecutiveFailures — emit when any job's consecutive failures
	// reaches or exceeds this value. 0 = disabled.
	CronConsecutiveFailures int

	// StuckRunMinutes — emit when any agent runtime run has been in Running
	// status for at least this many minutes. 0 = disabled.
	StuckRunMinutes int

	// DiskUsedPercentWarn — emit a warn signal when disk usage percent
	// reaches or exceeds this value. 0 = disabled.
	DiskUsedPercentWarn float64

	// DiskUsedPercentCritical — emit a critical signal above this value.
	// 0 = disabled.
	DiskUsedPercentCritical float64

	// DeliveryFailuresWithinWindow — emit when telegram delivery failures
	// in the last DeliveryFailureWindow duration reach this count.
	// 0 = disabled.
	DeliveryFailuresWithinWindow int

	// DeliveryFailureWindow — rolling window for counting delivery
	// failures. Zero defaults to 10 minutes.
	DeliveryFailureWindow time.Duration

	// ReflectionConsecutiveFailures — emit when the reflection health
	// source reports this many (or more) consecutive nightly failures.
	// 0 = disabled.
	ReflectionConsecutiveFailures int
}

Thresholds controls when signals are emitted. Zero values mean "disabled" for that signal (never emit).

type TickOutcome 

type TickOutcome struct {
	At              time.Time `json:"at"`
	Skipped         bool      `json:"skipped,omitempty"`
	SkipReason      string    `json:"skip_reason,omitempty"`
	Signals         []Signal  `json:"signals,omitempty"`
	DeciderInvoked  bool      `json:"decider_invoked,omitempty"`
	Decision        *Decision `json:"decision,omitempty"`
	AutofixAttempt  string    `json:"autofix_attempt,omitempty"`
	AutofixOK       bool      `json:"autofix_ok,omitempty"`
	AutofixErr      string    `json:"autofix_err,omitempty"`
	NotifyDelivered bool      `json:"notify_delivered,omitempty"`
	Err             string    `json:"err,omitempty"`
}

TickOutcome captures everything that happened in a single pulse tick, suitable for appending to a ring buffer of recent activity.

Source Files

View all Source files

Directories

Path Synopsis
Package autofix holds the whitelist of corrective actions pulse may invoke without further human approval.
 Package autofix holds the whitelist of corrective actions pulse may invoke without further human approval.

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.