schedule

Documentation

Overview

Package schedule provides a native, in-process task scheduler for odek.

It runs agent tasks on a cron schedule from inside a long-lived process (the Telegram bot, the `odek schedule daemon`, or `odek serve`) and delivers each result somewhere (Telegram, stdout, a log file). Running in-process is the whole point: the host process has already resolved its configuration (API key, model, bot token, default chat) into memory, so a scheduled task sees exactly what an interactive one does — no environment inheritance games, no external cron daemon, no container-only behaviour.

The package is deliberately decoupled from the agent and Telegram packages. The firing engine (Scheduler) talks to the rest of odek through two small interfaces, Runner and Deliverer, so it can be unit-tested against fakes and reused by every host process unchanged.

Layout on disk (mirrors the rest of ~/.odek):

~/.odek/schedules.json        job definitions (user-editable, 0600)
~/.odek/schedule-state.json   runtime state: last/next run, status (0600)

Definitions and runtime state are kept in separate files on purpose: the definitions file is something a human edits or the CLI rewrites, while the state file churns on every fire. Keeping them apart means a hand-edit never races with a state write and the definitions file stays diff-clean.

Index

• Constants
• type Deliverer
• type Delivery
• type Job
  • func (j Job) Validate() error
• type Logger
• type NopLogger
  • func (NopLogger) Error(string, ...any)
  • func (NopLogger) Info(string, ...any)
• type Options
• type RunState
• type Runner
• type Schedule
  • func Parse(expr string) (*Schedule, error)
  • func ParseInLocation(expr string, loc *time.Location) (*Schedule, error)
  • func (s *Schedule) Matches(t time.Time) bool
  • func (s *Schedule) Next(after time.Time) time.Time
  • func (s *Schedule) String() string
• type Scheduler
  • func New(store *Store, runner Runner, deliverer Deliverer, opts Options) *Scheduler
  • func (s *Scheduler) Reload()
  • func (s *Scheduler) Run(ctx context.Context) error
  • func (s *Scheduler) Wait()
• type Store
  • func NewStore() (*Store, error)
  • func NewStoreAt(dir string) (*Store, error)
  • func (s *Store) Add(job Job) (Job, error)
  • func (s *Store) Get(id string) (Job, bool, error)
  • func (s *Store) List() ([]Job, error)
  • func (s *Store) LoadState() (map[string]RunState, error)
  • func (s *Store) ModTime() time.Time
  • func (s *Store) Put(job Job) error
  • func (s *Store) Remove(id string) error
  • func (s *Store) SaveState(st RunState) error
  • func (s *Store) SetEnabled(id string, enabled bool) error

Constants

const (
	DeliverTelegram = "telegram" // send via the bot to ChatID (0 = default_chat_id)
	DeliverStdout   = "stdout"   // print to the daemon's stdout
	DeliverLog      = "log"      // append to the schedule run log
)

Delivery kinds. A job's result is routed to exactly one destination.

const (
	StatusOK      = "ok"      // task ran and delivered
	StatusError   = "error"   // task or delivery failed (see LastError)
	StatusSkipped = "skipped" // a due fire was intentionally not run (e.g. missed while down, catchup off)
)

Run-status values recorded in RunState.LastStatus.

Types

type Deliverer

type Deliverer interface {
	Deliver(ctx context.Context, job Job, result string) error
}

Deliverer routes a successful job result to its destination (Telegram chat, stdout, a log file). It is called only when Run succeeded. The context lets a slow delivery (e.g. an unreachable Telegram endpoint) be cancelled on shutdown instead of blocking the drain.

type Delivery

type Delivery struct {
	Kind   string `json:"kind"`              // one of the Deliver* constants
	ChatID int64  `json:"chat_id,omitempty"` // telegram only; 0 = use the configured default_chat_id
}

Delivery describes where a job's result is sent.

type Job

type Job struct {
	ID        string    `json:"id"`                 // stable short id, e.g. "jb-ab12cd"
	Name      string    `json:"name"`               // human-readable label
	Cron      string    `json:"cron"`               // 5-field expression or @macro (see cronexpr.go)
	Task      string    `json:"task"`               // the prompt handed to the agent
	Deliver   Delivery  `json:"deliver"`            // where the result goes
	Enabled   bool      `json:"enabled"`            // disabled jobs are parsed but never fired
	Catchup   bool      `json:"catchup,omitempty"`  // if a fire was missed while the process was down, run once on startup
	Timezone  string    `json:"timezone,omitempty"` // IANA name (e.g. "Europe/Berlin"); "" = scheduler default
	CreatedAt time.Time `json:"created_at"`         // when the job was added
}

