Changeset 0.28.6 (#192)
This commit is contained in:
@@ -51,6 +51,12 @@ func NewExecutor(stores store.Stores, vault *crypto.KeyResolver, hub *events.Hub
|
||||
func (e *Executor) Execute(ctx context.Context, task models.Task, run *models.TaskRun, channelID string) {
|
||||
startTime := time.Now()
|
||||
|
||||
// v0.28.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)
|
||||
@@ -259,6 +265,56 @@ func (e *Executor) Execute(ctx context.Context, task models.Task, run *models.Ta
|
||||
}
|
||||
}
|
||||
|
||||
// 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) {
|
||||
|
||||
130
server/scheduler/system_builtins.go
Normal file
130
server/scheduler/system_builtins.go
Normal file
@@ -0,0 +1,130 @@
|
||||
// Package scheduler — system_builtins.go
|
||||
//
|
||||
// v0.28.6: Built-in system functions registered at startup.
|
||||
// These mirror the background goroutines in main.go but run as visible,
|
||||
// scheduled, auditable tasks. The goroutines remain as fallback until
|
||||
// system tasks are validated in production.
|
||||
package scheduler
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"log"
|
||||
"time"
|
||||
|
||||
"git.gobha.me/xcaliber/chat-switchboard/database"
|
||||
"git.gobha.me/xcaliber/chat-switchboard/store"
|
||||
"git.gobha.me/xcaliber/chat-switchboard/taskutil"
|
||||
)
|
||||
|
||||
// RegisterBuiltins registers all built-in system functions.
|
||||
// Called once from main.go at startup.
|
||||
func RegisterBuiltins() {
|
||||
taskutil.RegisterSystemFunc("session_cleanup",
|
||||
"Remove expired anonymous visitor sessions",
|
||||
sessionCleanup)
|
||||
|
||||
taskutil.RegisterSystemFunc("staleness_check",
|
||||
"Mark idle workflow instances as stale",
|
||||
stalenessCheck)
|
||||
|
||||
taskutil.RegisterSystemFunc("retention_sweep",
|
||||
"Delete completed workflow instances past retention policy",
|
||||
retentionSweep)
|
||||
|
||||
taskutil.RegisterSystemFunc("health_prune",
|
||||
"Prune provider health windows older than 7 days",
|
||||
healthPrune)
|
||||
}
|
||||
|
||||
// ── session_cleanup ─────────────────────────
|
||||
|
||||
func sessionCleanup(ctx context.Context, stores store.Stores) (string, error) {
|
||||
if stores.Sessions == nil {
|
||||
return "skipped: session store not available", nil
|
||||
}
|
||||
// Default: expire sessions older than 7 days
|
||||
cutoff := time.Now().UTC().AddDate(0, 0, -7)
|
||||
n, err := stores.Sessions.DeleteExpired(ctx, cutoff)
|
||||
if err != nil {
|
||||
return "", fmt.Errorf("session cleanup failed: %w", err)
|
||||
}
|
||||
msg := fmt.Sprintf("cleaned up %d expired sessions (cutoff: %s)", n, cutoff.Format(time.RFC3339))
|
||||
if n > 0 {
|
||||
log.Printf("[system_task] %s", msg)
|
||||
}
|
||||
return msg, nil
|
||||
}
|
||||
|
||||
// ── staleness_check ─────────────────────────
|
||||
|
||||
func stalenessCheck(ctx context.Context, stores store.Stores) (string, error) {
|
||||
// Default: 48 hours idle → stale
|
||||
cutoff := time.Now().UTC().Add(-48 * time.Hour)
|
||||
res, err := database.DB.ExecContext(ctx, database.Q(`
|
||||
UPDATE channels
|
||||
SET workflow_status = 'stale'
|
||||
WHERE type = 'workflow'
|
||||
AND workflow_status = 'active'
|
||||
AND last_activity_at < $1
|
||||
`), cutoff)
|
||||
if err != nil {
|
||||
return "", fmt.Errorf("staleness sweep failed: %w", err)
|
||||
}
|
||||
n, _ := res.RowsAffected()
|
||||
msg := fmt.Sprintf("marked %d idle workflow instances as stale (cutoff: %s)", n, cutoff.Format(time.RFC3339))
|
||||
if n > 0 {
|
||||
log.Printf("[system_task] %s", msg)
|
||||
}
|
||||
return msg, nil
|
||||
}
|
||||
|
||||
// ── retention_sweep ─────────────────────────
|
||||
|
||||
func retentionSweep(ctx context.Context, stores store.Stores) (string, error) {
|
||||
if database.IsSQLite() {
|
||||
return "skipped: retention enforcement requires PostgreSQL (JSON operators)", nil
|
||||
}
|
||||
cutoff := time.Now().UTC().Add(-48 * time.Hour)
|
||||
res, err := database.DB.ExecContext(ctx, database.Q(`
|
||||
DELETE FROM channels
|
||||
WHERE type = 'workflow'
|
||||
AND workflow_status IN ('completed', 'archived')
|
||||
AND workflow_id IS NOT NULL
|
||||
AND last_activity_at < $1
|
||||
AND workflow_id IN (
|
||||
SELECT id FROM workflows
|
||||
WHERE retention IS NOT NULL
|
||||
AND retention->>'mode' = 'delete'
|
||||
AND (retention->>'delete_after_days')::int > 0
|
||||
AND channels.last_activity_at < now() - ((retention->>'delete_after_days')::int || ' days')::interval
|
||||
)
|
||||
`), cutoff)
|
||||
if err != nil {
|
||||
return "", fmt.Errorf("retention enforcement failed: %w", err)
|
||||
}
|
||||
n, _ := res.RowsAffected()
|
||||
msg := fmt.Sprintf("deleted %d expired workflow instances (retention policy)", n)
|
||||
if n > 0 {
|
||||
log.Printf("[system_task] %s", msg)
|
||||
}
|
||||
return msg, nil
|
||||
}
|
||||
|
||||
// ── health_prune ────────────────────────────
|
||||
|
||||
func healthPrune(ctx context.Context, stores store.Stores) (string, error) {
|
||||
cutoff := time.Now().UTC().Add(-7 * 24 * time.Hour)
|
||||
res, err := database.DB.ExecContext(ctx, database.Q(`
|
||||
DELETE FROM health_windows WHERE window_start < $1
|
||||
`), cutoff)
|
||||
if err != nil {
|
||||
return "", fmt.Errorf("health prune failed: %w", err)
|
||||
}
|
||||
n, _ := res.RowsAffected()
|
||||
msg := fmt.Sprintf("pruned %d old health windows (cutoff: %s)", n, cutoff.Format(time.RFC3339))
|
||||
if n > 0 {
|
||||
log.Printf("[system_task] %s", msg)
|
||||
}
|
||||
return msg, nil
|
||||
}
|
||||
Reference in New Issue
Block a user