Documentation
¶
Overview ¶
Package scheduler provides a cron-based task scheduler for Forge agents.
Index ¶
- Constants
- func CronJobName(agentID, scheduleID string) string
- func CronJobYAML(input CronJobManifestInput) string
- func InCluster() bool
- type AuditFunc
- type Backend
- type CronJobManifestInput
- type CronSchedule
- type FileBackend
- func (b *FileBackend) Delete(ctx context.Context, id string) error
- func (b *FileBackend) Get(ctx context.Context, id string) (*Schedule, error)
- func (b *FileBackend) History(ctx context.Context, scheduleID string, limit int) ([]HistoryEntry, error)
- func (b *FileBackend) List(ctx context.Context) ([]Schedule, error)
- func (b *FileBackend) Reload(ctx context.Context)
- func (b *FileBackend) Set(ctx context.Context, sched Schedule) error
- func (b *FileBackend) Start(ctx context.Context)
- func (b *FileBackend) Stop()
- func (b *FileBackend) Store() ScheduleStore
- func (b *FileBackend) Sync(ctx context.Context, declared []Schedule) error
- type HistoryEntry
- type IntervalSchedule
- type Logger
- type ParsedSchedule
- type Schedule
- type ScheduleStore
- type Scheduler
- type TaskDispatcher
Constants ¶
const ( // SourceYAML marks schedules synced in from forge.yaml's // `schedules[]` block at startup or hot-reload. SourceYAML = "yaml" // SourceLLM marks schedules created at runtime by the LLM via the // schedule_set builtin tool. SourceLLM = "llm" )
Source constants mark schedules by origin so Sync can reconcile declarative state without nuking LLM-set entries.
const ( AuditScheduleFire = "schedule_fire" AuditScheduleComplete = "schedule_complete" AuditScheduleSkip = "schedule_skip" AuditScheduleModify = "schedule_modify" )
Audit event constants for schedule operations.
const DefaultTriggerImage = "curlimages/curl:8.10.1"
DefaultTriggerImage is the curl image the CronJob's trigger container runs by default. Pinned to a specific tag so a registry pull is reproducible.
Variables ¶
This section is empty.
Functions ¶
func CronJobName ¶
CronJobName returns the deterministic K8s resource name for a schedule. K8s resource names are constrained to 63 chars with a restricted character set; we hash-suffix when the natural name would exceed the limit to keep the name unique.
func CronJobYAML ¶
func CronJobYAML(input CronJobManifestInput) string
CronJobYAML returns the apiVersion=batch/v1 CronJob manifest text for a Schedule. Reused by the KubernetesBackend (formatted with Sprintf and passed through yaml.Unmarshal → batchv1.CronJob for the Create/Patch API call) and by the `forge package` build stage in part 3 (written verbatim to the k8s/ directory for `kubectl apply -k`).
The trigger container's args are an A2A JSON-RPC tasks/send body that includes the schedule task as the user message. The cluster substitutes $(...) shell expansions via the curl image's shell; the $(date +%s) generates a unique task ID per fire.
All values are inlined (no Helm-template placeholders) so the manifest is operator-readable and `kubectl diff`-able against the running state.
func InCluster ¶
func InCluster() bool
InCluster reports whether the process appears to be running inside a Kubernetes pod. The signal is the presence of the projected ServiceAccount token at the well-known mount path. Override at test time by setting the FORGE_IN_CLUSTER env var ("true" / "false") — useful for unit tests on developer laptops and for forcing file-backend behavior inside a cluster (e.g. single-replica dev deploys that don't want CronJob CRUD).
Types ¶
type Backend ¶
type Backend interface {
// Start launches any backend-specific goroutines (file backend: the
// 30s ticker). For backends that delegate timing to an external
// system (kubernetes backend: the CronJob controller), Start is a
// no-op. Must be safe to call once per backend instance.
Start(ctx context.Context)
// Stop signals the backend to terminate any goroutines launched by
// Start and waits for them to exit. Idempotent.
Stop()
// Reload re-reads any cached state (file backend: the parsed-cron
// cache). For backends with no cached state (kubernetes backend:
// each operation hits the API), Reload is a no-op.
Reload(ctx context.Context)
// Sync reconciles the backend's state with the declarative list
// of schedules pulled from forge.yaml. Called once at startup
// after Start and again on hot-reload. Existing schedules with
// matching IDs are updated in-place; new ones are added;
// previously-yaml-sourced schedules no longer in the list are
// deleted (LLM-sourced schedules are left alone — they're owned
// by the agent's chat history, not the declarative manifest).
Sync(ctx context.Context, declared []Schedule) error
// List returns every active schedule the backend knows about.
List(ctx context.Context) ([]Schedule, error)
// Get returns a single schedule by ID, or nil when absent.
Get(ctx context.Context, id string) (*Schedule, error)
// Set creates or updates a schedule. Schedule.Source distinguishes
// declarative (forge.yaml) entries from LLM-set ones; backends use
// it to enforce RBAC in the kubernetes case and labeling in both.
Set(ctx context.Context, sched Schedule) error
// Delete removes a schedule by ID. Returns nil when the schedule
// did not exist (idempotent).
Delete(ctx context.Context, id string) error
// History returns recent run records for a schedule (or all
// schedules when scheduleID is empty). File backend reads from
// the SCHEDULES.md history block; kubernetes backend returns
// empty + a logger.Warn deferring to the audit stream (which
// already carries schedule_complete events with status +
// duration).
History(ctx context.Context, scheduleID string, limit int) ([]HistoryEntry, error)
}
Backend abstracts the persistence + timing layer the runner uses for scheduled tasks. Two implementations ship today (#162):
FileBackend: wraps the existing Scheduler ticker + MemoryScheduleStore. Persistence is a markdown file at <WorkDir>/.forge/memory/SCHEDULES.md; timing is a 30s goroutine ticker; overlap is prevented by an in-process map of "currently running" flags.
KubernetesBackend (forge-cli/runtime/scheduler_k8s.go): persists schedules as K8s CronJob resources via client-go. Timing is the cluster's CronJob controller. Overlap is prevented by CronJob.Spec.ConcurrencyPolicy=Forbid (K8s's native equivalent of the file backend's running map).
The Backend interface intentionally bundles "timing concerns" (Start, Stop, Reload) with "persistence concerns" (List, Get, Set, Delete, Sync) into one surface because the two are co-located in the file backend's existing implementation and entirely owned by the cluster in the kubernetes backend. Splitting them would force one or the other backend to implement no-op methods.
type CronJobManifestInput ¶
type CronJobManifestInput struct {
// AgentID is the operator's agent identifier, used as a label and
// as the prefix of the CronJob's name.
AgentID string
// Namespace is the target K8s namespace. Empty defaults to the
// agent pod's own namespace (resolved at runtime from the pod's
// downward API or the in-cluster config).
Namespace string
// ServiceURL is the in-cluster URL CronJob curl requests target.
// Typically `http://<agent-svc>.<ns>.svc:<port>/`.
ServiceURL string
// AuthSecretName is the K8s Secret holding the internal token the
// CronJob mounts and sends as Bearer auth. Defaults to
// `<agent-id>-internal-token` matching `forge auth secret-yaml`.
AuthSecretName string
// TriggerImage is the container image the CronJob runs to make
// the curl request. Defaults to `curlimages/curl:8.10.1`.
TriggerImage string
// Schedule is the Forge schedule entry to materialize.
Schedule Schedule
}
CronJobManifestInput is the data the manifest builders consume. The runtime KubernetesBackend (this file) and the upcoming `forge package` stage (#162 part 3) both feed it; the manifest text + the in-memory CronJob spec must agree byte-for-byte so the runtime's Sync reconciliation doesn't churn against manifest-applied resources.
type CronSchedule ¶
type CronSchedule struct {
Minute bitset
Hour bitset
Dom bitset
Month bitset
Dow bitset
}
CronSchedule implements 5-field cron matching.
type FileBackend ¶
type FileBackend struct {
// contains filtered or unexported fields
}
FileBackend is the default Backend implementation: wraps the existing Scheduler tick loop and ScheduleStore behind the unified Backend interface. Zero behavior change vs the pre-#162 wiring — constructed via NewFileBackend at the runner's existing scheduler init site. The wrapping is structural (delegates everything to the underlying Scheduler / Store), not a reimplementation.
func NewFileBackend ¶
func NewFileBackend(store ScheduleStore, sched *Scheduler) *FileBackend
NewFileBackend constructs a FileBackend wrapping the given store + scheduler. The caller still owns store + scheduler lifecycle outside of Start/Stop on the Backend.
func (*FileBackend) History ¶
func (b *FileBackend) History(ctx context.Context, scheduleID string, limit int) ([]HistoryEntry, error)
func (*FileBackend) Reload ¶
func (b *FileBackend) Reload(ctx context.Context)
func (*FileBackend) Start ¶
func (b *FileBackend) Start(ctx context.Context)
func (*FileBackend) Stop ¶
func (b *FileBackend) Stop()
func (*FileBackend) Store ¶
func (b *FileBackend) Store() ScheduleStore
Store exposes the underlying ScheduleStore for callers (the schedule_* builtin tools registered by the runner) that already speak the ScheduleStore vocabulary. KubernetesBackend exposes a thin adapter over its CronJob CRUD here so the builtin tools work in both modes without per-tool branching.
type HistoryEntry ¶
type HistoryEntry struct {
Timestamp time.Time `json:"timestamp"`
ScheduleID string `json:"schedule_id"`
Status string `json:"status"` // completed, error, skipped
Duration string `json:"duration"`
CorrelationID string `json:"correlation_id"`
Error string `json:"error,omitempty"`
}
HistoryEntry records a single execution of a scheduled task.
type IntervalSchedule ¶
IntervalSchedule fires at fixed intervals.
type Logger ¶
type Logger interface {
Info(msg string, fields map[string]any)
Warn(msg string, fields map[string]any)
Error(msg string, fields map[string]any)
}
Logger is the minimal logging interface used by the scheduler.
type ParsedSchedule ¶
ParsedSchedule computes the next fire time after a given reference.
func Parse ¶
func Parse(expr string) (ParsedSchedule, error)
Parse parses a cron expression and returns a ParsedSchedule. Supported formats:
- Standard 5-field: "minute hour dom month dow"
- Aliases: @hourly, @daily, @weekly, @monthly
- Intervals: @every 5m, @every 1h30m
type Schedule ¶
type Schedule struct {
ID string `json:"id"`
Cron string `json:"cron"`
Task string `json:"task"`
Skill string `json:"skill,omitempty"`
Channel string `json:"channel,omitempty"` // channel adapter name (e.g. "slack", "telegram")
ChannelTarget string `json:"channel_target,omitempty"` // destination ID (channel ID, chat ID)
Source string `json:"source"` // "yaml" or "llm"
Enabled bool `json:"enabled"`
Created time.Time `json:"created"`
LastRun time.Time `json:"last_run,omitempty"`
LastStatus string `json:"last_status,omitempty"` // completed, error, running, skipped
RunCount int `json:"run_count"`
}
Schedule represents a recurring scheduled task.
type ScheduleStore ¶
type ScheduleStore interface {
// List returns all schedules.
List(ctx context.Context) ([]Schedule, error)
// Get returns a single schedule by ID, or nil if not found.
Get(ctx context.Context, id string) (*Schedule, error)
// Set creates or updates a schedule.
Set(ctx context.Context, sched Schedule) error
// Delete removes a schedule by ID.
Delete(ctx context.Context, id string) error
// RecordRun records a history entry for a completed schedule execution.
RecordRun(ctx context.Context, entry HistoryEntry) error
// History returns recent history entries, optionally filtered by schedule ID.
History(ctx context.Context, scheduleID string, limit int) ([]HistoryEntry, error)
}
ScheduleStore defines the persistence interface for schedules and their history.
type Scheduler ¶
type Scheduler struct {
// contains filtered or unexported fields
}
Scheduler runs scheduled tasks on a tick loop.
func New ¶
func New(store ScheduleStore, dispatch TaskDispatcher, logger Logger, audit AuditFunc) *Scheduler
New creates a new Scheduler.