Job is a single scheduled agent task. Definitions live in schedules.json. All fields are exported so the CLI layer can construct and mutate jobs directly, matching the convention used by session.Session.

func (Job) Validate

func (j Job) Validate() error

Validate checks that a job is well-formed enough to persist and run: a parseable cron expression, a known delivery kind, a non-empty task, and a loadable timezone if one is set. It does not assign IDs or defaults.

type Logger

type Logger interface {
	Info(msg string, kv ...any)
	Error(msg string, kv ...any)
}

Logger is the minimal logging surface the engine needs, satisfied by the Telegram file logger and by NopLogger. Key/value variadics mirror slog.

type NopLogger

type NopLogger struct{}

NopLogger discards all log output.

func (NopLogger) Error

func (NopLogger) Error(string, ...any)

func (NopLogger) Info

func (NopLogger) Info(string, ...any)

type Options

type Options struct {
	MaxConcurrent int              // max jobs running at once (default 2)
	DefaultTZ     *time.Location   // timezone for jobs with no Timezone set (default UTC)
	Catchup       bool             // global default: run a job once if a fire was missed while down
	ReloadEvery   time.Duration    // how often to poll schedules.json mtime for changes (default 30s)
	RunTimeout    time.Duration    // max wall-clock per job run (default 15m; <=0 keeps the engine default)
	Logger        Logger           // defaults to NopLogger
	Now           func() time.Time // injectable clock for decisions (default time.Now); tests override
}

Options configures a Scheduler. Zero values fall back to sensible defaults.

type RunState

type RunState struct {
	JobID      string    `json:"job_id"`
	LastRun    time.Time `json:"last_run,omitzero"`     // omitzero (not omitempty) — time.Time is a struct
	LastStatus string    `json:"last_status,omitempty"` // one of the Status* constants
	LastError  string    `json:"last_error,omitempty"`  // populated when LastStatus == StatusError
	LastResult string    `json:"last_result,omitempty"` // truncated preview of the delivered text
	NextRun    time.Time `json:"next_run,omitzero"`     // computed projected next fire
	Runs       int       `json:"runs,omitempty"`        // total successful + failed fires
	Sig        string    `json:"sig,omitempty"`         // schedule signature that produced NextRun (cron|tz); detects stale state after a cron edit
}

RunState is the mutable runtime state for one job, persisted in schedule-state.json keyed by Job.ID. It is updated after every fire.

type Runner

type Runner interface {
	Run(ctx context.Context, job Job) (result string, tokens int64, err error)
}

Runner executes one scheduled job's task and returns the agent's final text, the tokens it consumed (for budgeting/telemetry; 0 if unknown), and any error. Implementations live outside this package — the daemon and the Telegram bot each build an agent-backed Runner — so the engine stays decoupled from the agent and is trivially faked in tests.

type Schedule

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

Schedule is a parsed cron expression bound to a timezone. It answers one question — "given an instant, when does this next fire?" — via Next.

Supported syntax is standard 5-field Vixie cron:

┌ minute        0-59
│ ┌ hour        0-23
│ │ ┌ dom        1-31
│ │ │ ┌ month    1-12 or JAN-DEC
│ │ │ │ ┌ dow    0-6 or SUN-SAT (0 and 7 both mean Sunday)
* * * * *

Each field accepts: a wildcard "*", a single value, a range "a-b", a step "*/n" / "a-b/n" / "a/n" (from a to the field max), and comma-separated lists of any of those. Month and day-of-week also accept three-letter names.

Macros: @yearly (@annually), @monthly, @weekly, @daily (@midnight), @hourly.

Day-of-month / day-of-week coupling follows Vixie semantics: when BOTH fields are restricted (neither is "*"), a day matches if EITHER field matches (union). When at least one is a wildcard, the usual intersection applies. This is why "0 0 13 * 5" fires on the 13th OR any Friday, not only Friday-the-13th.

func Parse

func Parse(expr string) (*Schedule, error)

