scheduler

package

Go to main page

Versions in this module

v1

v1.117.2

May 15, 2026

v1.117.1

May 15, 2026

v1.117.0

May 15, 2026

v1.116.0

May 15, 2026

v1.115.0

May 12, 2026

v1.114.0

May 11, 2026

v1.113.1

May 8, 2026

v1.113.0

May 8, 2026

v1.112.2

May 8, 2026

v1.112.1

May 5, 2026

v1.112.0

May 5, 2026

Changes in this version

type ActionType

+ const ActionAgentResume

+ const ActionParkPoll

type CLIBridge

+ NotifyParkComplete func(ctx context.Context, token, outcome, detail string) error

+ RunHTTPProbe func(ctx context.Context, url, method string, headers map[string]string, ...) (status int, body string, err error)

type ExecEnv

+ Enqueue func(ctx context.Context, job *Job) (*Job, error)

v1.111.1

Apr 30, 2026

v1.111.0

Apr 29, 2026

v1.110.2

Apr 28, 2026

v1.110.1

Apr 27, 2026

Changes in this version

type ExecutionResult

+ CycleNum int

type Job

+ CycleCount int

v1.110.0

Apr 27, 2026

v1.109.2

Apr 26, 2026

v1.109.1

Apr 25, 2026

Changes in this version

type EvalEnv

+ DangerousConfirmed bool

type JobSummary

+ DangerousConfirmed bool

+ type PruneFilter struct

+ NameSubstr string

+ OlderThan time.Duration

+ Owner *Owner

+ Statuses []JobStatus

type Scheduler

+ func (s *Scheduler) Prune(filter PruneFilter) ([]JobID, error)

v1.109.0

Apr 24, 2026

Changes in this version

+ const SchemaVersion

+ var ErrActionDisallowed = errors.New("scheduler: action type not in allowlist")

+ var ErrBreakerOpen = errors.New("scheduler: circuit breaker open")

+ var ErrDAGCycle = errors.New("scheduler: dag cycle")

+ var ErrDaemonRunning = errors.New("scheduler: daemon already running")

+ var ErrDangerousShell = errors.New("scheduler: dangerous shell command requires explicit confirmation")

+ var ErrDuplicateName = errors.New("scheduler: duplicate name")

+ var ErrIPCProtocol = errors.New("scheduler: ipc protocol error")

+ var ErrInvalidAction = errors.New("scheduler: invalid action")

+ var ErrInvalidCondition = errors.New("scheduler: invalid condition")

+ var ErrInvalidJob = errors.New("scheduler: invalid job")

+ var ErrInvalidSchedule = errors.New("scheduler: invalid schedule")

+ var ErrJobNotFound = errors.New("scheduler: job not found")

+ var ErrJobTerminal = errors.New("scheduler: job is terminal")

+ var ErrNoDaemon = errors.New("scheduler: no daemon running")

+ var ErrNotAuthorized = errors.New("scheduler: not authorized")

+ var ErrNotStarted = errors.New("scheduler: not started")

+ var ErrQueueFull = errors.New("scheduler: queue full")

+ var ErrRateLimited = errors.New("scheduler: rate limited")

+ var ErrSchedulerClosed = errors.New("scheduler: closed")

+ var ErrSchedulerDraining = errors.New("scheduler: draining")

+ var ErrShellPolicyAsk = errors.New(...)

+ var ErrShellPolicyDeny = errors.New("scheduler: shell command denied by policy")

+ var ErrWALCorrupted = errors.New("scheduler: wal record corrupted")

+ var ErrWaitTimeout = errors.New("scheduler: wait condition timed out")

+ func CheckDaemon(socketPath string) error

+ func DefaultSocketPath(cfg Config) string

+ func NewAuditFileWriter(dir string, maxSizeMB, maxBackups, maxAgeDays int, logger *zap.Logger, ...) auditWriter

+ func NopAuditWriter() auditWriter

+ func RenderList(summaries []JobSummary) string

+ func RenderShow(j *Job) string

+ func RenderTree(jobs []*Job) string

+ func StatusBadge(s JobStatus) string

+ func StatusLine(summaries []JobSummary) string

+ type Action struct

+ Payload map[string]any

+ Type ActionType

+ func ParseActionDSL(input string) (Action, error)

+ func ParseActionJSON(raw string) (Action, error)

+ func (a Action) PayloadBool(key string, def bool) bool

