scheduler

package
v0.40.0 Latest Latest
Warning

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

 Go to latest Published: Jun 3, 2026 License: AGPL-3.0 Imports: 19 Imported by: 0

Documentation

Index

Constants

View Source 
const (
	// SessionReuse reuses the same session across job executions (default).
	SessionReuse = "reuse"
	// SessionNew creates a fresh session for each execution.
	SessionNew = "new"
)
View Source 
const (
	// ExecScopeSystem runs once with no user context (system workspace).
	ExecScopeSystem = "system"
	// ExecScopeUser runs once with a specific user's context (UserID must be set).
	ExecScopeUser = "user"
	// ExecScopeAllUsers fans out: runs once per active user, each with that user's context.
	ExecScopeAllUsers = "all_users"
)

ExecScope constants define how a job runs at execution time.

View Source 
const (
	JobOwnerUser   = "user"
	JobOwnerPlugin = "plugin"
	JobOwnerSystem = "system"
)

Job is the persisted job definition.

View Source 
const (
	RunStatusRunning = "running"
	RunStatusSuccess = "success"
	RunStatusError   = "error"
)

Variables

View Source 
var RecallyDigestBuiltin = BuiltinJob{
	Name: "recally-digest",
	Message: `Load the recally skill. Then generate and send a daily reading digest:
1. Run "stella recally digest" to get today's digest data.
2. If saved_yesterday_count is 0 AND worth_revisiting_count is 0, stop — do NOT notify and do NOT save.
3. Check the user's language preference in memory. If no language preference is found, write the digest in English.
4. Write a newsletter-style narrative in the selected language. It should read like a personal curator wrote it — not a bullet list or a status report. Cover:
   - A short opening sentence that sets the tone for the day's reading (1–2 sentences).
   - For each article saved yesterday: weave the titles and summaries into 1–2 flowing sentences that explain why each piece is worth reading, grouping thematically where possible.
   - A brief "worth revisiting" section: mention the articles by title and why they might be relevant again now.
   - Close with the trending tags as a loose sentence: "Today's themes span X, Y, and Z."
   Keep the tone warm, curious, and concise — aim for 150–300 words total. No bullet points, no emoji, no section headers.
5. Call notify once with a short 1-sentence preview of the narrative (the opening sentence).
6. Save the full narrative through a private temporary file: tmpfile="$(mktemp -t stella_digest.XXXXXX)"; chmod 600 "$tmpfile"; write the narrative to "$tmpfile"; run stella recally digest save --narrative "$(cat "$tmpfile")"; then rm -f "$tmpfile".`,
	Schedule:    Schedule{Every: "24h"},
	SessionMode: SessionNew,
	ExecScope:   ExecScopeAllUsers,
}

RecallyDigestBuiltin is the spec for the daily recally reading digest builtin job. Registered from gateway wiring via (*Service).RegisterBuiltin.

View Source 
var RecallyRSSBuiltin = BuiltinJob{
	Name: "recally-rss",
	Message: `1. Poll all feeds: run "stella recally feed poll --limit 20 --json" (no feed-id polls every enabled feed).
2. Load the recally skill. For each pending entry, follow the RSS workflow defined in the recally skill (fetch → generate metadata → save → mark).
3. Notify only when at least one article was saved: count articles saved in step 2. If zero, do NOT call the notify tool — stop here. If one or more, call notify once:
   - For each article (up to 5): Worth-Reading label (emoji + text), title, author, and the "# Summary" section from the structured summary.
   - If more than 5 were saved, list the remaining as title + author only.`,
	Schedule:    Schedule{Every: "6h"},
	SessionMode: SessionNew,
	ExecScope:   ExecScopeAllUsers,
}

RecallyRSSBuiltin is the spec for the recurring recally RSS polling builtin job. Registered from gateway wiring via (*Service).RegisterBuiltin.

Functions

func RunSessionIDFromContext 

func RunSessionIDFromContext(ctx context.Context) string

func WithRunSessionID 

func WithRunSessionID(ctx context.Context, sid string) context.Context

Types

type BuiltinJob 

type BuiltinJob struct {
	Name        string
	Message     string
	Schedule    Schedule
	SessionMode string
	AgentID     string
	// ExecScope controls how the job runs: ExecScopeSystem (once, no user context),
	// ExecScopeUser (once for a specific user), or ExecScopeAllUsers (fan-out per active user).
	// Defaults to ExecScopeSystem when empty.
	ExecScope string
	// Handler, when set, is invoked directly when the job fires instead of
	// dispatching the Message through the default OnJob agent path.
	Handler OnJobFunc
}

BuiltinJob defines a job that is automatically seeded on scheduler startup.

