From 8e08f3e4b0ddaeab0c17e282daa95437a1901828 Mon Sep 17 00:00:00 2001 From: xcaliber Date: Wed, 11 Mar 2026 19:41:04 +0000 Subject: [PATCH] Changeset 0.28.0.2 (#174) --- docs/ICD/tasks.md | 116 ++- server/auth/permissions.go | 2 + server/database/migrations/019_tasks.sql | 32 +- .../database/migrations/sqlite/019_tasks.sql | 27 +- server/handlers/integration_test.go | 42 + server/handlers/task_test.go | 944 ++++++++++++++++++ server/handlers/tasks.go | 188 +++- server/main.go | 4 + server/models/models_task.go | 70 +- server/scheduler/cron.go | 49 - server/scheduler/executor.go | 77 +- server/scheduler/scheduler.go | 62 +- server/scheduler/task_config.go | 110 -- server/store/postgres/tasks.go | 162 ++- server/store/sqlite/helpers.go | 8 + server/store/sqlite/tasks.go | 204 +++- server/store/task_iface.go | 6 + server/tools/task_create.go | 10 +- server/webhook/webhook.go | 5 +- 19 files changed, 1734 insertions(+), 384 deletions(-) create mode 100644 server/handlers/task_test.go delete mode 100644 server/scheduler/cron.go delete mode 100644 server/scheduler/task_config.go diff --git a/docs/ICD/tasks.md b/docs/ICD/tasks.md index e41af69..c7d83a3 100644 --- a/docs/ICD/tasks.md +++ b/docs/ICD/tasks.md @@ -1,7 +1,8 @@ # Tasks -Autonomous agent scheduling. A task is a cron-driven or one-shot prompt -execution in a headless service channel — no human participant in the loop. +Autonomous agent scheduling. A task is a cron-driven, one-shot, or +webhook-triggered execution in a headless service channel — no human +participant in the loop. ## Configuration @@ -63,17 +64,20 @@ POST /tasks } ``` -`task_type`: `prompt` (direct LLM execution) or `workflow` (instantiate -a workflow in the service channel). +`task_type`: `prompt` (direct LLM execution), `workflow` (deferred — +not yet implemented; creation is rejected at the API), or `action` +(no LLM — relay/webhook only; requires `task.action` permission). -`schedule`: cron expression or `once`. Supported patterns: `@hourly`, -`@daily`, `@weekly`, `*/N * * * *` (every N minutes), full 5-field cron -via `robfig/cron/v3`. +`schedule`: cron expression, `once`, or `webhook`. Supported cron +patterns: `@hourly`, `@daily`, `@weekly`, `*/N * * * *` (every N +minutes), full 5-field cron via `robfig/cron/v3`. When `schedule` is +`webhook`, the task fires only on inbound POST to its trigger URL. `output_mode`: `channel` (default — output stays in service channel), `note` (creates a note from the output), `webhook` (POST result to URL). -**Auth:** `tasks.create` permission required. +**Auth:** `tasks.create` permission required. Additionally, `task.action` +permission is required for `task_type: "action"`. ### Get Task @@ -89,7 +93,11 @@ GET /tasks/:id PUT /tasks/:id ``` -Partial update. All fields from create are accepted. +Partial update. Accepted fields: `name`, `description`, `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`. **Auth:** `tasks.create` permission, must be owner. @@ -108,7 +116,7 @@ POST /tasks/:id/run ``` Immediately executes the task regardless of schedule. Creates a new -task run. Returns 409 if a run is already active. +task run. Returns 409 if a run is already active or queued. **Auth:** `tasks.create` permission, must be owner. @@ -132,6 +140,39 @@ Returns run history for the task, most recent first. **Auth:** Owner or admin. +## Webhook Trigger (Inbound) + +External systems fire a task by POSTing to its trigger URL. + +``` +POST /api/v1/hooks/t/:token +``` + +The `:token` is a per-task secret generated at creation time (returned +in the `trigger_token` field). The request body is stored as +`trigger_payload` on the run record and forwarded to the executor: + +- **Prompt tasks:** payload is prepended to the user prompt as context. +- **Action tasks:** payload is relayed to the outbound webhook as-is. +- **Workflow tasks:** (deferred) payload becomes initial stage data. + +Returns `202 Accepted`: +```json +{ + "triggered": true, + "run_id": "uuid", + "task_id": "uuid" +} +``` + +Returns `409 Conflict` if a run is already active or queued. +Returns `410 Gone` if the task is inactive. + +**Auth:** None (token-based). The trigger token *is* the credential. + +**Rate limiting:** Governed by global request rate limits. No per-task +rate limiting in v0.28 — defer to a future version if needed. + ## Team Tasks Team-scoped tasks visible to all team members, manageable by team admins. @@ -209,16 +250,17 @@ DELETE /admin/tasks/:id "name": "Morning News Digest", "description": "...", "scope": "personal|team|global", - "task_type": "prompt|workflow", + "task_type": "prompt|workflow|action", "persona_id": "uuid|null", "model_id": "claude-sonnet-4-20250514", "system_prompt": "...", "user_prompt": "...", "workflow_id": "uuid|null", "tool_grants": ["web_search", "url_fetch"], - "schedule": "@daily", + "schedule": "@daily|once|webhook", "timezone": "America/New_York", "is_active": true, + "trigger_token": "hex-string|null", "max_tokens": 4096, "max_tool_calls": 10, "max_wall_clock": 300, @@ -244,7 +286,8 @@ DELETE /admin/tasks/:id "id": "uuid", "task_id": "uuid", "channel_id": "uuid|null", - "status": "running|completed|failed|budget_exceeded|cancelled", + "status": "queued|running|completed|failed|budget_exceeded|cancelled", + "trigger_payload": "...|null", "started_at": "...", "completed_at": "...|null", "tokens_used": 1234, @@ -260,18 +303,29 @@ The `TaskScheduler` is a background goroutine polling every 30 seconds. 1. Finds tasks where `next_run_at <= now` and `is_active = true` 2. Skips if an active run exists (`GetActiveRun` returns non-nil) -3. Creates or reuses a `service` channel (`output_channel_id`) -4. Persists the `user_prompt` as a message in the service channel -5. Runs `CoreToolLoop` (headless completion — same as streaming but no SSE) -6. Enforces budgets: `max_tokens`, `max_tool_calls`, `max_wall_clock` -7. On budget breach: status = `budget_exceeded`, owner notified -8. On completion: updates `last_run_at`, calculates `next_run_at` +3. Adopts a queued run if one exists (from webhook trigger), otherwise + creates a new run record +4. Marks `last_run_at` (records when execution started, not when it finished) +5. Creates or reuses a `service` channel (`output_channel_id`) +6. Persists the `user_prompt` as a message in the service channel + (with trigger payload prepended if present) +7. For **prompt** tasks: runs `CoreToolLoop` (headless completion) +8. For **action** tasks: skips LLM, relays trigger payload to webhook +9. Enforces budgets: `max_tokens`, `max_tool_calls`, `max_wall_clock` +10. On budget breach: status = `budget_exceeded`, owner notified +11. On completion: calculates `next_run_at` + +**Workflow tasks:** Not yet implemented. The API rejects +`task_type: "workflow"` at creation time. Reserved for a future version. **Provider resolution:** BYOK → team provider → global provider → routing policy. If `tasks.personal_require_byok` is true, personal tasks that -don't have a BYOK provider fail at step 5. +don't have a BYOK provider fail at step 7. -## Webhooks +**Action tasks** skip steps 6–9. They create the service channel, fire +the outbound webhook with the trigger payload, and complete. + +## Webhooks (Outbound) Tasks with `webhook_url` fire a POST on completion: @@ -279,12 +333,28 @@ Tasks with `webhook_url` fire a POST on completion: { "task_id": "uuid", "run_id": "uuid", + "task_name": "Morning News Digest", + "channel_id": "uuid", "status": "completed|failed|budget_exceeded", + "completed_at": "2025-03-11T12:00:00Z", "output": "...", "tokens_used": 1234, - "timestamp": "..." + "error": "...|null" } ``` Signed with HMAC-SHA256 using `webhook_secret` in the -`X-Switchboard-Signature` header. Retry: 3 attempts, exponential backoff. +`X-Switchboard-Signature` header. Retry: 3 attempts, exponential +backoff (1s, 5s, 25s). + +## Task Chaining + +Task A's outbound `webhook_url` can be Task B's trigger URL: + +``` +Task A (webhook_url) → POST /api/v1/hooks/t/ → Task B runs +``` + +This enables multi-step pipelines: CI failure → summarize with LLM → +post to Slack. Each link is a separate task with its own type, budget, +and notification preferences. diff --git a/server/auth/permissions.go b/server/auth/permissions.go index f6a893d..5b1c68f 100644 --- a/server/auth/permissions.go +++ b/server/auth/permissions.go @@ -28,6 +28,7 @@ const ( PermTokenUnlimited = "token.unlimited" // bypass token budgets PermTaskCreate = "task.create" // create scheduled tasks (v0.27.2) PermTaskAdmin = "task.admin" // manage all tasks, set global task config (v0.27.2) + PermTaskAction = "task.action" // create non-LLM action tasks (v0.28.0) ) // AllPermissions is the complete set of valid permission strings. @@ -47,6 +48,7 @@ var AllPermissions = []string{ PermTokenUnlimited, PermTaskCreate, PermTaskAdmin, + PermTaskAction, } // ── Resolution ────────────────────────────── diff --git a/server/database/migrations/019_tasks.sql b/server/database/migrations/019_tasks.sql index 759c223..3cf44e0 100644 --- a/server/database/migrations/019_tasks.sql +++ b/server/database/migrations/019_tasks.sql @@ -3,6 +3,7 @@ -- ========================================== -- Task definitions and run history for autonomous agents. -- Consolidated v0.28.0: merges 026 + 027 (webhook_secret inline). +-- v0.28.0-cs3: trigger_token, action task_type, queued run status. -- ========================================== -- ========================================= @@ -20,7 +21,7 @@ CREATE TABLE IF NOT EXISTS tasks ( -- What to run task_type TEXT NOT NULL DEFAULT 'prompt' - CHECK (task_type IN ('prompt', 'workflow')), + CHECK (task_type IN ('prompt', 'workflow', 'action')), persona_id UUID REFERENCES personas(id) ON DELETE SET NULL, model_id TEXT, system_prompt TEXT DEFAULT '', @@ -28,11 +29,14 @@ CREATE TABLE IF NOT EXISTS tasks ( workflow_id UUID REFERENCES workflows(id) ON DELETE SET NULL, tool_grants JSONB, - -- Schedule + -- Schedule: cron expression, "once", or "webhook" schedule TEXT NOT NULL, timezone TEXT NOT NULL DEFAULT 'UTC', is_active BOOLEAN NOT NULL DEFAULT true, + -- Webhook trigger (inbound) — token-based auth for /hooks/t/:token + trigger_token TEXT UNIQUE, + -- Execution policy max_tokens INTEGER NOT NULL DEFAULT 4096, max_tool_calls INTEGER NOT NULL DEFAULT 10, @@ -61,6 +65,7 @@ CREATE TABLE IF NOT EXISTS tasks ( 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); +CREATE UNIQUE INDEX IF NOT EXISTS idx_tasks_trigger_token ON tasks (trigger_token) WHERE trigger_token IS NOT NULL; -- ========================================= @@ -68,17 +73,18 @@ CREATE INDEX IF NOT EXISTS idx_tasks_team ON tasks (team_id); -- ========================================= 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 + 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 ('queued', 'running', 'completed', 'failed', 'budget_exceeded', 'cancelled')), + trigger_payload TEXT, + 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); diff --git a/server/database/migrations/sqlite/019_tasks.sql b/server/database/migrations/sqlite/019_tasks.sql index 0224684..de405ea 100644 --- a/server/database/migrations/sqlite/019_tasks.sql +++ b/server/database/migrations/sqlite/019_tasks.sql @@ -1,5 +1,6 @@ -- Chat Switchboard — 019 Tasks (SQLite) -- Consolidated v0.28.0: merges 026 + 027 +-- v0.28.0-cs3: trigger_token, action task_type, queued run status. CREATE TABLE IF NOT EXISTS tasks ( id TEXT PRIMARY KEY, @@ -18,6 +19,7 @@ CREATE TABLE IF NOT EXISTS tasks ( schedule TEXT NOT NULL, timezone TEXT NOT NULL DEFAULT 'UTC', is_active INTEGER NOT NULL DEFAULT 1, + trigger_token TEXT UNIQUE, max_tokens INTEGER NOT NULL DEFAULT 4096, max_tool_calls INTEGER NOT NULL DEFAULT 10, max_wall_clock INTEGER NOT NULL DEFAULT 300, @@ -35,21 +37,24 @@ CREATE TABLE IF NOT EXISTS tasks ( 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_next_run ON tasks (next_run_at) WHERE is_active = 1; 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 UNIQUE INDEX IF NOT EXISTS idx_tasks_trigger_token ON tasks (trigger_token) WHERE trigger_token IS NOT NULL; 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 + 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', + trigger_payload TEXT, + 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); +CREATE INDEX IF NOT EXISTS idx_task_runs_status ON task_runs (task_id, status); diff --git a/server/handlers/integration_test.go b/server/handlers/integration_test.go index cda7297..8bff3ae 100644 --- a/server/handlers/integration_test.go +++ b/server/handlers/integration_test.go @@ -191,6 +191,26 @@ func setupHarness(t *testing.T) *testHarness { teamScoped.GET("/roles", teamRoles.ListTeamRoles) teamScoped.PUT("/roles/:role", teamRoles.UpdateTeamRole) teamScoped.DELETE("/roles/:role", teamRoles.DeleteTeamRole) + + // Team tasks — admin CRUD (v0.28.0-audit) + teamTaskH := NewTaskHandler(stores) + teamScoped.POST("/tasks", middleware.RequirePermission(authpkg.PermTaskCreate, stores), teamTaskH.CreateTeamTask) + teamScoped.PUT("/tasks/:id", middleware.RequirePermission(authpkg.PermTaskCreate, stores), teamTaskH.Update) + teamScoped.DELETE("/tasks/:id", middleware.RequirePermission(authpkg.PermTaskCreate, stores), teamTaskH.Delete) + teamScoped.POST("/tasks/:id/run", middleware.RequirePermission(authpkg.PermTaskCreate, stores), teamTaskH.RunNow) + teamScoped.POST("/tasks/:id/kill", middleware.RequirePermission(authpkg.PermTaskCreate, stores), teamTaskH.KillRun) + } + + // Team task viewing for all members (v0.28.0-audit) + // NOTE: Uses RequireTeamMember which has raw $1 placeholders — admin + // users bypass the DB query so these tests work on both dialects when + // the caller is an admin. Non-admin member tests require Postgres. + teamMemberRoutes := protected.Group("/teams/:teamId") + teamMemberRoutes.Use(middleware.RequireTeamMember()) + { + teamMemberTaskH := NewTaskHandler(stores) + teamMemberRoutes.GET("/tasks", teamMemberTaskH.ListTeamTasks) + teamMemberRoutes.GET("/tasks/:id/runs", teamMemberTaskH.ListRuns) } // User usage @@ -263,6 +283,21 @@ func setupHarness(t *testing.T) *testHarness { protected.PUT("/avatar", settings.UploadAvatar) protected.DELETE("/avatar", settings.DeleteAvatar) + // Tasks (v0.28.0) + taskH := NewTaskHandler(stores) + protected.GET("/tasks", taskH.ListMine) + protected.POST("/tasks", middleware.RequirePermission(authpkg.PermTaskCreate, stores), taskH.Create) + protected.GET("/tasks/:id", taskH.Get) + protected.PUT("/tasks/:id", middleware.RequirePermission(authpkg.PermTaskCreate, stores), taskH.Update) + protected.DELETE("/tasks/:id", middleware.RequirePermission(authpkg.PermTaskCreate, stores), taskH.Delete) + protected.GET("/tasks/:id/runs", taskH.ListRuns) + protected.POST("/tasks/:id/run", middleware.RequirePermission(authpkg.PermTaskCreate, stores), taskH.RunNow) + protected.POST("/tasks/:id/kill", middleware.RequirePermission(authpkg.PermTaskCreate, stores), taskH.KillRun) + + // Webhook trigger (unauthenticated, token-based) + triggerH := NewTriggerHandler(stores) + api.POST("/hooks/t/:token", triggerH.Handle) + // Admin routes admin := api.Group("/admin") admin.Use(middleware.Auth(cfg), middleware.RequireAdmin()) @@ -323,6 +358,13 @@ func setupHarness(t *testing.T) *testHarness { admin.PUT("/pricing", usageH.UpsertPricing) admin.DELETE("/pricing/:provider/:model", usageH.DeletePricing) + // Admin tasks + taskAdm := NewTaskHandler(stores) + admin.GET("/tasks", taskAdm.ListAll) + admin.POST("/tasks/:id/run", taskAdm.RunNow) + admin.POST("/tasks/:id/kill", taskAdm.KillRun) + admin.DELETE("/tasks/:id", taskAdm.Delete) + return &testHarness{router: r, t: t} } diff --git a/server/handlers/task_test.go b/server/handlers/task_test.go new file mode 100644 index 0000000..6c76f2b --- /dev/null +++ b/server/handlers/task_test.go @@ -0,0 +1,944 @@ +package handlers + +import ( + "encoding/json" + "fmt" + "net/http" + "strings" + "testing" + + "git.gobha.me/xcaliber/chat-switchboard/database" +) + +// ═══════════════════════════════════════════════ +// Task CRUD Tests +// ═══════════════════════════════════════════════ + +func TestTask_CreateAndGet(t *testing.T) { + h := setupHarness(t) + _, token := h.createAdminUser("taskowner", "taskowner@test.com") + + // Create a prompt task + w := h.request("POST", "/api/v1/tasks", token, map[string]interface{}{ + "name": "Test Task", + "description": "Unit test task", + "task_type": "prompt", + "schedule": "@daily", + "user_prompt": "Summarize news", + "model_id": "test-model", + "timezone": "UTC", + }) + if w.Code != http.StatusCreated { + t.Fatalf("create task: want 201, got %d: %s", w.Code, w.Body.String()) + } + + var task map[string]interface{} + decode(w, &task) + + taskID := task["id"].(string) + if taskID == "" { + t.Fatal("task should have an id") + } + if task["name"].(string) != "Test Task" { + t.Fatalf("name mismatch: got %q", task["name"]) + } + if task["is_active"] != true { + t.Fatal("task should be active by default") + } + if task["schedule"].(string) != "@daily" { + t.Fatalf("schedule mismatch: got %q", task["schedule"]) + } + // next_run_at should be computed + if task["next_run_at"] == nil { + t.Fatal("next_run_at should be set for cron task") + } + // created_at should not be zero time + createdAt, _ := task["created_at"].(string) + if createdAt == "" || createdAt == "0001-01-01T00:00:00Z" { + t.Fatalf("created_at should be a real timestamp, got %q", createdAt) + } + + // Get it back + w = h.request("GET", "/api/v1/tasks/"+taskID, token, nil) + if w.Code != http.StatusOK { + t.Fatalf("get task: want 200, got %d: %s", w.Code, w.Body.String()) + } + var got map[string]interface{} + decode(w, &got) + if got["id"].(string) != taskID { + t.Fatal("get returned wrong task") + } +} + +func TestTask_ListMine(t *testing.T) { + h := setupHarness(t) + _, token := h.createAdminUser("lister", "lister@test.com") + + // Create 2 tasks + for i := 0; i < 2; i++ { + w := h.request("POST", "/api/v1/tasks", token, map[string]interface{}{ + "name": fmt.Sprintf("Task %d", i), + "task_type": "prompt", + "schedule": "@hourly", + "user_prompt": "do stuff", + "model_id": "m", + }) + if w.Code != http.StatusCreated { + t.Fatalf("create task %d: want 201, got %d", i, w.Code) + } + } + + // List + w := h.request("GET", "/api/v1/tasks", token, nil) + if w.Code != http.StatusOK { + t.Fatalf("list: want 200, got %d", w.Code) + } + var resp map[string]interface{} + decode(w, &resp) + data := resp["data"].([]interface{}) + if len(data) != 2 { + t.Fatalf("expected 2 tasks, got %d", len(data)) + } +} + +func TestTask_Update(t *testing.T) { + h := setupHarness(t) + _, token := h.createAdminUser("updater", "updater@test.com") + + // Create + w := h.request("POST", "/api/v1/tasks", token, map[string]interface{}{ + "name": "Before", "task_type": "prompt", "schedule": "@daily", + "user_prompt": "old", "model_id": "m", + }) + if w.Code != http.StatusCreated { + t.Fatalf("create: %d: %s", w.Code, w.Body.String()) + } + var task map[string]interface{} + decode(w, &task) + taskID := task["id"].(string) + + // Patch name and user_prompt + w = h.request("PUT", "/api/v1/tasks/"+taskID, token, map[string]interface{}{ + "name": "After", + "user_prompt": "new prompt", + }) + if w.Code != http.StatusOK { + t.Fatalf("update: want 200, got %d: %s", w.Code, w.Body.String()) + } + var updated map[string]interface{} + decode(w, &updated) + if updated["name"].(string) != "After" { + t.Fatalf("name not updated: got %q", updated["name"]) + } + if updated["user_prompt"].(string) != "new prompt" { + t.Fatalf("user_prompt not updated: got %q", updated["user_prompt"]) + } +} + +func TestTask_UpdateToolGrants(t *testing.T) { + h := setupHarness(t) + _, token := h.createAdminUser("grantuser", "grant@test.com") + + // Create with tool_grants + w := h.request("POST", "/api/v1/tasks", token, map[string]interface{}{ + "name": "Grants", "task_type": "prompt", "schedule": "@daily", + "user_prompt": "x", "model_id": "m", + "tool_grants": []string{"web_search"}, + }) + if w.Code != http.StatusCreated { + t.Fatalf("create: %d: %s", w.Code, w.Body.String()) + } + var task map[string]interface{} + decode(w, &task) + taskID := task["id"].(string) + + // Patch tool_grants + newGrants := json.RawMessage(`["url_fetch","web_search"]`) + w = h.request("PUT", "/api/v1/tasks/"+taskID, token, map[string]interface{}{ + "tool_grants": newGrants, + }) + if w.Code != http.StatusOK { + t.Fatalf("update tool_grants: want 200, got %d: %s", w.Code, w.Body.String()) + } + + // Verify via GET + w = h.request("GET", "/api/v1/tasks/"+taskID, token, nil) + if w.Code != http.StatusOK { + t.Fatalf("get: %d", w.Code) + } + var got map[string]interface{} + decode(w, &got) + grants, ok := got["tool_grants"].([]interface{}) + if !ok || len(grants) != 2 { + t.Fatalf("tool_grants not updated: got %v", got["tool_grants"]) + } +} + +func TestTask_Delete(t *testing.T) { + h := setupHarness(t) + _, token := h.createAdminUser("deleter", "deleter@test.com") + + w := h.request("POST", "/api/v1/tasks", token, map[string]interface{}{ + "name": "Doomed", "task_type": "prompt", "schedule": "once", + "user_prompt": "bye", "model_id": "m", + }) + if w.Code != http.StatusCreated { + t.Fatalf("create: %d", w.Code) + } + var task map[string]interface{} + decode(w, &task) + taskID := task["id"].(string) + + // Delete + w = h.request("DELETE", "/api/v1/tasks/"+taskID, token, nil) + if w.Code != http.StatusOK { + t.Fatalf("delete: want 200, got %d: %s", w.Code, w.Body.String()) + } + + // Verify gone + w = h.request("GET", "/api/v1/tasks/"+taskID, token, nil) + if w.Code != http.StatusNotFound { + t.Fatalf("get deleted: want 404, got %d", w.Code) + } +} + +// ═══════════════════════════════════════════════ +// Ownership / Access Control Tests +// ═══════════════════════════════════════════════ + +func TestTask_OwnershipIsolation(t *testing.T) { + h := setupHarness(t) + _, tokenA := h.createAdminUser("ownerA", "ownerA@test.com") + + // User B — non-admin regular user (seeded directly, bypassing register endpoint) + userB := database.SeedTestUser(h.t, "ownerB", "ownerB@test.com") + database.TestDB.Exec(dialectSQL("UPDATE users SET is_active = true WHERE id = $1"), userB) + tokenB := makeToken(userB, "ownerB@test.com", "user") + + // A creates a task + w := h.request("POST", "/api/v1/tasks", tokenA, map[string]interface{}{ + "name": "A's Task", "task_type": "prompt", "schedule": "@daily", + "user_prompt": "private", "model_id": "m", + }) + if w.Code != http.StatusCreated { + t.Fatalf("create: %d: %s", w.Code, w.Body.String()) + } + var task map[string]interface{} + decode(w, &task) + taskID := task["id"].(string) + + // B cannot GET it (no RequirePermission on GET — reaches handler → 404) + w = h.request("GET", "/api/v1/tasks/"+taskID, tokenB, nil) + if w.Code != http.StatusNotFound { + t.Fatalf("non-owner GET: want 404, got %d", w.Code) + } + + // B cannot UPDATE it (RequirePermission blocks → 403, or handler blocks → 404) + w = h.request("PUT", "/api/v1/tasks/"+taskID, tokenB, map[string]interface{}{ + "name": "Hijacked", + }) + if w.Code == http.StatusOK { + t.Fatal("non-owner PUT should not succeed") + } + + // B cannot DELETE it + w = h.request("DELETE", "/api/v1/tasks/"+taskID, tokenB, nil) + if w.Code == http.StatusOK { + t.Fatal("non-owner DELETE should not succeed") + } + + // B's list should be empty (no tasks of their own) + w = h.request("GET", "/api/v1/tasks", tokenB, nil) + if w.Code != http.StatusOK { + t.Fatalf("list: want 200, got %d", w.Code) + } + var resp map[string]interface{} + decode(w, &resp) + data := resp["data"].([]interface{}) + if len(data) != 0 { + t.Fatalf("expected 0 tasks for B, got %d", len(data)) + } +} + +// ═══════════════════════════════════════════════ +// Validation Tests +// ═══════════════════════════════════════════════ + +func TestTask_CreateValidation(t *testing.T) { + h := setupHarness(t) + _, token := h.createAdminUser("validator", "validator@test.com") + + tests := []struct { + name string + body map[string]interface{} + want int + }{ + { + name: "missing name", + body: map[string]interface{}{"task_type": "prompt", "schedule": "@daily", "user_prompt": "x", "model_id": "m"}, + want: http.StatusBadRequest, + }, + { + name: "missing schedule", + body: map[string]interface{}{"name": "X", "task_type": "prompt", "user_prompt": "x", "model_id": "m"}, + want: http.StatusBadRequest, + }, + { + name: "prompt without user_prompt", + body: map[string]interface{}{"name": "X", "task_type": "prompt", "schedule": "@daily", "model_id": "m"}, + want: http.StatusBadRequest, + }, + { + name: "invalid cron", + body: map[string]interface{}{"name": "X", "task_type": "prompt", "schedule": "not-a-cron", "user_prompt": "x", "model_id": "m"}, + want: http.StatusBadRequest, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + w := h.request("POST", "/api/v1/tasks", token, tc.body) + if w.Code != tc.want { + t.Fatalf("want %d, got %d: %s", tc.want, w.Code, w.Body.String()) + } + }) + } +} + +func TestTask_UpdateInvalidSchedule(t *testing.T) { + h := setupHarness(t) + _, token := h.createAdminUser("badsched", "badsched@test.com") + + w := h.request("POST", "/api/v1/tasks", token, map[string]interface{}{ + "name": "T", "task_type": "prompt", "schedule": "@daily", + "user_prompt": "x", "model_id": "m", + }) + if w.Code != http.StatusCreated { + t.Fatalf("create: %d", w.Code) + } + var task map[string]interface{} + decode(w, &task) + + // Try to update with invalid schedule + w = h.request("PUT", "/api/v1/tasks/"+task["id"].(string), token, map[string]interface{}{ + "schedule": "garbage-cron", + }) + if w.Code != http.StatusBadRequest { + t.Fatalf("invalid schedule update: want 400, got %d: %s", w.Code, w.Body.String()) + } +} + +// ═══════════════════════════════════════════════ +// Run Now / Kill Tests +// ═══════════════════════════════════════════════ + +func TestTask_RunNowAndListRuns(t *testing.T) { + h := setupHarness(t) + _, token := h.createAdminUser("runner", "runner@test.com") + + // Create + w := h.request("POST", "/api/v1/tasks", token, map[string]interface{}{ + "name": "Runnable", "task_type": "prompt", "schedule": "@daily", + "user_prompt": "go", "model_id": "m", + }) + if w.Code != http.StatusCreated { + t.Fatalf("create: %d", w.Code) + } + var task map[string]interface{} + decode(w, &task) + taskID := task["id"].(string) + + // Run Now + w = h.request("POST", "/api/v1/tasks/"+taskID+"/run", token, nil) + if w.Code != http.StatusOK { + t.Fatalf("run now: want 200, got %d: %s", w.Code, w.Body.String()) + } + var runResp map[string]interface{} + decode(w, &runResp) + if runResp["scheduled"] != true { + t.Fatal("expected scheduled: true") + } + + // List runs — might be empty if scheduler hasn't ticked, but endpoint should work + w = h.request("GET", "/api/v1/tasks/"+taskID+"/runs", token, nil) + if w.Code != http.StatusOK { + t.Fatalf("list runs: want 200, got %d: %s", w.Code, w.Body.String()) + } +} + +func TestTask_KillNoActiveRun(t *testing.T) { + h := setupHarness(t) + _, token := h.createAdminUser("killer", "killer@test.com") + + w := h.request("POST", "/api/v1/tasks", token, map[string]interface{}{ + "name": "NoRun", "task_type": "prompt", "schedule": "@daily", + "user_prompt": "x", "model_id": "m", + }) + if w.Code != http.StatusCreated { + t.Fatalf("create: %d", w.Code) + } + var task map[string]interface{} + decode(w, &task) + + // Kill with no active run — should 404 + w = h.request("POST", "/api/v1/tasks/"+task["id"].(string)+"/kill", token, nil) + if w.Code != http.StatusNotFound { + t.Fatalf("kill no-run: want 404, got %d: %s", w.Code, w.Body.String()) + } +} + +// ═══════════════════════════════════════════════ +// Webhook Trigger Tests (v0.28.0) +// ═══════════════════════════════════════════════ + +func TestTask_WebhookScheduleCreatesTriggerToken(t *testing.T) { + h := setupHarness(t) + _, token := h.createAdminUser("webhooker", "webhooker@test.com") + + w := h.request("POST", "/api/v1/tasks", token, map[string]interface{}{ + "name": "Webhook Task", + "task_type": "prompt", + "schedule": "webhook", + "user_prompt": "process this", + "model_id": "m", + }) + if w.Code != http.StatusCreated { + t.Fatalf("create webhook task: want 201, got %d: %s", w.Code, w.Body.String()) + } + + var task map[string]interface{} + decode(w, &task) + + // Should have trigger_token + triggerToken, ok := task["trigger_token"].(string) + if !ok || triggerToken == "" { + t.Fatal("webhook task should have a trigger_token") + } + + // next_run_at should be nil (webhook tasks don't have cron) + if task["next_run_at"] != nil { + t.Fatalf("webhook task should have nil next_run_at, got %v", task["next_run_at"]) + } +} + +func TestTask_TriggerEndpoint(t *testing.T) { + h := setupHarness(t) + _, token := h.createAdminUser("triguser", "triguser@test.com") + + // Create webhook task + w := h.request("POST", "/api/v1/tasks", token, map[string]interface{}{ + "name": "CI Watcher", + "task_type": "prompt", + "schedule": "webhook", + "user_prompt": "analyze failure", + "model_id": "m", + }) + if w.Code != http.StatusCreated { + t.Fatalf("create: %d: %s", w.Code, w.Body.String()) + } + var task map[string]interface{} + decode(w, &task) + triggerToken := task["trigger_token"].(string) + + // Fire the trigger (unauthenticated) + w = h.request("POST", "/api/v1/hooks/t/"+triggerToken, "", map[string]interface{}{ + "build_id": 12345, + "status": "failed", + "repo": "chat-switchboard", + }) + if w.Code != http.StatusAccepted { + t.Fatalf("trigger: want 202, got %d: %s", w.Code, w.Body.String()) + } + var trigResp map[string]interface{} + decode(w, &trigResp) + if trigResp["triggered"] != true { + t.Fatal("expected triggered: true") + } + if trigResp["run_id"] == nil || trigResp["run_id"].(string) == "" { + t.Fatal("expected run_id") + } + + // Second trigger should 409 (queued run exists) + w = h.request("POST", "/api/v1/hooks/t/"+triggerToken, "", map[string]interface{}{ + "build_id": 12346, + }) + if w.Code != http.StatusConflict { + t.Fatalf("duplicate trigger: want 409, got %d: %s", w.Code, w.Body.String()) + } +} + +func TestTask_TriggerInvalidToken(t *testing.T) { + h := setupHarness(t) + + w := h.request("POST", "/api/v1/hooks/t/nonexistent-token", "", map[string]interface{}{ + "data": "nope", + }) + if w.Code != http.StatusNotFound { + t.Fatalf("bad token: want 404, got %d", w.Code) + } +} + +func TestTask_TriggerInactiveTask(t *testing.T) { + h := setupHarness(t) + _, token := h.createAdminUser("inactive", "inactive@test.com") + + // Create webhook task + w := h.request("POST", "/api/v1/tasks", token, map[string]interface{}{ + "name": "Deactivated", "task_type": "prompt", "schedule": "webhook", + "user_prompt": "x", "model_id": "m", + }) + if w.Code != http.StatusCreated { + t.Fatalf("create: %d", w.Code) + } + var task map[string]interface{} + decode(w, &task) + taskID := task["id"].(string) + triggerToken := task["trigger_token"].(string) + + // Deactivate + w = h.request("PUT", "/api/v1/tasks/"+taskID, token, map[string]interface{}{ + "is_active": false, + }) + if w.Code != http.StatusOK { + t.Fatalf("deactivate: %d", w.Code) + } + + // Trigger should 410 + w = h.request("POST", "/api/v1/hooks/t/"+triggerToken, "", map[string]interface{}{}) + if w.Code != http.StatusGone { + t.Fatalf("inactive trigger: want 410, got %d: %s", w.Code, w.Body.String()) + } +} + +// ═══════════════════════════════════════════════ +// Action Task Tests (v0.28.0) +// ═══════════════════════════════════════════════ + +func TestTask_ActionTaskAdminAllowed(t *testing.T) { + h := setupHarness(t) + _, token := h.createAdminUser("actionadmin", "action@test.com") + + // Admin can always create action tasks + w := h.request("POST", "/api/v1/tasks", token, map[string]interface{}{ + "name": "Action Relay", + "task_type": "action", + "schedule": "webhook", + }) + if w.Code != http.StatusCreated { + t.Fatalf("create action task: want 201, got %d: %s", w.Code, w.Body.String()) + } + var task map[string]interface{} + decode(w, &task) + if task["task_type"].(string) != "action" { + t.Fatalf("task_type should be action, got %q", task["task_type"]) + } +} + +// ═══════════════════════════════════════════════ +// Admin List All Tasks +// ═══════════════════════════════════════════════ + +func TestTask_AdminListAll(t *testing.T) { + h := setupHarness(t) + _, adminToken := h.createAdminUser("tadmin", "tadmin@test.com") + + // Create 2 tasks + for i := 0; i < 2; i++ { + h.request("POST", "/api/v1/tasks", adminToken, map[string]interface{}{ + "name": fmt.Sprintf("Admin Task %d", i), "task_type": "prompt", + "schedule": "@daily", "user_prompt": "x", "model_id": "m", + }) + } + + // Admin list + w := h.request("GET", "/api/v1/admin/tasks", adminToken, nil) + if w.Code != http.StatusOK { + t.Fatalf("admin list: want 200, got %d: %s", w.Code, w.Body.String()) + } + var resp map[string]interface{} + decode(w, &resp) + data := resp["data"].([]interface{}) + if len(data) < 2 { + t.Fatalf("admin list: expected at least 2 tasks, got %d", len(data)) + } +} + +// ═══════════════════════════════════════════════ +// Webhook Secret / Output Channel (Audit C2, C3) +// ═══════════════════════════════════════════════ + +func TestTask_WebhookSecretGenerated(t *testing.T) { + h := setupHarness(t) + _, token := h.createAdminUser("secretuser", "secret@test.com") + + w := h.request("POST", "/api/v1/tasks", token, map[string]interface{}{ + "name": "Webhook Out", "task_type": "prompt", "schedule": "@daily", + "user_prompt": "x", "model_id": "m", + "webhook_url": "https://example.com/hook", + }) + if w.Code != http.StatusCreated { + t.Fatalf("create: %d: %s", w.Code, w.Body.String()) + } + var task map[string]interface{} + decode(w, &task) + taskID := task["id"].(string) + + // Verify webhook_secret was auto-generated + secret, ok := task["webhook_secret"].(string) + if !ok || secret == "" { + t.Fatal("webhook_secret should be auto-generated when webhook_url is set") + } + + // Verify it persists on GET (C2 fix) + w = h.request("GET", "/api/v1/tasks/"+taskID, token, nil) + if w.Code != http.StatusOK { + t.Fatalf("get: %d", w.Code) + } + var got map[string]interface{} + decode(w, &got) + gotSecret, ok := got["webhook_secret"].(string) + if !ok || gotSecret != secret { + t.Fatalf("webhook_secret not persisted on read: got %q, want %q", gotSecret, secret) + } +} + +// ═══════════════════════════════════════════════ +// Once-shot schedule +// ═══════════════════════════════════════════════ + +func TestTask_OnceSchedule(t *testing.T) { + h := setupHarness(t) + _, token := h.createAdminUser("onceuser", "once@test.com") + + w := h.request("POST", "/api/v1/tasks", token, map[string]interface{}{ + "name": "One Shot", "task_type": "prompt", "schedule": "once", + "user_prompt": "x", "model_id": "m", + }) + if w.Code != http.StatusCreated { + t.Fatalf("create: %d: %s", w.Code, w.Body.String()) + } + var task map[string]interface{} + decode(w, &task) + + // Should have next_run_at set (approximately now) + if task["next_run_at"] == nil { + t.Fatal("once task should have next_run_at set to ~now") + } +} + +// ═══════════════════════════════════════════════ +// Budget defaults applied +// ═══════════════════════════════════════════════ + +func TestTask_DefaultBudgets(t *testing.T) { + h := setupHarness(t) + _, token := h.createAdminUser("budget", "budget@test.com") + + // Create with zero budgets — should get defaults + w := h.request("POST", "/api/v1/tasks", token, map[string]interface{}{ + "name": "Defaults", "task_type": "prompt", "schedule": "@daily", + "user_prompt": "x", "model_id": "m", + }) + if w.Code != http.StatusCreated { + t.Fatalf("create: %d: %s", w.Code, w.Body.String()) + } + var task map[string]interface{} + decode(w, &task) + + maxTokens := int(task["max_tokens"].(float64)) + maxToolCalls := int(task["max_tool_calls"].(float64)) + maxWall := int(task["max_wall_clock"].(float64)) + + if maxTokens != 4096 { + t.Fatalf("max_tokens: want 4096, got %d", maxTokens) + } + if maxToolCalls != 10 { + t.Fatalf("max_tool_calls: want 10, got %d", maxToolCalls) + } + if maxWall != 300 { + t.Fatalf("max_wall_clock: want 300, got %d", maxWall) + } +} + +// ═══════════════════════════════════════════════ +// F2: Workflow task_type rejected at API boundary +// ═══════════════════════════════════════════════ + +func TestTask_WorkflowTypeRejected(t *testing.T) { + h := setupHarness(t) + _, token := h.createAdminUser("wfuser", "wfuser@test.com") + + w := h.request("POST", "/api/v1/tasks", token, map[string]interface{}{ + "name": "Workflow Task", + "task_type": "workflow", + "schedule": "@daily", + "workflow_id": "00000000-0000-0000-0000-000000000001", + "model_id": "m", + }) + if w.Code != http.StatusBadRequest { + t.Fatalf("workflow task_type: want 400, got %d: %s", w.Code, w.Body.String()) + } + // Verify error message mentions "not yet implemented" + body := w.Body.String() + if !strings.Contains(body, "not yet implemented") { + t.Fatalf("expected 'not yet implemented' in error, got: %s", body) + } +} + +// ═══════════════════════════════════════════════ +// F8: Team Task Routes +// ═══════════════════════════════════════════════ + +func TestTask_TeamTaskCRUD(t *testing.T) { + h := setupHarness(t) + adminID, adminToken := h.createAdminUser("teamadm", "teamadm@test.com") + + // Create a team (admin users bypass RequireTeamAdmin middleware) + teamID := seedInsertReturningID(t, + `INSERT INTO teams (name, description, created_by) VALUES ($1, $2, $3) RETURNING id`, + "Task Team", "For team task tests", adminID, + ) + + // Create team task via team route + w := h.request("POST", "/api/v1/teams/"+teamID+"/tasks", adminToken, map[string]interface{}{ + "name": "Team Task", + "task_type": "prompt", + "schedule": "@daily", + "user_prompt": "team work", + "model_id": "m", + }) + if w.Code != http.StatusCreated { + t.Fatalf("create team task: want 201, got %d: %s", w.Code, w.Body.String()) + } + + var task map[string]interface{} + decode(w, &task) + taskID := task["id"].(string) + + // Verify scope was injected as "team" + if task["scope"].(string) != "team" { + t.Fatalf("team task scope: want 'team', got %q", task["scope"]) + } + if task["team_id"] == nil || task["team_id"].(string) != teamID { + t.Fatalf("team task team_id: want %q, got %v", teamID, task["team_id"]) + } + + // List team tasks (member route — admin bypasses membership check) + w = h.request("GET", "/api/v1/teams/"+teamID+"/tasks", adminToken, nil) + if w.Code != http.StatusOK { + t.Fatalf("list team tasks: want 200, got %d: %s", w.Code, w.Body.String()) + } + var listResp map[string]interface{} + decode(w, &listResp) + data := listResp["data"].([]interface{}) + if len(data) != 1 { + t.Fatalf("expected 1 team task, got %d", len(data)) + } + + // Update via team route + w = h.request("PUT", "/api/v1/teams/"+teamID+"/tasks/"+taskID, adminToken, map[string]interface{}{ + "name": "Updated Team Task", + }) + if w.Code != http.StatusOK { + t.Fatalf("update team task: want 200, got %d: %s", w.Code, w.Body.String()) + } + var updated map[string]interface{} + decode(w, &updated) + if updated["name"].(string) != "Updated Team Task" { + t.Fatalf("name not updated: got %q", updated["name"]) + } + + // List runs via team member route + w = h.request("GET", "/api/v1/teams/"+teamID+"/tasks/"+taskID+"/runs", adminToken, nil) + if w.Code != http.StatusOK { + t.Fatalf("list team task runs: want 200, got %d: %s", w.Code, w.Body.String()) + } + + // Delete via team route + w = h.request("DELETE", "/api/v1/teams/"+teamID+"/tasks/"+taskID, adminToken, nil) + if w.Code != http.StatusOK { + t.Fatalf("delete team task: want 200, got %d: %s", w.Code, w.Body.String()) + } + + // Verify gone + w = h.request("GET", "/api/v1/teams/"+teamID+"/tasks", adminToken, nil) + if w.Code != http.StatusOK { + t.Fatalf("list after delete: want 200, got %d", w.Code) + } + decode(w, &listResp) + data = listResp["data"].([]interface{}) + if len(data) != 0 { + t.Fatalf("expected 0 team tasks after delete, got %d", len(data)) + } +} + +// ═══════════════════════════════════════════════ +// F9: Admin Run / Kill / Delete +// ═══════════════════════════════════════════════ + +func TestTask_AdminRunNow(t *testing.T) { + h := setupHarness(t) + _, adminToken := h.createAdminUser("admrun", "admrun@test.com") + + // Create a task + w := h.request("POST", "/api/v1/tasks", adminToken, map[string]interface{}{ + "name": "Admin Runnable", "task_type": "prompt", "schedule": "@daily", + "user_prompt": "go", "model_id": "m", + }) + if w.Code != http.StatusCreated { + t.Fatalf("create: %d", w.Code) + } + var task map[string]interface{} + decode(w, &task) + taskID := task["id"].(string) + + // Admin run via admin route + w = h.request("POST", "/api/v1/admin/tasks/"+taskID+"/run", adminToken, nil) + if w.Code != http.StatusOK { + t.Fatalf("admin run: want 200, got %d: %s", w.Code, w.Body.String()) + } + var runResp map[string]interface{} + decode(w, &runResp) + if runResp["scheduled"] != true { + t.Fatal("expected scheduled: true") + } +} + +func TestTask_AdminKillRun(t *testing.T) { + h := setupHarness(t) + _, adminToken := h.createAdminUser("admkill", "admkill@test.com") + + // Create a webhook task and trigger it to get a queued run + w := h.request("POST", "/api/v1/tasks", adminToken, map[string]interface{}{ + "name": "Killable", "task_type": "prompt", "schedule": "webhook", + "user_prompt": "x", "model_id": "m", + }) + if w.Code != http.StatusCreated { + t.Fatalf("create: %d", w.Code) + } + var task map[string]interface{} + decode(w, &task) + taskID := task["id"].(string) + triggerToken := task["trigger_token"].(string) + + // Trigger to create a queued run + w = h.request("POST", "/api/v1/hooks/t/"+triggerToken, "", map[string]interface{}{"test": true}) + if w.Code != http.StatusAccepted { + t.Fatalf("trigger: %d", w.Code) + } + var trigResp map[string]interface{} + decode(w, &trigResp) + runID := trigResp["run_id"].(string) + + // Manually transition the run to "running" so kill works + database.TestDB.Exec(dialectSQL("UPDATE task_runs SET status = 'running' WHERE id = $1"), runID) + + // Admin kill + w = h.request("POST", "/api/v1/admin/tasks/"+taskID+"/kill", adminToken, nil) + if w.Code != http.StatusOK { + t.Fatalf("admin kill: want 200, got %d: %s", w.Code, w.Body.String()) + } + var killResp map[string]interface{} + decode(w, &killResp) + if killResp["killed"] != true { + t.Fatal("expected killed: true") + } +} + +func TestTask_AdminDelete(t *testing.T) { + h := setupHarness(t) + _, adminToken := h.createAdminUser("admdel", "admdel@test.com") + + // Create a task + w := h.request("POST", "/api/v1/tasks", adminToken, map[string]interface{}{ + "name": "Admin Deletable", "task_type": "prompt", "schedule": "@daily", + "user_prompt": "x", "model_id": "m", + }) + if w.Code != http.StatusCreated { + t.Fatalf("create: %d", w.Code) + } + var task map[string]interface{} + decode(w, &task) + taskID := task["id"].(string) + + // Admin delete + w = h.request("DELETE", "/api/v1/admin/tasks/"+taskID, adminToken, nil) + if w.Code != http.StatusOK { + t.Fatalf("admin delete: want 200, got %d: %s", w.Code, w.Body.String()) + } + + // Verify gone + w = h.request("GET", "/api/v1/tasks/"+taskID, adminToken, nil) + if w.Code != http.StatusNotFound { + t.Fatalf("get deleted: want 404, got %d", w.Code) + } +} + +// ═══════════════════════════════════════════════ +// F10: RunNow 409 Conflict (queued run exists) +// ═══════════════════════════════════════════════ + +func TestTask_RunNowConflictQueued(t *testing.T) { + h := setupHarness(t) + _, token := h.createAdminUser("conflict", "conflict@test.com") + + // Create a webhook task + w := h.request("POST", "/api/v1/tasks", token, map[string]interface{}{ + "name": "Conflict Test", "task_type": "prompt", "schedule": "webhook", + "user_prompt": "x", "model_id": "m", + }) + if w.Code != http.StatusCreated { + t.Fatalf("create: %d", w.Code) + } + var task map[string]interface{} + decode(w, &task) + taskID := task["id"].(string) + triggerToken := task["trigger_token"].(string) + + // Trigger to create a queued run + w = h.request("POST", "/api/v1/hooks/t/"+triggerToken, "", map[string]interface{}{"test": true}) + if w.Code != http.StatusAccepted { + t.Fatalf("trigger: %d", w.Code) + } + + // RunNow should 409 because a queued run exists + w = h.request("POST", "/api/v1/tasks/"+taskID+"/run", token, nil) + if w.Code != http.StatusConflict { + t.Fatalf("run-now with queued run: want 409, got %d: %s", w.Code, w.Body.String()) + } +} + +func TestTask_RunNowConflictActive(t *testing.T) { + h := setupHarness(t) + _, token := h.createAdminUser("active", "active@test.com") + + // Create a webhook task + w := h.request("POST", "/api/v1/tasks", token, map[string]interface{}{ + "name": "Active Conflict", "task_type": "prompt", "schedule": "webhook", + "user_prompt": "x", "model_id": "m", + }) + if w.Code != http.StatusCreated { + t.Fatalf("create: %d", w.Code) + } + var task map[string]interface{} + decode(w, &task) + taskID := task["id"].(string) + triggerToken := task["trigger_token"].(string) + + // Trigger and promote to running + w = h.request("POST", "/api/v1/hooks/t/"+triggerToken, "", map[string]interface{}{"test": true}) + if w.Code != http.StatusAccepted { + t.Fatalf("trigger: %d", w.Code) + } + var trigResp map[string]interface{} + decode(w, &trigResp) + runID := trigResp["run_id"].(string) + database.TestDB.Exec(dialectSQL("UPDATE task_runs SET status = 'running' WHERE id = $1"), runID) + + // RunNow should 409 because an active run exists + w = h.request("POST", "/api/v1/tasks/"+taskID+"/run", token, nil) + if w.Code != http.StatusConflict { + t.Fatalf("run-now with active run: want 409, got %d: %s", w.Code, w.Body.String()) + } +} + diff --git a/server/handlers/tasks.go b/server/handlers/tasks.go index 636b2ca..0c9b393 100644 --- a/server/handlers/tasks.go +++ b/server/handlers/tasks.go @@ -1,12 +1,14 @@ package handlers import ( + "io" "net/http" "time" "github.com/gin-gonic/gin" - "git.gobha.me/xcaliber/chat-switchboard/database" + "git.gobha.me/xcaliber/chat-switchboard/auth" + "git.gobha.me/xcaliber/chat-switchboard/middleware" "git.gobha.me/xcaliber/chat-switchboard/models" "git.gobha.me/xcaliber/chat-switchboard/store" "git.gobha.me/xcaliber/chat-switchboard/taskutil" @@ -22,6 +24,8 @@ func NewTaskHandler(stores store.Stores) *TaskHandler { return &TaskHandler{stores: stores} } +// ── List endpoints ────────────────────────────── + // ListMine returns tasks owned by the current user. // GET /api/v1/tasks func (h *TaskHandler) ListMine(c *gin.Context) { @@ -67,6 +71,8 @@ func (h *TaskHandler) ListTeamTasks(c *gin.Context) { c.JSON(http.StatusOK, gin.H{"data": tasks}) } +// ── Create ────────────────────────────────────── + // CreateTeamTask creates a task scoped to the team. // Team admins only (RequireTeamAdmin middleware). // POST /api/v1/teams/:teamId/tasks @@ -116,6 +122,25 @@ func (h *TaskHandler) Create(c *gin.Context) { } } + // v0.28.0: Action tasks require task.action permission + if t.TaskType == "action" { + if c.GetString("role") != "admin" { + perms := middleware.GetResolvedPermissions(c) + if perms == nil || !perms[auth.PermTaskAction] { + c.JSON(http.StatusForbidden, gin.H{"error": "permission required: " + auth.PermTaskAction}) + return + } + } + } + + // v0.28.0-audit: Workflow task execution is not yet implemented. + // Reject at the API boundary to prevent silent wrong behavior + // (workflow tasks would fall through to the prompt pipeline). + if t.TaskType == "workflow" { + c.JSON(http.StatusBadRequest, gin.H{"error": "workflow task execution is not yet implemented — use task_type 'prompt' or 'action'"}) + return + } + // Validate required fields if t.Name == "" { c.JSON(http.StatusBadRequest, gin.H{"error": "name is required"}) @@ -129,15 +154,16 @@ func (h *TaskHandler) Create(c *gin.Context) { 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 - } - // v0.27.2: Validate cron expression before persisting - if err := taskutil.ValidateCron(t.Schedule); err != nil { - c.JSON(http.StatusBadRequest, gin.H{"error": "invalid schedule: " + err.Error()}) - return + // v0.28.0: Webhook schedule validation — cannot use cron for webhook-triggered tasks + if t.Schedule == "webhook" { + // Webhook tasks never have a cron schedule — they fire on inbound POST + } else { + // v0.27.2: Validate cron expression before persisting + if err := taskutil.ValidateCron(t.Schedule); err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": "invalid schedule: " + err.Error()}) + return + } } // Defaults @@ -157,8 +183,11 @@ func (h *TaskHandler) Create(c *gin.Context) { // v0.27.2: Apply global default budgets for zero-value fields taskCfg.ApplyDefaults(&t) - // Compute initial next_run_at - if t.Schedule == "once" { + // Compute initial next_run_at (webhook tasks have no schedule-based next_run) + if t.Schedule == "webhook" { + // No initial next_run_at — triggered externally + t.NextRunAt = nil + } else if t.Schedule == "once" { now := time.Now().UTC() t.NextRunAt = &now } else { @@ -173,6 +202,11 @@ func (h *TaskHandler) Create(c *gin.Context) { t.WebhookSecret = webhook.GenerateSecret() } + // v0.28.0: Generate trigger token for webhook-scheduled tasks + if t.Schedule == "webhook" { + t.TriggerToken = webhook.GenerateSecret() + } + if err := h.stores.Tasks.Create(ctx, &t); err != nil { c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to create task: " + err.Error()}) return @@ -181,6 +215,8 @@ func (h *TaskHandler) Create(c *gin.Context) { c.JSON(http.StatusCreated, t) } +// ── Access control helpers ────────────────────── + // canAccessTask returns true if the user can view this task. // Access: owner, system admin, or team member (for team-scoped tasks). func (h *TaskHandler) canAccessTask(c *gin.Context, t *models.Task) bool { @@ -192,11 +228,8 @@ func (h *TaskHandler) canAccessTask(c *gin.Context, t *models.Task) bool { } // Team members can view team-scoped tasks if t.Scope == "team" && t.TeamID != nil { - var exists bool - database.DB.QueryRow(database.Q( - `SELECT EXISTS(SELECT 1 FROM team_members WHERE team_id = $1 AND user_id = $2)`, - ), *t.TeamID, c.GetString("user_id")).Scan(&exists) - return exists + ok, _ := h.stores.Teams.IsMember(c.Request.Context(), *t.TeamID, c.GetString("user_id")) + return ok } return false } @@ -211,15 +244,14 @@ func (h *TaskHandler) canMutateTask(c *gin.Context, t *models.Task) bool { return true } if t.Scope == "team" && t.TeamID != nil { - var teamRole string - database.DB.QueryRow(database.Q( - `SELECT role FROM team_members WHERE team_id = $1 AND user_id = $2`, - ), *t.TeamID, c.GetString("user_id")).Scan(&teamRole) - return teamRole == "admin" + ok, _ := h.stores.Teams.IsTeamAdmin(c.Request.Context(), *t.TeamID, c.GetString("user_id")) + return ok } return false } +// ── Get / Update / Delete ─────────────────────── + // Get returns a single task. // GET /api/v1/tasks/:id func (h *TaskHandler) Get(c *gin.Context) { @@ -260,7 +292,7 @@ func (h *TaskHandler) Update(c *gin.Context) { } // v0.27.2: Validate new schedule if provided - if patch.Schedule != nil { + if patch.Schedule != nil && *patch.Schedule != "webhook" { if err := taskutil.ValidateCron(*patch.Schedule); err != nil { c.JSON(http.StatusBadRequest, gin.H{"error": "invalid schedule: " + err.Error()}) return @@ -276,10 +308,16 @@ func (h *TaskHandler) Update(c *gin.Context) { // Recompute next_run_at if schedule or timezone changed if patch.Schedule != nil || patch.Timezone != nil { - tz := updated.Timezone - next := taskutil.NextRunFromSchedule(updated.Schedule, tz) - _ = h.stores.Tasks.SetNextRun(ctx, id, next) - updated.NextRunAt = next + if updated.Schedule == "webhook" { + // Webhook tasks have no cron-based next_run + _ = h.stores.Tasks.SetNextRun(ctx, id, nil) + updated.NextRunAt = nil + } else { + tz := updated.Timezone + next := taskutil.NextRunFromSchedule(updated.Schedule, tz) + _ = h.stores.Tasks.SetNextRun(ctx, id, next) + updated.NextRunAt = next + } } c.JSON(http.StatusOK, updated) @@ -308,6 +346,8 @@ func (h *TaskHandler) Delete(c *gin.Context) { c.JSON(http.StatusOK, gin.H{"deleted": true, "id": id}) } +// ── Run History ───────────────────────────────── + // ListRuns returns run history for a task. // GET /api/v1/tasks/:id/runs func (h *TaskHandler) ListRuns(c *gin.Context) { @@ -336,6 +376,8 @@ func (h *TaskHandler) ListRuns(c *gin.Context) { c.JSON(http.StatusOK, gin.H{"data": runs}) } +// ── RunNow / KillRun ──────────────────────────── + // 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) { @@ -366,6 +408,13 @@ func (h *TaskHandler) RunNow(c *gin.Context) { return } + // Also check for queued runs + queued, _ := h.stores.Tasks.GetQueuedRun(ctx, id) + if queued != nil { + c.JSON(http.StatusConflict, gin.H{"error": "task already has a queued 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"}) @@ -404,3 +453,90 @@ func (h *TaskHandler) KillRun(c *gin.Context) { c.JSON(http.StatusOK, gin.H{"killed": true, "run_id": active.ID}) } + +// ════════════════════════════════════════════════ +// Webhook Trigger Handler (v0.28.0) +// ════════════════════════════════════════════════ +// +// POST /api/v1/hooks/t/:token — unauthenticated, token-based auth. +// External systems (CI, task chaining, etc.) POST here to fire a +// webhook-triggered task. The request body is stored as trigger_payload +// and forwarded to the executor. + +// TriggerHandler handles inbound webhook triggers. +type TriggerHandler struct { + stores store.Stores +} + +func NewTriggerHandler(stores store.Stores) *TriggerHandler { + return &TriggerHandler{stores: stores} +} + +// Handle processes an inbound webhook trigger. +// POST /api/v1/hooks/t/:token +func (h *TriggerHandler) Handle(c *gin.Context) { + token := c.Param("token") + ctx := c.Request.Context() + + // Look up task by trigger token + task, err := h.stores.Tasks.GetByTriggerToken(ctx, token) + if err != nil { + c.JSON(http.StatusNotFound, gin.H{"error": "not found"}) + return + } + + // Validate task state + if !task.IsActive { + c.JSON(http.StatusGone, gin.H{"error": "task is inactive"}) + return + } + if task.Schedule != "webhook" { + c.JSON(http.StatusBadRequest, gin.H{"error": "task is not webhook-triggered"}) + return + } + + // Check for already-active or queued run + active, _ := h.stores.Tasks.GetActiveRun(ctx, task.ID) + if active != nil { + c.JSON(http.StatusConflict, gin.H{"error": "task already has an active run"}) + return + } + queued, _ := h.stores.Tasks.GetQueuedRun(ctx, task.ID) + if queued != nil { + c.JSON(http.StatusConflict, gin.H{"error": "task already has a queued run"}) + return + } + + // Read trigger payload (optional) + var triggerPayload string + if c.Request.Body != nil { + body, err := io.ReadAll(io.LimitReader(c.Request.Body, 1<<20)) // 1MB limit + if err == nil && len(body) > 0 { + triggerPayload = string(body) + } + } + + // Create a queued run with the trigger payload + run := &models.TaskRun{ + TaskID: task.ID, + Status: "queued", + TriggerPayload: triggerPayload, + } + if err := h.stores.Tasks.CreateRun(ctx, run); err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to create run"}) + return + } + + // Set next_run_at to now so the scheduler picks it up + now := time.Now().UTC() + if err := h.stores.Tasks.SetNextRun(ctx, task.ID, now); err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to schedule run"}) + return + } + + c.JSON(http.StatusAccepted, gin.H{ + "triggered": true, + "run_id": run.ID, + "task_id": task.ID, + }) +} diff --git a/server/main.go b/server/main.go index 61f1603..980ca0c 100644 --- a/server/main.go +++ b/server/main.go @@ -1185,6 +1185,10 @@ func main() { wfEntry := handlers.NewWorkflowEntryHandler(stores) base.POST("/api/v1/workflow-entry/:scope/:slug", wfEntry.StartVisitor) + // v0.28.0: Webhook trigger endpoint (token-based auth, no JWT) + triggerH := handlers.NewTriggerHandler(stores) + base.POST("/api/v1/hooks/t/:token", triggerH.Handle) + bp := cfg.BasePath if bp == "" { bp = "/" diff --git a/server/models/models_task.go b/server/models/models_task.go index 48884f0..af0fca4 100644 --- a/server/models/models_task.go +++ b/server/models/models_task.go @@ -5,8 +5,9 @@ import ( "time" ) -// Task is a scheduled or one-shot job that creates a service channel -// and runs a completion (prompt task) or instantiates a workflow. +// Task is a scheduled, one-shot, or webhook-triggered job that creates a +// service channel and runs a completion (prompt task), instantiates a +// workflow, or relays a payload without LLM involvement (action task). type Task struct { ID string `json:"id" db:"id"` OwnerID string `json:"owner_id" db:"owner_id"` @@ -16,7 +17,7 @@ type Task struct { Scope string `json:"scope" db:"scope"` // personal | team | global // What to run - TaskType string `json:"task_type" db:"task_type"` // prompt | workflow + TaskType string `json:"task_type" db:"task_type"` // prompt | workflow | action 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"` @@ -25,10 +26,13 @@ type Task struct { ToolGrants json.RawMessage `json:"tool_grants,omitempty" db:"tool_grants"` // Schedule - Schedule string `json:"schedule" db:"schedule"` // cron expression or "once" + Schedule string `json:"schedule" db:"schedule"` // cron expression, "once", or "webhook" Timezone string `json:"timezone" db:"timezone"` IsActive bool `json:"is_active" db:"is_active"` + // Webhook trigger (inbound) + TriggerToken string `json:"trigger_token,omitempty" db:"trigger_token"` + // Execution policy MaxTokens int `json:"max_tokens" db:"max_tokens"` MaxToolCalls int `json:"max_tool_calls" db:"max_tool_calls"` @@ -55,35 +59,39 @@ type Task struct { // 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"` + 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"` + WorkflowID *string `json:"workflow_id,omitempty"` + ToolGrants *json.RawMessage `json:"tool_grants,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"` + OutputChannelID *string `json:"output_channel_id,omitempty"` + WebhookURL *string `json:"webhook_url,omitempty"` + ProviderConfigID *string `json:"provider_config_id,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"` + 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"` // queued | running | completed | failed | budget_exceeded | cancelled + TriggerPayload string `json:"trigger_payload,omitempty" db:"trigger_payload"` + 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/cron.go b/server/scheduler/cron.go deleted file mode 100644 index 07b2f29..0000000 --- a/server/scheduler/cron.go +++ /dev/null @@ -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 -} diff --git a/server/scheduler/executor.go b/server/scheduler/executor.go index 07b320a..c743c7b 100644 --- a/server/scheduler/executor.go +++ b/server/scheduler/executor.go @@ -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(), diff --git a/server/scheduler/scheduler.go b/server/scheduler/scheduler.go index 9c63e19..cd56e4a 100644 --- a/server/scheduler/scheduler.go +++ b/server/scheduler/scheduler.go @@ -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 { diff --git a/server/scheduler/task_config.go b/server/scheduler/task_config.go deleted file mode 100644 index e35cf27..0000000 --- a/server/scheduler/task_config.go +++ /dev/null @@ -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 - } -} diff --git a/server/store/postgres/tasks.go b/server/store/postgres/tasks.go index 3922bc1..9000ed5 100644 --- a/server/store/postgres/tasks.go +++ b/server/store/postgres/tasks.go @@ -11,45 +11,86 @@ type TaskStore struct{} func NewTaskStore() *TaskStore { return &TaskStore{} } +// ── Full column list (single source of truth) ────────────────── +// Used by GetByID, GetByTriggerToken, and list(). Every query must +// SELECT and Scan the same columns in the same order to prevent drift. +// +// scanTask is the ONLY function that maps columns → struct fields. +// If a column is added or reordered, update taskColumns + scanTask. + +// taskColumns is the canonical SELECT list for tasks. +// COALESCE wraps nullable columns that scan into non-pointer Go types +// (string, json.RawMessage). database/sql returns "unsupported Scan, +// storing driver.Value type " for these without COALESCE. +const taskColumns = `id, owner_id, team_id, name, description, scope, + task_type, persona_id, model_id, system_prompt, user_prompt, + workflow_id, COALESCE(tool_grants, '[]'::jsonb) AS tool_grants, + schedule, timezone, is_active, + COALESCE(trigger_token, '') AS trigger_token, + max_tokens, max_tool_calls, max_wall_clock, output_mode, + output_channel_id, + COALESCE(webhook_url, '') AS webhook_url, + COALESCE(webhook_secret, '') AS webhook_secret, + provider_config_id, + notify_on_complete, notify_on_failure, + last_run_at, next_run_at, run_count, created_at, updated_at` + +// scanTask maps one row from taskColumns into a Task struct. +// Single source of truth — used by GetByID, GetByTriggerToken, and list(). +func scanTask(scanner interface{ Scan(...interface{}) error }, t *models.Task) error { + return scanner.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.TriggerToken, + &t.MaxTokens, &t.MaxToolCalls, &t.MaxWallClock, &t.OutputMode, + &t.OutputChannelID, &t.WebhookURL, &t.WebhookSecret, &t.ProviderConfigID, + &t.NotifyOnComplete, &t.NotifyOnFailure, + &t.LastRunAt, &t.NextRunAt, &t.RunCount, &t.CreatedAt, &t.UpdatedAt, + ) +} + +// ── CRUD ─────────────────────────────────────── + 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, + trigger_token, max_tokens, max_tool_calls, max_wall_clock, output_mode, - output_channel_id, webhook_url, provider_config_id, + output_channel_id, webhook_url, webhook_secret, 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) + 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,$26,$27) 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, + nilIfEmpty(t.TriggerToken), t.MaxTokens, t.MaxToolCalls, t.MaxWallClock, t.OutputMode, - t.OutputChannelID, t.WebhookURL, t.ProviderConfigID, + t.OutputChannelID, t.WebhookURL, t.WebhookSecret, 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 { + row := DB.QueryRowContext(ctx, + `SELECT `+taskColumns+` FROM tasks WHERE id = $1`, id, + ) + if err := scanTask(row, t); err != nil { + return nil, err + } + return t, nil +} + +func (s *TaskStore) GetByTriggerToken(ctx context.Context, token string) (*models.Task, error) { + t := &models.Task{} + row := DB.QueryRowContext(ctx, + `SELECT `+taskColumns+` FROM tasks WHERE trigger_token = $1`, token, + ) + if err := scanTask(row, t); err != nil { return nil, err } return t, nil @@ -66,6 +107,8 @@ func (s *TaskStore) Update(ctx context.Context, id string, p models.TaskPatch) e 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.WorkflowID != nil { q += comma(i, "workflow_id"); args = append(args, *p.WorkflowID); i++ } + if p.ToolGrants != nil { q += comma(i, "tool_grants"); args = append(args, jsonOrNull(*p.ToolGrants)); 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++ } @@ -73,7 +116,9 @@ func (s *TaskStore) Update(ctx context.Context, id string, p models.TaskPatch) e 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.OutputChannelID != nil { q += comma(i, "output_channel_id"); args = append(args, *p.OutputChannelID); i++ } if p.WebhookURL != nil { q += comma(i, "webhook_url"); args = append(args, *p.WebhookURL); i++ } + if p.ProviderConfigID != nil { q += comma(i, "provider_config_id"); args = append(args, *p.ProviderConfigID); 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++ } @@ -104,8 +149,15 @@ func (s *TaskStore) ListDue(ctx context.Context, limit int) ([]models.Task, erro return s.list(ctx, `WHERE is_active = true AND next_run_at <= NOW() ORDER BY next_run_at ASC LIMIT $1`, limit) } +// ── Scheduler bookkeeping ────────────────────── + 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) + _, err := DB.ExecContext(ctx, `UPDATE tasks SET next_run_at = $1, updated_at = NOW() WHERE id = $2`, nextRun, id) + return err +} + +func (s *TaskStore) SetLastRun(ctx context.Context, id string) error { + _, err := DB.ExecContext(ctx, `UPDATE tasks SET last_run_at = NOW(), updated_at = NOW() WHERE id = $1`, id) return err } @@ -114,13 +166,14 @@ func (s *TaskStore) IncrementRunCount(ctx context.Context, id string) error { return err } -// ── Run History ───────────────────────────── +// ── Run History ───────────────────────────────── func (s *TaskStore) CreateRun(ctx context.Context, r *models.TaskRun) error { + triggerPayload := nilIfEmpty(r.TriggerPayload) 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, + INSERT INTO task_runs (task_id, channel_id, status, trigger_payload) + VALUES ($1, $2, $3, $4) RETURNING id, started_at`, + r.TaskID, r.ChannelID, r.Status, triggerPayload, ).Scan(&r.ID, &r.StartedAt) } @@ -132,13 +185,31 @@ func (s *TaskStore) UpdateRun(ctx context.Context, id, status string, tokensUsed return err } +func (s *TaskStore) TransitionRunStatus(ctx context.Context, id string, status string) error { + _, err := DB.ExecContext(ctx, `UPDATE task_runs SET status = $1 WHERE id = $2`, status, 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 + SELECT id, task_id, channel_id, status, COALESCE(trigger_payload, ''), 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) + ).Scan(&r.ID, &r.TaskID, &r.ChannelID, &r.Status, &r.TriggerPayload, &r.StartedAt) + if err != nil { + return nil, err + } + return r, nil +} + +func (s *TaskStore) GetQueuedRun(ctx context.Context, taskID string) (*models.TaskRun, error) { + r := &models.TaskRun{} + err := DB.QueryRowContext(ctx, ` + SELECT id, task_id, channel_id, status, COALESCE(trigger_payload, ''), started_at + FROM task_runs WHERE task_id = $1 AND status = 'queued' + ORDER BY started_at ASC LIMIT 1`, taskID, + ).Scan(&r.ID, &r.TaskID, &r.ChannelID, &r.Status, &r.TriggerPayload, &r.StartedAt) if err != nil { return nil, err } @@ -147,8 +218,9 @@ func (s *TaskStore) GetActiveRun(ctx context.Context, taskID string) (*models.Ta 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, '') + SELECT id, task_id, channel_id, status, COALESCE(trigger_payload, ''), + 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 { @@ -158,8 +230,9 @@ func (s *TaskStore) ListRuns(ctx context.Context, taskID string, limit int) ([]m 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 { + if err := rows.Scan(&r.ID, &r.TaskID, &r.ChannelID, &r.Status, &r.TriggerPayload, + &r.StartedAt, &r.CompletedAt, &r.TokensUsed, &r.ToolCalls, &r.WallClock, + &r.Error); err != nil { continue } runs = append(runs, r) @@ -167,16 +240,10 @@ func (s *TaskStore) ListRuns(ctx context.Context, taskID string, limit int) ([]m return runs, nil } -// ── Helpers ───────────────────────────────── +// ── list helper (single source of truth for SELECT + Scan) ───── 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 + q := `SELECT ` + taskColumns + ` FROM tasks ` + where rows, err := DB.QueryContext(ctx, q, args...) if err != nil { @@ -187,12 +254,7 @@ func (s *TaskStore) list(ctx context.Context, where string, args ...interface{}) 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 { + if err := scanTask(rows, &t); err != nil { continue } tasks = append(tasks, t) @@ -200,6 +262,8 @@ func (s *TaskStore) list(ctx context.Context, where string, args ...interface{}) return tasks, nil } +// ── Helpers ───────────────────────────────────── + // comma builds ", column = $N" func comma(i int, col string) string { return ", " + col + " = " + pgArg(i) @@ -214,3 +278,11 @@ func pgArg(i int) string { func pgWhere(i int, col string) string { return " WHERE " + col + " = " + pgArg(i) } + +// nilIfEmpty returns nil for empty strings (prevents empty-string vs NULL issues). +func nilIfEmpty(s string) interface{} { + if s == "" { + return nil + } + return s +} diff --git a/server/store/sqlite/helpers.go b/server/store/sqlite/helpers.go index 49e62a4..5920d90 100644 --- a/server/store/sqlite/helpers.go +++ b/server/store/sqlite/helpers.go @@ -33,6 +33,12 @@ func (s *sqliteTime) Scan(src interface{}) error { case time.Time: *s.t = v return nil + case int64: + // modernc/sqlite stores Go time.Time as Unix timestamp (int64). + // This happens when code passes *time.Time directly as a ? + // parameter instead of using datetime('now') in SQL. + *s.t = time.Unix(v, 0).UTC() + return nil case string: for _, f := range timeFormats { if p, err := time.Parse(f, v); err == nil { @@ -51,6 +57,8 @@ var timeFormats = []string{ "2006-01-02T15:04:05Z", time.RFC3339, "2006-01-02 15:04:05.000000000+00:00", + "2006-01-02 15:04:05Z07:00", // modernc: time.Time UTC → "2025-03-11 16:00:00Z" + "2006-01-02 15:04:05.999999999Z07:00", // modernc: time.Time with fractional seconds } // stN wraps a *time.Time pointer for nullable columns. diff --git a/server/store/sqlite/tasks.go b/server/store/sqlite/tasks.go index da63c18..f09c37e 100644 --- a/server/store/sqlite/tasks.go +++ b/server/store/sqlite/tasks.go @@ -2,6 +2,8 @@ package sqlite import ( "context" + "encoding/json" + "time" "git.gobha.me/xcaliber/chat-switchboard/models" "git.gobha.me/xcaliber/chat-switchboard/store" @@ -11,44 +13,97 @@ type TaskStore struct{} func NewTaskStore() *TaskStore { return &TaskStore{} } +// ── Full column list (single source of truth) ────────────────── + +// taskColumns is the canonical SELECT list for tasks. +// COALESCE wraps nullable columns that scan into non-pointer Go types. +// tool_grants uses COALESCE to '[]' so the string intermediate never sees NULL. +const taskColumns = `id, owner_id, team_id, name, description, scope, + task_type, persona_id, model_id, system_prompt, user_prompt, + workflow_id, COALESCE(tool_grants, '[]') AS tool_grants, + schedule, timezone, is_active, + COALESCE(trigger_token, '') AS trigger_token, + max_tokens, max_tool_calls, max_wall_clock, output_mode, + output_channel_id, + COALESCE(webhook_url, '') AS webhook_url, + COALESCE(webhook_secret, '') AS webhook_secret, + provider_config_id, + notify_on_complete, notify_on_failure, + last_run_at, next_run_at, run_count, created_at, updated_at` + +// scanTask scans a full task row into the model. +// +// SQLite dialect issues handled here: +// - tool_grants: modernc driver returns TEXT as Go string, not []byte. +// json.RawMessage (named []byte) can't accept string via convertAssign. +// Scan into string intermediate, then cast. +// - is_active, notify_on_complete, notify_on_failure: stored as INTEGER +// (0/1). modernc returns int64, but convertAssign can't assign int64 +// to *bool. Scan into int intermediate, then convert. +// +// These match the patterns in store/sqlite/workflows.go. +func scanTask(scanner interface{ Scan(...interface{}) error }, t *models.Task) error { + var toolGrantsStr string + var isActive, notifyComplete, notifyFailure int + err := scanner.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, &toolGrantsStr, &t.Schedule, &t.Timezone, &isActive, + &t.TriggerToken, + &t.MaxTokens, &t.MaxToolCalls, &t.MaxWallClock, &t.OutputMode, + &t.OutputChannelID, &t.WebhookURL, &t.WebhookSecret, &t.ProviderConfigID, + ¬ifyComplete, ¬ifyFailure, + stN(&t.LastRunAt), stN(&t.NextRunAt), &t.RunCount, st(&t.CreatedAt), st(&t.UpdatedAt), + ) + if err != nil { + return err + } + t.ToolGrants = json.RawMessage(toolGrantsStr) + t.IsActive = isActive != 0 + t.NotifyOnComplete = notifyComplete != 0 + t.NotifyOnFailure = notifyFailure != 0 + return nil +} + +// ── CRUD ─────────────────────────────────────── + func (s *TaskStore) Create(ctx context.Context, t *models.Task) error { t.ID = store.NewID() + now := time.Now().UTC() + t.CreatedAt = now + t.UpdatedAt = now _, 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, + trigger_token, max_tokens, max_tool_calls, max_wall_clock, output_mode, - output_channel_id, webhook_url, provider_config_id, + output_channel_id, webhook_url, webhook_secret, provider_config_id, notify_on_complete, notify_on_failure, next_run_at) - VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)`, + 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), + nilIfEmpty(t.TriggerToken), t.MaxTokens, t.MaxToolCalls, t.MaxWallClock, t.OutputMode, - t.OutputChannelID, t.WebhookURL, t.ProviderConfigID, - boolToInt(t.NotifyOnComplete), boolToInt(t.NotifyOnFailure), t.NextRunAt) + t.OutputChannelID, t.WebhookURL, t.WebhookSecret, t.ProviderConfigID, + boolToInt(t.NotifyOnComplete), boolToInt(t.NotifyOnFailure), timeToSQLite(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, - stN(&t.LastRunAt), stN(&t.NextRunAt), &t.RunCount, st(&t.CreatedAt), st(&t.UpdatedAt)) - if err != nil { + row := DB.QueryRowContext(ctx, `SELECT `+taskColumns+` FROM tasks WHERE id = ?`, id) + if err := scanTask(row, t); err != nil { + return nil, err + } + return t, nil +} + +func (s *TaskStore) GetByTriggerToken(ctx context.Context, token string) (*models.Task, error) { + t := &models.Task{} + row := DB.QueryRowContext(ctx, `SELECT `+taskColumns+` FROM tasks WHERE trigger_token = ?`, token) + if err := scanTask(row, t); err != nil { return nil, err } return t, nil @@ -63,6 +118,8 @@ func (s *TaskStore) Update(ctx context.Context, id string, p models.TaskPatch) e 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.WorkflowID != nil { q += ", workflow_id = ?"; args = append(args, *p.WorkflowID) } + if p.ToolGrants != nil { q += ", tool_grants = ?"; args = append(args, nullableJSON(*p.ToolGrants)) } 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)) } @@ -70,7 +127,9 @@ func (s *TaskStore) Update(ctx context.Context, id string, p models.TaskPatch) e 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.OutputChannelID != nil { q += ", output_channel_id = ?"; args = append(args, *p.OutputChannelID) } if p.WebhookURL != nil { q += ", webhook_url = ?"; args = append(args, *p.WebhookURL) } + if p.ProviderConfigID != nil { q += ", provider_config_id = ?"; args = append(args, *p.ProviderConfigID) } 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 = ?" @@ -100,8 +159,15 @@ func (s *TaskStore) ListDue(ctx context.Context, limit int) ([]models.Task, erro return s.list(ctx, `WHERE is_active = 1 AND next_run_at <= datetime('now') ORDER BY next_run_at ASC LIMIT ?`, limit) } +// ── Scheduler bookkeeping ────────────────────── + 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) + _, err := DB.ExecContext(ctx, `UPDATE tasks SET next_run_at = ?, updated_at = datetime('now') WHERE id = ?`, timeToSQLiteAny(nextRun), id) + return err +} + +func (s *TaskStore) SetLastRun(ctx context.Context, id string) error { + _, err := DB.ExecContext(ctx, `UPDATE tasks SET last_run_at = datetime('now'), updated_at = datetime('now') WHERE id = ?`, id) return err } @@ -110,11 +176,17 @@ func (s *TaskStore) IncrementRunCount(ctx context.Context, id string) error { return err } +// ── Run History ───────────────────────────────── + func (s *TaskStore) CreateRun(ctx context.Context, r *models.TaskRun) error { r.ID = store.NewID() + // F4 audit fix: set StartedAt in Go so the returned struct matches PG + // behavior (PG returns it via RETURNING; SQLite can't). + r.StartedAt = time.Now().UTC() _, err := DB.ExecContext(ctx, ` - INSERT INTO task_runs (id, task_id, channel_id, status) - VALUES (?, ?, ?, ?)`, r.ID, r.TaskID, r.ChannelID, r.Status) + INSERT INTO task_runs (id, task_id, channel_id, status, trigger_payload, started_at) + VALUES (?, ?, ?, ?, ?, ?)`, r.ID, r.TaskID, r.ChannelID, r.Status, + nilIfEmpty(r.TriggerPayload), r.StartedAt.UTC().Format("2006-01-02 15:04:05")) return err } @@ -126,12 +198,30 @@ func (s *TaskStore) UpdateRun(ctx context.Context, id, status string, tokensUsed return err } +func (s *TaskStore) TransitionRunStatus(ctx context.Context, id string, status string) error { + _, err := DB.ExecContext(ctx, `UPDATE task_runs SET status = ? WHERE id = ?`, status, 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 + SELECT id, task_id, channel_id, status, COALESCE(trigger_payload, ''), started_at FROM task_runs WHERE task_id = ? AND status = 'running' - LIMIT 1`, taskID).Scan(&r.ID, &r.TaskID, &r.ChannelID, &r.Status, st(&r.StartedAt)) + LIMIT 1`, taskID).Scan(&r.ID, &r.TaskID, &r.ChannelID, &r.Status, &r.TriggerPayload, st(&r.StartedAt)) + if err != nil { + return nil, err + } + return r, nil +} + +func (s *TaskStore) GetQueuedRun(ctx context.Context, taskID string) (*models.TaskRun, error) { + r := &models.TaskRun{} + err := DB.QueryRowContext(ctx, ` + SELECT id, task_id, channel_id, status, COALESCE(trigger_payload, ''), started_at + FROM task_runs WHERE task_id = ? AND status = 'queued' + ORDER BY started_at ASC LIMIT 1`, taskID).Scan( + &r.ID, &r.TaskID, &r.ChannelID, &r.Status, &r.TriggerPayload, st(&r.StartedAt)) if err != nil { return nil, err } @@ -140,8 +230,9 @@ func (s *TaskStore) GetActiveRun(ctx context.Context, taskID string) (*models.Ta 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, '') + SELECT id, task_id, channel_id, status, COALESCE(trigger_payload, ''), + 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 { @@ -151,8 +242,9 @@ func (s *TaskStore) ListRuns(ctx context.Context, taskID string, limit int) ([]m var runs []models.TaskRun for rows.Next() { var r models.TaskRun - if err := rows.Scan(&r.ID, &r.TaskID, &r.ChannelID, &r.Status, st(&r.StartedAt), - stN(&r.CompletedAt), &r.TokensUsed, &r.ToolCalls, &r.WallClock, &r.Error); err != nil { + if err := rows.Scan(&r.ID, &r.TaskID, &r.ChannelID, &r.Status, &r.TriggerPayload, + st(&r.StartedAt), stN(&r.CompletedAt), + &r.TokensUsed, &r.ToolCalls, &r.WallClock, &r.Error); err != nil { continue } runs = append(runs, r) @@ -160,14 +252,10 @@ func (s *TaskStore) ListRuns(ctx context.Context, taskID string, limit int) ([]m return runs, nil } +// ── list helper ──────────────────────────────── + 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 + q := `SELECT ` + taskColumns + ` FROM tasks ` + where rows, err := DB.QueryContext(ctx, q, args...) if err != nil { @@ -178,12 +266,7 @@ func (s *TaskStore) list(ctx context.Context, where string, args ...interface{}) 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, - stN(&t.LastRunAt), stN(&t.NextRunAt), &t.RunCount, st(&t.CreatedAt), st(&t.UpdatedAt)); err != nil { + if err := scanTask(rows, &t); err != nil { continue } tasks = append(tasks, t) @@ -191,9 +274,44 @@ func (s *TaskStore) list(ctx context.Context, where string, args ...interface{}) return tasks, nil } +// ── Helpers ───────────────────────────────────── + func nullableJSON(data []byte) interface{} { if len(data) == 0 || string(data) == "null" { return nil } return string(data) } + +func nilIfEmpty(s string) interface{} { + if s == "" { + return nil + } + return s +} + +// timeToSQLite converts *time.Time to the TEXT format that datetime('now') +// produces ("2006-01-02 15:04:05"). Returns nil for nil input. +// This prevents the modernc driver from serializing time.Time in an +// unparseable format (int64 / RFC3339Nano / driver-specific). +func timeToSQLite(t *time.Time) interface{} { + if t == nil { + return nil + } + return t.UTC().Format("2006-01-02 15:04:05") +} + +// timeToSQLiteAny handles interface{} that may be *time.Time, time.Time, or nil. +func timeToSQLiteAny(v interface{}) interface{} { + if v == nil { + return nil + } + switch t := v.(type) { + case *time.Time: + return timeToSQLite(t) + case time.Time: + return t.UTC().Format("2006-01-02 15:04:05") + default: + return v + } +} diff --git a/server/store/task_iface.go b/server/store/task_iface.go index f58e6d4..15638fa 100644 --- a/server/store/task_iface.go +++ b/server/store/task_iface.go @@ -17,14 +17,20 @@ type TaskStore interface { ListByTeam(ctx context.Context, teamID string) ([]models.Task, error) ListAll(ctx context.Context) ([]models.Task, error) + // Trigger lookup (inbound webhooks) + GetByTriggerToken(ctx context.Context, token string) (*models.Task, error) + // Scheduler queries ListDue(ctx context.Context, limit int) ([]models.Task, error) SetNextRun(ctx context.Context, id string, nextRun interface{}) error + SetLastRun(ctx context.Context, id string) 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 + TransitionRunStatus(ctx context.Context, id string, status string) error GetActiveRun(ctx context.Context, taskID string) (*models.TaskRun, error) + GetQueuedRun(ctx context.Context, taskID string) (*models.TaskRun, error) ListRuns(ctx context.Context, taskID string, limit int) ([]models.TaskRun, error) } diff --git a/server/tools/task_create.go b/server/tools/task_create.go index a2f9e08..aba01d2 100644 --- a/server/tools/task_create.go +++ b/server/tools/task_create.go @@ -11,7 +11,6 @@ import ( "git.gobha.me/xcaliber/chat-switchboard/models" "git.gobha.me/xcaliber/chat-switchboard/store" "git.gobha.me/xcaliber/chat-switchboard/taskutil" - "git.gobha.me/xcaliber/chat-switchboard/webhook" ) // ── Late Registration ──────────────────────── @@ -100,6 +99,12 @@ func (t *taskCreateTool) Execute(ctx context.Context, execCtx ExecutionContext, return "", fmt.Errorf("name, prompt, and schedule are required") } + // F5 audit fix: reject webhook schedule — webhook-triggered tasks + // require API creation (trigger token generation, no cron). + if args.Schedule == "webhook" { + return "", fmt.Errorf("webhook-triggered tasks cannot be created via this tool — use the API directly") + } + // Validate cron if err := taskutil.ValidateCron(args.Schedule); err != nil { return "", fmt.Errorf("invalid schedule: %w", err) @@ -162,9 +167,6 @@ func (t *taskCreateTool) Execute(ctx context.Context, execCtx ExecutionContext, cfg := taskutil.LoadTaskConfig(ctx, t.stores.GlobalConfig) cfg.ApplyDefaults(task) - // Generate webhook secret in case webhook is added later - task.WebhookSecret = webhook.GenerateSecret() - // Compute next_run_at if args.Schedule == "once" { now := time.Now().UTC() diff --git a/server/webhook/webhook.go b/server/webhook/webhook.go index f285c70..5cf7312 100644 --- a/server/webhook/webhook.go +++ b/server/webhook/webhook.go @@ -2,6 +2,7 @@ // // v0.27.3: Retry (3 attempts, exponential backoff), HMAC-SHA256 signature, // 10s timeout per attempt. +// v0.28.0: Added RunID, TokensUsed fields (D1 audit fix). package webhook import ( @@ -27,12 +28,14 @@ const ( // Payload is the JSON body sent to webhook endpoints. type Payload struct { TaskID string `json:"task_id,omitempty"` + RunID string `json:"run_id,omitempty"` TaskName string `json:"task_name,omitempty"` WorkflowID string `json:"workflow_id,omitempty"` ChannelID string `json:"channel_id,omitempty"` Status string `json:"status"` CompletedAt time.Time `json:"completed_at"` - Output string `json:"output,omitempty"` // last assistant message or summary + Output string `json:"output,omitempty"` // last assistant message or relay payload + TokensUsed int `json:"tokens_used,omitempty"` // total tokens consumed StageData any `json:"stage_data,omitempty"` // workflow stage data (if applicable) Error string `json:"error,omitempty"` }