+ func (a Action) PayloadInt(key string, def int) int

+ func (a Action) PayloadMap(key string) map[string]any

+ func (a Action) PayloadString(key, def string) string

+ func (a Action) PayloadStringSlice(key string) []string

+ func (a Action) Validate() error

+ type ActionExecutor interface

+ Execute func(ctx context.Context, action Action, env *ExecEnv) ActionResult

+ Type func() ActionType

+ ValidateSpec func(payload map[string]any) error

+ type ActionRegistry struct

+ func NewActionRegistry() *ActionRegistry

+ func (r *ActionRegistry) Get(t ActionType) (ActionExecutor, bool)

+ func (r *ActionRegistry) MustRegister(e ActionExecutor)

+ func (r *ActionRegistry) Register(e ActionExecutor) error

+ func (r *ActionRegistry) Types() []ActionType

+ type ActionResult struct

+ Cost float64

+ Err error

+ Output string

+ Tokens int

+ Transient bool

+ type ActionType string

+ const ActionAgentTask

+ const ActionHook

+ const ActionLLMPrompt

+ const ActionNoop

+ const ActionShell

+ const ActionSlashCmd

+ const ActionWebhook

+ const ActionWorkerDispatch

+ type Breaker struct

+ func (b *Breaker) Acquire() (release func(success bool), err error)

+ func (b *Breaker) Key() string

+ func (b *Breaker) State() BreakerState

+ type BreakerConfig struct

+ Cooldown time.Duration

+ FailureThreshold int

+ HalfOpenSuccessRequired int

+ Window time.Duration

+ type BreakerState int32

+ const BreakerClosed

+ const BreakerHalfOpen

+ const BreakerOpen

+ func (s BreakerState) String() string

+ type Budget struct

+ ActionTimeout time.Duration

+ BackoffInitial time.Duration

+ BackoffJitter float64

+ BackoffMax time.Duration

+ BackoffMult float64

+ MaxPolls int

+ MaxRetries int

+ PollInterval time.Duration

+ WaitTimeout time.Duration

+ func DefaultBackoff() Budget

+ func (b Budget) Merge(other Budget) Budget

+ type CLIBridge interface

+ AppendHistory func(msg models.Message)

+ ClassifyShellCommand func(cmd string) ShellPolicy

+ DispatchWorker func(ctx context.Context, agentType, task string) (string, error)

+ DockerSocketPath func() string

+ ExecuteSlashCommand func(ctx context.Context, line string) (output string, exit bool, err error)

+ FireHook func(event hooks.HookEvent) *hooks.HookResult

+ KubeconfigPath func() string

+ LLMClient func() client.LLMClient

+ PublishEvent func(evt Event)

+ RunAgentTask func(ctx context.Context, task, systemHint string) (string, error)

+ RunShell func(ctx context.Context, cmd string, envOverrides map[string]string, ...) (stdout string, stderr string, exitCode int, err error)

+ SendLLMPrompt func(ctx context.Context, system, prompt string, maxTokens int) (text string, tokens int, cost float64, err error)

+ WorkspaceDir func() string

+ func NewNoopBridge() CLIBridge

+ type Condition struct

+ Children []Condition

+ Negate bool

+ Spec map[string]any

+ Type string

+ func ParseConditionDSL(input string) (Condition, error)

+ func ParseConditionJSON(raw string) (Condition, error)

+ func (c Condition) SpecBool(key string, def bool) bool

+ func (c Condition) SpecDuration(key string, def time.Duration) time.Duration

+ func (c Condition) SpecInt(key string, def int) int

+ func (c Condition) SpecString(key, def string) string

+ func (c Condition) Validate() error

+ type ConditionEvaluator interface

+ Evaluate func(ctx context.Context, cond Condition, env *EvalEnv) EvalOutcome

+ Type func() string

+ ValidateSpec func(spec map[string]any) error

+ type ConditionRegistry struct

+ func NewConditionRegistry() *ConditionRegistry

+ func (r *ConditionRegistry) Get(t string) (ConditionEvaluator, bool)

+ func (r *ConditionRegistry) MustRegister(e ConditionEvaluator)

+ func (r *ConditionRegistry) Register(e ConditionEvaluator) error

+ func (r *ConditionRegistry) Types() []string

+ type Config struct

+ ActionAllowlist map[ActionType]bool

+ AllowAgents bool

