scheduler

package
v0.49.3 Latest Latest
Warning

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

 Go to latest Published: Jun 17, 2026 License: AGPL-3.0 Imports: 14 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"
)

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 ErrAlreadySubscribed = errors.New("already subscribed to template")

ErrAlreadySubscribed is returned by Subscribe when the user already has an active subscription instance for the given template key.

View Source 
var ErrSubscriptionMessageReadOnly = errors.New("subscription message is read-only")

ErrSubscriptionMessageReadOnly is returned by UpdateUserJob when the caller attempts to change the message on a subscription instance (job_key non-empty). The message is owned by the template registry, not the row.

View Source 
var ErrTemplateNotFound = errors.New("template not found")

ErrTemplateNotFound is returned by Subscribe when the requested template key is not registered.

View Source 
var RecallyDigestTemplate = JobTemplate{
	Key:         "recally-digest",
	Name:        "recally-digest",
	Description: "Generate and deliver a daily reading digest from your Recally library.",
	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".`,
	DefaultSchedule: Schedule{Every: "24h"},
	SessionMode:     SessionNew,
}

RecallyDigestTemplate is the job template for the daily recally reading digest. Users subscribe to this template instead of receiving it as a broadcast. Registered via (*Service).RegisterTemplate before Start.

View Source 
var RecallyRSSTemplate = JobTemplate{
	Key:         "recally-rss",
	Name:        "recally-rss",
	Description: "Poll your Recally feeds every 6 hours and save new articles.",
	Message: `1. Discover new entries for every enabled feed. Load the recally skill, then dispatch by feed kind:
   - rss: run "stella recally feed poll --limit 20 --json" once (no feed-id polls every enabled feed; non-rss feeds are skipped server-side).
   - twitter / website (and other non-rss kinds): run "stella recally feed list --json", and for each feed whose kind is not "rss", follow that kind's workflow in the recally skill (e.g. the Twitter or website workflow) to list items and push them via "stella recally feed entry add" (Go dedups on guid).
2. For each pending entry (across all feeds), follow the save workflow defined in the recally skill (fetch → generate metadata → save with the entry's source type → 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.`,
	DefaultSchedule: Schedule{Every: "6h"},
	SessionMode:     SessionNew,
}

RecallyRSSTemplate is the job template for recurring recally RSS polling. Users subscribe to this template instead of receiving it as a broadcast. Registered via (*Service).RegisterTemplate before Start.

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
	Schedule    Schedule
	SessionMode string
	AgentID     string
	// Handler is invoked directly when the job fires instead of dispatching
	// through the default OnJob agent path. Required.
	Handler OnJobFunc
}

BuiltinJob defines a handler-mode job that is automatically seeded on scheduler startup. The Handler is invoked directly when the job fires, bypassing the agent dispatch. Used for internal subsystems (e.g. reflect) that run native Go code on a schedule.

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 job executions.

func (Job) UserSessionID 

func (j Job) UserSessionID(userID string) string

UserSessionID returns a user-scoped session ID for user-owned job executions. 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"`
	Output     string     `json:"output,omitempty"`
}

type JobTemplate added in v0.46.0 

type JobTemplate struct {
	// Key is a stable, unique identifier used as job_key on subscription rows.
	Key string
	// Name is the human-readable display name. Must not collide with any
	// registered builtin name — the two registries share a reserved-name check.
	Name string
	// Description is shown to users when browsing available templates.
	Description string
	// Message is the prompt dispatched when a subscription fires.
	Message string
	// DefaultSchedule is the schedule pre-filled when creating a subscription.
	DefaultSchedule Schedule
	// SessionMode is the default session_mode for new subscriptions.
	SessionMode string
}

JobTemplate defines a platform-managed job recipe that users can subscribe to. Unlike a BuiltinJob the template itself is never scheduled directly; instead each subscriber gets an ordinary user-owned sched_job row whose job_key field names the template. The message (prompt) lives here and is resolved at fire time so prompt improvements propagate automatically to all existing subscribers on the next upgrade.

type JobUpdate added in v0.46.0 

type JobUpdate struct {
	Name        *string
	Message     *string
	Schedule    *Schedule
	SessionMode *string
	Enabled     *bool
	AgentID     *string
}

JobUpdate carries optional field overrides for UpdateUserJob. A nil pointer means "leave unchanged".

type OnJobFunc 

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

OnJobFunc is called when a scheduled job fires.

type RunOutputSink added in v0.46.0 

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

RunOutputSink is a context-carried slot the dispatch callback fills with the run's final assistant text so the run record can persist it. Mirrors the WithRunSessionID pattern: the scheduler owns the run lifecycle, the callback owns the agent conversation.

func RunOutputSinkFromContext added in v0.46.0 

func RunOutputSinkFromContext(ctx context.Context) *RunOutputSink

RunOutputSinkFromContext returns the output sink for the current run, or nil when the job was not dispatched through the run lifecycle.

func (*RunOutputSink) Set added in v0.46.0 

func (s *RunOutputSink) Set(text string)

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, and idempotently retires legacy system job rows whose names are in retiredBuiltinNames (cascade deletes their run history too, via the sched_job_run FK ON DELETE CASCADE).

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.

Mutual exclusion with templates: RegisterBuiltin rejects any name that already appears as a template key or name, and RegisterTemplate likewise rejects the reverse. Whichever registers second for a conflicting key/name errors.

func (*Service) RegisterTemplate added in v0.46.0 

func (s *Service) RegisterTemplate(tmpl JobTemplate) error

RegisterTemplate registers a job template before the scheduler is started.

Rules:

  • Key must be unique across all templates.
  • Name must not collide with any registered builtin name (and vice-versa — RegisterBuiltin likewise rejects names that match a template key/name).
  • Must be called BEFORE (*Service).Start.

func (*Service) RemoveJob 

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

RemoveJob unschedules and removes a job.

func (*Service) ResolveTemplateMessage added in v0.46.0 

func (s *Service) ResolveTemplateMessage(key string) (string, bool)

ResolveTemplateMessage returns the message for the given template key. Returns ("", false) when the key is not registered.

func (*Service) RunJobNow 

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

RunJobNow triggers an immediate execution of the given job asynchronously. 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) 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 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.

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.

func (*Service) Subscribe added in v0.46.0 

func (s *Service) Subscribe(ctx context.Context, userID, agentID, key string, schedOverride Schedule) (Job, error)

Subscribe creates a user-owned subscription instance for the given template. A user may have at most one subscription per template; the uniqueness check and the insert + gocron registration happen inside a single critical section because there is no DB unique constraint to fall back on.

schedOverride, when non-zero, replaces the template's DefaultSchedule.

func (*Service) Templates added in v0.46.0 

func (s *Service) Templates() []JobTemplate

Templates returns a snapshot of all registered job templates.

func (*Service) UpdateUserJob added in v0.46.0 

func (s *Service) UpdateUserJob(ctx context.Context, id string, update JobUpdate) (Job, error)

UpdateUserJob merges the provided field overrides into the existing job, persists the result, and reschedules the live gocron entry — all inside a single critical section.

Subscription instances (job_key non-empty) may not have their message changed; that field is owned by the template registry.

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.