Run scheduled tasks through workflows

internal/store/migrateschedules.go

@@ -0,0 +1,55 @@
+package store
+
+import (
+	"context"
+	"database/sql"
+	"fmt"
+
+	"g.fc.run/ai/byre/internal/api"
+)
+
+func migrateScheduleWorkflows(tx *sql.Tx, _ string) error {
+	rows, err := tx.Query(`SELECT id, title, model, prompt FROM schedules WHERE workflow = ''`)
+	if err != nil {
+		return err
+	}
+	type row struct{ id, title, model, prompt string }
+	all := []row{}
+	for rows.Next() {
+		r := row{}
+		err = rows.Scan(&r.id, &r.title, &r.model, &r.prompt)
+		if err != nil {
+			rows.Close()
+			return err
+		}
+		all = append(all, r)
+	}
+	rows.Close()
+	err = rows.Err()
+	if err != nil {
+		return err
+	}
+
+	ctx := context.Background()
+	for _, r := range all {
+		def := api.WorkflowDefinition{
+			Description: r.title,
+			Steps:       []api.WorkflowStep{{Stage: r.title, Model: r.model, Prompt: r.prompt}},
+		}
+		id, err := insertDefinition(ctx, tx, def)
+		if err != nil {
+			return fmt.Errorf("schedule %s: %w", r.id, err)
+		}
+		_, err = tx.Exec(`INSERT INTO workflows (name, definition_id) VALUES (?, ?)`, r.title, id)
+		if err != nil {
+			return fmt.Errorf("schedule %s: workflow %q: %w", r.id, r.title, err)
+		}
+		_, err = tx.Exec(`UPDATE schedules SET workflow = ?, prompt = '' WHERE id = ?`, r.title, r.id)
+		if err != nil {
+			return err
+		}
+	}
+
+	_, err = tx.Exec(`ALTER TABLE schedules DROP COLUMN model`)
+	return err
+}

internal/store/schedule.go

@@ -13,7 +13,7 @@ import (
 	"g.fc.run/ai/byre/internal/schedule"
 )
 
-const scheduleColumns = `id, project_id, owner, title, every, hour, minute, day_of_month, prompt, model, workflow, next_run_at`
+const scheduleColumns = `id, project_id, owner, title, every, hour, minute, day_of_month, prompt, workflow, next_run_at`
 
 func querySchedules(ctx context.Context, q querier, where string, args ...any) ([]api.Schedule, error) {
 	rows, err := q.QueryContext(ctx, `SELECT `+scheduleColumns+` FROM schedules `+where+` ORDER BY next_run_at, id`, args...)
@@ -24,7 +24,7 @@ func querySchedules(ctx context.Context, q querier, where string, args ...any) (
 	for rows.Next() {
 		sc := api.Schedule{Weekdays: []api.Weekday{}}
 		next := ""
-		err = rows.Scan(&sc.Id, &sc.ProjectId, &sc.Owner, &sc.Title, &sc.Every, &sc.Hour, &sc.Minute, &sc.DayOfMonth, &sc.Prompt, &sc.Model, &sc.Workflow, &next)
+		err = rows.Scan(&sc.Id, &sc.ProjectId, &sc.Owner, &sc.Title, &sc.Every, &sc.Hour, &sc.Minute, &sc.DayOfMonth, &sc.Prompt, &sc.Workflow, &next)
 		if err != nil {
 			rows.Close()
 			return nil, err
@@ -95,14 +95,14 @@ func (s *Store) SaveSchedule(ctx context.Context, sc api.Schedule, actor string)
 	err := s.write(ctx, func(tx *sql.Tx, emit emitFunc) error {
 		if sc.Id == "" {
 			sc.Id = ids.New(ids.Schedule)
-			_, err := tx.ExecContext(ctx, `INSERT INTO schedules (`+scheduleColumns+`, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
-				sc.Id, sc.ProjectId, sc.Owner, sc.Title, string(sc.Every), sc.Hour, sc.Minute, sc.DayOfMonth, sc.Prompt, sc.Model, sc.Workflow, formatTime(sc.NextRun), now())
+			_, err := tx.ExecContext(ctx, `INSERT INTO schedules (`+scheduleColumns+`, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
+				sc.Id, sc.ProjectId, sc.Owner, sc.Title, string(sc.Every), sc.Hour, sc.Minute, sc.DayOfMonth, sc.Prompt, sc.Workflow, formatTime(sc.NextRun), now())
 			if err != nil {
 				return err
 			}
 		} else {
-			_, err := tx.ExecContext(ctx, `UPDATE schedules SET title = ?, every = ?, hour = ?, minute = ?, day_of_month = ?, prompt = ?, model = ?, workflow = ?, next_run_at = ? WHERE id = ?`,
-				sc.Title, string(sc.Every), sc.Hour, sc.Minute, sc.DayOfMonth, sc.Prompt, sc.Model, sc.Workflow, formatTime(sc.NextRun), sc.Id)
+			_, err := tx.ExecContext(ctx, `UPDATE schedules SET title = ?, every = ?, hour = ?, minute = ?, day_of_month = ?, prompt = ?, workflow = ?, next_run_at = ? WHERE id = ?`,
+				sc.Title, string(sc.Every), sc.Hour, sc.Minute, sc.DayOfMonth, sc.Prompt, sc.Workflow, formatTime(sc.NextRun), sc.Id)
 			if err != nil {
 				return err
 			}

internal/store/store.go

@@ -35,6 +35,7 @@ var goMigrations = map[string]func(tx *sql.Tx, dataDir string) error{
 	"078-normalize-workflow-definitions": migrateWorkflowDefinitions,
 	"085-prefix-ids":                     migratePrefixIDs,
 	"088-workflow-run-tasks":             migrateRunTasks,
+	"100-schedule-workflows":             migrateScheduleWorkflows,
 }
 
 func loadMigrations() ([]migration, error) {