Parse compiles a cron expression in UTC.

func ParseInLocation

func ParseInLocation(expr string, loc *time.Location) (*Schedule, error)

ParseInLocation compiles a cron expression bound to loc. A nil loc defaults to UTC. The location only affects Next/Matches, not parsing.

func (*Schedule) Matches

func (s *Schedule) Matches(t time.Time) bool

Matches reports whether t (in the schedule's location, to the minute) is a firing time.

func (*Schedule) Next

func (s *Schedule) Next(after time.Time) time.Time

Next returns the first firing time strictly after the given instant, or the zero time if none occurs within the search horizon. The result is in the schedule's location and has zero seconds/nanoseconds.

It advances by the coarsest non-matching unit (month → day → hour → minute) so even rare expressions converge in a handful of iterations rather than stepping minute-by-minute across years.

func (*Schedule) String

func (s *Schedule) String() string

String returns the original expression the schedule was parsed from.

type Scheduler

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

Scheduler fires jobs from a Store on their cron schedule, runs them through a Runner, and routes results through a Deliverer. It is safe for a single Run call; do not call Run concurrently on the same Scheduler.

func New

func New(store *Store, runner Runner, deliverer Deliverer, opts Options) *Scheduler

New builds a Scheduler. The store, runner, and deliverer are required.

func (*Scheduler) Reload

func (s *Scheduler) Reload()

Reload asks a running Run loop to re-read job definitions immediately instead of waiting for the next mtime poll — used after an out-of-band edit (e.g. the Telegram `/schedule` commands) so changes take effect at once. Safe to call from any goroutine; if a reload is already pending it coalesces, and if Run isn't active the buffered signal is consumed on the next loop iteration.

func (*Scheduler) Run

func (s *Scheduler) Run(ctx context.Context) error

Run drives the scheduler until ctx is cancelled. On cancellation it stops scheduling new fires and waits for in-flight executions to finish before returning ctx.Err().

func (*Scheduler) Wait

func (s *Scheduler) Wait()

Wait blocks until all in-flight executions complete. Intended for tests.

type Store

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

Store persists schedule definitions and runtime state as two JSON files under a directory (normally ~/.odek). It is a thin, mutex-guarded file manager in the same spirit as session.Store: all Job fields are public, so callers read a Job, mutate it, and write it back.

func NewStore

func NewStore() (*Store, error)

NewStore opens the schedule store rooted at ~/.odek, creating the directory if needed.

func NewStoreAt

func NewStoreAt(dir string) (*Store, error)

NewStoreAt opens the schedule store rooted at dir. Used by tests and by callers that resolve ~/.odek themselves. The directory is created with 0700 permissions so schedule/state filenames are not listable by other local users.

func (*Store) Add

func (s *Store) Add(job Job) (Job, error)

Add validates and appends a job. If job.ID is empty a stable ID is generated; if job.CreatedAt is zero it is stamped with now. The stored job (with ID/CreatedAt filled in) is returned.

func (*Store) Get

func (s *Store) Get(id string) (Job, bool, error)

Get returns the job with the given ID. The bool reports whether it was found.

func (*Store) List

func (s *Store) List() ([]Job, error)

List returns all jobs, sorted by creation time then ID for stable output.

func (*Store) LoadState

func (s *Store) LoadState() (map[string]RunState, error)

LoadState returns runtime state for all jobs, keyed by job ID. A missing state file yields an empty (non-nil) map.

func (*Store) ModTime

func (s *Store) ModTime() time.Time

ModTime returns the last-modified time of the schedules file, or the zero time if it does not exist yet. The engine polls this for cheap hot-reload detection without parsing the file.

func (*Store) Put

func (s *Store) Put(job Job) error

Put upserts a job by ID: it replaces an existing job with the same ID, or appends it if absent. The job is validated first.

func (*Store) Remove

func (s *Store) Remove(id string) error

Remove deletes a job (and its runtime state) by ID. Removing a job that does not exist returns an error so the CLI can report it.

func (*Store) SaveState

func (s *Store) SaveState(st RunState) error

SaveState writes (or replaces) the runtime state for a single job. Other jobs' state is preserved.

func (*Store) SetEnabled

func (s *Store) SetEnabled(id string, enabled bool) error

SetEnabled flips a job's Enabled flag.