A spec must run in exactly one mode:

  • Message mode: Message != "" — the scheduler fires the default OnJob callback, which routes the message through the agent pool. Used for prompt-driven builtins like recally-rss and recally-digest.
  • Handler mode: Handler != nil — the scheduler invokes the Go callback directly, bypassing the agent dispatch. Used for internal subsystems (e.g. reflect) that run native Go code on a schedule.

Setting both, or neither, is rejected at registration time.

type ChatFunc 

type ChatFunc func(ctx context.Context, sessionID, message, model string) <-chan agent.Event

ChatFunc streams runner events for heartbeat decision/execution prompts.

type Decision 

type Decision struct {
	Action string `json:"action"`
	Reason string `json:"reason,omitempty"`
}

Decision is the gate-keeper response from the LLM.

type HeartbeatConfig 

type HeartbeatConfig struct {
	File      string
	FastModel string
}

HeartbeatConfig holds heartbeat-specific settings.

type Job 

type Job struct {
	ID          string         `json:"id"`
	OwnerKind   string         `json:"owner_kind,omitempty"`
	ExecScope   string         `json:"exec_scope,omitempty"`
	PluginID    string         `json:"plugin_id,omitempty"`
	JobKey      string         `json:"job_key,omitempty"`
	RuntimeName string         `json:"runtime_name,omitempty"`
	Name        string         `json:"name"`
	Description string         `json:"description,omitempty"`
	Schedule    Schedule       `json:"schedule"`
	Message     string         `json:"message,omitempty"`
	Payload     map[string]any `json:"payload,omitempty"`
	SessionMode string         `json:"session_mode"` // "reuse" (default) or "new"
	Enabled     bool           `json:"enabled"`
	AgentID     string         `json:"agent_id,omitempty"` // agent to route to (empty = default)
	UserID      string         `json:"user_id,omitempty"`  // user context (empty = none)
	CreatedAt   time.Time      `json:"created_at"`
	UpdatedAt   time.Time      `json:"updated_at"`
	LastRunAt   *time.Time     `json:"last_run_at,omitempty"`
	LastError   string         `json:"last_error,omitempty"`
}

func (Job) SessionID 

func (j Job) SessionID() string

SessionID returns the stable session identifier for system or single-user job executions. For all_users fan-out runs, use UserSessionID instead.

func (Job) UserSessionID 

func (j Job) UserSessionID(userID string) string

UserSessionID returns a user-scoped session ID for all_users fan-out sub-runs. In reuse mode the ID is stable per user; in new mode a timestamp suffix ensures freshness.

type JobRun 

type JobRun struct {
	ID         string     `json:"id"`
	JobID      string     `json:"job_id"`
	SessionID  string     `json:"session_id"`
	UserID     string     `json:"user_id,omitempty"`
	Status     string     `json:"status"`
	StartedAt  time.Time  `json:"started_at"`
	FinishedAt *time.Time `json:"finished_at,omitempty"`
	Error      string     `json:"error,omitempty"`
}

type ListActiveUsersFunc 

type ListActiveUsersFunc func(ctx context.Context) ([]string, error)

ListActiveUsersFunc returns the IDs of all currently active users. Used by the scheduler to fan out ExecScopeAllUsers jobs.

type OnJobFunc 

type OnJobFunc func(ctx context.Context, job Job) error

OnJobFunc is called when a scheduled job fires.

type Schedule 

type Schedule struct {
	Cron  string `json:"cron,omitempty"`  // "0 9 * * 1-5"
	Every string `json:"every,omitempty"` // "30m", "2h"
	At    string `json:"at,omitempty"`    // RFC3339: "2024-01-15T14:30:00+08:00"
}

Schedule defines when a job runs. Exactly one field must be set.

type Service 

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

Service manages scheduled jobs backed by gocron/v2 with database persistence.

func New 

func New(db *sql.DB) (*Service, error)

New creates a scheduler service backed by the given database. Call Start to load persisted jobs and begin scheduling.

func NewFromPath 

func NewFromPath(dbPath string) (*Service, error)

NewFromPath creates a scheduler service that opens its own SQLite database at the given path. The database is closed when Stop is called.

func (*Service) AddJobForContext 

func (s *Service) AddJobForContext(ctx context.Context, name, message string, sched Schedule, sessionMode string) (Job, error)

AddJobForContext creates a user-owned job bound to the current execution context.

func (*Service) AddJobWithOwner 

func (s *Service) AddJobWithOwner(name, message string, sched Schedule, sessionMode, agentID string, userID string) (Job, error)

AddJobWithOwner creates a user-owned job with explicit owner parameters.