+ AuditEnabled bool

+ AuditMaxAgeDays int

+ AuditMaxBackups int

+ AuditMaxSizeMB int

+ BreakerConfig BreakerConfig

+ DaemonAutoConnect bool

+ DaemonSocket string

+ DataDir string

+ DefaultActionTimeout time.Duration

+ DefaultBackoffInitial time.Duration

+ DefaultBackoffJitter float64

+ DefaultBackoffMax time.Duration

+ DefaultBackoffMult float64

+ DefaultMaxPolls int

+ DefaultMaxRetries int

+ DefaultPollInterval time.Duration

+ DefaultTTL time.Duration

+ DefaultWaitTimeout time.Duration

+ Enabled bool

+ HistoryLimit int

+ MaxJobs int

+ RateLimitGlobalBurst int

+ RateLimitGlobalRPS float64

+ RateLimitOwnerBurst int

+ RateLimitOwnerRPS float64

+ SnapshotInterval time.Duration

+ WALGCInterval time.Duration

+ WaitWorkerCount int

+ WorkerCount int

+ func DefaultConfig() Config

+ func LoadConfigFromEnv() Config

+ type Daemon struct

+ func NewDaemon(s *Scheduler, socketPath string, logger *zap.Logger) *Daemon

+ func (d *Daemon) Run(ctx context.Context) error

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

+ func (d *Daemon) Stop(drainTimeout time.Duration) error

+ type DaemonStats struct

+ Breakers map[string]BreakerState

+ Connections int

+ JobsActive int

+ QueueDepth int

+ Started time.Time

+ Uptime string

+ Version string

+ WALSegments int

+ type EvalEnv struct

+ Bridge CLIBridge

+ Logger *zap.Logger

+ type EvalOutcome struct

+ Details string

+ Err error

+ Satisfied bool

+ Transient bool

+ type Event struct

+ Data map[string]any

+ Execution *ExecutionResult

+ JobID JobID

+ Message string

+ Name string

+ Outcome Outcome

+ Owner Owner

+ Status JobStatus

+ Timestamp time.Time

+ Type EventType

+ func NewEvent(t EventType) Event

+ func (e Event) WithData(k string, v any) Event

+ func (e Event) WithExecution(r ExecutionResult) Event

+ func (e Event) WithJob(j *Job) Event

+ func (e Event) WithMessage(msg string) Event

+ type EventType string

+ const EventBreakerClosed

+ const EventBreakerHalfOpen

+ const EventBreakerOpened

+ const EventDaemonStarted

+ const EventDaemonStopped

+ const EventJobCancelled

+ const EventJobCompleted

+ const EventJobCreated

+ const EventJobDependency

+ const EventJobFailed

+ const EventJobFired

+ const EventJobPaused

+ const EventJobResumed

+ const EventJobRetryQueued

+ const EventJobRunning

+ const EventJobScheduled

+ const EventJobSkipped

+ const EventJobTimedOut

+ const EventJobWaitSatisfied

+ const EventJobWaitStarted

+ const EventJobWaitTick

+ type ExecEnv struct

+ Bridge CLIBridge

+ Job JobSummary

+ Logger *zap.Logger

+ type ExecutionResult struct

+ AttemptNum int

+ ConditionDetails string

+ ConditionSatisfied bool

+ Cost float64

+ Duration time.Duration

+ Error string

+ FinishedAt time.Time

+ Outcome Outcome

+ Output string

+ StartedAt time.Time

+ Tokens int

+ func (r ExecutionResult) Succeeded() bool

+ type Frame struct

+ Error string

+ Event *Event

+ ID string

+ Input json.RawMessage

+ Kind FrameKind

+ Owner *Owner

+ Result json.RawMessage

+ TS time.Time

+ type FrameKind string

+ const KindBye

+ const KindCancel

+ const KindEnqueue

+ const KindError

+ const KindEvent

+ const KindList

+ const KindOK

+ const KindPause

+ const KindPing

+ const KindQuery

+ const KindResume

+ const KindSnapshot

+ const KindStats

+ const KindSubscribe

+ type Job struct

+ Action Action

+ Attempts int

+ Budget Budget

+ CancelReason string

+ CreatedAt time.Time

+ DangerousConfirmed bool

+ DependsOn []JobID

+ Description string

+ FinishedAt time.Time

+ History []ExecutionResult

+ HistoryLimit int

