diff --git a/VERSION b/VERSION index 81566e4..4daf42f 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -0.27.0 \ No newline at end of file +0.27.1 \ No newline at end of file diff --git a/docs/ROADMAP.md b/docs/ROADMAP.md index 89acb35..a785323 100644 --- a/docs/ROADMAP.md +++ b/docs/ROADMAP.md @@ -154,11 +154,11 @@ v0.26.0 Workflow Engine ✅ human + AI collab, visitor intake, assignment queue) │ -v0.27.0 Debt Clearance +v0.27.0 Debt Clearance ✅ (extension surface routes, workflow polish, editor debt) │ -v0.27.1 Tasks Foundation +v0.27.1 Tasks Foundation ✅ (service channels, scheduler, task definitions, cron) │ @@ -917,21 +917,39 @@ See [DESIGN-0.27.0.md](DESIGN-0.27.0.md) for full spec. --- -## v0.27.1 — Tasks Foundation: Service Channels + Scheduler +## v0.27.1 — Tasks Foundation: Service Channels + Scheduler ✅ Core primitive for autonomous agents: a channel with no human participant, driven by a cron scheduler. Depends on: v0.27.0 (debt clearance). -- [ ] `tasks` table + store interface + PG/SQLite implementations -- [ ] `type: 'service'` channel type (no human participants) -- [ ] `TaskScheduler` background goroutine (poll interval: 30s) -- [ ] Cron expression parsing (`robfig/cron/v3`) + `next_run_at` computation -- [ ] Prompt task execution (create service channel, send prompt, collect response) -- [ ] Workflow task execution (instantiate workflow in service channel) +- [x] `tasks` + `task_runs` tables + store interface + PG (UUID) / SQLite implementations +- [x] `type: 'service'` channel type (PG CHECK extended, SQLite by convention) +- [x] `TaskStore` interface: CRUD, `ListDue`, `SetNextRun`, `IncrementRunCount`, run history +- [x] `TaskScheduler` background goroutine (30s poll, skip-if-running, one-shot + cron) +- [x] Minimal cron parser (hourly, N-minute intervals, daily H:M, weekly DOW patterns) +- [x] Service channel creation per task (reuse `output_channel_id` on subsequent runs) +- [x] User prompt persisted as message in service channel +- [x] Task CRUD handler with ownership checks + admin bypass +- [x] `POST /tasks/:id/run` — manual "Run Now" trigger +- [x] Migration 026: `tasks` + `task_runs` tables, `service` channel type + +**Deferred to v0.27.2:** +- [ ] Completion invocation from scheduler (wire into `streamWithToolLoop`) +- [ ] Full cron parsing via `robfig/cron/v3` (replacing minimal parser) - [ ] Provider resolution for task context (BYOK → team → global → routing policy) +- [ ] Workflow task execution (instantiate workflow in service channel) - [ ] Tasks sidebar section (read-only view of service channels) -- [ ] Migration 026: `tasks` table, `service` channel type + +**API routes (v0.27.1):** +- `GET /api/v1/tasks` — list my tasks +- `POST /api/v1/tasks` — create task +- `GET /api/v1/tasks/:id` — get task +- `PUT /api/v1/tasks/:id` — update task +- `DELETE /api/v1/tasks/:id` — delete task +- `GET /api/v1/tasks/:id/runs` — run history +- `POST /api/v1/tasks/:id/run` — manual trigger +- `GET /api/v1/admin/tasks` — list all tasks (admin) --- @@ -941,9 +959,9 @@ Guardrails for unattended execution. Depends on: v0.27.1 (task scheduler). - [ ] Execution budget enforcement: `max_tokens`, `max_tool_calls`, `max_wall_clock` -- [ ] `task_runs` table + store (run history, status, budget usage) +- [x] ~~`task_runs` table + store~~ (shipped v0.27.1 — table, PG+SQLite store, run history API) - [ ] Budget breach → `budget_exceeded` status + owner notification -- [ ] Concurrent run skip (if previous run still active, skip with warning) +- [x] ~~Concurrent run skip~~ (shipped v0.27.1 — scheduler skips if `GetActiveRun` returns non-nil) - [ ] Global config: `tasks.enabled`, `tasks.allow_personal`, `tasks.max_concurrent` - [ ] Default budget ceilings in global config (overridable per task) - [ ] `tasks.create` and `tasks.admin` permissions diff --git a/server/database/migrations/026_tasks.sql b/server/database/migrations/026_tasks.sql new file mode 100644 index 0000000..9152875 --- /dev/null +++ b/server/database/migrations/026_tasks.sql @@ -0,0 +1,77 @@ +-- 026_tasks.sql — Task scheduling for autonomous agents (v0.27.1) + +-- Extend channel type to include 'service' (no human participants) +ALTER TABLE channels DROP CONSTRAINT IF EXISTS channels_type_check; +ALTER TABLE channels ADD CONSTRAINT channels_type_check + CHECK (type IN ('direct', 'dm', 'group', 'channel', 'workflow', 'service')); + +-- Task definitions +CREATE TABLE IF NOT EXISTS tasks ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + owner_id UUID NOT NULL REFERENCES users(id) ON DELETE CASCADE, + team_id UUID REFERENCES teams(id) ON DELETE SET NULL, + name TEXT NOT NULL, + description TEXT DEFAULT '', + scope TEXT NOT NULL DEFAULT 'personal' + CHECK (scope IN ('personal', 'team', 'global')), + + -- What to run + task_type TEXT NOT NULL DEFAULT 'prompt' + CHECK (task_type IN ('prompt', 'workflow')), + persona_id UUID REFERENCES personas(id) ON DELETE SET NULL, + model_id TEXT, + system_prompt TEXT DEFAULT '', + user_prompt TEXT DEFAULT '', + workflow_id UUID REFERENCES workflows(id) ON DELETE SET NULL, + tool_grants JSONB, + + -- Schedule + schedule TEXT NOT NULL, + timezone TEXT NOT NULL DEFAULT 'UTC', + is_active BOOLEAN NOT NULL DEFAULT true, + + -- Execution policy + max_tokens INTEGER NOT NULL DEFAULT 4096, + max_tool_calls INTEGER NOT NULL DEFAULT 10, + max_wall_clock INTEGER NOT NULL DEFAULT 300, + output_mode TEXT NOT NULL DEFAULT 'channel' + CHECK (output_mode IN ('channel', 'note', 'webhook')), + output_channel_id UUID REFERENCES channels(id) ON DELETE SET NULL, + webhook_url TEXT, + + -- Provider routing + provider_config_id UUID REFERENCES provider_configs(id) ON DELETE SET NULL, + + -- Notifications + notify_on_complete BOOLEAN NOT NULL DEFAULT false, + notify_on_failure BOOLEAN NOT NULL DEFAULT true, + + -- Bookkeeping + last_run_at TIMESTAMPTZ, + next_run_at TIMESTAMPTZ, + run_count INTEGER NOT NULL DEFAULT 0, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT now() +); + +CREATE INDEX IF NOT EXISTS idx_tasks_next_run ON tasks (next_run_at) WHERE is_active = true; +CREATE INDEX IF NOT EXISTS idx_tasks_owner ON tasks (owner_id); +CREATE INDEX IF NOT EXISTS idx_tasks_team ON tasks (team_id); + +-- Task run history +CREATE TABLE IF NOT EXISTS task_runs ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + task_id UUID NOT NULL REFERENCES tasks(id) ON DELETE CASCADE, + channel_id UUID REFERENCES channels(id) ON DELETE SET NULL, + status TEXT NOT NULL DEFAULT 'running' + CHECK (status IN ('running', 'completed', 'failed', 'budget_exceeded', 'cancelled')), + started_at TIMESTAMPTZ NOT NULL DEFAULT now(), + completed_at TIMESTAMPTZ, + tokens_used INTEGER DEFAULT 0, + tool_calls INTEGER DEFAULT 0, + wall_clock INTEGER DEFAULT 0, + error TEXT +); + +CREATE INDEX IF NOT EXISTS idx_task_runs_task ON task_runs (task_id); +CREATE INDEX IF NOT EXISTS idx_task_runs_status ON task_runs (task_id, status); diff --git a/server/database/migrations/sqlite/026_tasks.sql b/server/database/migrations/sqlite/026_tasks.sql new file mode 100644 index 0000000..3f1ccd0 --- /dev/null +++ b/server/database/migrations/sqlite/026_tasks.sql @@ -0,0 +1,63 @@ +-- 026_tasks.sql — Task scheduling for autonomous agents (v0.27.1) — SQLite + +-- Note: SQLite CHECK on channels.type is in the CREATE TABLE (005), +-- which cannot be altered. The 'service' type is accepted by convention; +-- the Go layer validates. + +CREATE TABLE IF NOT EXISTS tasks ( + id TEXT PRIMARY KEY, + owner_id TEXT NOT NULL REFERENCES users(id) ON DELETE CASCADE, + team_id TEXT REFERENCES teams(id) ON DELETE SET NULL, + name TEXT NOT NULL, + description TEXT DEFAULT '', + scope TEXT NOT NULL DEFAULT 'personal', + + task_type TEXT NOT NULL DEFAULT 'prompt', + persona_id TEXT REFERENCES personas(id) ON DELETE SET NULL, + model_id TEXT, + system_prompt TEXT DEFAULT '', + user_prompt TEXT DEFAULT '', + workflow_id TEXT REFERENCES workflows(id) ON DELETE SET NULL, + tool_grants TEXT, + + schedule TEXT NOT NULL, + timezone TEXT NOT NULL DEFAULT 'UTC', + is_active INTEGER NOT NULL DEFAULT 1, + + max_tokens INTEGER NOT NULL DEFAULT 4096, + max_tool_calls INTEGER NOT NULL DEFAULT 10, + max_wall_clock INTEGER NOT NULL DEFAULT 300, + output_mode TEXT NOT NULL DEFAULT 'channel', + output_channel_id TEXT REFERENCES channels(id) ON DELETE SET NULL, + webhook_url TEXT, + + provider_config_id TEXT REFERENCES provider_configs(id) ON DELETE SET NULL, + + notify_on_complete INTEGER NOT NULL DEFAULT 0, + notify_on_failure INTEGER NOT NULL DEFAULT 1, + + last_run_at TEXT, + next_run_at TEXT, + run_count INTEGER NOT NULL DEFAULT 0, + created_at TEXT NOT NULL DEFAULT (datetime('now')), + updated_at TEXT NOT NULL DEFAULT (datetime('now')) +); + +CREATE INDEX IF NOT EXISTS idx_tasks_next_run ON tasks (next_run_at); +CREATE INDEX IF NOT EXISTS idx_tasks_owner ON tasks (owner_id); +CREATE INDEX IF NOT EXISTS idx_tasks_team ON tasks (team_id); + +CREATE TABLE IF NOT EXISTS task_runs ( + id TEXT PRIMARY KEY, + task_id TEXT NOT NULL REFERENCES tasks(id) ON DELETE CASCADE, + channel_id TEXT REFERENCES channels(id) ON DELETE SET NULL, + status TEXT NOT NULL DEFAULT 'running', + started_at TEXT NOT NULL DEFAULT (datetime('now')), + completed_at TEXT, + tokens_used INTEGER DEFAULT 0, + tool_calls INTEGER DEFAULT 0, + wall_clock INTEGER DEFAULT 0, + error TEXT +); + +CREATE INDEX IF NOT EXISTS idx_task_runs_task ON task_runs (task_id); diff --git a/server/handlers/tasks.go b/server/handlers/tasks.go new file mode 100644 index 0000000..53d2dd8 --- /dev/null +++ b/server/handlers/tasks.go @@ -0,0 +1,240 @@ +package handlers + +import ( + "net/http" + "time" + + "github.com/gin-gonic/gin" + + "git.gobha.me/xcaliber/chat-switchboard/models" + "git.gobha.me/xcaliber/chat-switchboard/store" +) + +// TaskHandler manages task CRUD and manual execution. +type TaskHandler struct { + stores store.Stores +} + +func NewTaskHandler(stores store.Stores) *TaskHandler { + return &TaskHandler{stores: stores} +} + +// ListMine returns tasks owned by the current user. +// GET /api/v1/tasks +func (h *TaskHandler) ListMine(c *gin.Context) { + userID := c.GetString("user_id") + tasks, err := h.stores.Tasks.ListByOwner(c.Request.Context(), userID) + if err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to list tasks"}) + return + } + if tasks == nil { + tasks = []models.Task{} + } + c.JSON(http.StatusOK, gin.H{"data": tasks}) +} + +// ListAll returns all tasks (admin only). +// GET /api/v1/admin/tasks +func (h *TaskHandler) ListAll(c *gin.Context) { + tasks, err := h.stores.Tasks.ListAll(c.Request.Context()) + if err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to list tasks"}) + return + } + if tasks == nil { + tasks = []models.Task{} + } + c.JSON(http.StatusOK, gin.H{"data": tasks}) +} + +// Create creates a new task. +// POST /api/v1/tasks +func (h *TaskHandler) Create(c *gin.Context) { + var t models.Task + if err := c.ShouldBindJSON(&t); err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()}) + return + } + + t.OwnerID = c.GetString("user_id") + + // Validate required fields + if t.Name == "" { + c.JSON(http.StatusBadRequest, gin.H{"error": "name is required"}) + return + } + if t.Schedule == "" { + c.JSON(http.StatusBadRequest, gin.H{"error": "schedule is required"}) + return + } + if t.TaskType == "prompt" && t.UserPrompt == "" { + c.JSON(http.StatusBadRequest, gin.H{"error": "user_prompt is required for prompt tasks"}) + return + } + if t.TaskType == "workflow" && (t.WorkflowID == nil || *t.WorkflowID == "") { + c.JSON(http.StatusBadRequest, gin.H{"error": "workflow_id is required for workflow tasks"}) + return + } + + // Defaults + if t.Scope == "" { + t.Scope = "personal" + } + if t.TaskType == "" { + t.TaskType = "prompt" + } + if t.Timezone == "" { + t.Timezone = "UTC" + } + if t.OutputMode == "" { + t.OutputMode = "channel" + } + if t.MaxTokens == 0 { + t.MaxTokens = 4096 + } + if t.MaxToolCalls == 0 { + t.MaxToolCalls = 10 + } + if t.MaxWallClock == 0 { + t.MaxWallClock = 300 + } + + // Compute initial next_run_at + // For "once" tasks, next_run_at = now (runs on next poll) + // For cron tasks, compute from schedule + if t.Schedule == "once" { + now := time.Now().UTC() + t.NextRunAt = &now + } + // TODO: compute next_run_at from cron expression (requires robfig/cron/v3) + // For now, "once" tasks run immediately; cron tasks need next_run_at set by scheduler + + t.IsActive = true + + if err := h.stores.Tasks.Create(c.Request.Context(), &t); err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to create task: " + err.Error()}) + return + } + + c.JSON(http.StatusCreated, t) +} + +// Get returns a single task. +// GET /api/v1/tasks/:id +func (h *TaskHandler) Get(c *gin.Context) { + id := c.Param("id") + t, err := h.stores.Tasks.GetByID(c.Request.Context(), id) + if err != nil { + c.JSON(http.StatusNotFound, gin.H{"error": "task not found"}) + return + } + // Ownership check (admin bypasses) + if c.GetString("role") != "admin" && t.OwnerID != c.GetString("user_id") { + c.JSON(http.StatusNotFound, gin.H{"error": "task not found"}) + return + } + c.JSON(http.StatusOK, t) +} + +// Update patches a task. +// PUT /api/v1/tasks/:id +func (h *TaskHandler) Update(c *gin.Context) { + id := c.Param("id") + ctx := c.Request.Context() + + // Ownership check + t, err := h.stores.Tasks.GetByID(ctx, id) + if err != nil { + c.JSON(http.StatusNotFound, gin.H{"error": "task not found"}) + return + } + if c.GetString("role") != "admin" && t.OwnerID != c.GetString("user_id") { + c.JSON(http.StatusNotFound, gin.H{"error": "task not found"}) + return + } + + var patch models.TaskPatch + if err := c.ShouldBindJSON(&patch); err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()}) + return + } + + if err := h.stores.Tasks.Update(ctx, id, patch); err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to update task"}) + return + } + + updated, _ := h.stores.Tasks.GetByID(ctx, id) + c.JSON(http.StatusOK, updated) +} + +// Delete removes a task. +// DELETE /api/v1/tasks/:id +func (h *TaskHandler) Delete(c *gin.Context) { + id := c.Param("id") + ctx := c.Request.Context() + + t, err := h.stores.Tasks.GetByID(ctx, id) + if err != nil { + c.JSON(http.StatusNotFound, gin.H{"error": "task not found"}) + return + } + if c.GetString("role") != "admin" && t.OwnerID != c.GetString("user_id") { + c.JSON(http.StatusNotFound, gin.H{"error": "task not found"}) + return + } + + if err := h.stores.Tasks.Delete(ctx, id); err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to delete task"}) + return + } + c.JSON(http.StatusOK, gin.H{"deleted": true, "id": id}) +} + +// ListRuns returns run history for a task. +// GET /api/v1/tasks/:id/runs +func (h *TaskHandler) ListRuns(c *gin.Context) { + id := c.Param("id") + runs, err := h.stores.Tasks.ListRuns(c.Request.Context(), id, 50) + if err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to list runs"}) + return + } + if runs == nil { + runs = []models.TaskRun{} + } + c.JSON(http.StatusOK, gin.H{"data": runs}) +} + +// RunNow triggers immediate execution of a task (sets next_run_at to now). +// POST /api/v1/tasks/:id/run +func (h *TaskHandler) RunNow(c *gin.Context) { + id := c.Param("id") + ctx := c.Request.Context() + + t, err := h.stores.Tasks.GetByID(ctx, id) + if err != nil { + c.JSON(http.StatusNotFound, gin.H{"error": "task not found"}) + return + } + if c.GetString("role") != "admin" && t.OwnerID != c.GetString("user_id") { + c.JSON(http.StatusNotFound, gin.H{"error": "task not found"}) + return + } + + // Check for already-running execution + active, _ := h.stores.Tasks.GetActiveRun(ctx, id) + if active != nil { + c.JSON(http.StatusConflict, gin.H{"error": "task already has an active run"}) + return + } + + now := time.Now().UTC() + if err := h.stores.Tasks.SetNextRun(ctx, id, now); err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to schedule run"}) + return + } + + c.JSON(http.StatusOK, gin.H{"scheduled": true, "next_run_at": now}) +} diff --git a/server/main.go b/server/main.go index ee1eed8..8e00262 100644 --- a/server/main.go +++ b/server/main.go @@ -30,6 +30,7 @@ import ( "git.gobha.me/xcaliber/chat-switchboard/providers" "git.gobha.me/xcaliber/chat-switchboard/roles" "git.gobha.me/xcaliber/chat-switchboard/routing" + "git.gobha.me/xcaliber/chat-switchboard/scheduler" "git.gobha.me/xcaliber/chat-switchboard/storage" "git.gobha.me/xcaliber/chat-switchboard/store" postgres "git.gobha.me/xcaliber/chat-switchboard/store/postgres" @@ -215,6 +216,13 @@ func main() { }() } + // v0.27.1: Task scheduler — polls for due tasks every 30s + if stores.Tasks != nil { + taskSched := scheduler.New(stores) + go taskSched.Run() + log.Println(" ⏰ Task scheduler started") + } + // Bootstrap admin from env (K8s secret) — upserts on every restart handlers.BootstrapAdmin(cfg, stores) @@ -596,6 +604,16 @@ func main() { protected.POST("/workflow-assignments/:id/claim", wfAssignH.Claim) protected.POST("/workflow-assignments/:id/complete", wfAssignH.Complete) + // Tasks (v0.27.1) + taskH := handlers.NewTaskHandler(stores) + protected.GET("/tasks", taskH.ListMine) + protected.POST("/tasks", taskH.Create) + protected.GET("/tasks/:id", taskH.Get) + protected.PUT("/tasks/:id", taskH.Update) + protected.DELETE("/tasks/:id", taskH.Delete) + protected.GET("/tasks/:id/runs", taskH.ListRuns) + protected.POST("/tasks/:id/run", taskH.RunNow) + // Channel models (v0.20.0 — multi-model @mention routing) chModelH := handlers.NewChannelModelHandler(stores) protected.GET("/channels/:id/models", chModelH.List) @@ -1074,6 +1092,10 @@ func main() { admin.PUT("/surfaces/:id/enable", surfaceAdm.EnableSurface) admin.PUT("/surfaces/:id/disable", surfaceAdm.DisableSurface) admin.DELETE("/surfaces/:id", surfaceAdm.DeleteSurface) + + // Task management — admin (v0.27.1) + taskAdm := handlers.NewTaskHandler(stores) + admin.GET("/tasks", taskAdm.ListAll) } } diff --git a/server/models/models_task.go b/server/models/models_task.go new file mode 100644 index 0000000..f5ff936 --- /dev/null +++ b/server/models/models_task.go @@ -0,0 +1,88 @@ +package models + +import ( + "encoding/json" + "time" +) + +// Task is a scheduled or one-shot job that creates a service channel +// and runs a completion (prompt task) or instantiates a workflow. +type Task struct { + ID string `json:"id" db:"id"` + OwnerID string `json:"owner_id" db:"owner_id"` + TeamID *string `json:"team_id,omitempty" db:"team_id"` + Name string `json:"name" db:"name"` + Description string `json:"description" db:"description"` + Scope string `json:"scope" db:"scope"` // personal | team | global + + // What to run + TaskType string `json:"task_type" db:"task_type"` // prompt | workflow + PersonaID *string `json:"persona_id,omitempty" db:"persona_id"` + ModelID string `json:"model_id" db:"model_id"` + SystemPrompt string `json:"system_prompt" db:"system_prompt"` + UserPrompt string `json:"user_prompt" db:"user_prompt"` + WorkflowID *string `json:"workflow_id,omitempty" db:"workflow_id"` + ToolGrants json.RawMessage `json:"tool_grants,omitempty" db:"tool_grants"` + + // Schedule + Schedule string `json:"schedule" db:"schedule"` // cron expression or "once" + Timezone string `json:"timezone" db:"timezone"` + IsActive bool `json:"is_active" db:"is_active"` + + // Execution policy + MaxTokens int `json:"max_tokens" db:"max_tokens"` + MaxToolCalls int `json:"max_tool_calls" db:"max_tool_calls"` + MaxWallClock int `json:"max_wall_clock" db:"max_wall_clock"` // seconds + OutputMode string `json:"output_mode" db:"output_mode"` // channel | note | webhook + OutputChannelID *string `json:"output_channel_id,omitempty" db:"output_channel_id"` + WebhookURL string `json:"webhook_url" db:"webhook_url"` + + // Provider routing + ProviderConfigID *string `json:"provider_config_id,omitempty" db:"provider_config_id"` + + // Notifications + NotifyOnComplete bool `json:"notify_on_complete" db:"notify_on_complete"` + NotifyOnFailure bool `json:"notify_on_failure" db:"notify_on_failure"` + + // Bookkeeping + LastRunAt *time.Time `json:"last_run_at,omitempty" db:"last_run_at"` + NextRunAt *time.Time `json:"next_run_at,omitempty" db:"next_run_at"` + RunCount int `json:"run_count" db:"run_count"` + CreatedAt time.Time `json:"created_at" db:"created_at"` + UpdatedAt time.Time `json:"updated_at" db:"updated_at"` +} + +// TaskPatch contains optional fields for updating a task. +type TaskPatch struct { + Name *string `json:"name,omitempty"` + Description *string `json:"description,omitempty"` + PersonaID *string `json:"persona_id,omitempty"` + ModelID *string `json:"model_id,omitempty"` + SystemPrompt *string `json:"system_prompt,omitempty"` + UserPrompt *string `json:"user_prompt,omitempty"` + Schedule *string `json:"schedule,omitempty"` + Timezone *string `json:"timezone,omitempty"` + IsActive *bool `json:"is_active,omitempty"` + MaxTokens *int `json:"max_tokens,omitempty"` + MaxToolCalls *int `json:"max_tool_calls,omitempty"` + MaxWallClock *int `json:"max_wall_clock,omitempty"` + OutputMode *string `json:"output_mode,omitempty"` + WebhookURL *string `json:"webhook_url,omitempty"` + ToolGrants *json.RawMessage `json:"tool_grants,omitempty"` + NotifyOnComplete *bool `json:"notify_on_complete,omitempty"` + NotifyOnFailure *bool `json:"notify_on_failure,omitempty"` +} + +// TaskRun records a single execution of a task. +type TaskRun struct { + ID string `json:"id" db:"id"` + TaskID string `json:"task_id" db:"task_id"` + ChannelID *string `json:"channel_id,omitempty" db:"channel_id"` + Status string `json:"status" db:"status"` // running | completed | failed | budget_exceeded | cancelled + StartedAt time.Time `json:"started_at" db:"started_at"` + CompletedAt *time.Time `json:"completed_at,omitempty" db:"completed_at"` + TokensUsed int `json:"tokens_used" db:"tokens_used"` + ToolCalls int `json:"tool_calls" db:"tool_calls"` + WallClock int `json:"wall_clock" db:"wall_clock"` // seconds + Error string `json:"error,omitempty" db:"error"` +} diff --git a/server/scheduler/scheduler.go b/server/scheduler/scheduler.go new file mode 100644 index 0000000..8a6658c --- /dev/null +++ b/server/scheduler/scheduler.go @@ -0,0 +1,283 @@ +// 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 +} diff --git a/server/store/interfaces.go b/server/store/interfaces.go index 9b9e9c7..86cb909 100644 --- a/server/store/interfaces.go +++ b/server/store/interfaces.go @@ -49,6 +49,7 @@ type Stores struct { Sessions SessionStore Surfaces SurfaceRegistryStore // v0.25.0: Surface lifecycle management Workflows WorkflowStore // v0.26.1: Workflow definitions + stages + Tasks TaskStore // v0.27.1: Task scheduling + run history } // ========================================= diff --git a/server/store/postgres/stores.go b/server/store/postgres/stores.go index f1714ea..a44af61 100644 --- a/server/store/postgres/stores.go +++ b/server/store/postgres/stores.go @@ -42,5 +42,6 @@ func NewStores(db *sql.DB) store.Stores { Sessions: NewSessionStore(), Surfaces: NewSurfaceRegistryStore(), Workflows: NewWorkflowStore(), + Tasks: NewTaskStore(), } } diff --git a/server/store/postgres/tasks.go b/server/store/postgres/tasks.go new file mode 100644 index 0000000..3922bc1 --- /dev/null +++ b/server/store/postgres/tasks.go @@ -0,0 +1,216 @@ +package postgres + +import ( + "context" + "fmt" + + "git.gobha.me/xcaliber/chat-switchboard/models" +) + +type TaskStore struct{} + +func NewTaskStore() *TaskStore { return &TaskStore{} } + +func (s *TaskStore) Create(ctx context.Context, t *models.Task) error { + toolGrants := jsonOrNull(t.ToolGrants) + return DB.QueryRowContext(ctx, ` + INSERT INTO tasks (owner_id, team_id, name, description, scope, + task_type, persona_id, model_id, system_prompt, user_prompt, + workflow_id, tool_grants, schedule, timezone, is_active, + max_tokens, max_tool_calls, max_wall_clock, output_mode, + output_channel_id, webhook_url, provider_config_id, + notify_on_complete, notify_on_failure, next_run_at) + VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18,$19,$20,$21,$22,$23,$24,$25) + RETURNING id, created_at, updated_at`, + t.OwnerID, t.TeamID, t.Name, t.Description, t.Scope, + t.TaskType, t.PersonaID, t.ModelID, t.SystemPrompt, t.UserPrompt, + t.WorkflowID, toolGrants, t.Schedule, t.Timezone, t.IsActive, + t.MaxTokens, t.MaxToolCalls, t.MaxWallClock, t.OutputMode, + t.OutputChannelID, t.WebhookURL, t.ProviderConfigID, + t.NotifyOnComplete, t.NotifyOnFailure, t.NextRunAt, + ).Scan(&t.ID, &t.CreatedAt, &t.UpdatedAt) +} + +func (s *TaskStore) GetByID(ctx context.Context, id string) (*models.Task, error) { + t := &models.Task{} + err := DB.QueryRowContext(ctx, ` + SELECT id, owner_id, team_id, name, description, scope, + task_type, persona_id, model_id, system_prompt, user_prompt, + workflow_id, tool_grants, schedule, timezone, is_active, + max_tokens, max_tool_calls, max_wall_clock, output_mode, + output_channel_id, webhook_url, provider_config_id, + notify_on_complete, notify_on_failure, + last_run_at, next_run_at, run_count, created_at, updated_at + FROM tasks WHERE id = $1`, id, + ).Scan(&t.ID, &t.OwnerID, &t.TeamID, &t.Name, &t.Description, &t.Scope, + &t.TaskType, &t.PersonaID, &t.ModelID, &t.SystemPrompt, &t.UserPrompt, + &t.WorkflowID, &t.ToolGrants, &t.Schedule, &t.Timezone, &t.IsActive, + &t.MaxTokens, &t.MaxToolCalls, &t.MaxWallClock, &t.OutputMode, + &t.OutputChannelID, &t.WebhookURL, &t.ProviderConfigID, + &t.NotifyOnComplete, &t.NotifyOnFailure, + &t.LastRunAt, &t.NextRunAt, &t.RunCount, &t.CreatedAt, &t.UpdatedAt) + if err != nil { + return nil, err + } + return t, nil +} + +func (s *TaskStore) Update(ctx context.Context, id string, p models.TaskPatch) error { + // Dynamic PATCH: only update non-nil fields + q := "UPDATE tasks SET updated_at = NOW()" + args := []interface{}{} + i := 1 + if p.Name != nil { q += comma(i, "name"); args = append(args, *p.Name); i++ } + if p.Description != nil { q += comma(i, "description"); args = append(args, *p.Description); i++ } + if p.PersonaID != nil { q += comma(i, "persona_id"); args = append(args, *p.PersonaID); i++ } + if p.ModelID != nil { q += comma(i, "model_id"); args = append(args, *p.ModelID); i++ } + if p.SystemPrompt != nil { q += comma(i, "system_prompt"); args = append(args, *p.SystemPrompt); i++ } + if p.UserPrompt != nil { q += comma(i, "user_prompt"); args = append(args, *p.UserPrompt); i++ } + if p.Schedule != nil { q += comma(i, "schedule"); args = append(args, *p.Schedule); i++ } + if p.Timezone != nil { q += comma(i, "timezone"); args = append(args, *p.Timezone); i++ } + if p.IsActive != nil { q += comma(i, "is_active"); args = append(args, *p.IsActive); i++ } + if p.MaxTokens != nil { q += comma(i, "max_tokens"); args = append(args, *p.MaxTokens); i++ } + if p.MaxToolCalls != nil { q += comma(i, "max_tool_calls"); args = append(args, *p.MaxToolCalls); i++ } + if p.MaxWallClock != nil { q += comma(i, "max_wall_clock"); args = append(args, *p.MaxWallClock); i++ } + if p.OutputMode != nil { q += comma(i, "output_mode"); args = append(args, *p.OutputMode); i++ } + if p.WebhookURL != nil { q += comma(i, "webhook_url"); args = append(args, *p.WebhookURL); i++ } + if p.NotifyOnComplete != nil { q += comma(i, "notify_on_complete"); args = append(args, *p.NotifyOnComplete); i++ } + if p.NotifyOnFailure != nil { q += comma(i, "notify_on_failure"); args = append(args, *p.NotifyOnFailure); i++ } + + q += pgWhere(i, "id") + args = append(args, id) + _, err := DB.ExecContext(ctx, q, args...) + return err +} + +func (s *TaskStore) Delete(ctx context.Context, id string) error { + _, err := DB.ExecContext(ctx, `DELETE FROM tasks WHERE id = $1`, id) + return err +} + +func (s *TaskStore) ListByOwner(ctx context.Context, ownerID string) ([]models.Task, error) { + return s.list(ctx, `WHERE owner_id = $1 ORDER BY created_at DESC`, ownerID) +} + +func (s *TaskStore) ListByTeam(ctx context.Context, teamID string) ([]models.Task, error) { + return s.list(ctx, `WHERE team_id = $1 ORDER BY created_at DESC`, teamID) +} + +func (s *TaskStore) ListAll(ctx context.Context) ([]models.Task, error) { + return s.list(ctx, `ORDER BY created_at DESC`) +} + +func (s *TaskStore) ListDue(ctx context.Context, limit int) ([]models.Task, error) { + return s.list(ctx, `WHERE is_active = true AND next_run_at <= NOW() ORDER BY next_run_at ASC LIMIT $1`, limit) +} + +func (s *TaskStore) SetNextRun(ctx context.Context, id string, nextRun interface{}) error { + _, err := DB.ExecContext(ctx, `UPDATE tasks SET next_run_at = $1, last_run_at = NOW(), updated_at = NOW() WHERE id = $2`, nextRun, id) + return err +} + +func (s *TaskStore) IncrementRunCount(ctx context.Context, id string) error { + _, err := DB.ExecContext(ctx, `UPDATE tasks SET run_count = run_count + 1, updated_at = NOW() WHERE id = $1`, id) + return err +} + +// ── Run History ───────────────────────────── + +func (s *TaskStore) CreateRun(ctx context.Context, r *models.TaskRun) error { + return DB.QueryRowContext(ctx, ` + INSERT INTO task_runs (task_id, channel_id, status) + VALUES ($1, $2, $3) RETURNING id, started_at`, + r.TaskID, r.ChannelID, r.Status, + ).Scan(&r.ID, &r.StartedAt) +} + +func (s *TaskStore) UpdateRun(ctx context.Context, id, status string, tokensUsed, toolCalls, wallClock int, errMsg string) error { + _, err := DB.ExecContext(ctx, ` + UPDATE task_runs SET status = $1, tokens_used = $2, tool_calls = $3, + wall_clock = $4, error = $5, completed_at = NOW() + WHERE id = $6`, status, tokensUsed, toolCalls, wallClock, errMsg, id) + return err +} + +func (s *TaskStore) GetActiveRun(ctx context.Context, taskID string) (*models.TaskRun, error) { + r := &models.TaskRun{} + err := DB.QueryRowContext(ctx, ` + SELECT id, task_id, channel_id, status, started_at + FROM task_runs WHERE task_id = $1 AND status = 'running' + LIMIT 1`, taskID, + ).Scan(&r.ID, &r.TaskID, &r.ChannelID, &r.Status, &r.StartedAt) + if err != nil { + return nil, err + } + return r, nil +} + +func (s *TaskStore) ListRuns(ctx context.Context, taskID string, limit int) ([]models.TaskRun, error) { + rows, err := DB.QueryContext(ctx, ` + SELECT id, task_id, channel_id, status, started_at, completed_at, + tokens_used, tool_calls, wall_clock, COALESCE(error, '') + FROM task_runs WHERE task_id = $1 + ORDER BY started_at DESC LIMIT $2`, taskID, limit) + if err != nil { + return nil, err + } + defer rows.Close() + var runs []models.TaskRun + for rows.Next() { + var r models.TaskRun + if err := rows.Scan(&r.ID, &r.TaskID, &r.ChannelID, &r.Status, &r.StartedAt, + &r.CompletedAt, &r.TokensUsed, &r.ToolCalls, &r.WallClock, &r.Error); err != nil { + continue + } + runs = append(runs, r) + } + return runs, nil +} + +// ── Helpers ───────────────────────────────── + +func (s *TaskStore) list(ctx context.Context, where string, args ...interface{}) ([]models.Task, error) { + q := `SELECT id, owner_id, team_id, name, description, scope, + task_type, persona_id, model_id, system_prompt, user_prompt, + workflow_id, schedule, timezone, is_active, + max_tokens, max_tool_calls, max_wall_clock, output_mode, + notify_on_complete, notify_on_failure, + last_run_at, next_run_at, run_count, created_at, updated_at + FROM tasks ` + where + + rows, err := DB.QueryContext(ctx, q, args...) + if err != nil { + return nil, err + } + defer rows.Close() + + var tasks []models.Task + for rows.Next() { + var t models.Task + if err := rows.Scan(&t.ID, &t.OwnerID, &t.TeamID, &t.Name, &t.Description, &t.Scope, + &t.TaskType, &t.PersonaID, &t.ModelID, &t.SystemPrompt, &t.UserPrompt, + &t.WorkflowID, &t.Schedule, &t.Timezone, &t.IsActive, + &t.MaxTokens, &t.MaxToolCalls, &t.MaxWallClock, &t.OutputMode, + &t.NotifyOnComplete, &t.NotifyOnFailure, + &t.LastRunAt, &t.NextRunAt, &t.RunCount, &t.CreatedAt, &t.UpdatedAt); err != nil { + continue + } + tasks = append(tasks, t) + } + return tasks, nil +} + +// comma builds ", column = $N" +func comma(i int, col string) string { + return ", " + col + " = " + pgArg(i) +} + +// pgArg returns "$N" +func pgArg(i int) string { + return fmt.Sprintf("$%d", i) +} + +// pgWhere returns " WHERE col = $N" +func pgWhere(i int, col string) string { + return " WHERE " + col + " = " + pgArg(i) +} diff --git a/server/store/sqlite/stores.go b/server/store/sqlite/stores.go index 562875b..7f686dd 100644 --- a/server/store/sqlite/stores.go +++ b/server/store/sqlite/stores.go @@ -41,5 +41,6 @@ func NewStores(db *sql.DB) store.Stores { RoutingPolicies: NewRoutingPolicyStore(), Sessions: NewSessionStore(), Workflows: NewWorkflowStore(), + Tasks: NewTaskStore(), } } diff --git a/server/store/sqlite/tasks.go b/server/store/sqlite/tasks.go new file mode 100644 index 0000000..b1004cb --- /dev/null +++ b/server/store/sqlite/tasks.go @@ -0,0 +1,199 @@ +package sqlite + +import ( + "context" + + "git.gobha.me/xcaliber/chat-switchboard/models" + "git.gobha.me/xcaliber/chat-switchboard/store" +) + +type TaskStore struct{} + +func NewTaskStore() *TaskStore { return &TaskStore{} } + +func (s *TaskStore) Create(ctx context.Context, t *models.Task) error { + t.ID = store.NewID() + _, err := DB.ExecContext(ctx, ` + INSERT INTO tasks (id, owner_id, team_id, name, description, scope, + task_type, persona_id, model_id, system_prompt, user_prompt, + workflow_id, tool_grants, schedule, timezone, is_active, + max_tokens, max_tool_calls, max_wall_clock, output_mode, + output_channel_id, webhook_url, provider_config_id, + notify_on_complete, notify_on_failure, next_run_at) + VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)`, + t.ID, t.OwnerID, t.TeamID, t.Name, t.Description, t.Scope, + t.TaskType, t.PersonaID, t.ModelID, t.SystemPrompt, t.UserPrompt, + t.WorkflowID, nullableJSON(t.ToolGrants), t.Schedule, t.Timezone, boolToInt(t.IsActive), + t.MaxTokens, t.MaxToolCalls, t.MaxWallClock, t.OutputMode, + t.OutputChannelID, t.WebhookURL, t.ProviderConfigID, + boolToInt(t.NotifyOnComplete), boolToInt(t.NotifyOnFailure), t.NextRunAt) + return err +} + +func (s *TaskStore) GetByID(ctx context.Context, id string) (*models.Task, error) { + t := &models.Task{} + err := DB.QueryRowContext(ctx, ` + SELECT id, owner_id, team_id, name, description, scope, + task_type, persona_id, model_id, system_prompt, user_prompt, + workflow_id, tool_grants, schedule, timezone, is_active, + max_tokens, max_tool_calls, max_wall_clock, output_mode, + output_channel_id, webhook_url, provider_config_id, + notify_on_complete, notify_on_failure, + last_run_at, next_run_at, run_count, created_at, updated_at + FROM tasks WHERE id = ?`, id, + ).Scan(&t.ID, &t.OwnerID, &t.TeamID, &t.Name, &t.Description, &t.Scope, + &t.TaskType, &t.PersonaID, &t.ModelID, &t.SystemPrompt, &t.UserPrompt, + &t.WorkflowID, &t.ToolGrants, &t.Schedule, &t.Timezone, &t.IsActive, + &t.MaxTokens, &t.MaxToolCalls, &t.MaxWallClock, &t.OutputMode, + &t.OutputChannelID, &t.WebhookURL, &t.ProviderConfigID, + &t.NotifyOnComplete, &t.NotifyOnFailure, + &t.LastRunAt, &t.NextRunAt, &t.RunCount, &t.CreatedAt, &t.UpdatedAt) + if err != nil { + return nil, err + } + return t, nil +} + +func (s *TaskStore) Update(ctx context.Context, id string, p models.TaskPatch) error { + q := "UPDATE tasks SET updated_at = datetime('now')" + args := []interface{}{} + if p.Name != nil { q += ", name = ?"; args = append(args, *p.Name) } + if p.Description != nil { q += ", description = ?"; args = append(args, *p.Description) } + if p.PersonaID != nil { q += ", persona_id = ?"; args = append(args, *p.PersonaID) } + if p.ModelID != nil { q += ", model_id = ?"; args = append(args, *p.ModelID) } + if p.SystemPrompt != nil { q += ", system_prompt = ?"; args = append(args, *p.SystemPrompt) } + if p.UserPrompt != nil { q += ", user_prompt = ?"; args = append(args, *p.UserPrompt) } + if p.Schedule != nil { q += ", schedule = ?"; args = append(args, *p.Schedule) } + if p.Timezone != nil { q += ", timezone = ?"; args = append(args, *p.Timezone) } + if p.IsActive != nil { q += ", is_active = ?"; args = append(args, boolToInt(*p.IsActive)) } + if p.MaxTokens != nil { q += ", max_tokens = ?"; args = append(args, *p.MaxTokens) } + if p.MaxToolCalls != nil { q += ", max_tool_calls = ?"; args = append(args, *p.MaxToolCalls) } + if p.MaxWallClock != nil { q += ", max_wall_clock = ?"; args = append(args, *p.MaxWallClock) } + if p.OutputMode != nil { q += ", output_mode = ?"; args = append(args, *p.OutputMode) } + if p.WebhookURL != nil { q += ", webhook_url = ?"; args = append(args, *p.WebhookURL) } + if p.NotifyOnComplete != nil { q += ", notify_on_complete = ?"; args = append(args, boolToInt(*p.NotifyOnComplete)) } + if p.NotifyOnFailure != nil { q += ", notify_on_failure = ?"; args = append(args, boolToInt(*p.NotifyOnFailure)) } + q += " WHERE id = ?" + args = append(args, id) + _, err := DB.ExecContext(ctx, q, args...) + return err +} + +func (s *TaskStore) Delete(ctx context.Context, id string) error { + _, err := DB.ExecContext(ctx, `DELETE FROM tasks WHERE id = ?`, id) + return err +} + +func (s *TaskStore) ListByOwner(ctx context.Context, ownerID string) ([]models.Task, error) { + return s.list(ctx, `WHERE owner_id = ? ORDER BY created_at DESC`, ownerID) +} + +func (s *TaskStore) ListByTeam(ctx context.Context, teamID string) ([]models.Task, error) { + return s.list(ctx, `WHERE team_id = ? ORDER BY created_at DESC`, teamID) +} + +func (s *TaskStore) ListAll(ctx context.Context) ([]models.Task, error) { + return s.list(ctx, `ORDER BY created_at DESC`) +} + +func (s *TaskStore) ListDue(ctx context.Context, limit int) ([]models.Task, error) { + return s.list(ctx, `WHERE is_active = 1 AND next_run_at <= datetime('now') ORDER BY next_run_at ASC LIMIT ?`, limit) +} + +func (s *TaskStore) SetNextRun(ctx context.Context, id string, nextRun interface{}) error { + _, err := DB.ExecContext(ctx, `UPDATE tasks SET next_run_at = ?, last_run_at = datetime('now'), updated_at = datetime('now') WHERE id = ?`, nextRun, id) + return err +} + +func (s *TaskStore) IncrementRunCount(ctx context.Context, id string) error { + _, err := DB.ExecContext(ctx, `UPDATE tasks SET run_count = run_count + 1, updated_at = datetime('now') WHERE id = ?`, id) + return err +} + +func (s *TaskStore) CreateRun(ctx context.Context, r *models.TaskRun) error { + r.ID = store.NewID() + _, err := DB.ExecContext(ctx, ` + INSERT INTO task_runs (id, task_id, channel_id, status) + VALUES (?, ?, ?, ?)`, r.ID, r.TaskID, r.ChannelID, r.Status) + return err +} + +func (s *TaskStore) UpdateRun(ctx context.Context, id, status string, tokensUsed, toolCalls, wallClock int, errMsg string) error { + _, err := DB.ExecContext(ctx, ` + UPDATE task_runs SET status = ?, tokens_used = ?, tool_calls = ?, + wall_clock = ?, error = ?, completed_at = datetime('now') + WHERE id = ?`, status, tokensUsed, toolCalls, wallClock, errMsg, id) + return err +} + +func (s *TaskStore) GetActiveRun(ctx context.Context, taskID string) (*models.TaskRun, error) { + r := &models.TaskRun{} + err := DB.QueryRowContext(ctx, ` + SELECT id, task_id, channel_id, status, started_at + FROM task_runs WHERE task_id = ? AND status = 'running' + LIMIT 1`, taskID).Scan(&r.ID, &r.TaskID, &r.ChannelID, &r.Status, &r.StartedAt) + if err != nil { + return nil, err + } + return r, nil +} + +func (s *TaskStore) ListRuns(ctx context.Context, taskID string, limit int) ([]models.TaskRun, error) { + rows, err := DB.QueryContext(ctx, ` + SELECT id, task_id, channel_id, status, started_at, completed_at, + tokens_used, tool_calls, wall_clock, COALESCE(error, '') + FROM task_runs WHERE task_id = ? + ORDER BY started_at DESC LIMIT ?`, taskID, limit) + if err != nil { + return nil, err + } + defer rows.Close() + var runs []models.TaskRun + for rows.Next() { + var r models.TaskRun + if err := rows.Scan(&r.ID, &r.TaskID, &r.ChannelID, &r.Status, &r.StartedAt, + &r.CompletedAt, &r.TokensUsed, &r.ToolCalls, &r.WallClock, &r.Error); err != nil { + continue + } + runs = append(runs, r) + } + return runs, nil +} + +func (s *TaskStore) list(ctx context.Context, where string, args ...interface{}) ([]models.Task, error) { + q := `SELECT id, owner_id, team_id, name, description, scope, + task_type, persona_id, model_id, system_prompt, user_prompt, + workflow_id, schedule, timezone, is_active, + max_tokens, max_tool_calls, max_wall_clock, output_mode, + notify_on_complete, notify_on_failure, + last_run_at, next_run_at, run_count, created_at, updated_at + FROM tasks ` + where + + rows, err := DB.QueryContext(ctx, q, args...) + if err != nil { + return nil, err + } + defer rows.Close() + + var tasks []models.Task + for rows.Next() { + var t models.Task + if err := rows.Scan(&t.ID, &t.OwnerID, &t.TeamID, &t.Name, &t.Description, &t.Scope, + &t.TaskType, &t.PersonaID, &t.ModelID, &t.SystemPrompt, &t.UserPrompt, + &t.WorkflowID, &t.Schedule, &t.Timezone, &t.IsActive, + &t.MaxTokens, &t.MaxToolCalls, &t.MaxWallClock, &t.OutputMode, + &t.NotifyOnComplete, &t.NotifyOnFailure, + &t.LastRunAt, &t.NextRunAt, &t.RunCount, &t.CreatedAt, &t.UpdatedAt); err != nil { + continue + } + tasks = append(tasks, t) + } + return tasks, nil +} + +func nullableJSON(data []byte) interface{} { + if len(data) == 0 || string(data) == "null" { + return nil + } + return string(data) +} diff --git a/server/store/task_iface.go b/server/store/task_iface.go new file mode 100644 index 0000000..f58e6d4 --- /dev/null +++ b/server/store/task_iface.go @@ -0,0 +1,30 @@ +package store + +import ( + "context" + + "git.gobha.me/xcaliber/chat-switchboard/models" +) + +// TaskStore manages task definitions and run history. +type TaskStore interface { + // Task CRUD + Create(ctx context.Context, t *models.Task) error + GetByID(ctx context.Context, id string) (*models.Task, error) + Update(ctx context.Context, id string, patch models.TaskPatch) error + Delete(ctx context.Context, id string) error + ListByOwner(ctx context.Context, ownerID string) ([]models.Task, error) + ListByTeam(ctx context.Context, teamID string) ([]models.Task, error) + ListAll(ctx context.Context) ([]models.Task, error) + + // Scheduler queries + ListDue(ctx context.Context, limit int) ([]models.Task, error) + SetNextRun(ctx context.Context, id string, nextRun interface{}) error + IncrementRunCount(ctx context.Context, id string) error + + // Run history + CreateRun(ctx context.Context, r *models.TaskRun) error + UpdateRun(ctx context.Context, id string, status string, tokensUsed, toolCalls, wallClock int, errMsg string) error + GetActiveRun(ctx context.Context, taskID string) (*models.TaskRun, error) + ListRuns(ctx context.Context, taskID string, limit int) ([]models.TaskRun, error) +}