func (*Service) AddOnJobListener 

func (s *Service) AddOnJobListener(fn OnJobFunc)

AddOnJobListener appends an additional callback invoked when a job fires.

func (*Service) AddPluginJob 

func (s *Service) AddPluginJob(ctx context.Context, pluginID, key, runtimeName, name, description string, sched Schedule, payload map[string]any) (Job, error)

AddPluginJob creates, persists, and schedules a plugin-owned job.

func (*Service) EnsureBuiltinJobs 

func (s *Service) EnsureBuiltinJobs()

EnsureBuiltinJobs creates or updates all registered builtin jobs. For ExecScopeAllUsers jobs the scheduler fans out to all active users at execution time.

func (*Service) EnsureJob 

func (s *Service) EnsureJob(name, message string, sched Schedule, sessionMode, agentID, execScope string) (Job, error)

EnsureJob creates a job if no job with the same name exists, or updates the existing job when any field has changed. It is intended for builtin jobs that should be seeded on startup. Jobs created by EnsureJob are owned by the system (JobOwnerSystem).

func (*Service) ListJobRuns 

func (s *Service) ListJobRuns(ctx context.Context, jobID string, limit int) ([]JobRun, error)

ListJobRuns returns recent runs for a job. ListJobRuns returns the latest `limit` runs for a job (first page only, across all users). HTTP callers paginate via the handler's direct query instead; this accessor is for internal callers that just want recent runs.

func (*Service) ListJobs 

func (s *Service) ListJobs() []Job

ListJobs returns all jobs.

func (*Service) RegisterBuiltin added in v0.38.0 

func (s *Service) RegisterBuiltin(job BuiltinJob) error

RegisterBuiltin registers a builtin job spec on this Service instance.

Called from gateway wiring during setup. Returns an error on a malformed spec so the gateway can fail startup cleanly. Must be called BEFORE (*Service).Start — handler-mode dispatch is keyed on Name, so persisted jobs loaded by Start can only be routed correctly if the handler is already registered.

func (*Service) RemoveJob 

func (s *Service) RemoveJob(id string) error

RemoveJob unschedules and removes a job.

func (*Service) RunJobNow 

func (s *Service) RunJobNow(ctx context.Context, jobID string) (string, error)

RunJobNow triggers an immediate execution of the given job asynchronously. For ExecScopeAllUsers jobs all user sub-runs are launched; no run ID is returned. For other scopes, returns the run ID of the newly created run record. Returns errJobAlreadyRunning (wrapped) if a run is already active for the job.

func (*Service) ScheduleEvery 

func (s *Service) ScheduleEvery(ctx context.Context, every string, fn TaskFunc) error

ScheduleEvery registers a non-persisted recurring task on the existing scheduler.

func (*Service) SetHeartbeat 

func (s *Service) SetHeartbeat(cfg HeartbeatConfig, chat ChatFunc, notifier notify.Notifier)

SetHeartbeat configures the heartbeat on the scheduler service.

func (*Service) SetListActiveUsersFunc 

func (s *Service) SetListActiveUsersFunc(fn ListActiveUsersFunc)

SetListActiveUsersFunc registers the function used to enumerate active users when fanning out ExecScopeAllUsers jobs.

func (*Service) SetOnJob 

func (s *Service) SetOnJob(fn OnJobFunc)

SetOnJob sets the primary callback invoked when a job fires.

func (*Service) SetUserJobsEnabled 

func (s *Service) SetUserJobsEnabled(enabled bool)

SetUserJobsEnabled controls whether persisted user-owned and all_users scheduler jobs are loaded.

func (*Service) Start 

func (s *Service) Start(ctx context.Context) error

Start loads persisted jobs and starts the scheduler.

func (*Service) StartEphemeral 

func (s *Service) StartEphemeral(ctx context.Context) error

StartEphemeral starts the shared scheduler without loading persisted jobs. Use this when the scheduler is only needed for internal tasks such as heartbeat.

func (*Service) StartHeartbeat 

func (s *Service) StartHeartbeat(ctx context.Context, every string) error

StartHeartbeat schedules the heartbeat poll on the shared scheduler.

func (*Service) Stop 

func (s *Service) Stop() error

Stop shuts down the scheduler and closes the database if owned.

Service is single-shot: Start after Stop is not supported. gocron's Shutdown is terminal, and s.started is not reset, so any subsequent RegisterBuiltin will reject with "called after Start". Construct a new Service if you need a fresh lifecycle.

type TaskFunc 

type TaskFunc func(ctx context.Context)

TaskFunc is a lightweight scheduled callback that is not persisted as a scheduled job.

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.