+ ID JobID

+ LastResult *ExecutionResult

+ Name string

+ NextFireAt time.Time

+ Owner Owner

+ ParentID JobID

+ PauseReason string

+ Schedule Schedule

+ Status JobStatus

+ TTL time.Duration

+ Tags map[string]string

+ Transitions []TransitionReason

+ Triggers []JobID

+ UpdatedAt time.Time

+ Version int

+ Wait *WaitSpec

+ func NewJob(name string, owner Owner, sched Schedule, action Action) *Job

+ func (j *Job) Clone() *Job

+ func (j *Job) IsExpired(now time.Time) bool

+ func (j *Job) Summary() JobSummary

+ func (j *Job) Validate() error

+ type JobID string

+ func DeriveJobID(name string, owner Owner, nonce string) JobID

+ func NewJobID() JobID

+ func (j JobID) IsZero() bool

+ func (j JobID) String() string

+ type JobStatus string

+ const StatusBlocked

+ const StatusCancelled

+ const StatusCompleted

+ const StatusFailed

+ const StatusPaused

+ const StatusPending

+ const StatusRunning

+ const StatusSkipped

+ const StatusTimedOut

+ const StatusWaiting

+ func (s JobStatus) IsActive() bool

+ func (s JobStatus) IsTerminal() bool

+ type JobSummary struct

+ Attempts int

+ CreatedAt time.Time

+ Description string

+ ID JobID

+ LastOutcome Outcome

+ Name string

+ NextFireAt time.Time

+ Owner Owner

+ Status JobStatus

+ Tags []string

+ Type string

+ UpdatedAt time.Time

+ type ListFilter struct

+ IncludeTerminal bool

+ NameSubstr string

+ Owner *Owner

+ Statuses []JobStatus

+ Tag string

+ type Metrics struct

+ ActionDuration *prometheus.HistogramVec

+ ActiveJobs prometheus.Gauge

+ AuditWrites prometheus.Counter

+ BreakerState *prometheus.GaugeVec

+ DaemonConnections prometheus.Gauge

+ EnqueueErrors *prometheus.CounterVec

+ JobsCreated *prometheus.CounterVec

+ JobsFired *prometheus.CounterVec

+ QueueDepth prometheus.Gauge

+ RetryCount *prometheus.CounterVec

+ WALSegments prometheus.Gauge

+ WaitChecks *prometheus.CounterVec

+ WaitDuration *prometheus.HistogramVec

+ func GetMetrics() *Metrics

+ type MissPolicy string

+ const MissFireAll

+ const MissFireOnce

+ const MissSkip

+ type Outcome string

+ const OutcomeBreakerOff

+ const OutcomeCancelled

+ const OutcomeFailed

+ const OutcomeSkipped

+ const OutcomeSuccess

+ const OutcomeTimeout

+ type Owner struct

+ ID string

+ Kind OwnerKind

+ Tag string

+ func (o Owner) Equal(other Owner) bool

+ func (o Owner) String() string

+ type OwnerKind string

+ const OwnerAgent

+ const OwnerHook

+ const OwnerSystem

+ const OwnerUser

+ const OwnerWorker

+ type RemoteClient struct

+ func Dial(socketPath string) (*RemoteClient, error)

+ func (c *RemoteClient) Cancel(ctx context.Context, owner Owner, id JobID, reason string) error

+ func (c *RemoteClient) Close() error

+ func (c *RemoteClient) Enqueue(ctx context.Context, owner Owner, in ToolInput) (*ToolOutput, error)

+ func (c *RemoteClient) List(ctx context.Context, filter ListFilter) ([]JobSummary, error)

+ func (c *RemoteClient) Pause(ctx context.Context, owner Owner, id JobID, reason string) error

+ func (c *RemoteClient) Ping(ctx context.Context) error

+ func (c *RemoteClient) Query(ctx context.Context, id JobID) (*Job, error)

+ func (c *RemoteClient) Resume(ctx context.Context, owner Owner, id JobID) error

+ func (c *RemoteClient) Snapshot(ctx context.Context) error

+ func (c *RemoteClient) Stats(ctx context.Context) (DaemonStats, error)

+ func (c *RemoteClient) Subscribe(ctx context.Context) (<-chan Event, error)

+ type Schedule struct

+ Cron string

+ ExactTime time.Time

+ Interval time.Duration

+ Kind ScheduleKind

