Documentation
¶
Overview ¶
Package cron — activity tracking for gateway-aware cron fire dispatch. Cron jobs produce output asynchronously; we hold that output until the user is idle (no agent activity for DefaultIdleWindow), then flush through the gateway dispatcher so the user sees a tidy digest.
Package cron — timezone-aware cron scheduler with Postgres-backed jobs. Wired via daemon.go; busy-gate + output buffer prevent message storms.
Index ¶
- Constants
- Variables
- func SetIdleWindow(d time.Duration)
- type ActivityTracker
- type Config
- type CronExpr
- type ExecutionStyle
- type Job
- type JobStatus
- type JobStore
- type OutputBuffer
- type PostgresJobStore
- func (s *PostgresJobStore) DeleteJob(ctx context.Context, id string) error
- func (s *PostgresJobStore) GetRunLogs(ctx context.Context, jobID string, limit int) ([]RunLog, error)
- func (s *PostgresJobStore) LoadJobs(ctx context.Context) ([]Job, error)
- func (s *PostgresJobStore) LogRun(ctx context.Context, jobID, output, errMsg string, success bool) error
- func (s *PostgresJobStore) SaveJob(ctx context.Context, job *Job) error
- type RunLog
- type Scheduler
- func (s *Scheduler) AddJob(job *Job) error
- func (s *Scheduler) GetJob(id string) (*Job, bool)
- func (s *Scheduler) ListJobs() []*Job
- func (s *Scheduler) NextRunTime(job *Job) (time.Time, error)
- func (s *Scheduler) PauseJob(id string) error
- func (s *Scheduler) RemoveJob(id string) error
- func (s *Scheduler) ResumeJob(id string) error
- func (s *Scheduler) Start()
- func (s *Scheduler) Stop()
Constants ¶
const DefaultIdleWindow = 5 * time.Minute
DefaultIdleWindow is how long the agent must be idle before cron outputs are flushed through the gateway.
Variables ¶
var ( ErrJobNotFound = errors.New("cron: job not found") ErrJobExists = errors.New("cron: job already exists") ErrInvalidJob = errors.New("cron: invalid job") )
var IdleWindow = DefaultIdleWindow
IdleWindow is the active idle window used by the tracker. Defaults to DefaultIdleWindow; override with SetIdleWindow.
Functions ¶
func SetIdleWindow ¶
SetIdleWindow overrides the idle window at runtime.
Types ¶
type ActivityTracker ¶
type ActivityTracker struct {
// contains filtered or unexported fields
}
ActivityTracker records the most recent agent activity timestamp. Cron outputs are held while the user is actively chatting; when the agent goes idle (no Record calls within DefaultIdleWindow), buffered cron outputs are flushed through the gateway.
func NewActivityTracker ¶
func NewActivityTracker() *ActivityTracker
NewActivityTracker returns a tracker seeded with the current time so a freshly-started daemon doesn't immediately flush on the first tick.
func (*ActivityTracker) IdleFor ¶
func (t *ActivityTracker) IdleFor(dur time.Duration) bool
IdleFor returns true when no activity has been recorded for at least dur. Alias kept for compatibility with existing callers; prefer IdleFor.
func (*ActivityTracker) IdleSince ¶
func (t *ActivityTracker) IdleSince() time.Duration
IdleSince returns how long since the last recorded activity.
func (*ActivityTracker) IsIdle ¶
func (t *ActivityTracker) IsIdle(dur time.Duration) bool
IsIdle returns true when no activity has been recorded for at least dur.
func (*ActivityTracker) Record ¶
func (t *ActivityTracker) Record()
Record marks an activity event, resetting the idle timer. Call this from the gateway dispatcher every time a user message is processed (each agent turn).
type Config ¶
type Config struct {
Timezone string
Store JobStore
OnFire func(job *Job) error
// OnRunLog is called after each job execution to persist the run record.
// Receives job, output text (may be empty), error, and success flag.
// Best-effort — errors from this callback are logged but not surfaced.
OnRunLog func(job *Job, output, errMsg string, success bool)
}
type CronExpr ¶
type CronExpr struct {
Minute field
Hour field
Day field
Month field
Weekday field
}
type ExecutionStyle ¶
type ExecutionStyle string
const ( StyleMain ExecutionStyle = "main" StyleIsolated ExecutionStyle = "isolated" StyleCurrent ExecutionStyle = "current" StyleSession ExecutionStyle = "session" )
type Job ¶
type Job struct {
ID string `json:"id"`
Name string `json:"name"`
Schedule string `json:"schedule"`
Timezone string `json:"timezone"`
ExecutionStyle ExecutionStyle `json:"execution_style"`
SessionID string `json:"session_id,omitempty"`
Command string `json:"command"`
Status JobStatus `json:"status"`
LastRun time.Time `json:"last_run"`
NextRun time.Time `json:"next_run"`
RunCount int `json:"run_count"`
FailureCount int `json:"failure_count"`
MaxRetries int `json:"max_retries"`
// BackoffBaseSec is the initial retry delay in seconds (default 60).
// Doubles on each subsequent failure up to BackoffMaxSec.
BackoffBaseSec int `json:"backoff_base_sec"`
// BackoffMaxSec caps the retry delay (default 300).
BackoffMaxSec int `json:"backoff_max_sec"`
CreatedAt time.Time `json:"created_at"`
Metadata map[string]string `json:"metadata"`
// DeliveryTarget restricts delivery of cron output to a specific
// gateway channel. Valid values: "telegram", "slack", "discord",
// "" (default — dispatch to the cron channel only).
DeliveryTarget string `json:"delivery_target"`
// NotifyGateway, when true, dispatches cron output through the
// gateway dispatcher so the agent processes it as an inbound
// message. When false, output goes to shell only (legacy).
NotifyGateway bool `json:"notify_gateway"`
}
type OutputBuffer ¶
type OutputBuffer struct {
// contains filtered or unexported fields
}
OutputBuffer holds queued cron output when the agent is busy. When the agent becomes idle (tracker.IdleFor returns true for the configured window), all queued outputs are flushed via onFlush in FIFO order.
func NewOutputBuffer ¶
func NewOutputBuffer(idleWindow time.Duration, tracker *ActivityTracker, onFlush func(job *Job, output string)) *OutputBuffer
NewOutputBuffer creates a buffer that checks idle status against the given tracker. When idleWindow has passed since the last activity, queued outputs are flushed via onFlush. A background goroutine polls every 30 seconds and auto-flushes when the idle window is met.
func (*OutputBuffer) MaybeFire ¶
func (b *OutputBuffer) MaybeFire(j *Job, output string)
MaybeFire delivers output immediately if the agent is idle, or queues it for later delivery if the agent is busy.
func (*OutputBuffer) Stop ¶
func (b *OutputBuffer) Stop()
Stop shuts down the background flush goroutine. Call once before the buffer goes out of scope.
type PostgresJobStore ¶
type PostgresJobStore struct {
// contains filtered or unexported fields
}
PostgresJobStore persists cron jobs in PostgreSQL.
func NewPostgresJobStore ¶
func NewPostgresJobStore(db *sql.DB) (*PostgresJobStore, error)
func (*PostgresJobStore) DeleteJob ¶
func (s *PostgresJobStore) DeleteJob(ctx context.Context, id string) error
func (*PostgresJobStore) GetRunLogs ¶
func (s *PostgresJobStore) GetRunLogs(ctx context.Context, jobID string, limit int) ([]RunLog, error)
GetRunLogs returns the most recent run log entries for a job, newest first.
func (*PostgresJobStore) LoadJobs ¶
func (s *PostgresJobStore) LoadJobs(ctx context.Context) ([]Job, error)
func (*PostgresJobStore) LogRun ¶
func (s *PostgresJobStore) LogRun(ctx context.Context, jobID, output, errMsg string, success bool) error
LogRun persists a cron job execution record. output may be empty when the onFire callback doesn't produce captured output text (gateway delivery is separate); the error field and success flag tell the operational story.
type RunLog ¶
type RunLog struct {
ID int64 `json:"id"`
JobID string `json:"job_id"`
Output string `json:"output"`
Error string `json:"error"`
Success bool `json:"success"`
RunAt time.Time `json:"run_at"`
}
RunLog represents a single cron job execution record.
type Scheduler ¶
type Scheduler struct {
// contains filtered or unexported fields
}
func NewScheduler ¶
func (*Scheduler) GetJob ¶
GetJob returns a defensive copy of the job so the caller cannot observe torn writes from fireJob mutating NextRun / RunCount / LastRun under the scheduler's lock. Returns (nil, false) when the id is unknown.
func (*Scheduler) ListJobs ¶
ListJobs returns defensive copies of every job. See GetJob for the race rationale — the live *Job pointers are mutated by fireJob under the scheduler's lock; returning the pointers directly lets external readers race against those writes on multi-word fields like time.Time.