295 lines
8.5 KiB
Go
295 lines
8.5 KiB
Go
// Package scheduler — executor.go
|
|
//
|
|
// v0.27.2: Headless task execution via coreToolLoop.
|
|
//
|
|
// The Executor bridges the task scheduler with the completion pipeline.
|
|
// It resolves providers, builds tool definitions, runs the core tool loop
|
|
// with budget enforcement, persists results, and sends notifications.
|
|
package scheduler
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"log"
|
|
"time"
|
|
|
|
capspkg "git.gobha.me/xcaliber/chat-switchboard/capabilities"
|
|
"git.gobha.me/xcaliber/chat-switchboard/crypto"
|
|
"git.gobha.me/xcaliber/chat-switchboard/events"
|
|
"git.gobha.me/xcaliber/chat-switchboard/handlers"
|
|
"git.gobha.me/xcaliber/chat-switchboard/models"
|
|
"git.gobha.me/xcaliber/chat-switchboard/notifications"
|
|
"git.gobha.me/xcaliber/chat-switchboard/providers"
|
|
"git.gobha.me/xcaliber/chat-switchboard/store"
|
|
"git.gobha.me/xcaliber/chat-switchboard/tools"
|
|
"git.gobha.me/xcaliber/chat-switchboard/webhook"
|
|
)
|
|
|
|
// Executor runs task completions headlessly (no HTTP client).
|
|
type Executor struct {
|
|
stores store.Stores
|
|
vault *crypto.KeyResolver
|
|
hub *events.Hub
|
|
health handlers.HealthRecorder
|
|
}
|
|
|
|
// NewExecutor creates a task executor. All fields are optional except stores.
|
|
func NewExecutor(stores store.Stores, vault *crypto.KeyResolver, hub *events.Hub, health handlers.HealthRecorder) *Executor {
|
|
return &Executor{
|
|
stores: stores,
|
|
vault: vault,
|
|
hub: hub,
|
|
health: health,
|
|
}
|
|
}
|
|
|
|
// Execute runs a single task to completion.
|
|
// Called from the scheduler's execute() goroutine.
|
|
func (e *Executor) Execute(ctx context.Context, task models.Task, run *models.TaskRun, channelID string) {
|
|
startTime := time.Now()
|
|
|
|
// ── 1. Resolve provider ────────────────────
|
|
providerConfigID := ""
|
|
if task.ProviderConfigID != nil {
|
|
providerConfigID = *task.ProviderConfigID
|
|
}
|
|
res, err := handlers.ResolveProviderConfig(e.vault, task.OwnerID, channelID, providerConfigID, task.ModelID)
|
|
if err != nil {
|
|
e.failRun(ctx, task, run, "provider resolution failed: "+err.Error())
|
|
return
|
|
}
|
|
|
|
provider, err := providers.Get(res.ProviderID)
|
|
if err != nil {
|
|
e.failRun(ctx, task, run, "provider unavailable: "+err.Error())
|
|
return
|
|
}
|
|
|
|
// ── 2. Build messages ──────────────────────
|
|
messages := make([]providers.Message, 0, 3)
|
|
|
|
// System prompt: task-level > persona > empty
|
|
systemPrompt := task.SystemPrompt
|
|
if systemPrompt == "" && task.PersonaID != nil && e.stores.Personas != nil {
|
|
if persona, err := e.stores.Personas.GetByID(ctx, *task.PersonaID); err == nil {
|
|
systemPrompt = persona.SystemPrompt
|
|
}
|
|
}
|
|
if systemPrompt != "" {
|
|
messages = append(messages, providers.Message{
|
|
Role: "system",
|
|
Content: systemPrompt,
|
|
})
|
|
}
|
|
|
|
// User prompt
|
|
if task.UserPrompt != "" {
|
|
messages = append(messages, providers.Message{
|
|
Role: "user",
|
|
Content: task.UserPrompt,
|
|
})
|
|
}
|
|
|
|
// ── 3. Build tool definitions ──────────────
|
|
caps := capspkg.InferCapabilities(res.Model)
|
|
caps.MaxOutputTokens = capspkg.ResolveMaxOutput(res.Model, caps)
|
|
|
|
var toolDefs []providers.ToolDef
|
|
if caps.ToolCalling {
|
|
personaID := ""
|
|
if task.PersonaID != nil {
|
|
personaID = *task.PersonaID
|
|
}
|
|
tctx := tools.ToolContext{
|
|
ChannelType: "service",
|
|
PersonaID: personaID,
|
|
}
|
|
// No browser tools for headless execution
|
|
toolDefs = handlers.BuildToolDefs(ctx, e.stores, task.OwnerID, false, nil, tctx, personaID)
|
|
|
|
// Apply task-level tool grants
|
|
if len(task.ToolGrants) > 0 {
|
|
var grants []string
|
|
if json.Unmarshal(task.ToolGrants, &grants) == nil && len(grants) > 0 {
|
|
toolDefs = handlers.FilterToolDefsByGrants(toolDefs, grants)
|
|
}
|
|
}
|
|
}
|
|
|
|
// ── 4. Build completion request ────────────
|
|
req := providers.CompletionRequest{
|
|
Model: res.Model,
|
|
Messages: messages,
|
|
Tools: toolDefs,
|
|
}
|
|
if task.MaxTokens > 0 {
|
|
req.MaxTokens = task.MaxTokens
|
|
} else if caps.MaxOutputTokens > 0 {
|
|
req.MaxTokens = caps.MaxOutputTokens
|
|
}
|
|
|
|
// Apply provider-specific request hooks
|
|
if hooks := providers.GetHooks(res.ProviderID); hooks != nil {
|
|
hooks.PreRequest(res.Config, &req)
|
|
}
|
|
|
|
// ── 5. Execute via core tool loop ──────────
|
|
personaID := ""
|
|
if task.PersonaID != nil {
|
|
personaID = *task.PersonaID
|
|
}
|
|
|
|
sink := handlers.NewHeadlessSink(task.ID)
|
|
result := handlers.CoreToolLoop(ctx, handlers.LoopConfig{
|
|
Provider: provider,
|
|
Cfg: res.Config,
|
|
Req: &req,
|
|
Model: res.Model,
|
|
ProviderType: res.ProviderID,
|
|
ExecCtx: tools.ExecutionContext{
|
|
UserID: task.OwnerID,
|
|
ChannelID: channelID,
|
|
PersonaID: personaID,
|
|
},
|
|
Hub: e.hub,
|
|
Health: e.health,
|
|
ConfigID: res.ConfigID,
|
|
Budget: handlers.LoopBudget{
|
|
MaxRounds: 0, // use default
|
|
MaxToolCalls: task.MaxToolCalls,
|
|
MaxTokens: task.MaxTokens,
|
|
},
|
|
Streaming: false, // headless — use ChatCompletion
|
|
}, sink)
|
|
|
|
// ── 6. Persist assistant response ──────────
|
|
wallClock := int(time.Since(startTime).Seconds())
|
|
|
|
if result.Content != "" && e.stores.Messages != nil {
|
|
_ = e.stores.Messages.Create(ctx, &models.Message{
|
|
ChannelID: channelID,
|
|
Role: "assistant",
|
|
Content: result.Content,
|
|
Model: res.Model,
|
|
})
|
|
}
|
|
|
|
// ── 7. Determine terminal status ───────────
|
|
status := "completed"
|
|
errMsg := ""
|
|
|
|
if result.Error != nil {
|
|
status = "failed"
|
|
errMsg = result.Error.Error()
|
|
} else if result.BudgetExceeded != "" {
|
|
status = "budget_exceeded"
|
|
errMsg = "budget exceeded: " + result.BudgetExceeded
|
|
}
|
|
|
|
// ── 8. Update run record ───────────────────
|
|
_ = e.stores.Tasks.UpdateRun(ctx, run.ID, status,
|
|
result.InputTokens+result.OutputTokens,
|
|
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)
|
|
|
|
// ── 9. Owner notification ──────────────────
|
|
e.notifyOwner(ctx, task, status, errMsg)
|
|
|
|
// ── 10. Webhook delivery (v0.27.3) ─────────
|
|
if task.WebhookURL != "" {
|
|
go webhook.Deliver(task.WebhookURL, task.WebhookSecret, webhook.Payload{
|
|
TaskID: task.ID,
|
|
TaskName: task.Name,
|
|
ChannelID: channelID,
|
|
Status: status,
|
|
CompletedAt: time.Now().UTC(),
|
|
Output: result.Content,
|
|
Error: errMsg,
|
|
})
|
|
}
|
|
}
|
|
|
|
// 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)
|
|
_ = e.stores.Tasks.UpdateRun(ctx, run.ID, "failed", 0, 0, 0, errMsg)
|
|
e.notifyOwner(ctx, task, "failed", errMsg)
|
|
if task.WebhookURL != "" {
|
|
go webhook.Deliver(task.WebhookURL, task.WebhookSecret, webhook.Payload{
|
|
TaskID: task.ID,
|
|
TaskName: task.Name,
|
|
Status: "failed",
|
|
CompletedAt: time.Now().UTC(),
|
|
Error: errMsg,
|
|
})
|
|
}
|
|
}
|
|
|
|
// notifyOwner sends a notification based on task outcome and preferences.
|
|
func (e *Executor) notifyOwner(ctx context.Context, task models.Task, status, errMsg string) {
|
|
notifSvc := notifications.Default()
|
|
if notifSvc == nil {
|
|
return
|
|
}
|
|
|
|
switch status {
|
|
case "completed":
|
|
if !task.NotifyOnComplete {
|
|
return
|
|
}
|
|
_ = notifSvc.Notify(ctx, &models.Notification{
|
|
UserID: task.OwnerID,
|
|
Type: "task.completed",
|
|
Title: "Task completed: " + task.Name,
|
|
Body: "Scheduled task finished successfully.",
|
|
ResourceType: "channel",
|
|
ResourceID: stringVal(task.OutputChannelID),
|
|
})
|
|
|
|
case "failed":
|
|
if !task.NotifyOnFailure {
|
|
return
|
|
}
|
|
body := "Scheduled task failed."
|
|
if errMsg != "" {
|
|
body = errMsg
|
|
if len(body) > 200 {
|
|
body = body[:200] + "…"
|
|
}
|
|
}
|
|
_ = notifSvc.Notify(ctx, &models.Notification{
|
|
UserID: task.OwnerID,
|
|
Type: "task.failed",
|
|
Title: "Task failed: " + task.Name,
|
|
Body: body,
|
|
ResourceType: "channel",
|
|
ResourceID: stringVal(task.OutputChannelID),
|
|
})
|
|
|
|
case "budget_exceeded":
|
|
// Always notify on budget breach regardless of preference
|
|
_ = notifSvc.Notify(ctx, &models.Notification{
|
|
UserID: task.OwnerID,
|
|
Type: "task.budget_exceeded",
|
|
Title: "Task budget exceeded: " + task.Name,
|
|
Body: errMsg,
|
|
ResourceType: "channel",
|
|
ResourceID: stringVal(task.OutputChannelID),
|
|
})
|
|
}
|
|
}
|
|
|
|
func stringVal(s *string) string {
|
|
if s == nil {
|
|
return ""
|
|
}
|
|
return *s
|
|
}
|