+ MissPolicy MissPolicy

+ Relative time.Duration

+ Timezone string

+ func ParseScheduleDSL(input string) (Schedule, error)

+ func (s *Schedule) IsRecurring() bool

+ func (s *Schedule) Next(after, createdAt time.Time) time.Time

+ func (s *Schedule) Validate() error

+ type ScheduleKind string

+ const ScheduleAbsolute

+ const ScheduleCron

+ const ScheduleInterval

+ const ScheduleManual

+ const ScheduleOnCondition

+ const ScheduleRelative

+ type Scheduler struct

+ func New(cfg Config, bridge CLIBridge, deps SchedulerDeps, logger *zap.Logger) (*Scheduler, error)

+ func (s *Scheduler) Actions() *ActionRegistry

+ func (s *Scheduler) Bridge() CLIBridge

+ func (s *Scheduler) Cancel(id JobID, reason string, requester Owner) error

+ func (s *Scheduler) Conditions() *ConditionRegistry

+ func (s *Scheduler) Config() Config

+ func (s *Scheduler) DrainAndShutdown(timeout time.Duration)

+ func (s *Scheduler) Enqueue(ctx context.Context, job *Job) (*Job, error)

+ func (s *Scheduler) FindByName(name string) (*Job, bool)

+ func (s *Scheduler) IsClosed() bool

+ func (s *Scheduler) List(filter ListFilter) []JobSummary

+ func (s *Scheduler) Metrics() *Metrics

+ func (s *Scheduler) Pause(id JobID, reason string, requester Owner) error

+ func (s *Scheduler) Query(id JobID) (*Job, error)

+ func (s *Scheduler) Resume(id JobID, requester Owner) error

+ func (s *Scheduler) SetBridge(b CLIBridge)

+ func (s *Scheduler) Snapshot() error

+ func (s *Scheduler) Start(ctx context.Context) error

+ func (s *Scheduler) Version() uint64

+ type SchedulerDeps struct

+ EventBus *bus.MessageBus

+ Hooks *hooks.Manager

+ type ShellPolicy int

+ const ShellPolicyAllow

+ const ShellPolicyAsk

+ const ShellPolicyDeny

+ func (p ShellPolicy) IsTerminal() bool

+ func (p ShellPolicy) String() string

+ type Suggestion struct

+ Description string

+ Text string

+ func CommandSuggestions(command string) []Suggestion

+ func JobsListFlags() []Suggestion

+ type TimeoutBehavior string

+ const TimeoutFail

+ const TimeoutFallback

+ const TimeoutFireAnyway

+ type ToolAdapter struct

+ func NewToolAdapter(s *Scheduler) *ToolAdapter

+ func (t *ToolAdapter) CancelJob(_ context.Context, owner Owner, rawIn string) (string, error)

+ func (t *ToolAdapter) ListJobs(_ context.Context, owner Owner, rawIn string) (string, error)

+ func (t *ToolAdapter) QueryJob(_ context.Context, _ Owner, rawIn string) (string, error)

+ func (t *ToolAdapter) ScheduleJob(ctx context.Context, owner Owner, rawIn string) (string, error)

+ func (t *ToolAdapter) WaitUntil(ctx context.Context, owner Owner, rawIn string) (string, error)

+ type ToolInput struct

+ Action *Action

+ Async bool

+ DependsOn []string

+ Description string

+ Do string

+ Filter ListFilter

+ ID string

+ IKnow bool

+ MaxPolls int

+ MaxRetries int

+ Name string

+ OnTimeout string

+ PollInterval string

+ Reason string

+ Schedule *Schedule

+ TTL string

+ Tags map[string]string

+ Timeout string

+ Triggers []string

+ Until string

+ Wait *WaitSpec

+ WaitTimeout string

+ When string

+ type ToolOutput struct

+ Details string

+ Error string

+ Job *Job

+ JobID JobID

+ Jobs []JobSummary

+ OK bool

+ Outcome Outcome

+ Status JobStatus

+ Summary *JobSummary

+ type TransitionReason struct

+ At time.Time

+ From JobStatus

+ Message string

+ To JobStatus

+ type WaitSpec struct

+ Condition Condition

+ Fallback *Action

+ OnTimeout TimeoutBehavior

+ func (w *WaitSpec) Validate() error

?	: This menu
/	: Search site
f or F	: Jump to
y or Y	: Canonical URL