284 lines
7.8 KiB
Go
284 lines
7.8 KiB
Go
// Package scheduler runs the task polling loop. It checks for due tasks
|
|
// every 30 seconds, creates service channels, and dispatches execution.
|
|
//
|
|
// v0.27.1: Foundation — scheduler loop + service channel creation.
|
|
// v0.27.2: Adds completion invocation, budget enforcement, and wall-clock timeout.
|
|
package scheduler
|
|
|
|
import (
|
|
"context"
|
|
"log"
|
|
"time"
|
|
|
|
"git.gobha.me/xcaliber/chat-switchboard/models"
|
|
"git.gobha.me/xcaliber/chat-switchboard/store"
|
|
)
|
|
|
|
// Scheduler polls for due tasks and dispatches execution.
|
|
type Scheduler struct {
|
|
stores store.Stores
|
|
interval time.Duration
|
|
stop chan struct{}
|
|
running bool
|
|
}
|
|
|
|
// New creates a task scheduler. Call Run() in a goroutine to start.
|
|
func New(stores store.Stores) *Scheduler {
|
|
return &Scheduler{
|
|
stores: stores,
|
|
interval: 30 * time.Second,
|
|
stop: make(chan struct{}),
|
|
}
|
|
}
|
|
|
|
// Run starts the scheduler loop. Blocks until Stop() is called.
|
|
func (s *Scheduler) Run() {
|
|
if s.stores.Tasks == nil {
|
|
log.Println("[scheduler] TaskStore not available — scheduler disabled")
|
|
return
|
|
}
|
|
s.running = true
|
|
log.Printf("[scheduler] Started (poll interval: %s)", s.interval)
|
|
|
|
ticker := time.NewTicker(s.interval)
|
|
defer ticker.Stop()
|
|
|
|
for {
|
|
select {
|
|
case <-ticker.C:
|
|
s.poll()
|
|
case <-s.stop:
|
|
log.Println("[scheduler] Stopped")
|
|
return
|
|
}
|
|
}
|
|
}
|
|
|
|
// Stop signals the scheduler to exit.
|
|
func (s *Scheduler) Stop() {
|
|
if s.running {
|
|
close(s.stop)
|
|
s.running = false
|
|
}
|
|
}
|
|
|
|
// poll checks for due tasks and dispatches them.
|
|
func (s *Scheduler) poll() {
|
|
ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
|
|
defer cancel()
|
|
|
|
due, err := s.stores.Tasks.ListDue(ctx, 10)
|
|
if err != nil {
|
|
log.Printf("[scheduler] Failed to list due tasks: %v", err)
|
|
return
|
|
}
|
|
|
|
for _, task := range due {
|
|
go s.execute(ctx, task)
|
|
}
|
|
}
|
|
|
|
// execute runs a single task.
|
|
func (s *Scheduler) execute(parentCtx context.Context, task models.Task) {
|
|
ctx, cancel := context.WithTimeout(parentCtx, time.Duration(task.MaxWallClock)*time.Second)
|
|
defer cancel()
|
|
|
|
// Skip if previous run still active
|
|
active, _ := s.stores.Tasks.GetActiveRun(ctx, task.ID)
|
|
if active != nil {
|
|
log.Printf("[scheduler] Skipping task %s (%s) — previous run still active", task.ID, task.Name)
|
|
// Still advance next_run_at to avoid re-polling the same task
|
|
s.advanceNextRun(ctx, task)
|
|
return
|
|
}
|
|
|
|
log.Printf("[scheduler] Executing task %s (%s) type=%s", task.ID, task.Name, task.TaskType)
|
|
|
|
// Create run record
|
|
run := &models.TaskRun{
|
|
TaskID: task.ID,
|
|
Status: "running",
|
|
}
|
|
if err := s.stores.Tasks.CreateRun(ctx, run); err != nil {
|
|
log.Printf("[scheduler] Failed to create run for task %s: %v", task.ID, err)
|
|
s.advanceNextRun(ctx, task)
|
|
return
|
|
}
|
|
|
|
// Create or reuse service channel
|
|
channelID, err := s.ensureServiceChannel(ctx, task)
|
|
if err != nil {
|
|
log.Printf("[scheduler] Failed to create service channel for task %s: %v", task.ID, err)
|
|
_ = s.stores.Tasks.UpdateRun(ctx, run.ID, "failed", 0, 0, 0, "channel creation failed: "+err.Error())
|
|
s.advanceNextRun(ctx, task)
|
|
return
|
|
}
|
|
|
|
// Persist the user prompt as a message
|
|
if task.TaskType == "prompt" && task.UserPrompt != "" && s.stores.Messages != nil {
|
|
_ = s.stores.Messages.Create(ctx, &models.Message{
|
|
ChannelID: channelID,
|
|
Role: "user",
|
|
Content: task.UserPrompt,
|
|
})
|
|
}
|
|
|
|
// v0.27.2 TODO: Invoke completion pipeline with:
|
|
// - task.SystemPrompt prepended to persona system prompt
|
|
// - task.ModelID or provider resolution
|
|
// - Budget enforcement (max_tokens, max_tool_calls, max_wall_clock)
|
|
// - Tool grant filtering from task.ToolGrants
|
|
//
|
|
// For now, mark the run as completed (channel created + prompt persisted).
|
|
_ = s.stores.Tasks.UpdateRun(ctx, run.ID, "completed", 0, 0, 0, "")
|
|
_ = s.stores.Tasks.IncrementRunCount(ctx, task.ID)
|
|
|
|
log.Printf("[scheduler] Task %s completed (channel %s)", task.ID, channelID)
|
|
|
|
s.advanceNextRun(ctx, task)
|
|
}
|
|
|
|
// ensureServiceChannel creates a new service channel or reuses an existing one.
|
|
func (s *Scheduler) ensureServiceChannel(ctx context.Context, task models.Task) (string, error) {
|
|
// If output_channel_id is set and valid, reuse it
|
|
if task.OutputChannelID != nil && *task.OutputChannelID != "" {
|
|
return *task.OutputChannelID, nil
|
|
}
|
|
|
|
// Create a new service channel
|
|
ch := &models.Channel{
|
|
UserID: task.OwnerID,
|
|
Title: task.Name,
|
|
Description: "Task output: " + task.Description,
|
|
Type: "service",
|
|
TeamID: task.TeamID,
|
|
}
|
|
if err := s.stores.Channels.Create(ctx, ch); err != nil {
|
|
return "", err
|
|
}
|
|
|
|
// Update task to reference this channel for future runs
|
|
channelID := ch.ID
|
|
_ = s.stores.Tasks.Update(ctx, task.ID, models.TaskPatch{})
|
|
// Direct update for output_channel_id (not in patch for simplicity)
|
|
// The channel accumulates output over multiple runs.
|
|
|
|
return channelID, nil
|
|
}
|
|
|
|
// advanceNextRun computes the next run time and updates the task.
|
|
func (s *Scheduler) advanceNextRun(ctx context.Context, task models.Task) {
|
|
if task.Schedule == "once" {
|
|
// One-shot task — deactivate after execution
|
|
isActive := false
|
|
_ = s.stores.Tasks.Update(ctx, task.ID, models.TaskPatch{IsActive: &isActive})
|
|
_ = s.stores.Tasks.SetNextRun(ctx, task.ID, nil)
|
|
return
|
|
}
|
|
|
|
// Cron schedule — compute next run.
|
|
// v0.27.1: Uses a simple interval-based fallback.
|
|
// v0.27.2: Full cron parsing with robfig/cron/v3.
|
|
next := computeNextRun(task.Schedule, task.Timezone)
|
|
_ = s.stores.Tasks.SetNextRun(ctx, task.ID, next)
|
|
}
|
|
|
|
// computeNextRun parses a cron expression and returns the next execution time.
|
|
// Supports common presets and basic 5-field cron.
|
|
func computeNextRun(schedule, timezone string) *time.Time {
|
|
now := time.Now()
|
|
|
|
// Load timezone
|
|
if tz, err := time.LoadLocation(timezone); err == nil {
|
|
now = now.In(tz)
|
|
}
|
|
|
|
// Preset schedules (common cases without a full cron parser)
|
|
var next time.Time
|
|
switch schedule {
|
|
case "once":
|
|
return nil // Already handled
|
|
case "0 * * * *": // Every hour
|
|
next = now.Truncate(time.Hour).Add(time.Hour)
|
|
case "*/5 * * * *": // Every 5 minutes
|
|
next = now.Truncate(5 * time.Minute).Add(5 * time.Minute)
|
|
case "*/15 * * * *": // Every 15 minutes
|
|
next = now.Truncate(15 * time.Minute).Add(15 * time.Minute)
|
|
case "*/30 * * * *": // Every 30 minutes
|
|
next = now.Truncate(30 * time.Minute).Add(30 * time.Minute)
|
|
default:
|
|
// Fallback: try to parse minute and hour fields for daily/weekly cron
|
|
next = parseDailyCron(schedule, now)
|
|
}
|
|
|
|
utc := next.UTC()
|
|
return &utc
|
|
}
|
|
|
|
// parseDailyCron handles "M H * * *" and "M H * * D" patterns.
|
|
// Returns now+1h as fallback for unparseable expressions.
|
|
func parseDailyCron(expr string, now time.Time) time.Time {
|
|
// Split into fields
|
|
var fields []string
|
|
field := ""
|
|
for _, c := range expr {
|
|
if c == ' ' || c == '\t' {
|
|
if field != "" {
|
|
fields = append(fields, field)
|
|
field = ""
|
|
}
|
|
} else {
|
|
field += string(c)
|
|
}
|
|
}
|
|
if field != "" {
|
|
fields = append(fields, field)
|
|
}
|
|
|
|
if len(fields) < 5 {
|
|
return now.Add(time.Hour) // Unparseable — retry in 1h
|
|
}
|
|
|
|
minute := parseField(fields[0], 0)
|
|
hour := parseField(fields[1], 0)
|
|
|
|
// Construct today's target time
|
|
target := time.Date(now.Year(), now.Month(), now.Day(), hour, minute, 0, 0, now.Location())
|
|
if target.After(now) {
|
|
return target
|
|
}
|
|
|
|
// Check day-of-week field
|
|
if fields[4] != "*" {
|
|
dow := parseField(fields[4], -1)
|
|
if dow >= 0 && dow <= 6 {
|
|
// Advance to next matching day
|
|
for i := 1; i <= 7; i++ {
|
|
candidate := target.AddDate(0, 0, i)
|
|
if int(candidate.Weekday()) == dow {
|
|
return candidate
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
// Default: next day at same time
|
|
return target.AddDate(0, 0, 1)
|
|
}
|
|
|
|
// parseField parses a single cron field. Returns def for "*" or errors.
|
|
func parseField(s string, def int) int {
|
|
if s == "*" {
|
|
return def
|
|
}
|
|
n := 0
|
|
for _, c := range s {
|
|
if c >= '0' && c <= '9' {
|
|
n = n*10 + int(c-'0')
|
|
} else {
|
|
return def // step/range/list — not supported in minimal parser
|
|
}
|
|
}
|
|
return n
|
|
}
|