core/server/scheduler/executor.go

// 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
// 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/sandbox"
	"git.gobha.me/xcaliber/chat-switchboard/store"
	"git.gobha.me/xcaliber/chat-switchboard/taskutil"
	"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
	runner *sandbox.Runner // v0.29.0: Starlark task execution
}

// 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,
	}
}

// SetRunner attaches the Starlark sandbox runner for starlark task execution.
func (e *Executor) SetRunner(r *sandbox.Runner) {
	e.runner = r
}

// 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()

	// v0.28.6: System tasks run a built-in Go function. No LLM, no provider, no channel.
	if task.TaskType == "system" {
		e.executeSystem(ctx, task, run, startTime)
		return
	}

	// v0.28.0: Action tasks skip the LLM pipeline entirely.
	if task.TaskType == "action" {
		e.executeAction(ctx, task, run, channelID, startTime)
		return
	}

	// v0.29.0: Starlark tasks run a sandboxed extension script.
	if task.TaskType == "starlark" {
		e.executeStarlark(ctx, task, run, channelID, startTime)
		return
	}

	// ── 1. Resolve provider ────────────────────
	providerConfigID := ""
	if task.ProviderConfigID != nil {
		providerConfigID = *task.ProviderConfigID
	}
	res, err := handlers.ResolveProviderConfig(e.stores, e.vault, task.OwnerID, channelID, providerConfigID, task.ModelID)
	if err != nil {
		e.failRun(ctx, task, run, "provider resolution failed: "+err.Error())
		return
	}

	// v0.27.4: Enforce personal_require_byok — personal tasks must use BYOK provider
	if task.Scope == "personal" && res.ProviderScope != "personal" {
		cfg := taskutil.LoadTaskConfig(ctx, e.stores.GlobalConfig)
		if cfg.PersonalRequireBYOK {
			e.failRun(ctx, task, run, "personal tasks require a BYOK provider — add an API key in Settings → Providers")
			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 — 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: userContent,
		})
	}

	// ── 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
	}

	extTools := handlers.BuildExtToolMap(ctx, e.stores, task.OwnerID)
	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
		Runner:    e.runner,
		ExtTools:  extTools,
	}, sink)

	// ── 6. Persist output based on output_mode ──
	wallClock := int(time.Since(startTime).Seconds())

	if result.Content != "" {
		switch task.OutputMode {
		case "note":
			// v0.27.4: Save output as a note
			if e.stores.Notes != nil {
				noteTitle := task.Name + " — " + time.Now().Format("2006-01-02 15:04")
				_ = e.stores.Notes.Create(ctx, &models.Note{
					UserID:          task.OwnerID,
					Title:           noteTitle,
					Content:         result.Content,
					SourceChannelID: &channelID,
					TeamID:          task.TeamID,
					Tags:            []string{"task-output"},
				})
			}
		case "webhook":
			// Webhook delivery handled in step 10 below
		default: // "channel"
			if 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
	}

	tokensUsed := result.InputTokens + result.OutputTokens

	// ── 8. Update run record ───────────────────
	_ = e.stores.Tasks.UpdateRun(ctx, run.ID, status,
		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, tokensUsed, 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,
			RunID:       run.ID,
			TaskName:    task.Name,
			ChannelID:   channelID,
			Status:      status,
			CompletedAt: time.Now().UTC(),
			Output:      result.Content,
			TokensUsed:  tokensUsed,
			Error:       errMsg,
		})
	}
}

// executeStarlark runs a sandboxed Starlark extension script.
// The task's system_function field holds the package ID.
// The script's on_run(ctx) entry point is called with task context.
func (e *Executor) executeStarlark(ctx context.Context, task models.Task, run *models.TaskRun, channelID string, startTime time.Time) {
	if e.runner == nil {
		e.failRun(ctx, task, run, "starlark runner not configured")
		return
	}

	packageID := task.SystemFunction
	if packageID == "" {
		e.failRun(ctx, task, run, "starlark task missing package_id (system_function field)")
		return
	}

	// Load the package
	pkg, err := e.stores.Packages.Get(ctx, packageID)
	if err != nil || pkg == nil {
		e.failRun(ctx, task, run, "package not found: "+packageID)
		return
	}

	// Run the script and call on_run entry point
	// Build a context dict with task info
	rc := &sandbox.RunContext{UserID: task.OwnerID, ChannelID: channelID}
	val, output, err := e.runner.CallEntryPoint(ctx, pkg, "on_run", nil, nil, rc)

	wallClock := int(time.Since(startTime).Seconds())
	status := "completed"
	errMsg := ""
	result := ""

	if err != nil {
		status = "failed"
		errMsg = err.Error()
	} else if val != nil {
		result = val.String()
	}

	// Append print output to result
	if output != "" {
		if result != "" {
			result = result + "\n---\n" + output
		} else {
			result = output
		}
	}

	// Persist to channel if output_mode == "channel"
	if status == "completed" && task.OutputMode == "channel" && e.stores.Messages != nil && result != "" {
		_ = e.stores.Messages.Create(ctx, &models.Message{
			ChannelID: channelID,
			Role:      "system",
			Content:   "Starlark task output:\n```\n" + result + "\n```",
		})
	}

	_ = e.stores.Tasks.UpdateRun(ctx, run.ID, status, 0, 0, wallClock, errMsg)
	_ = e.stores.Tasks.IncrementRunCount(ctx, task.ID)

	log.Printf("[executor] Starlark task %s (%s → %s) → %s (wall=%ds)", task.ID, task.Name, packageID, status, wallClock)

	e.notifyOwner(ctx, task, status, errMsg)

	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,
			Error:       errMsg,
		})
	}
}

// executeSystem runs a built-in Go function from the system registry.
// No LLM, no provider resolution, no channel needed.
func (e *Executor) executeSystem(ctx context.Context, task models.Task, run *models.TaskRun, startTime time.Time) {
	fn, ok := taskutil.GetSystemFunc(task.SystemFunction)
	if !ok {
		e.failRun(ctx, task, run, "unknown system function: "+task.SystemFunction)
		return
	}

	result, err := fn(ctx, e.stores)

	wallClock := int(time.Since(startTime).Seconds())
	status := "completed"
	errMsg := ""

	if err != nil {
		status = "failed"
		errMsg = err.Error()
	}

	// Store result as the run's output (tokens=0, tools=0 for system tasks)
	_ = e.stores.Tasks.UpdateRun(ctx, run.ID, status, 0, 0, wallClock, errMsg)
	_ = e.stores.Tasks.IncrementRunCount(ctx, task.ID)

	log.Printf("[executor] System task %s (%s → %s) → %s (wall=%ds, result=%s)",
		task.ID, task.Name, task.SystemFunction, status, wallClock, truncate(result, 200))

	e.notifyOwner(ctx, task, status, errMsg)

	// Outbound webhook with result
	if task.WebhookURL != "" {
		go webhook.Deliver(task.WebhookURL, task.WebhookSecret, webhook.Payload{
			TaskID:      task.ID,
			RunID:       run.ID,
			TaskName:    task.Name,
			Status:      status,
			CompletedAt: time.Now().UTC(),
			Output:      result,
			Error:       errMsg,
		})
	}
}

func truncate(s string, n int) string {
	if len(s) <= n {
		return s
	}
	return s[:n] + "…"
}

// 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)
	_ = 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,
			RunID:       run.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
}