switchboard/core
Archived
1
0
Fork 0
Code Issues Pull Requests Actions Releases 9 Wiki Activity

Changeset 0.28.0.2 (#174)

Browse Source
This commit is contained in:
xcaliber
2026-03-11 19:41:04 +00:00
parent 58313f7e31
commit 8e08f3e4b0
19 changed files with 1734 additions and 384 deletions
Download Patch File Download Diff File Expand all files Collapse all files

49
server/scheduler/cron.go
View File

@@ -1,49 +0,0 @@
package scheduler
import (
"time"
"github.com/robfig/cron/v3"
)
// cronParser is a shared parser instance. Standard 5-field cron with
// optional descriptors (@hourly, @daily, @weekly, @monthly, etc.).
var cronParser = cron.NewParser(
cron.Minute | cron.Hour | cron.Dom | cron.Month | cron.Dow | cron.Descriptor,
)
// NextRunFromSchedule computes the next run time from a cron expression
// and timezone. Returns nil for "once" schedules (one-shot tasks).
//
// Replaces the v0.27.1 hand-rolled parseDailyCron with full 5-field
// cron support via robfig/cron/v3.
func NextRunFromSchedule(schedule, timezone string) *time.Time {
if schedule == "once" {
return nil
}
now := time.Now()
if tz, err := time.LoadLocation(timezone); err == nil {
now = now.In(tz)
}
sched, err := cronParser.Parse(schedule)
if err != nil {
// Unparseable — log at call site, caller decides fallback
return nil
}
next := sched.Next(now).UTC()
return &next
}
// ValidateCron checks whether a cron expression is valid.
// Returns nil for valid expressions, error describing the problem otherwise.
// "once" is always valid (one-shot schedule).
func ValidateCron(schedule string) error {
if schedule == "once" {
return nil
}
_, err := cronParser.Parse(schedule)
return err
}

77
server/scheduler/executor.go
View File

@@ -1,6 +1,8 @@
// Package scheduler — executor.go
//
// v0.27.2: Headless task execution via coreToolLoop.
// v0.28.0: Action task type (no LLM), trigger payload passthrough,
// webhook payload shape fix (D1).
//
// The Executor bridges the task scheduler with the completion pipeline.
// It resolves providers, builds tool definitions, runs the core tool loop
@@ -49,6 +51,12 @@ func NewExecutor(stores store.Stores, vault *crypto.KeyResolver, hub *events.Hub
func (e *Executor) Execute(ctx context.Context, task models.Task, run *models.TaskRun, channelID string) {
startTime := time.Now()
// v0.28.0: Action tasks skip the LLM pipeline entirely.
if task.TaskType == "action" {
e.executeAction(ctx, task, run, channelID, startTime)
return
}
// ── 1. Resolve provider ────────────────────
providerConfigID := ""
if task.ProviderConfigID != nil {
@@ -92,11 +100,17 @@ func (e *Executor) Execute(ctx context.Context, task models.Task, run *models.Ta
})
}
// User prompt
if task.UserPrompt != "" {
// User prompt — with optional trigger payload prepended
userContent := task.UserPrompt
if run.TriggerPayload != "" && userContent != "" {
userContent = "[Webhook trigger data]\n```json\n" + run.TriggerPayload + "\n```\n\n[Task instructions]\n" + userContent
} else if run.TriggerPayload != "" {
userContent = run.TriggerPayload
}
if userContent != "" {
messages = append(messages, providers.Message{
Role: "user",
Content: task.UserPrompt,
Content: userContent,
})
}
@@ -216,18 +230,15 @@ func (e *Executor) Execute(ctx context.Context, task models.Task, run *models.Ta
errMsg = "budget exceeded: " + result.BudgetExceeded
}
tokensUsed := result.InputTokens + result.OutputTokens
// ── 8. Update run record ───────────────────
_ = e.stores.Tasks.UpdateRun(ctx, run.ID, status,
result.InputTokens+result.OutputTokens,
result.ToolCallCount,
wallClock,
errMsg)
tokensUsed, result.ToolCallCount, wallClock, errMsg)
_ = e.stores.Tasks.IncrementRunCount(ctx, task.ID)
log.Printf("[executor] Task %s (%s) → %s (tokens=%d, tools=%d, wall=%ds)",
task.ID, task.Name, status,
result.InputTokens+result.OutputTokens,
result.ToolCallCount, wallClock)
task.ID, task.Name, status, tokensUsed, result.ToolCallCount, wallClock)
// ── 9. Owner notification ──────────────────
e.notifyOwner(ctx, task, status, errMsg)
@@ -236,16 +247,61 @@ func (e *Executor) Execute(ctx context.Context, task models.Task, run *models.Ta
if task.WebhookURL != "" {
go webhook.Deliver(task.WebhookURL, task.WebhookSecret, webhook.Payload{
TaskID: task.ID,
RunID: run.ID,
TaskName: task.Name,
ChannelID: channelID,
Status: status,
CompletedAt: time.Now().UTC(),
Output: result.Content,
TokensUsed: tokensUsed,
Error: errMsg,
})
}
}
// executeAction handles non-LLM action tasks.
// Skips provider resolution and completion entirely.
func (e *Executor) executeAction(ctx context.Context, task models.Task, run *models.TaskRun, channelID string, startTime time.Time) {
wallClock := int(time.Since(startTime).Seconds())
status := "completed"
// v0.28.0: Action tasks relay trigger payload to outbound webhook.
// No LLM execution — the value is in the automation plumbing.
output := run.TriggerPayload
if output == "" {
output = "{}"
}
// Persist to channel if output_mode == "channel"
if task.OutputMode == "channel" && e.stores.Messages != nil {
_ = e.stores.Messages.Create(ctx, &models.Message{
ChannelID: channelID,
Role: "system",
Content: "Action task executed. Trigger payload:\n```json\n" + output + "\n```",
})
}
_ = e.stores.Tasks.UpdateRun(ctx, run.ID, status, 0, 0, wallClock, "")
_ = e.stores.Tasks.IncrementRunCount(ctx, task.ID)
log.Printf("[executor] Action task %s (%s) → %s (wall=%ds)", task.ID, task.Name, status, wallClock)
e.notifyOwner(ctx, task, status, "")
// Fire outbound webhook with trigger payload
if task.WebhookURL != "" {
go webhook.Deliver(task.WebhookURL, task.WebhookSecret, webhook.Payload{
TaskID: task.ID,
RunID: run.ID,
TaskName: task.Name,
ChannelID: channelID,
Status: status,
CompletedAt: time.Now().UTC(),
Output: output,
})
}
}
// failRun marks a run as failed before completion was attempted.
func (e *Executor) failRun(ctx context.Context, task models.Task, run *models.TaskRun, errMsg string) {
log.Printf("[executor] Task %s (%s) pre-execution failure: %s", task.ID, task.Name, errMsg)
@@ -254,6 +310,7 @@ func (e *Executor) failRun(ctx context.Context, task models.Task, run *models.Ta
if task.WebhookURL != "" {
go webhook.Deliver(task.WebhookURL, task.WebhookSecret, webhook.Payload{
TaskID: task.ID,
RunID: run.ID,
TaskName: task.Name,
Status: "failed",
CompletedAt: time.Now().UTC(),

62
server/scheduler/scheduler.go
View File

@@ -4,6 +4,7 @@
// v0.27.1: Foundation — scheduler loop + service channel creation.
// v0.27.2: Adds global config checks (enabled, max_concurrent), full cron
// parsing via robfig/cron/v3, and completion invocation via executor.
// v0.28.0: Adopt queued runs (webhook triggers), action tasks, C3/C4 audit fixes.
package scheduler
import (
@@ -104,18 +105,31 @@ func (s *Scheduler) execute(parentCtx context.Context, task models.Task) {
return
}
// v0.28.0: Check for queued run (from webhook trigger) — adopt it instead
// of creating a new one so the trigger_payload is preserved.
run, _ := s.stores.Tasks.GetQueuedRun(ctx, task.ID)
if run != nil {
// Adopt: transition queued → running
_ = s.stores.Tasks.TransitionRunStatus(ctx, run.ID, "running")
run.Status = "running"
log.Printf("[scheduler] Adopting queued run %s for task %s (%s)", run.ID, task.ID, task.Name)
} else {
// Normal cron-triggered execution — create a new run
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
}
}
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
}
// Mark execution start (C4 fix: SetLastRun is now separate from SetNextRun)
_ = s.stores.Tasks.SetLastRun(ctx, task.ID)
// Create or reuse service channel
channelID, err := s.ensureServiceChannel(ctx, task)
@@ -126,20 +140,25 @@ func (s *Scheduler) execute(parentCtx context.Context, task models.Task) {
return
}
// Persist the user prompt as a message
// Persist the user prompt as a message (prompt tasks only)
if task.TaskType == "prompt" && task.UserPrompt != "" && s.stores.Messages != nil {
content := task.UserPrompt
// v0.28.0: Prepend trigger payload as context if present
if run.TriggerPayload != "" {
content = "[Webhook trigger data]\n```json\n" + run.TriggerPayload + "\n```\n\n[Task instructions]\n" + task.UserPrompt
}
_ = s.stores.Messages.Create(ctx, &models.Message{
ChannelID: channelID,
Role: "user",
Content: task.UserPrompt,
Content: content,
})
}
// v0.27.2: Invoke completion via executor
if s.executor != nil && task.TaskType == "prompt" {
if s.executor != nil {
s.executor.Execute(ctx, task, run, channelID)
} else {
// No executor or non-prompt task type — mark completed (channel + prompt persisted)
// No executor — mark completed (channel + prompt persisted)
_ = s.stores.Tasks.UpdateRun(ctx, run.ID, "completed", 0, 0, 0, "")
_ = s.stores.Tasks.IncrementRunCount(ctx, task.ID)
}
@@ -168,11 +187,11 @@ func (s *Scheduler) ensureServiceChannel(ctx context.Context, task models.Task)
return "", err
}
// Update task to reference this channel for future runs
// C3 fix: Persist output_channel_id on the task so future runs reuse this channel.
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.
_ = s.stores.Tasks.Update(ctx, task.ID, models.TaskPatch{
OutputChannelID: &channelID,
})
return channelID, nil
}
@@ -187,6 +206,13 @@ func (s *Scheduler) advanceNextRun(ctx context.Context, task models.Task) {
return
}
if task.Schedule == "webhook" {
// Webhook tasks have no cron schedule — clear next_run_at.
// They only fire when triggered externally.
_ = s.stores.Tasks.SetNextRun(ctx, task.ID, nil)
return
}
// v0.27.2: Full cron parsing via robfig/cron/v3 (replaces hand-rolled parser).
next := taskutil.NextRunFromSchedule(task.Schedule, task.Timezone)
if next == nil {

110
server/scheduler/task_config.go
View File

@@ -1,110 +0,0 @@
package scheduler
import (
"context"
"log"
"git.gobha.me/xcaliber/chat-switchboard/models"
"git.gobha.me/xcaliber/chat-switchboard/store"
)
// TaskConfig holds the runtime task configuration read from global_settings.
// Keys: tasks.enabled, tasks.allow_personal, tasks.max_concurrent,
// tasks.default_max_tokens, tasks.default_max_tool_calls,
// tasks.default_max_wall_clock
type TaskConfig struct {
Enabled bool
AllowPersonal bool
MaxConcurrent int
DefaultMaxTokens int
DefaultMaxToolCalls int
DefaultMaxWallClock int // seconds
}
// DefaultTaskConfig returns sensible defaults when no global config is set.
func DefaultTaskConfig() TaskConfig {
return TaskConfig{
Enabled: true,
AllowPersonal: true,
MaxConcurrent: 5,
DefaultMaxTokens: 4096,
DefaultMaxToolCalls: 10,
DefaultMaxWallClock: 300,
}
}
// LoadTaskConfig reads task configuration from global_settings.
// Falls back to defaults for missing keys.
func LoadTaskConfig(ctx context.Context, gc store.GlobalConfigStore) TaskConfig {
cfg := DefaultTaskConfig()
if gc == nil {
return cfg
}
raw, err := gc.Get(ctx, "tasks")
if err != nil || raw == nil {
return cfg
}
if v, ok := boolVal(raw, "enabled"); ok {
cfg.Enabled = v
}
if v, ok := boolVal(raw, "allow_personal"); ok {
cfg.AllowPersonal = v
}
if v, ok := intVal(raw, "max_concurrent"); ok && v > 0 {
cfg.MaxConcurrent = v
}
if v, ok := intVal(raw, "default_max_tokens"); ok && v > 0 {
cfg.DefaultMaxTokens = v
}
if v, ok := intVal(raw, "default_max_tool_calls"); ok && v > 0 {
cfg.DefaultMaxToolCalls = v
}
if v, ok := intVal(raw, "default_max_wall_clock"); ok && v > 0 {
cfg.DefaultMaxWallClock = v
}
return cfg
}
// ApplyDefaults fills zero-value budget fields on a task with the global defaults.
func (tc TaskConfig) ApplyDefaults(t *models.Task) {
if t.MaxTokens == 0 {
t.MaxTokens = tc.DefaultMaxTokens
}
if t.MaxToolCalls == 0 {
t.MaxToolCalls = tc.DefaultMaxToolCalls
}
if t.MaxWallClock == 0 {
t.MaxWallClock = tc.DefaultMaxWallClock
}
}
// ── helpers ────────────────────────────────────
func boolVal(m models.JSONMap, key string) (bool, bool) {
v, ok := m[key]
if !ok {
return false, false
}
b, ok := v.(bool)
return b, ok
}
func intVal(m models.JSONMap, key string) (int, bool) {
v, ok := m[key]
if !ok {
return 0, false
}
// JSON numbers are float64 after Unmarshal
switch n := v.(type) {
case float64:
return int(n), true
case int:
return n, true
default:
log.Printf("[task_config] unexpected type for %s: %T", key, v)
return 0, false
}
}