diff --git a/CHANGELOG.md b/CHANGELOG.md index b1df2a1..ddb94d0 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,96 @@ # Changelog +## [0.27.2] — 2026-03-10 + +### Summary + +Task execution — the scheduler can now actually run completions. Headless +executor wires tasks into the full completion pipeline (provider resolution, +tool execution, budget enforcement, notifications). Admin panel gains a Tasks +section for CRUD, run history, kill switch, and global task configuration. + +### Added + +#### Core Tool Loop Extraction +- **`CoreToolLoop` + `LoopSink` interface.** Single canonical implementation of + multi-round LLM completion with tool execution. Replaces three duplicated + implementations (SSE streaming, multi-model streaming, sync JSON). Budget + enforcement built in: `MaxTokens`, `MaxToolCalls`, `MaxRounds`. +- **`LoopResult.BudgetExceeded`** field tracks which limit was hit. Callers + get budget status without inspecting loop internals. +- **Four sink implementations:** `sseSink` (interactive SSE), `sseModelSink` + (multi-model SSE), `accumSink` (sync JSON), `HeadlessSink` (scheduler). + +#### Provider Resolution Extraction +- **`ResolveProviderConfig()`** standalone function. Resolution chain: + explicit config → channel config → owner's personal → global. Shared by + completion handler and task executor. +- **`BuildToolDefs()`** standalone function. Server-registered tools filtered + by context predicates + browser extension tools + persona tool grant allowlist. +- **`FilterToolDefsByGrants()`** for task-level tool grant narrowing. + +#### Task Executor (`scheduler/executor.go`) +- **Headless completion pipeline.** Resolves provider, builds messages + tools, + runs `CoreToolLoop` with `Streaming: false`, persists assistant response in + service channel, updates run record with tokens/tools/wall-clock, sends + owner notification. +- **Budget enforcement.** `max_tokens`, `max_tool_calls` from task definition. + Wall clock via `context.WithTimeout`. Budget breach → `budget_exceeded` + status + mandatory owner notification. +- **Provider routing.** Task `provider_config_id` → channel → owner's + personal BYOK → team → global. Full vault decryption path. +- **Tool grants.** Persona grants + task-level grants applied as second-pass + allowlist. No browser tools in headless mode. + +#### Task Configuration (`taskutil/`) +- **`LoadTaskConfig()`** reads from `global_settings` key `"tasks"`. Fields: + `enabled`, `allow_personal`, `max_concurrent`, `default_max_tokens`, + `default_max_tool_calls`, `default_max_wall_clock`. +- **`ApplyDefaults()`** fills zero-value budget fields on new tasks. +- **Full cron parsing** via `robfig/cron/v3`. Replaces hand-rolled + `parseDailyCron`. Supports 5-field cron + descriptors (`@hourly`, `@daily`). +- **`ValidateCron()`** pre-create validation rejects invalid expressions. + +#### Permissions + Global Config +- **`task.create`** permission — gates task create/update/delete/run routes. +- **`task.admin`** permission — for future admin-level task management. +- **Global config keys:** `tasks.enabled` (kill switch), `tasks.allow_personal`, + `tasks.max_concurrent` (scheduler poll limit). + +#### Admin Panel +- **Tasks section** under Workflows category. Two tabs: task list + configuration. +- **Task list:** all tasks across users with status, schedule, run count, next + run time. Action buttons: run now, view history, delete. +- **Run history:** drill-down per task showing start/end times, tokens used, + tool calls, wall clock, status badge, error messages. +- **Kill switch:** cancel active run from run history view. +- **Configuration tab:** toggle tasks enabled/personal, set max concurrent and + default budget ceilings. Writes to `global_settings` key `"tasks"`. + +#### Routes +- `POST /api/v1/tasks/:id/kill` — cancel active run (user + admin) +- `POST /api/v1/admin/tasks/:id/run` — admin trigger +- `POST /api/v1/admin/tasks/:id/kill` — admin kill +- `DELETE /api/v1/admin/tasks/:id` — admin delete + +### Changed +- **`stream_loop.go`** rewritten from 559 lines to 181 — thin wrappers over + `CoreToolLoop`. Identical behavior, zero duplication. +- **`completion.go`** `syncCompletion` rewritten as wrapper. `resolveConfig` + and `buildToolDefs` delegate to standalone functions. `-237 lines`. +- **Scheduler startup** moved from early init to after hub + notification + service initialization. Executor needs both. +- **`scheduler.New()`** signature: `New(stores) → New(stores, executor)`. + Executor is optional (nil = v0.27.1 behavior). +- **Cron computation** in scheduler and task handler now uses `robfig/cron/v3` + via `taskutil.NextRunFromSchedule`. Create and update validate schedule + before persisting. + +### Dependencies +- Added `github.com/robfig/cron/v3 v3.0.1` + + + # Changelog Additions — v0.26.0 ## [0.26.0] — 2026-03-10 diff --git a/VERSION b/VERSION index 4daf42f..76cab36 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -0.27.1 \ No newline at end of file +0.27.2 \ No newline at end of file diff --git a/docs/ROADMAP.md b/docs/ROADMAP.md index a785323..13e7a0b 100644 --- a/docs/ROADMAP.md +++ b/docs/ROADMAP.md @@ -162,7 +162,7 @@ v0.27.1 Tasks Foundation ✅ (service channels, scheduler, task definitions, cron) │ -v0.27.2 Task Execution +v0.27.2 Task Execution ✅ (budgets, admin controls, run history, kill switch) │ @@ -953,19 +953,23 @@ Depends on: v0.27.0 (debt clearance). --- -## v0.27.2 — Task Execution: Budgets + Admin Controls +## v0.27.2 — Task Execution: Budgets + Admin Controls ✅ Guardrails for unattended execution. Depends on: v0.27.1 (task scheduler). -- [ ] Execution budget enforcement: `max_tokens`, `max_tool_calls`, `max_wall_clock` +- [x] Execution budget enforcement: `max_tokens`, `max_tool_calls`, `max_wall_clock` - [x] ~~`task_runs` table + store~~ (shipped v0.27.1 — table, PG+SQLite store, run history API) -- [ ] Budget breach → `budget_exceeded` status + owner notification +- [x] Budget breach → `budget_exceeded` status + owner notification - [x] ~~Concurrent run skip~~ (shipped v0.27.1 — scheduler skips if `GetActiveRun` returns non-nil) -- [ ] Global config: `tasks.enabled`, `tasks.allow_personal`, `tasks.max_concurrent` -- [ ] Default budget ceilings in global config (overridable per task) -- [ ] `tasks.create` and `tasks.admin` permissions -- [ ] Admin panel → Tasks section (CRUD, history, kill switch, budget config) +- [x] Global config: `tasks.enabled`, `tasks.allow_personal`, `tasks.max_concurrent` +- [x] Default budget ceilings in global config (overridable per task) +- [x] `tasks.create` and `tasks.admin` permissions +- [x] Admin panel → Tasks section (CRUD, history, kill switch, budget config) +- [x] Completion invocation from scheduler (headless executor via `CoreToolLoop`) +- [x] Full cron parsing via `robfig/cron/v3` (replacing minimal parser) +- [x] Provider resolution for task context (BYOK → team → global → routing policy) +- [x] Kill switch: `POST /tasks/:id/kill` cancels active run --- diff --git a/server/auth/permissions.go b/server/auth/permissions.go index 0747116..f6a893d 100644 --- a/server/auth/permissions.go +++ b/server/auth/permissions.go @@ -26,6 +26,8 @@ const ( PermWorkflowCreate = "workflow.create" // create workflow definitions (v0.25.0) PermAdminView = "admin.view" // read-only admin panel access 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) ) // AllPermissions is the complete set of valid permission strings. @@ -43,6 +45,8 @@ var AllPermissions = []string{ PermWorkflowCreate, PermAdminView, PermTokenUnlimited, + PermTaskCreate, + PermTaskAdmin, } // ── Resolution ────────────────────────────── diff --git a/server/go.mod b/server/go.mod index bcab0e4..2d9ae87 100644 --- a/server/go.mod +++ b/server/go.mod @@ -10,6 +10,7 @@ require ( github.com/joho/godotenv v1.5.1 github.com/lib/pq v1.10.9 github.com/minio/minio-go/v7 v7.0.82 + github.com/robfig/cron/v3 v3.0.1 golang.org/x/crypto v0.28.0 golang.org/x/net v0.30.0 modernc.org/sqlite v1.34.5 diff --git a/server/handlers/completion.go b/server/handlers/completion.go index 326cdbf..601db03 100644 --- a/server/handlers/completion.go +++ b/server/handlers/completion.go @@ -2,7 +2,6 @@ package handlers import ( "context" - "database/sql" "encoding/base64" "encoding/json" "fmt" @@ -857,97 +856,10 @@ func (h *CompletionHandler) multiModelStream( sendSSE("[DONE]") } -// buildToolDefs converts registered tools to provider-format tool definitions. -// If includeBrowser is true, also fetches tool schemas from enabled browser extensions. -// Tools whose names appear in disabledTools are excluded. -// v0.25.0: Tools self-declare availability via predicates on ToolContext. -// v0.25.0: Persona tool grants apply as a second-pass allowlist when personaID -// has explicit grants. Empty grants = inherit all (backward compat). +// buildToolDefs delegates to the standalone BuildToolDefs function. +// See resolve.go for the full implementation. func (h *CompletionHandler) buildToolDefs(ctx context.Context, userID string, includeBrowser bool, disabledTools []string, tctx tools.ToolContext, personaID string) []providers.ToolDef { - // Build disabled set for O(1) lookup - disabled := make(map[string]bool, len(disabledTools)) - for _, name := range disabledTools { - disabled[name] = true - } - - // v0.25.0: Tools self-declare availability via predicates. - // AvailableFor evaluates each tool's Availability() against tctx, - // then applies the disabled set. Replaces manual WorkspaceToolNames() - // and GitToolNames() suppression. - allTools := tools.AvailableFor(tctx, disabled) - defs := make([]providers.ToolDef, 0, len(allTools)) - for _, t := range allTools { - defs = append(defs, providers.ToolDef{ - Type: "function", - Function: providers.FunctionDef{ - Name: t.Name, - Description: t.Description, - Parameters: t.Parameters, - }, - }) - } - - // Append browser-defined tool schemas from extensions - if includeBrowser { - exts, err := h.stores.Extensions.ListForUser(ctx, userID) - if err != nil { - log.Printf("⚠️ Failed to load extensions for tools: %v", err) - return defs - } - for _, ext := range exts { - if ext.Tier != "browser" { - continue - } - var manifest struct { - Tools []struct { - Name string `json:"name"` - Description string `json:"description"` - Parameters json.RawMessage `json:"parameters"` - Tier string `json:"tier"` - } `json:"tools"` - } - if err := json.Unmarshal(ext.Manifest, &manifest); err != nil { - continue - } - for _, t := range manifest.Tools { - if disabled[t.Name] { - continue - } - defs = append(defs, providers.ToolDef{ - Type: "function", - Function: providers.FunctionDef{ - Name: t.Name, - Description: t.Description, - Parameters: t.Parameters, - }, - }) - } - } - } - - // v0.25.0: Persona tool grants — second-pass allowlist. - // If the active persona has explicit tool grants, restrict to only those tools. - // Empty grants = persona inherits all context-available tools (backward compat). - if personaID != "" && h.stores.Personas != nil { - grants, err := h.stores.Personas.GetToolGrants(ctx, personaID) - if err != nil { - log.Printf("⚠️ Failed to load tool grants for persona %s: %v", personaID, err) - } else if len(grants) > 0 { - allowed := make(map[string]bool, len(grants)) - for _, g := range grants { - allowed[g] = true - } - filtered := defs[:0] - for _, d := range defs { - if allowed[d.Function.Name] { - filtered = append(filtered, d) - } - } - defs = filtered - } - } - - return defs + return BuildToolDefs(ctx, h.stores, userID, includeBrowser, disabledTools, tctx, personaID) } // ── List Available Tools ──────────────────── @@ -1017,8 +929,6 @@ func (h *CompletionHandler) ListTools(c *gin.Context) { // ── Streaming Completion (SSE) with Tool Loop ── -const maxToolIterations = 10 // safety limit on tool call rounds - func (h *CompletionHandler) streamCompletion( c *gin.Context, provider providers.Provider, @@ -1071,125 +981,92 @@ func (h *CompletionHandler) syncCompletion( hooks.PreRequest(cfg, &req) } - var totalInput, totalOutput int - var totalCacheCreation, totalCacheRead int - var finalContent string - var allToolActivity []map[string]interface{} - - for iteration := 0; iteration < maxToolIterations; iteration++ { - callStart := time.Now() - resp, err := provider.ChatCompletion(c.Request.Context(), cfg, req) - h.recordHealth(configID, callStart, err) - if err != nil { - c.JSON(http.StatusBadGateway, gin.H{"error": "provider error: " + err.Error()}) - return - } - - totalInput += resp.InputTokens - totalOutput += resp.OutputTokens - totalCacheCreation += resp.CacheCreationTokens - totalCacheRead += resp.CacheReadTokens - - // No tool calls — normal response - if len(resp.ToolCalls) == 0 || resp.FinishReason != "tool_calls" { - finalContent = resp.Content - - // Persist assistant response - if _, err := h.persistMessage(channelID, userID, "assistant", finalContent, model, totalInput, totalOutput, nil, toolActivityJSON(allToolActivity), configID, personaID); err != nil { - log.Printf("Failed to persist assistant message: %v", err) - } - - // AI-to-AI chaining - go func() { - defer func() { - if r := recover(); r != nil { - log.Printf("[chain] PANIC recovered: %v", r) - } - }() - h.chainIfMentioned(channelID, userID, personaID, finalContent, 0) - }() - - // Log usage - h.logUsage(c, channelID, userID, configID, providerScope, model, - totalInput, totalOutput, totalCacheCreation, totalCacheRead) - - // Return OpenAI-compatible response - c.JSON(http.StatusOK, gin.H{ - "model": resp.Model, - "choices": []gin.H{ - { - "message": gin.H{ - "role": "assistant", - "content": finalContent, - }, - "finish_reason": resp.FinishReason, - }, - }, - "usage": gin.H{ - "prompt_tokens": totalInput, - "completion_tokens": totalOutput, - "total_tokens": totalInput + totalOutput, - }, - }) - return - } - - // ── Tool execution round ──────────────── - assistantMsg := providers.Message{ - Role: "assistant", - Content: resp.Content, - } - for _, tc := range resp.ToolCalls { - assistantMsg.ToolCalls = append(assistantMsg.ToolCalls, tc) - } - req.Messages = append(req.Messages, assistantMsg) - - execCtx := tools.ExecutionContext{ + result := CoreToolLoop(c.Request.Context(), LoopConfig{ + Provider: provider, + Cfg: cfg, + Req: &req, + Model: model, + ProviderType: providerID, + ExecCtx: tools.ExecutionContext{ UserID: userID, ChannelID: channelID, PersonaID: personaID, WorkspaceID: workspaceID, TeamID: teamID, - } - for _, tc := range resp.ToolCalls { - call := tools.ToolCall{ - ID: tc.ID, - Name: tc.Function.Name, - Arguments: tc.Function.Arguments, - } - log.Printf("🔧 Executing tool: %s (call %s)", call.Name, call.ID) - result := tools.ExecuteCall(c.Request.Context(), execCtx, call) + }, + Hub: h.hub, + Health: h.health, + ConfigID: configID, + Budget: LoopBudget{}, + Streaming: false, + }, accumSink{}) - // Collect for persistence - allToolActivity = append(allToolActivity, map[string]interface{}{ - "id": tc.ID, - "name": tc.Function.Name, - "arguments": tc.Function.Arguments, - "result": result.Content, - "is_error": result.IsError, - }) - - req.Messages = append(req.Messages, providers.Message{ - Role: "tool", - Content: result.Content, - ToolCallID: result.ToolCallID, - Name: result.Name, - }) - } + // Provider error + if result.Error != nil { + c.JSON(http.StatusBadGateway, gin.H{"error": "provider error: " + result.Error.Error()}) + return } - // Max iterations reached + // Max iterations hit with no final content + if result.BudgetExceeded == "max_rounds" && result.Content == "" { + c.JSON(http.StatusOK, gin.H{ + "model": model, + "choices": []gin.H{ + { + "message": gin.H{ + "role": "assistant", + "content": "[Tool execution limit reached]", + }, + "finish_reason": "stop", + }, + }, + }) + return + } + + finalContent := result.Content + + // Persist assistant response + if _, err := h.persistMessage(channelID, userID, "assistant", finalContent, model, result.InputTokens, result.OutputTokens, nil, toolActivityJSON(result.ToolActivity), configID, personaID); err != nil { + log.Printf("Failed to persist assistant message: %v", err) + } + + // AI-to-AI chaining + go func() { + defer func() { + if r := recover(); r != nil { + log.Printf("[chain] PANIC recovered: %v", r) + } + }() + h.chainIfMentioned(channelID, userID, personaID, finalContent, 0) + }() + + // Log usage + h.logUsage(c, channelID, userID, configID, providerScope, model, + result.InputTokens, result.OutputTokens, + result.CacheCreationTokens, result.CacheReadTokens) + + // Return OpenAI-compatible response + finishReason := "stop" + if result.BudgetExceeded != "" { + finishReason = "budget_exceeded" + } c.JSON(http.StatusOK, gin.H{ "model": model, "choices": []gin.H{ { "message": gin.H{ "role": "assistant", - "content": "[Tool execution limit reached]", + "content": finalContent, }, - "finish_reason": "stop", + "finish_reason": finishReason, }, }, + "usage": gin.H{ + "prompt_tokens": result.InputTokens, + "completion_tokens": result.OutputTokens, + "total_tokens": result.InputTokens + result.OutputTokens, + }, }) } @@ -1629,125 +1506,11 @@ func extractFirstMention(content string) string { // Priority: request.provider_config_id → chat.provider_config_id → user's first active config func (h *CompletionHandler) resolveConfig(userID string, channelID string, req completionRequest) (providers.ProviderConfig, string, string, string, string, error) { - var configID string - - // 1. Explicit config from request - if req.ProviderConfigID != "" { - configID = req.ProviderConfigID - } - - // 2. Config from channel - if configID == "" { - var channelConfigID *string - err := database.DB.QueryRow( - database.Q(`SELECT provider_config_id FROM channels WHERE id = $1`), channelID, - ).Scan(&channelConfigID) - if err == nil && channelConfigID != nil { - configID = *channelConfigID - } - } - - // 3. User's first active config (personal first, then global — excludes team providers) - if configID == "" { - err := database.DB.QueryRow(database.Q(` - SELECT id FROM provider_configs - WHERE is_active = true AND ( - (scope = 'personal' AND owner_id = $1) - OR scope = 'global' - ) - ORDER BY scope ASC, created_at ASC - LIMIT 1 - `), userID).Scan(&configID) - if err != nil { - return providers.ProviderConfig{}, "", "", "", "", fmt.Errorf("no API config found — add one at /api-configs") - } - } - - // Load the config — allow personal, global, OR team configs the user belongs to - var providerID, endpoint string - var providerScope string - var modelDefault *string - var apiKeyEnc, keyNonce []byte - var keyScope string - var customHeadersJSON, providerSettingsJSON []byte - var proxyMode string - var proxyURL *string - // $2/userID appears twice in the query; Postgres reuses positional params, - // SQLite needs each ? bound separately. - configArgs := []interface{}{configID, userID} - if database.IsSQLite() { - configArgs = append(configArgs, userID) - } - err := database.DB.QueryRow(database.Q(` - SELECT provider, endpoint, scope, api_key_enc, key_nonce, key_scope, - model_default, headers, settings, COALESCE(proxy_mode, 'system'), proxy_url - FROM provider_configs - WHERE id = $1 AND is_active = true - AND (scope = 'global' - OR (scope = 'personal' AND owner_id = $2) - OR (scope = 'team' AND owner_id IN (SELECT team_id FROM team_members WHERE user_id = $2))) - `), configArgs...).Scan(&providerID, &endpoint, &providerScope, &apiKeyEnc, &keyNonce, &keyScope, - &modelDefault, &customHeadersJSON, &providerSettingsJSON, &proxyMode, &proxyURL) - - if err == sql.ErrNoRows { - return providers.ProviderConfig{}, "", "", "", "", fmt.Errorf("API config not found or not accessible") - } + res, err := ResolveProviderConfig(h.vault, userID, channelID, req.ProviderConfigID, req.Model) if err != nil { - return providers.ProviderConfig{}, "", "", "", "", fmt.Errorf("failed to load API config: %w", err) + return providers.ProviderConfig{}, "", "", "", "", err } - - // Resolve model: request > config default - model := req.Model - if model == "" && modelDefault != nil { - model = *modelDefault - } - if model == "" { - return providers.ProviderConfig{}, "", "", "", "", fmt.Errorf("no model specified and no default model in config") - } - - // Decrypt API key using the appropriate tier - key := "" - if len(apiKeyEnc) > 0 { - if h.vault != nil { - var err error - key, err = h.vault.Decrypt(apiKeyEnc, keyNonce, keyScope, userID) - if err != nil { - if err == crypto.ErrVaultLocked { - return providers.ProviderConfig{}, "", "", "", "", fmt.Errorf("personal vault is locked — please log in again") - } - return providers.ProviderConfig{}, "", "", "", "", fmt.Errorf("failed to decrypt API key: %w", err) - } - } else { - // No vault — key stored as raw bytes (unencrypted fallback) - key = string(apiKeyEnc) - } - } - - // Parse custom headers - customHeaders := make(map[string]string) - if customHeadersJSON != nil { - _ = json.Unmarshal(customHeadersJSON, &customHeaders) - } - - // Parse provider-specific settings - providerSettings := make(map[string]interface{}) - if providerSettingsJSON != nil { - _ = json.Unmarshal(providerSettingsJSON, &providerSettings) - } - - proxyURLStr := "" - if proxyURL != nil { - proxyURLStr = *proxyURL - } - - return providers.ProviderConfig{ - Endpoint: endpoint, - APIKey: key, - CustomHeaders: customHeaders, - Settings: providerSettings, - ProxyMode: proxyMode, - ProxyURL: proxyURLStr, - }, providerID, model, configID, providerScope, nil + return res.Config, res.ProviderID, res.Model, res.ConfigID, res.ProviderScope, nil } // ── Conversation Loader ───────────────────── diff --git a/server/handlers/resolve.go b/server/handlers/resolve.go new file mode 100644 index 0000000..eb8e867 --- /dev/null +++ b/server/handlers/resolve.go @@ -0,0 +1,289 @@ +// Package handlers — resolve.go +// +// v0.27.2: Extracted provider resolution and tool definition building +// from CompletionHandler methods to standalone functions. Both the HTTP +// completion handler and the headless task scheduler need these paths. +// +// The CompletionHandler methods (resolveConfig, buildToolDefs) remain as +// thin wrappers for backward compat — existing callers are unchanged. +package handlers + +import ( + "context" + "database/sql" + "encoding/json" + "fmt" + "log" + + "git.gobha.me/xcaliber/chat-switchboard/crypto" + "git.gobha.me/xcaliber/chat-switchboard/database" + "git.gobha.me/xcaliber/chat-switchboard/providers" + "git.gobha.me/xcaliber/chat-switchboard/store" + "git.gobha.me/xcaliber/chat-switchboard/tools" +) + +// ── Provider Resolution ──────────────────────── + +// ProviderResolution holds the resolved provider configuration and metadata. +type ProviderResolution struct { + Config providers.ProviderConfig + ProviderID string // e.g. "anthropic", "openai" + Model string // resolved model ID + ConfigID string // provider_configs.id + ProviderScope string // "personal", "team", "global" +} + +// ResolveProviderConfig resolves the provider configuration for a completion +// request. Resolution order: +// 1. Explicit providerConfigID (from request or task definition) +// 2. Channel's configured provider (provider_config_id on channels table) +// 3. User's first active config (personal first, then global) +// +// The vault is used to decrypt API keys. Pass nil for unencrypted fallback. +func ResolveProviderConfig( + vault *crypto.KeyResolver, + userID, channelID, providerConfigID, modelID string, +) (ProviderResolution, error) { + configID := providerConfigID + + // 2. Config from channel + if configID == "" && channelID != "" { + var channelConfigID *string + err := database.DB.QueryRow( + database.Q(`SELECT provider_config_id FROM channels WHERE id = $1`), channelID, + ).Scan(&channelConfigID) + if err == nil && channelConfigID != nil { + configID = *channelConfigID + } + } + + // 3. User's first active config (personal first, then global — excludes team providers) + if configID == "" { + err := database.DB.QueryRow(database.Q(` + SELECT id FROM provider_configs + WHERE is_active = true AND ( + (scope = 'personal' AND owner_id = $1) + OR scope = 'global' + ) + ORDER BY scope ASC, created_at ASC + LIMIT 1 + `), userID).Scan(&configID) + if err != nil { + return ProviderResolution{}, fmt.Errorf("no API config found — add one at /api-configs") + } + } + + // Load the config — allow personal, global, OR team configs the user belongs to + var providerID, endpoint string + var providerScope string + var modelDefault *string + var apiKeyEnc, keyNonce []byte + var keyScope string + var customHeadersJSON, providerSettingsJSON []byte + var proxyMode string + var proxyURL *string + + // $2/userID appears twice in the query; Postgres reuses positional params, + // SQLite needs each ? bound separately. + configArgs := []interface{}{configID, userID} + if database.IsSQLite() { + configArgs = append(configArgs, userID) + } + err := database.DB.QueryRow(database.Q(` + SELECT provider, endpoint, scope, api_key_enc, key_nonce, key_scope, + model_default, headers, settings, COALESCE(proxy_mode, 'system'), proxy_url + FROM provider_configs + WHERE id = $1 AND is_active = true + AND (scope = 'global' + OR (scope = 'personal' AND owner_id = $2) + OR (scope = 'team' AND owner_id IN (SELECT team_id FROM team_members WHERE user_id = $2))) + `), configArgs...).Scan(&providerID, &endpoint, &providerScope, &apiKeyEnc, &keyNonce, &keyScope, + &modelDefault, &customHeadersJSON, &providerSettingsJSON, &proxyMode, &proxyURL) + + if err == sql.ErrNoRows { + return ProviderResolution{}, fmt.Errorf("API config not found or not accessible") + } + if err != nil { + return ProviderResolution{}, fmt.Errorf("failed to load API config: %w", err) + } + + // Resolve model: explicit > config default + model := modelID + if model == "" && modelDefault != nil { + model = *modelDefault + } + if model == "" { + return ProviderResolution{}, fmt.Errorf("no model specified and no default model in config") + } + + // Decrypt API key using the appropriate tier + key := "" + if len(apiKeyEnc) > 0 { + if vault != nil { + var err error + key, err = vault.Decrypt(apiKeyEnc, keyNonce, keyScope, userID) + if err != nil { + if err == crypto.ErrVaultLocked { + return ProviderResolution{}, fmt.Errorf("personal vault is locked — please log in again") + } + return ProviderResolution{}, fmt.Errorf("failed to decrypt API key: %w", err) + } + } else { + // No vault — key stored as raw bytes (unencrypted fallback) + key = string(apiKeyEnc) + } + } + + // Parse custom headers + customHeaders := make(map[string]string) + if customHeadersJSON != nil { + _ = json.Unmarshal(customHeadersJSON, &customHeaders) + } + + // Parse provider-specific settings + providerSettings := make(map[string]interface{}) + if providerSettingsJSON != nil { + _ = json.Unmarshal(providerSettingsJSON, &providerSettings) + } + + proxyURLStr := "" + if proxyURL != nil { + proxyURLStr = *proxyURL + } + + return ProviderResolution{ + Config: providers.ProviderConfig{ + Endpoint: endpoint, + APIKey: key, + CustomHeaders: customHeaders, + Settings: providerSettings, + ProxyMode: proxyMode, + ProxyURL: proxyURLStr, + }, + ProviderID: providerID, + Model: model, + ConfigID: configID, + ProviderScope: providerScope, + }, nil +} + +// ── Tool Definition Building ─────────────────── + +// BuildToolDefs assembles the tool definitions for a completion request. +// Includes server-registered tools filtered by context predicates, +// browser extension tools (if includeBrowser), and persona tool grant +// allowlisting. +// +// Additional tool grant filtering (e.g. task-level grants) can be applied +// by the caller after this function returns. +func BuildToolDefs( + ctx context.Context, + stores store.Stores, + userID string, + includeBrowser bool, + disabledTools []string, + tctx tools.ToolContext, + personaID string, +) []providers.ToolDef { + // Build disabled set for O(1) lookup + disabled := make(map[string]bool, len(disabledTools)) + for _, name := range disabledTools { + disabled[name] = true + } + + // v0.25.0: Tools self-declare availability via predicates. + allTools := tools.AvailableFor(tctx, disabled) + defs := make([]providers.ToolDef, 0, len(allTools)) + for _, t := range allTools { + defs = append(defs, providers.ToolDef{ + Type: "function", + Function: providers.FunctionDef{ + Name: t.Name, + Description: t.Description, + Parameters: t.Parameters, + }, + }) + } + + // Append browser-defined tool schemas from extensions + if includeBrowser && stores.Extensions != nil { + exts, err := stores.Extensions.ListForUser(ctx, userID) + if err != nil { + log.Printf("⚠️ Failed to load extensions for tools: %v", err) + return defs + } + for _, ext := range exts { + if ext.Tier != "browser" { + continue + } + var manifest struct { + Tools []struct { + Name string `json:"name"` + Description string `json:"description"` + Parameters json.RawMessage `json:"parameters"` + Tier string `json:"tier"` + } `json:"tools"` + } + if err := json.Unmarshal(ext.Manifest, &manifest); err != nil { + continue + } + for _, t := range manifest.Tools { + if disabled[t.Name] { + continue + } + defs = append(defs, providers.ToolDef{ + Type: "function", + Function: providers.FunctionDef{ + Name: t.Name, + Description: t.Description, + Parameters: t.Parameters, + }, + }) + } + } + } + + // v0.25.0: Persona tool grants — second-pass allowlist. + // If the active persona has explicit tool grants, restrict to only those tools. + // Empty grants = persona inherits all context-available tools (backward compat). + if personaID != "" && stores.Personas != nil { + grants, err := stores.Personas.GetToolGrants(ctx, personaID) + if err != nil { + log.Printf("⚠️ Failed to load tool grants for persona %s: %v", personaID, err) + } else if len(grants) > 0 { + allowed := make(map[string]bool, len(grants)) + for _, g := range grants { + allowed[g] = true + } + filtered := defs[:0] + for _, d := range defs { + if allowed[d.Function.Name] { + filtered = append(filtered, d) + } + } + defs = filtered + } + } + + return defs +} + +// FilterToolDefsByGrants applies an additional allowlist to tool defs. +// Used by the task scheduler to enforce task-level tool grants on top +// of persona-level grants. Passing nil or empty grants returns defs unchanged. +func FilterToolDefsByGrants(defs []providers.ToolDef, grants []string) []providers.ToolDef { + if len(grants) == 0 { + return defs + } + allowed := make(map[string]bool, len(grants)) + for _, g := range grants { + allowed[g] = true + } + filtered := make([]providers.ToolDef, 0, len(defs)) + for _, d := range defs { + if allowed[d.Function.Name] { + filtered = append(filtered, d) + } + } + return filtered +} diff --git a/server/handlers/stream_loop.go b/server/handlers/stream_loop.go index c1459e7..14e691d 100644 --- a/server/handlers/stream_loop.go +++ b/server/handlers/stream_loop.go @@ -4,7 +4,6 @@ import ( "crypto/rand" "encoding/hex" "encoding/json" - "fmt" "log" "net/http" "strings" @@ -17,17 +16,6 @@ import ( "git.gobha.me/xcaliber/chat-switchboard/tools" ) -// streamResult holds the accumulated output from a streaming completion -// with tool execution. Callers are responsible for persistence. -type streamResult struct { - Content string - ToolActivity []map[string]interface{} - InputTokens int - OutputTokens int - CacheCreationTokens int - CacheReadTokens int -} - // recordHealthFn records provider health from standalone streaming functions. func recordHealthFn(hr HealthRecorder, configID string, start time.Time, err error) { if hr == nil || configID == "" { @@ -46,17 +34,13 @@ func recordHealthFn(hr HealthRecorder, configID string, start time.Time, err err } } -// streamWithToolLoop is the single, canonical streaming implementation. -// It handles SSE setup, multi-round tool execution, reasoning_content -// forwarding, and client notifications. Returns the accumulated content -// and tool activity for the caller to persist however it needs to. +// streamWithToolLoop is the SSE streaming entry point for single-model +// completion with tool execution. Sets SSE headers, delegates to +// CoreToolLoop with an sseSink, and returns the accumulated result. // // Used by: // - streamCompletion (normal chat — persists via persistMessage) // - Regenerate (regen/edit — persists as sibling with tree metadata) -// -// If you add features here (new event types, new tool capabilities, new -// persistence fields), both callers get them automatically. func streamWithToolLoop( c *gin.Context, provider providers.Provider, @@ -65,244 +49,42 @@ func streamWithToolLoop( model, providerType, userID, channelID, personaID, workspaceID, configID, teamID string, hub *events.Hub, health HealthRecorder, -) streamResult { +) LoopResult { // Set SSE headers c.Header("Content-Type", "text/event-stream") c.Header("Cache-Control", "no-cache") c.Header("Connection", "keep-alive") c.Header("X-Accel-Buffering", "no") - c.Status(http.StatusOK) - flusher, _ := c.Writer.(http.Flusher) - flush := func() { - if flusher != nil { - flusher.Flush() - } - } + sink := newSSESink(c, model) - sendSSE := func(data string) { - fmt.Fprintf(c.Writer, "data: %s\n\n", data) - flush() - } - - sendEvent := func(event, data string) { - fmt.Fprintf(c.Writer, "event: %s\ndata: %s\n\n", event, data) - flush() - } - - var result streamResult - - for iteration := 0; iteration < maxToolIterations; iteration++ { - callStart := time.Now() - ch, err := provider.StreamCompletion(c.Request.Context(), cfg, *req) - if err != nil { - recordHealthFn(health, configID, callStart, err) - sendSSE(fmt.Sprintf(`{"error":"%s"}`, escapeJSON(err.Error()))) - return result - } - - var iterContent string - var iterReasoning string - var toolCalls []providers.ToolCall - - for event := range ch { - // Apply provider-specific post-processing hooks (v0.22.1) - if hooks := providers.GetHooks(providerType); hooks != nil { - hooks.PostStreamEvent(cfg, &event) - } - - if event.Error != nil { - recordHealthFn(health, configID, callStart, event.Error) - sendSSE(fmt.Sprintf(`{"error":"%s"}`, escapeJSON(event.Error.Error()))) - return result - } - - // Stream reasoning deltas to client - if event.Reasoning != "" { - iterReasoning += event.Reasoning - sendSSE(fmt.Sprintf( - `{"choices":[{"delta":{"reasoning_content":%q},"finish_reason":null}],"model":%q}`, - event.Reasoning, model)) - } - - // Stream text deltas to client - if event.Delta != "" { - iterContent += event.Delta - sendSSE(fmt.Sprintf( - `{"choices":[{"delta":{"content":%q},"finish_reason":null}],"model":%q}`, - event.Delta, model)) - } - - if event.Done { - recordHealthFn(health, configID, callStart, nil) - if event.FinishReason == "tool_calls" && len(event.ToolCalls) > 0 { - toolCalls = event.ToolCalls - // Accumulate tokens from this tool-call iteration - result.InputTokens += event.InputTokens - result.OutputTokens += event.OutputTokens - result.CacheCreationTokens += event.CacheCreationTokens - result.CacheReadTokens += event.CacheReadTokens - } else { - // Normal completion — send finish event - finishReason := event.FinishReason - if finishReason == "" { - finishReason = "stop" - } - if iterReasoning != "" { - result.Content += "" + iterReasoning + "" - } - result.Content += iterContent - // Accumulate final tokens - result.InputTokens += event.InputTokens - result.OutputTokens += event.OutputTokens - result.CacheCreationTokens += event.CacheCreationTokens - result.CacheReadTokens += event.CacheReadTokens - sendSSE(fmt.Sprintf( - `{"choices":[{"delta":{},"finish_reason":%q}],"model":%q}`, - finishReason, model)) - sendSSE("[DONE]") - return result - } - break - } - } - - // ── Tool execution round ──────────────── - if len(toolCalls) == 0 { - // Stream ended without tool calls or finish — shouldn't happen - sendSSE("[DONE]") - if iterReasoning != "" { - result.Content += "" + iterReasoning + "" - } - result.Content += iterContent - return result - } - - // Notify client about tool calls - toolCallsJSON, _ := json.Marshal(toolCalls) - sendEvent("tool_use", string(toolCallsJSON)) - - // Append assistant message (with tool_calls, possibly with text) to conversation - assistantMsg := providers.Message{ - Role: "assistant", - Content: iterContent, - } - for _, tc := range toolCalls { - assistantMsg.ToolCalls = append(assistantMsg.ToolCalls, tc) - } - req.Messages = append(req.Messages, assistantMsg) - - // Execute all tool calls - execCtx := tools.ExecutionContext{ + return CoreToolLoop(c.Request.Context(), LoopConfig{ + Provider: provider, + Cfg: cfg, + Req: req, + Model: model, + ProviderType: providerType, + ExecCtx: tools.ExecutionContext{ UserID: userID, ChannelID: channelID, PersonaID: personaID, WorkspaceID: workspaceID, TeamID: teamID, - } - for _, tc := range toolCalls { - call := tools.ToolCall{ - ID: tc.ID, - Name: tc.Function.Name, - Arguments: tc.Function.Arguments, - } - - var toolResult tools.ToolResult - - // Check if tool is registered locally (server-side) - toolStart := time.Now() - if tools.Get(call.Name) != nil { - log.Printf("🔧 Executing tool (server): %s (call %s)", call.Name, call.ID) - toolResult = tools.ExecuteCall(c.Request.Context(), execCtx, call) - // Record tool health (v0.22.4) - if health != nil { - toolLatency := int(time.Since(toolStart).Milliseconds()) - if toolResult.IsError { - health.RecordToolError(call.Name, toolLatency, toolResult.Content) - } else { - health.RecordToolSuccess(call.Name, toolLatency) - } - } - } else if hub != nil && hub.IsConnected(userID) { - // Route to browser via WebSocket bridge - log.Printf("🔧 Executing tool (browser): %s (call %s)", call.Name, call.ID) - toolResult = executeBrowserTool(hub, userID, call) - } else { - log.Printf("⚠️ Unknown tool: %s (call %s)", call.Name, call.ID) - toolResult = tools.ToolResult{ - ToolCallID: call.ID, - Name: call.Name, - Content: `{"error":"unknown tool or browser not connected"}`, - IsError: true, - } - } - - // Notify client about tool result - resultJSON, _ := json.Marshal(map[string]interface{}{ - "tool_call_id": toolResult.ToolCallID, - "name": toolResult.Name, - "content": toolResult.Content, - "is_error": toolResult.IsError, - }) - sendEvent("tool_result", string(resultJSON)) - - // Emit workspace.file.changed for live editor updates (v0.21.5) - if hub != nil && !toolResult.IsError && workspaceID != "" { - if call.Name == "workspace_write" || call.Name == "workspace_patch" { - var toolArgs struct{ Path string `json:"path"` } - if json.Unmarshal([]byte(call.Arguments), &toolArgs) == nil && toolArgs.Path != "" { - hub.SendToUser(userID, events.Event{ - Label: "workspace.file.changed", - Payload: events.MustJSON(map[string]string{ - "workspace_id": workspaceID, - "path": toolArgs.Path, - }), - }) - } - } - } - - // Collect for persistence - result.ToolActivity = append(result.ToolActivity, map[string]interface{}{ - "id": tc.ID, - "name": tc.Function.Name, - "arguments": tc.Function.Arguments, - "result": toolResult.Content, - "is_error": toolResult.IsError, - }) - - // Append tool result to conversation for next iteration - req.Messages = append(req.Messages, providers.Message{ - Role: "tool", - Content: toolResult.Content, - ToolCallID: toolResult.ToolCallID, - Name: toolResult.Name, - }) - } - - result.Content += iterContent - // Loop: send updated messages back to provider - } - - // Hit max iterations - log.Printf("⚠️ Tool loop hit max iterations (%d) for channel %s", maxToolIterations, channelID) - sendSSE(fmt.Sprintf( - `{"choices":[{"delta":{"content":"%s"},"finish_reason":"stop"}],"model":%q}`, - escapeJSON("[Tool execution limit reached]"), model)) - sendSSE("[DONE]") - - return result + }, + Hub: hub, + Health: health, + ConfigID: configID, + Budget: LoopBudget{}, + Streaming: true, + }, sink) } const browserToolTimeout = 30 * time.Second -// streamModelResponse is a variant of streamWithToolLoop used by multi-model -// fan-out (v0.20.0). It streams a single model's response within an already- -// established SSE connection — it does NOT set SSE headers or send [DONE]. -// -// SSE deltas include a "model_display" field so the frontend can attribute -// each response to the correct model. +// streamModelResponse is the multi-model streaming variant. Runs on an +// already-established SSE connection — does NOT set SSE headers or send +// [DONE]. SSE deltas include a "model_display" field for frontend attribution. func streamModelResponse( c *gin.Context, provider providers.Provider, @@ -311,187 +93,28 @@ func streamModelResponse( model, providerType, displayName, userID, channelID, personaID, workspaceID, configID, teamID string, hub *events.Hub, health HealthRecorder, -) streamResult { - flusher, _ := c.Writer.(http.Flusher) - flush := func() { - if flusher != nil { - flusher.Flush() - } - } - sendSSE := func(data string) { - fmt.Fprintf(c.Writer, "data: %s\n\n", data) - flush() - } - sendEvent := func(event, data string) { - fmt.Fprintf(c.Writer, "event: %s\ndata: %s\n\n", event, data) - flush() - } +) LoopResult { + sink := newSSEModelSink(c, model, displayName) - var result streamResult - - for iteration := 0; iteration < maxToolIterations; iteration++ { - callStart := time.Now() - ch, err := provider.StreamCompletion(c.Request.Context(), cfg, *req) - if err != nil { - recordHealthFn(health, configID, callStart, err) - sendSSE(fmt.Sprintf(`{"error":"%s"}`, escapeJSON(err.Error()))) - return result - } - - var iterContent string - var iterReasoning string - var toolCalls []providers.ToolCall - - for event := range ch { - // Apply provider-specific post-processing hooks (v0.22.1) - if hooks := providers.GetHooks(providerType); hooks != nil { - hooks.PostStreamEvent(cfg, &event) - } - - if event.Error != nil { - recordHealthFn(health, configID, callStart, event.Error) - sendSSE(fmt.Sprintf(`{"error":"%s"}`, escapeJSON(event.Error.Error()))) - return result - } - - if event.Reasoning != "" { - iterReasoning += event.Reasoning - sendSSE(fmt.Sprintf( - `{"choices":[{"delta":{"reasoning_content":%q},"finish_reason":null}],"model":%q,"model_display":%q}`, - event.Reasoning, model, displayName)) - } - - if event.Delta != "" { - iterContent += event.Delta - sendSSE(fmt.Sprintf( - `{"choices":[{"delta":{"content":%q},"finish_reason":null}],"model":%q,"model_display":%q}`, - event.Delta, model, displayName)) - } - - if event.Done { - recordHealthFn(health, configID, callStart, nil) - if event.FinishReason == "tool_calls" && len(event.ToolCalls) > 0 { - toolCalls = event.ToolCalls - result.InputTokens += event.InputTokens - result.OutputTokens += event.OutputTokens - result.CacheCreationTokens += event.CacheCreationTokens - result.CacheReadTokens += event.CacheReadTokens - } else { - finishReason := event.FinishReason - if finishReason == "" { - finishReason = "stop" - } - if iterReasoning != "" { - result.Content += "" + iterReasoning + "" - } - result.Content += iterContent - result.InputTokens += event.InputTokens - result.OutputTokens += event.OutputTokens - result.CacheCreationTokens += event.CacheCreationTokens - result.CacheReadTokens += event.CacheReadTokens - sendSSE(fmt.Sprintf( - `{"choices":[{"delta":{},"finish_reason":%q}],"model":%q,"model_display":%q}`, - finishReason, model, displayName)) - // Do NOT send [DONE] — caller manages that for multi-model - return result - } - break - } - } - - // Tool execution (same logic as streamWithToolLoop) - if len(toolCalls) == 0 { - if iterReasoning != "" { - result.Content += "" + iterReasoning + "" - } - result.Content += iterContent - return result - } - - toolCallsJSON, _ := json.Marshal(toolCalls) - sendEvent("tool_use", string(toolCallsJSON)) - - assistantMsg := providers.Message{ - Role: "assistant", - Content: iterContent, - } - for _, tc := range toolCalls { - assistantMsg.ToolCalls = append(assistantMsg.ToolCalls, tc) - } - req.Messages = append(req.Messages, assistantMsg) - - execCtx := tools.ExecutionContext{ + return CoreToolLoop(c.Request.Context(), LoopConfig{ + Provider: provider, + Cfg: cfg, + Req: req, + Model: model, + ProviderType: providerType, + ExecCtx: tools.ExecutionContext{ UserID: userID, ChannelID: channelID, PersonaID: personaID, WorkspaceID: workspaceID, TeamID: teamID, - } - for _, tc := range toolCalls { - call := tools.ToolCall{ - ID: tc.ID, - Name: tc.Function.Name, - Arguments: tc.Function.Arguments, - } - - var toolResult tools.ToolResult - toolStart := time.Now() - if tools.Get(call.Name) != nil { - toolResult = tools.ExecuteCall(c.Request.Context(), execCtx, call) - // Record tool health (v0.22.4) - if health != nil { - toolLatency := int(time.Since(toolStart).Milliseconds()) - if toolResult.IsError { - health.RecordToolError(call.Name, toolLatency, toolResult.Content) - } else { - health.RecordToolSuccess(call.Name, toolLatency) - } - } - } else if hub != nil && hub.IsConnected(userID) { - toolResult = executeBrowserTool(hub, userID, call) - } else { - toolResult = tools.ToolResult{ - ToolCallID: call.ID, - Name: call.Name, - Content: `{"error":"unknown tool or browser not connected"}`, - IsError: true, - } - } - - resultJSON, _ := json.Marshal(map[string]interface{}{ - "tool_call_id": toolResult.ToolCallID, - "name": toolResult.Name, - "content": toolResult.Content, - "is_error": toolResult.IsError, - }) - sendEvent("tool_result", string(resultJSON)) - - result.ToolActivity = append(result.ToolActivity, map[string]interface{}{ - "id": tc.ID, - "name": tc.Function.Name, - "arguments": tc.Function.Arguments, - "result": toolResult.Content, - "is_error": toolResult.IsError, - }) - - req.Messages = append(req.Messages, providers.Message{ - Role: "tool", - Content: toolResult.Content, - ToolCallID: toolResult.ToolCallID, - Name: toolResult.Name, - }) - } - - result.Content += iterContent - } - - // Hit max iterations - log.Printf("⚠️ Multi-model tool loop hit max iterations (%d) for model %s", maxToolIterations, displayName) - sendSSE(fmt.Sprintf( - `{"choices":[{"delta":{"content":"%s"},"finish_reason":"stop"}],"model":%q,"model_display":%q}`, - escapeJSON("[Tool execution limit reached]"), model, displayName)) - - return result + }, + Hub: hub, + Health: health, + ConfigID: configID, + Budget: LoopBudget{}, + Streaming: true, + }, sink) } // executeBrowserTool sends a tool call to the user's browser via WebSocket diff --git a/server/handlers/tasks.go b/server/handlers/tasks.go index 53d2dd8..08196db 100644 --- a/server/handlers/tasks.go +++ b/server/handlers/tasks.go @@ -7,6 +7,7 @@ import ( "github.com/gin-gonic/gin" "git.gobha.me/xcaliber/chat-switchboard/models" + "git.gobha.me/xcaliber/chat-switchboard/taskutil" "git.gobha.me/xcaliber/chat-switchboard/store" ) @@ -51,6 +52,15 @@ func (h *TaskHandler) ListAll(c *gin.Context) { // Create creates a new task. // POST /api/v1/tasks func (h *TaskHandler) Create(c *gin.Context) { + ctx := c.Request.Context() + + // v0.27.2: Check global task configuration + taskCfg := taskutil.LoadTaskConfig(ctx, h.stores.GlobalConfig) + if !taskCfg.Enabled { + c.JSON(http.StatusForbidden, gin.H{"error": "tasks are disabled by the administrator"}) + return + } + var t models.Task if err := c.ShouldBindJSON(&t); err != nil { c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()}) @@ -59,6 +69,14 @@ func (h *TaskHandler) Create(c *gin.Context) { t.OwnerID = c.GetString("user_id") + // v0.27.2: Personal task check — non-admin users need tasks.allow_personal + if t.Scope == "personal" || t.Scope == "" { + if c.GetString("role") != "admin" && !taskCfg.AllowPersonal { + c.JSON(http.StatusForbidden, gin.H{"error": "personal tasks are disabled — contact your administrator"}) + return + } + } + // Validate required fields if t.Name == "" { c.JSON(http.StatusBadRequest, gin.H{"error": "name is required"}) @@ -77,6 +95,12 @@ func (h *TaskHandler) Create(c *gin.Context) { 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 + } + // Defaults if t.Scope == "" { t.Scope = "personal" @@ -90,29 +114,22 @@ func (h *TaskHandler) Create(c *gin.Context) { if t.OutputMode == "" { t.OutputMode = "channel" } - if t.MaxTokens == 0 { - t.MaxTokens = 4096 - } - if t.MaxToolCalls == 0 { - t.MaxToolCalls = 10 - } - if t.MaxWallClock == 0 { - t.MaxWallClock = 300 - } + + // v0.27.2: Apply global default budgets for zero-value fields + taskCfg.ApplyDefaults(&t) // Compute initial next_run_at - // For "once" tasks, next_run_at = now (runs on next poll) - // For cron tasks, compute from schedule if t.Schedule == "once" { now := time.Now().UTC() t.NextRunAt = &now + } else { + // v0.27.2: Full cron parsing via robfig/cron/v3 + t.NextRunAt = taskutil.NextRunFromSchedule(t.Schedule, t.Timezone) } - // TODO: compute next_run_at from cron expression (requires robfig/cron/v3) - // For now, "once" tasks run immediately; cron tasks need next_run_at set by scheduler t.IsActive = true - if err := h.stores.Tasks.Create(c.Request.Context(), &t); err != nil { + if err := h.stores.Tasks.Create(ctx, &t); err != nil { c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to create task: " + err.Error()}) return } @@ -160,12 +177,29 @@ func (h *TaskHandler) Update(c *gin.Context) { return } + // v0.27.2: Validate new schedule if provided + if patch.Schedule != nil { + if err := taskutil.ValidateCron(*patch.Schedule); err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": "invalid schedule: " + err.Error()}) + return + } + } + if err := h.stores.Tasks.Update(ctx, id, patch); err != nil { c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to update task"}) return } updated, _ := h.stores.Tasks.GetByID(ctx, id) + + // 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 + } + c.JSON(http.StatusOK, updated) } @@ -213,6 +247,13 @@ func (h *TaskHandler) RunNow(c *gin.Context) { id := c.Param("id") ctx := c.Request.Context() + // v0.27.2: Check tasks enabled + taskCfg := taskutil.LoadTaskConfig(ctx, h.stores.GlobalConfig) + if !taskCfg.Enabled { + c.JSON(http.StatusForbidden, gin.H{"error": "tasks are disabled by the administrator"}) + return + } + t, err := h.stores.Tasks.GetByID(ctx, id) if err != nil { c.JSON(http.StatusNotFound, gin.H{"error": "task not found"}) @@ -238,3 +279,33 @@ func (h *TaskHandler) RunNow(c *gin.Context) { c.JSON(http.StatusOK, gin.H{"scheduled": true, "next_run_at": now}) } + +// KillRun cancels the active run of a task. +// POST /api/v1/tasks/:id/kill +func (h *TaskHandler) KillRun(c *gin.Context) { + id := c.Param("id") + ctx := c.Request.Context() + + t, err := h.stores.Tasks.GetByID(ctx, id) + if err != nil { + c.JSON(http.StatusNotFound, gin.H{"error": "task not found"}) + return + } + if c.GetString("role") != "admin" && t.OwnerID != c.GetString("user_id") { + c.JSON(http.StatusNotFound, gin.H{"error": "task not found"}) + return + } + + active, _ := h.stores.Tasks.GetActiveRun(ctx, id) + if active == nil { + c.JSON(http.StatusNotFound, gin.H{"error": "no active run to cancel"}) + return + } + + if err := h.stores.Tasks.UpdateRun(ctx, active.ID, "cancelled", active.TokensUsed, active.ToolCalls, active.WallClock, "killed by user"); err != nil { + c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to cancel run"}) + return + } + + c.JSON(http.StatusOK, gin.H{"killed": true, "run_id": active.ID}) +} diff --git a/server/handlers/tool_loop.go b/server/handlers/tool_loop.go new file mode 100644 index 0000000..115e4d8 --- /dev/null +++ b/server/handlers/tool_loop.go @@ -0,0 +1,612 @@ +// Package handlers — tool_loop.go +// +// v0.27.2: Extracted core tool loop with sink abstraction. +// +// coreToolLoop is the single implementation of multi-round LLM completion +// with tool execution. All callers (SSE streaming, multi-model streaming, +// sync JSON, headless scheduler) provide a LoopSink for I/O and a LoopConfig +// for dependencies. The loop handles provider calls, tool dispatch (server + +// browser bridge), health recording, budget enforcement, and workspace events. +// +// Prior to this extraction, the same logic was duplicated in three places: +// - streamWithToolLoop (SSE streaming) +// - streamModelResponse (multi-model SSE) +// - syncCompletion (non-streaming JSON) +// +// Those functions are now thin wrappers over coreToolLoop. +package handlers + +import ( + "context" + "encoding/json" + "fmt" + "log" + "net/http" + "time" + + "github.com/gin-gonic/gin" + + "git.gobha.me/xcaliber/chat-switchboard/events" + "git.gobha.me/xcaliber/chat-switchboard/providers" + "git.gobha.me/xcaliber/chat-switchboard/tools" +) + +// ── Types ────────────────────────────────────── + +// defaultMaxRounds is the safety limit on tool call iterations. +// Callers can override via LoopBudget.MaxRounds. +const defaultMaxRounds = 10 + +// LoopResult holds the accumulated output from a completion with tool +// execution. Callers are responsible for persistence. +type LoopResult struct { + Content string + ToolActivity []map[string]interface{} + InputTokens int + OutputTokens int + CacheCreationTokens int + CacheReadTokens int + ToolCallCount int // total tool calls across all rounds + BudgetExceeded string // "" | "tokens" | "tool_calls" | "max_rounds" + Error error // non-nil if the loop terminated due to a provider error +} + +// LoopBudget controls resource limits for the tool loop. +// Zero values mean "use default" for MaxRounds and "unlimited" for the rest. +type LoopBudget struct { + MaxRounds int // max tool-call iterations; 0 = defaultMaxRounds + MaxToolCalls int // total tool calls across all rounds; 0 = unlimited + MaxTokens int // total input+output tokens; 0 = unlimited +} + +func (b LoopBudget) maxRounds() int { + if b.MaxRounds > 0 { + return b.MaxRounds + } + return defaultMaxRounds +} + +// LoopConfig bundles all dependencies for the core tool loop. +type LoopConfig struct { + Provider providers.Provider + Cfg providers.ProviderConfig + Req *providers.CompletionRequest + Model string + ProviderType string + ExecCtx tools.ExecutionContext + Hub *events.Hub // nil for headless (scheduler) + Health HealthRecorder // nil for headless + ConfigID string + Budget LoopBudget + Streaming bool // true = StreamCompletion, false = ChatCompletion +} + +// ── Sink Interface ───────────────────────────── + +// LoopSink receives events from the core tool loop for I/O. +// Implementations control what happens with each event: write SSE to +// a gin.Context, accumulate silently, log to stdout, etc. +type LoopSink interface { + // OnDelta receives a text content delta from the LLM. + OnDelta(delta string) + + // OnReasoning receives a reasoning/thinking delta from the LLM. + OnReasoning(delta string) + + // OnError receives a fatal error (provider failure, stream error). + OnError(err error) + + // OnToolUse is called when the LLM requests tool calls. + OnToolUse(calls []providers.ToolCall) + + // OnToolResult is called after each tool execution completes. + OnToolResult(result map[string]interface{}) + + // OnFinish is called when the LLM produces a finish_reason. + OnFinish(reason string) + + // OnDone signals end of the entire completion (after finish or budget breach). + OnDone() + + // OnMaxRounds is called when the loop hits its iteration limit. + OnMaxRounds() +} + +// ── Core Tool Loop ───────────────────────────── + +// coreToolLoop is the single, canonical tool-loop implementation. +// It drives multi-round LLM completion with tool execution, budget +// enforcement, and health recording. I/O is delegated to the sink. +// +// The caller is responsible for: +// - Setting SSE headers (if streaming to a client) +// - Calling providers.GetHooks().PreRequest() before entry +// - Persisting the result after return +// - Sending [DONE] or HTTP response after return (via sink.OnDone) +func CoreToolLoop(ctx context.Context, lcfg LoopConfig, sink LoopSink) LoopResult { + var result LoopResult + maxRounds := lcfg.Budget.maxRounds() + + for iteration := 0; iteration < maxRounds; iteration++ { + var iterContent string + var iterReasoning string + var toolCalls []providers.ToolCall + var iterInput, iterOutput, iterCacheCreate, iterCacheRead int + + if lcfg.Streaming { + done := runStreamingRound(ctx, lcfg, sink, &result, + &iterContent, &iterReasoning, &toolCalls, + &iterInput, &iterOutput, &iterCacheCreate, &iterCacheRead) + if done { + return result + } + } else { + done := runSyncRound(ctx, lcfg, sink, &result, + &iterContent, &iterReasoning, &toolCalls, + &iterInput, &iterOutput, &iterCacheCreate, &iterCacheRead) + if done { + return result + } + } + + // ── Tool execution round ──────────────── + if len(toolCalls) == 0 { + // Stream/call ended without tool calls or finish — edge case + if iterReasoning != "" { + result.Content += "" + iterReasoning + "" + } + result.Content += iterContent + sink.OnDone() + return result + } + + // Budget check: tool calls (pre-execution) + result.ToolCallCount += len(toolCalls) + if lcfg.Budget.MaxToolCalls > 0 && result.ToolCallCount > lcfg.Budget.MaxToolCalls { + result.BudgetExceeded = "tool_calls" + result.Content += iterContent + sink.OnFinish("budget_exceeded") + sink.OnDone() + return result + } + + sink.OnToolUse(toolCalls) + + // Append assistant message (with tool_calls, possibly with text) to conversation + assistantMsg := providers.Message{ + Role: "assistant", + Content: iterContent, + } + for _, tc := range toolCalls { + assistantMsg.ToolCalls = append(assistantMsg.ToolCalls, tc) + } + lcfg.Req.Messages = append(lcfg.Req.Messages, assistantMsg) + + // Execute all tool calls + for _, tc := range toolCalls { + call := tools.ToolCall{ + ID: tc.ID, + Name: tc.Function.Name, + Arguments: tc.Function.Arguments, + } + + toolResult := executeToolCall(ctx, lcfg, call) + + // Notify sink + resultMap := map[string]interface{}{ + "tool_call_id": toolResult.ToolCallID, + "name": toolResult.Name, + "content": toolResult.Content, + "is_error": toolResult.IsError, + } + sink.OnToolResult(resultMap) + + // Emit workspace.file.changed for live editor updates (v0.21.5) + emitWorkspaceEvent(lcfg, call, toolResult) + + // Collect for persistence + result.ToolActivity = append(result.ToolActivity, map[string]interface{}{ + "id": tc.ID, + "name": tc.Function.Name, + "arguments": tc.Function.Arguments, + "result": toolResult.Content, + "is_error": toolResult.IsError, + }) + + // Append tool result to conversation for next iteration + lcfg.Req.Messages = append(lcfg.Req.Messages, providers.Message{ + Role: "tool", + Content: toolResult.Content, + ToolCallID: toolResult.ToolCallID, + Name: toolResult.Name, + }) + } + + result.Content += iterContent + + // Budget check: tokens (post-round) + if lcfg.Budget.MaxTokens > 0 && (result.InputTokens+result.OutputTokens) > lcfg.Budget.MaxTokens { + result.BudgetExceeded = "tokens" + sink.OnFinish("budget_exceeded") + sink.OnDone() + return result + } + + // Loop: send updated messages back to provider + } + + // Hit max iterations + result.BudgetExceeded = "max_rounds" + sink.OnMaxRounds() + sink.OnDone() + return result +} + +// ── Streaming Round ──────────────────────────── + +// runStreamingRound executes one provider StreamCompletion call and +// processes the event stream. Returns true if the loop should return +// (normal finish or error), false if tool execution should follow. +func runStreamingRound( + ctx context.Context, + lcfg LoopConfig, + sink LoopSink, + result *LoopResult, + iterContent, iterReasoning *string, + toolCalls *[]providers.ToolCall, + iterInput, iterOutput, iterCacheCreate, iterCacheRead *int, +) bool { + callStart := time.Now() + ch, err := lcfg.Provider.StreamCompletion(ctx, lcfg.Cfg, *lcfg.Req) + if err != nil { + recordHealthFn(lcfg.Health, lcfg.ConfigID, callStart, err) + result.Error = err + sink.OnError(err) + return true + } + + for event := range ch { + // Apply provider-specific post-processing hooks (v0.22.1) + if hooks := providers.GetHooks(lcfg.ProviderType); hooks != nil { + hooks.PostStreamEvent(lcfg.Cfg, &event) + } + + if event.Error != nil { + recordHealthFn(lcfg.Health, lcfg.ConfigID, callStart, event.Error) + result.Error = event.Error + sink.OnError(event.Error) + return true + } + + // Reasoning deltas + if event.Reasoning != "" { + *iterReasoning += event.Reasoning + sink.OnReasoning(event.Reasoning) + } + + // Content deltas + if event.Delta != "" { + *iterContent += event.Delta + sink.OnDelta(event.Delta) + } + + if event.Done { + recordHealthFn(lcfg.Health, lcfg.ConfigID, callStart, nil) + + if event.FinishReason == "tool_calls" && len(event.ToolCalls) > 0 { + *toolCalls = event.ToolCalls + result.InputTokens += event.InputTokens + result.OutputTokens += event.OutputTokens + result.CacheCreationTokens += event.CacheCreationTokens + result.CacheReadTokens += event.CacheReadTokens + return false // continue to tool execution + } + + // Normal completion + finishReason := event.FinishReason + if finishReason == "" { + finishReason = "stop" + } + if *iterReasoning != "" { + result.Content += "" + *iterReasoning + "" + } + result.Content += *iterContent + result.InputTokens += event.InputTokens + result.OutputTokens += event.OutputTokens + result.CacheCreationTokens += event.CacheCreationTokens + result.CacheReadTokens += event.CacheReadTokens + + sink.OnFinish(finishReason) + sink.OnDone() + return true + } + } + + // Channel closed without Done event — shouldn't happen + return false +} + +// ── Sync Round ───────────────────────────────── + +// runSyncRound executes one provider ChatCompletion call. Returns true +// if the loop should return, false if tool execution should follow. +func runSyncRound( + ctx context.Context, + lcfg LoopConfig, + sink LoopSink, + result *LoopResult, + iterContent, iterReasoning *string, + toolCalls *[]providers.ToolCall, + iterInput, iterOutput, iterCacheCreate, iterCacheRead *int, +) bool { + callStart := time.Now() + resp, err := lcfg.Provider.ChatCompletion(ctx, lcfg.Cfg, *lcfg.Req) + recordHealthFn(lcfg.Health, lcfg.ConfigID, callStart, err) + if err != nil { + result.Error = err + sink.OnError(err) + return true + } + + *iterContent = resp.Content + *iterInput = resp.InputTokens + *iterOutput = resp.OutputTokens + *iterCacheCreate = resp.CacheCreationTokens + *iterCacheRead = resp.CacheReadTokens + + result.InputTokens += resp.InputTokens + result.OutputTokens += resp.OutputTokens + result.CacheCreationTokens += resp.CacheCreationTokens + result.CacheReadTokens += resp.CacheReadTokens + + if resp.FinishReason == "tool_calls" && len(resp.ToolCalls) > 0 { + *toolCalls = resp.ToolCalls + return false // continue to tool execution + } + + // Normal completion — deliver full content as single delta + result.Content += resp.Content + sink.OnDelta(resp.Content) + finishReason := resp.FinishReason + if finishReason == "" { + finishReason = "stop" + } + sink.OnFinish(finishReason) + sink.OnDone() + return true +} + +// ── Tool Execution ───────────────────────────── + +// executeToolCall dispatches a single tool call: server-side first, +// browser bridge fallback, unknown tool error as last resort. +func executeToolCall(ctx context.Context, lcfg LoopConfig, call tools.ToolCall) tools.ToolResult { + toolStart := time.Now() + + // Server-side tool + if tools.Get(call.Name) != nil { + log.Printf("🔧 Executing tool (server): %s (call %s)", call.Name, call.ID) + result := tools.ExecuteCall(ctx, lcfg.ExecCtx, call) + if lcfg.Health != nil { + toolLatency := int(time.Since(toolStart).Milliseconds()) + if result.IsError { + lcfg.Health.RecordToolError(call.Name, toolLatency, result.Content) + } else { + lcfg.Health.RecordToolSuccess(call.Name, toolLatency) + } + } + return result + } + + // Browser bridge (only available when hub is connected) + if lcfg.Hub != nil && lcfg.Hub.IsConnected(lcfg.ExecCtx.UserID) { + log.Printf("🔧 Executing tool (browser): %s (call %s)", call.Name, call.ID) + return executeBrowserTool(lcfg.Hub, lcfg.ExecCtx.UserID, call) + } + + // Unknown tool + log.Printf("⚠️ Unknown tool: %s (call %s)", call.Name, call.ID) + return tools.ToolResult{ + ToolCallID: call.ID, + Name: call.Name, + Content: `{"error":"unknown tool or browser not connected"}`, + IsError: true, + } +} + +// emitWorkspaceEvent sends workspace.file.changed when a write tool succeeds. +func emitWorkspaceEvent(lcfg LoopConfig, call tools.ToolCall, result tools.ToolResult) { + if lcfg.Hub == nil || result.IsError || lcfg.ExecCtx.WorkspaceID == "" { + return + } + if call.Name != "workspace_write" && call.Name != "workspace_patch" { + return + } + var toolArgs struct { + Path string `json:"path"` + } + if json.Unmarshal([]byte(call.Arguments), &toolArgs) == nil && toolArgs.Path != "" { + lcfg.Hub.SendToUser(lcfg.ExecCtx.UserID, events.Event{ + Label: "workspace.file.changed", + Payload: events.MustJSON(map[string]string{ + "workspace_id": lcfg.ExecCtx.WorkspaceID, + "path": toolArgs.Path, + }), + }) + } +} + +// ── Sink Implementations ─────────────────────── + +// sseSink writes SSE events to a gin.Context writer for interactive streaming. +type sseSink struct { + w http.ResponseWriter + flush func() + model string +} + +func newSSESink(c *gin.Context, model string) *sseSink { + flusher, _ := c.Writer.(http.Flusher) + return &sseSink{ + w: c.Writer, + model: model, + flush: func() { + if flusher != nil { + flusher.Flush() + } + }, + } +} + +func (s *sseSink) sendData(data string) { + fmt.Fprintf(s.w, "data: %s\n\n", data) + s.flush() +} + +func (s *sseSink) sendEvent(event, data string) { + fmt.Fprintf(s.w, "event: %s\ndata: %s\n\n", event, data) + s.flush() +} + +func (s *sseSink) OnDelta(delta string) { + s.sendData(fmt.Sprintf( + `{"choices":[{"delta":{"content":%q},"finish_reason":null}],"model":%q}`, + delta, s.model)) +} + +func (s *sseSink) OnReasoning(delta string) { + s.sendData(fmt.Sprintf( + `{"choices":[{"delta":{"reasoning_content":%q},"finish_reason":null}],"model":%q}`, + delta, s.model)) +} + +func (s *sseSink) OnError(err error) { + s.sendData(fmt.Sprintf(`{"error":"%s"}`, escapeJSON(err.Error()))) +} + +func (s *sseSink) OnToolUse(calls []providers.ToolCall) { + toolCallsJSON, _ := json.Marshal(calls) + s.sendEvent("tool_use", string(toolCallsJSON)) +} + +func (s *sseSink) OnToolResult(result map[string]interface{}) { + resultJSON, _ := json.Marshal(result) + s.sendEvent("tool_result", string(resultJSON)) +} + +func (s *sseSink) OnFinish(reason string) { + s.sendData(fmt.Sprintf( + `{"choices":[{"delta":{},"finish_reason":%q}],"model":%q}`, + reason, s.model)) +} + +func (s *sseSink) OnDone() { + s.sendData("[DONE]") +} + +func (s *sseSink) OnMaxRounds() { + log.Printf("⚠️ Tool loop hit max iterations for model %s", s.model) + s.sendData(fmt.Sprintf( + `{"choices":[{"delta":{"content":"%s"},"finish_reason":"stop"}],"model":%q}`, + escapeJSON("[Tool execution limit reached]"), s.model)) +} + +// sseModelSink extends sseSink with model_display attribution for +// multi-model streaming. Does NOT send [DONE] — the multiModelStream +// caller sends it after all models finish. +type sseModelSink struct { + sseSink + displayName string +} + +func newSSEModelSink(c *gin.Context, model, displayName string) *sseModelSink { + base := newSSESink(c, model) + return &sseModelSink{ + sseSink: *base, + displayName: displayName, + } +} + +func (s *sseModelSink) OnDelta(delta string) { + s.sendData(fmt.Sprintf( + `{"choices":[{"delta":{"content":%q},"finish_reason":null}],"model":%q,"model_display":%q}`, + delta, s.model, s.displayName)) +} + +func (s *sseModelSink) OnReasoning(delta string) { + s.sendData(fmt.Sprintf( + `{"choices":[{"delta":{"reasoning_content":%q},"finish_reason":null}],"model":%q,"model_display":%q}`, + delta, s.model, s.displayName)) +} + +func (s *sseModelSink) OnFinish(reason string) { + s.sendData(fmt.Sprintf( + `{"choices":[{"delta":{},"finish_reason":%q}],"model":%q,"model_display":%q}`, + reason, s.model, s.displayName)) +} + +// OnDone is a no-op for multi-model — caller sends [DONE] after all models. +func (s *sseModelSink) OnDone() {} + +func (s *sseModelSink) OnMaxRounds() { + log.Printf("⚠️ Multi-model tool loop hit max iterations for model %s", s.displayName) + s.sendData(fmt.Sprintf( + `{"choices":[{"delta":{"content":"%s"},"finish_reason":"stop"}],"model":%q,"model_display":%q}`, + escapeJSON("[Tool execution limit reached]"), s.model, s.displayName)) +} + +// accumSink accumulates silently — used by syncCompletion where the +// caller formats the HTTP response after the loop returns. +type accumSink struct{} + +func (accumSink) OnDelta(string) {} +func (accumSink) OnReasoning(string) {} +func (accumSink) OnError(error) {} +func (accumSink) OnToolUse([]providers.ToolCall) {} +func (accumSink) OnToolResult(map[string]interface{}) {} +func (accumSink) OnFinish(string) {} +func (accumSink) OnDone() {} +func (accumSink) OnMaxRounds() {} + +// HeadlessSink is used by the task scheduler — accumulates and logs. +// No client connection, no SSE output. +type HeadlessSink struct { + taskID string +} + +func NewHeadlessSink(taskID string) *HeadlessSink { + return &HeadlessSink{taskID: taskID} +} + +func (s *HeadlessSink) OnDelta(string) {} +func (s *HeadlessSink) OnReasoning(string) {} + +func (s *HeadlessSink) OnError(err error) { + log.Printf("[task:%s] Error: %v", s.taskID, err) +} + +func (s *HeadlessSink) OnToolUse(calls []providers.ToolCall) { + names := make([]string, len(calls)) + for i, tc := range calls { + names[i] = tc.Function.Name + } + log.Printf("[task:%s] Tool calls: %v", s.taskID, names) +} + +func (s *HeadlessSink) OnToolResult(result map[string]interface{}) { + name, _ := result["name"].(string) + isErr, _ := result["is_error"].(bool) + if isErr { + log.Printf("[task:%s] Tool %s failed: %v", s.taskID, name, result["content"]) + } +} + +func (s *HeadlessSink) OnFinish(reason string) { + log.Printf("[task:%s] Finished: %s", s.taskID, reason) +} + +func (s *HeadlessSink) OnDone() {} + +func (s *HeadlessSink) OnMaxRounds() { + log.Printf("[task:%s] Hit max tool iterations", s.taskID) +} diff --git a/server/main.go b/server/main.go index 8e00262..8bf9763 100644 --- a/server/main.go +++ b/server/main.go @@ -216,12 +216,7 @@ func main() { }() } - // v0.27.1: Task scheduler — polls for due tasks every 30s - if stores.Tasks != nil { - taskSched := scheduler.New(stores) - go taskSched.Run() - log.Println(" ⏰ Task scheduler started") - } + // v0.27.2: Task scheduler startup deferred to after hub/notification init — see below. // Bootstrap admin from env (K8s secret) — upserts on every restart handlers.BootstrapAdmin(cfg, stores) @@ -407,6 +402,14 @@ func main() { defer notifSvc.StopCleanup() } + // v0.27.2: Task scheduler with executor — needs hub + notification service + if stores.Tasks != nil { + exec := scheduler.NewExecutor(stores, keyResolver, hub, healthAccum) + taskSched := scheduler.New(stores, exec) + go taskSched.Run() + log.Println(" ⏰ Task scheduler started (with executor)") + } + // Health check (k8s probes hit this directly) base.GET("/health", func(c *gin.Context) { c.JSON(200, gin.H{ @@ -604,15 +607,16 @@ func main() { protected.POST("/workflow-assignments/:id/claim", wfAssignH.Claim) protected.POST("/workflow-assignments/:id/complete", wfAssignH.Complete) - // Tasks (v0.27.1) + // Tasks (v0.27.1, permissions v0.27.2) taskH := handlers.NewTaskHandler(stores) protected.GET("/tasks", taskH.ListMine) - protected.POST("/tasks", taskH.Create) + protected.POST("/tasks", middleware.RequirePermission(auth.PermTaskCreate, stores), taskH.Create) protected.GET("/tasks/:id", taskH.Get) - protected.PUT("/tasks/:id", taskH.Update) - protected.DELETE("/tasks/:id", taskH.Delete) + protected.PUT("/tasks/:id", middleware.RequirePermission(auth.PermTaskCreate, stores), taskH.Update) + protected.DELETE("/tasks/:id", middleware.RequirePermission(auth.PermTaskCreate, stores), taskH.Delete) protected.GET("/tasks/:id/runs", taskH.ListRuns) - protected.POST("/tasks/:id/run", taskH.RunNow) + protected.POST("/tasks/:id/run", middleware.RequirePermission(auth.PermTaskCreate, stores), taskH.RunNow) + protected.POST("/tasks/:id/kill", middleware.RequirePermission(auth.PermTaskCreate, stores), taskH.KillRun) // Channel models (v0.20.0 — multi-model @mention routing) chModelH := handlers.NewChannelModelHandler(stores) @@ -1093,9 +1097,12 @@ func main() { admin.PUT("/surfaces/:id/disable", surfaceAdm.DisableSurface) admin.DELETE("/surfaces/:id", surfaceAdm.DeleteSurface) - // Task management — admin (v0.27.1) + // Task management — admin (v0.27.1, extended v0.27.2) taskAdm := handlers.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) } } diff --git a/server/pages/templates/surfaces/admin.html b/server/pages/templates/surfaces/admin.html index ec99dd4..5994884 100644 --- a/server/pages/templates/surfaces/admin.html +++ b/server/pages/templates/surfaces/admin.html @@ -218,6 +218,7 @@ +