Store workflow definitions as table rows and drop YAML

2026-10-20 21:30:39 -07:00
parent 5823af6ff9
commit 32ebf08b66
17 changed files with 624 additions and 454 deletions
@@ -0,0 +1,134 @@
+package store
+
+import (
+	"context"
+	"database/sql"
+	"errors"
+	"fmt"
+
+	"g.fc.run/ai/byre/internal/api"
+)
+
+func insertDefinition(ctx context.Context, tx *sql.Tx, def api.WorkflowDefinition) (string, error) {
+	id := NewID()
+	_, err := tx.ExecContext(ctx, `INSERT INTO workflow_definitions (id, description) VALUES (?, ?)`, id, def.Description)
+	if err != nil {
+		return "", err
+	}
+	next := 0
+	var insert func(steps []api.WorkflowStep, parent *int) error
+	insert = func(steps []api.WorkflowStep, parent *int) error {
+		for pos, s := range steps {
+			stepID := next
+			next++
+			kind, children := "stage", (*[]api.WorkflowStep)(nil)
+			switch {
+			case s.Order != nil:
+				kind, children = "order", s.Order
+			case s.Parallel != nil:
+				kind, children = "parallel", s.Parallel
+			}
+			_, err := tx.ExecContext(ctx, `INSERT INTO workflow_steps (definition_id, id, parent, position, kind, stage, model, prompt) VALUES (?, ?, ?, ?, ?, ?, ?, ?)`,
+				id, stepID, parent, pos, kind, s.Stage, s.Model, s.Prompt)
+			if err != nil {
+				return err
+			}
+			if s.Agents != nil {
+				for i, a := range *s.Agents {
+					_, err = tx.ExecContext(ctx, `INSERT INTO workflow_agents (definition_id, step_id, position, name, model, prompt) VALUES (?, ?, ?, ?, ?, ?)`,
+						id, stepID, i, a.Name, a.Model, a.Prompt)
+					if err != nil {
+						return err
+					}
+				}
+			}
+			if children != nil {
+				err = insert(*children, &stepID)
+				if err != nil {
+					return err
+				}
+			}
+		}
+		return nil
+	}
+	return id, insert(def.Steps, nil)
+}
+
+type stepRow struct {
+	id   int
+	kind string
+	step api.WorkflowStep
+}
+
+func loadDefinition(ctx context.Context, q querier, id string) (api.WorkflowDefinition, error) {
+	def := api.WorkflowDefinition{}
+	err := q.QueryRowContext(ctx, `SELECT description FROM workflow_definitions WHERE id = ?`, id).Scan(&def.Description)
+	if errors.Is(err, sql.ErrNoRows) {
+		return def, fmt.Errorf("%w: no workflow definition %s", ErrNotFound, id)
+	}
+	if err != nil {
+		return def, err
+	}
+
+	agents := map[int][]api.TeamMember{}
+	rows, err := q.QueryContext(ctx, `SELECT step_id, name, model, prompt FROM workflow_agents WHERE definition_id = ? ORDER BY step_id, position`, id)
+	if err != nil {
+		return def, err
+	}
+	for rows.Next() {
+		step := 0
+		a := api.TeamMember{}
+		err = rows.Scan(&step, &a.Name, &a.Model, &a.Prompt)
+		if err != nil {
+			rows.Close()
+			return def, err
+		}
+		agents[step] = append(agents[step], a)
+	}
+	err = errors.Join(rows.Err(), rows.Close())
+	if err != nil {
+		return def, err
+	}
+
+	children := map[int][]stepRow{}
+	rows, err = q.QueryContext(ctx, `SELECT id, coalesce(parent, -1), kind, stage, model, prompt FROM workflow_steps WHERE definition_id = ? ORDER BY coalesce(parent, -1), position`, id)
+	if err != nil {
+		return def, err
+	}
+	for rows.Next() {
+		r := stepRow{}
+		parent := 0
+		err = rows.Scan(&r.id, &parent, &r.kind, &r.step.Stage, &r.step.Model, &r.step.Prompt)
+		if err != nil {
+			rows.Close()
+			return def, err
+		}
+		if a, found := agents[r.id]; found {
+			r.step.Agents = &a
+		}
+		children[parent] = append(children[parent], r)
+	}
+	err = errors.Join(rows.Err(), rows.Close())
+	if err != nil {
+		return def, err
+	}
+
+	var build func(parent int) []api.WorkflowStep
+	build = func(parent int) []api.WorkflowStep {
+		steps := []api.WorkflowStep{}
+		for _, r := range children[parent] {
+			switch r.kind {
+			case "order":
+				sub := build(r.id)
+				r.step.Order = &sub
+			case "parallel":
+				sub := build(r.id)
+				r.step.Parallel = &sub
+			}
+			steps = append(steps, r.step)
+		}
+		return steps
+	}
+	def.Steps = build(-1)
+	return def, nil
+}
@@ -0,0 +1,135 @@
+package store
+
+import (
+	"bytes"
+	"context"
+	"database/sql"
+	"encoding/json"
+	"fmt"
+
+	"g.fc.run/ai/byre/internal/api"
+)
+
+func migrateWorkflowDefinitions(tx *sql.Tx) error {
+	_, err := tx.Exec(`
+		CREATE TABLE workflow_definitions (
+			id TEXT PRIMARY KEY,
+			description TEXT NOT NULL
+		);
+		CREATE TABLE workflow_steps (
+			definition_id TEXT NOT NULL REFERENCES workflow_definitions(id),
+			id INTEGER NOT NULL,
+			parent INTEGER,
+			position INTEGER NOT NULL,
+			kind TEXT NOT NULL CHECK (kind IN ('stage', 'order', 'parallel')),
+			stage TEXT NOT NULL,
+			model TEXT NOT NULL,
+			prompt TEXT NOT NULL,
+			PRIMARY KEY (definition_id, id),
+			UNIQUE (definition_id, parent, position)
+		);
+		CREATE TABLE workflow_agents (
+			definition_id TEXT NOT NULL REFERENCES workflow_definitions(id),
+			step_id INTEGER NOT NULL,
+			position INTEGER NOT NULL,
+			name TEXT NOT NULL,
+			model TEXT NOT NULL,
+			prompt TEXT NOT NULL,
+			PRIMARY KEY (definition_id, step_id, position),
+			FOREIGN KEY (definition_id, step_id) REFERENCES workflow_steps(definition_id, id)
+		);
+		ALTER TABLE tasks ADD COLUMN workflow_definition_id TEXT REFERENCES workflow_definitions(id);
+		ALTER TABLE reviews ADD COLUMN definition_id TEXT REFERENCES workflow_definitions(id);
+		ALTER TABLE workflows ADD COLUMN definition_id TEXT REFERENCES workflow_definitions(id);
+	`)
+	if err != nil {
+		return err
+	}
+
+	ids := map[string]string{}
+	for _, c := range []struct{ table, key, from, to string }{
+		{"workflows", "name", "definition", "definition_id"},
+		{"tasks", "id", "workflow_definition", "workflow_definition_id"},
+		{"reviews", "id", "definition", "definition_id"},
+	} {
+		err = moveDefinitions(tx, ids, c.table, c.key, c.from, c.to)
+		if err != nil {
+			return err
+		}
+	}
+
+	_, err = tx.Exec(`
+		CREATE TABLE workflows_next (
+			name TEXT PRIMARY KEY,
+			definition_id TEXT NOT NULL REFERENCES workflow_definitions(id)
+		);
+		INSERT INTO workflows_next (name, definition_id) SELECT name, definition_id FROM workflows;
+		DROP TABLE workflows;
+		ALTER TABLE workflows_next RENAME TO workflows;
+		ALTER TABLE tasks DROP COLUMN workflow_definition;
+		ALTER TABLE reviews DROP COLUMN definition;
+	`)
+	return err
+}
+
+func moveDefinitions(tx *sql.Tx, ids map[string]string, table, key, from, to string) error {
+	rows, err := tx.Query(fmt.Sprintf(`SELECT %s, %s FROM %s WHERE %s != ''`, key, from, table, from))
+	if err != nil {
+		return err
+	}
+	type row struct{ key, def string }
+	all := []row{}
+	for rows.Next() {
+		r := row{}
+		err = rows.Scan(&r.key, &r.def)
+		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 {
+		id, found := ids[r.def]
+		if !found {
+			def := api.WorkflowDefinition{}
+			dec := json.NewDecoder(bytes.NewReader([]byte(r.def)))
+			dec.DisallowUnknownFields()
+			err = dec.Decode(&def)
+			if err != nil {
+				return fmt.Errorf("%s %s: %w", table, r.key, err)
+			}
+			id, err = insertDefinition(ctx, tx, def)
+			if err != nil {
+				return fmt.Errorf("%s %s: %w", table, r.key, err)
+			}
+			back, err := loadDefinition(ctx, tx, id)
+			if err != nil {
+				return fmt.Errorf("%s %s: %w", table, r.key, err)
+			}
+			want, err := json.Marshal(def)
+			if err != nil {
+				return err
+			}
+			got, err := json.Marshal(back)
+			if err != nil {
+				return err
+			}
+			if !bytes.Equal(want, got) {
+				return fmt.Errorf("%s %s: definition reads back as %s, not %s", table, r.key, got, want)
+			}
+			ids[r.def] = id
+		}
+		_, err = tx.Exec(fmt.Sprintf(`UPDATE %s SET %s = ? WHERE %s = ?`, table, to, key), id, r.key)
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}
@@ -12,24 +12,16 @@ import (

 const reviewColumns = `id, task_id, snapshot_n, stage, coalesce(agent_session_id, ''), vendor, model, state, result, error, created_at, finished_at, finding_states, finding_responses`

-func (s *Store) CreateReview(ctx context.Context, r api.Review, def *api.WorkflowDefinition) error {
+func (s *Store) CreateReview(ctx context.Context, r api.Review, definitionID string) error {
 	finishedAt := ""
 	if r.FinishedAt != nil {
 		finishedAt = formatTime(*r.FinishedAt)
 	}
-	defText := ""
-	if def != nil {
-		b, err := json.Marshal(def)
-		if err != nil {
-			return err
-		}
-		defText = string(b)
-	}
 	return s.write(ctx, func(tx *sql.Tx, emit emitFunc) error {
 		_, err := tx.ExecContext(ctx, `
-			INSERT INTO reviews (id, task_id, snapshot_n, stage, definition, agent_session_id, vendor, model, state, error, created_at, finished_at)
-			VALUES (?, ?, ?, ?, ?, nullif(?, ''), ?, ?, ?, ?, ?, ?)
-		`, r.Id, r.TaskId, r.SnapshotN, r.Stage, defText, r.AgentSessionId, r.Vendor, r.Model, string(r.State), r.Error, formatTime(r.CreatedAt), finishedAt)
+			INSERT INTO reviews (id, task_id, snapshot_n, stage, definition_id, agent_session_id, vendor, model, state, error, created_at, finished_at)
+			VALUES (?, ?, ?, ?, nullif(?, ''), nullif(?, ''), ?, ?, ?, ?, ?, ?)
+		`, r.Id, r.TaskId, r.SnapshotN, r.Stage, definitionID, r.AgentSessionId, r.Vendor, r.Model, string(r.State), r.Error, formatTime(r.CreatedAt), finishedAt)
 		if err != nil {
 			return err
 		}
@@ -40,21 +32,17 @@ func (s *Store) CreateReview(ctx context.Context, r api.Review, def *api.Workflo
 	})
 }

-func (s *Store) ReviewDefinition(ctx context.Context, id string) (api.WorkflowDefinition, error) {
-	text := ""
-	err := s.db.QueryRowContext(ctx, `SELECT definition FROM reviews WHERE id = ?`, id).Scan(&text)
+func (s *Store) ReviewDefinition(ctx context.Context, id string) (string, api.WorkflowDefinition, error) {
+	defID := sql.NullString{}
+	err := s.db.QueryRowContext(ctx, `SELECT definition_id FROM reviews WHERE id = ?`, id).Scan(&defID)
 	if errors.Is(err, sql.ErrNoRows) {
-		return api.WorkflowDefinition{}, ErrNotFound
+		return "", api.WorkflowDefinition{}, ErrNotFound
 	}
-	if err != nil {
-		return api.WorkflowDefinition{}, err
+	if err != nil || !defID.Valid {
+		return "", api.WorkflowDefinition{}, err
 	}
-	def := api.WorkflowDefinition{}
-	if text == "" {
-		return def, nil
-	}
-	err = json.Unmarshal([]byte(text), &def)
-	return def, err
+	def, err := loadDefinition(ctx, s.db, defID.String)
+	return defID.String, def, err
 }

 func (s *Store) GetReview(ctx context.Context, id string) (api.Review, error) {
@@ -92,7 +80,7 @@ func (s *Store) RunningReviews(ctx context.Context) ([]api.Review, error) {
 func (s *Store) LatestDoneStages(ctx context.Context) ([]string, error) {
 	rows, err := s.db.QueryContext(ctx, `
 		SELECT r.id FROM reviews r JOIN tasks t ON t.id = r.task_id
-		WHERE t.status = ? AND r.state = ? AND r.definition != ''
+		WHERE t.status = ? AND r.state = ? AND r.definition_id IS NOT NULL
 		AND r.snapshot_n = (SELECT max(snapshot_n) FROM reviews WHERE task_id = r.task_id)
 		ORDER BY r.created_at`, string(api.TaskStatusActive), string(api.ReviewStateDone))
 	if err != nil {
@@ -33,7 +33,8 @@ type migration struct {
 }

 var goMigrations = map[string]func(*sql.Tx) error{
-	"071-workflow-steps": migrateWorkflowSteps,
+	"071-workflow-steps":                 migrateWorkflowSteps,
+	"078-normalize-workflow-definitions": migrateWorkflowDefinitions,
 }

 func loadMigrations() ([]migration, error) {
@@ -337,18 +338,14 @@ func (s *Store) GetProject(ctx context.Context, id string) (api.Project, error)

 func (s *Store) CreateTask(ctx context.Context, t api.Task, frozen *FrozenWorkflow, actor string) error {
 	workflowName := ""
-	workflowDefinition := ""
+	var workflowDefinitionID *string
 	workflowBrief := ""
 	workflowTitle := ""
 	workflowRun := ""
 	workflowStage := ""
 	if frozen != nil {
-		b, err := json.Marshal(frozen.Definition)
-		if err != nil {
-			return err
-		}
 		workflowName = frozen.Name
-		workflowDefinition = string(b)
+		workflowDefinitionID = &frozen.DefinitionID
 		workflowBrief = frozen.Brief
 		workflowTitle = frozen.Title
 		workflowRun = *t.WorkflowRun
@@ -411,9 +408,9 @@ func (s *Store) CreateTask(ctx context.Context, t api.Task, frozen *FrozenWorkfl
 			return err
 		}
 		_, err = tx.ExecContext(ctx, `
-			INSERT INTO tasks (id, project_id, title, merge_title, branch, worktree_path, base_ref, port_base, color, status, created_by, created_at, todo_item, forked_from, forked_at, workflow_name, workflow_definition, workflow_brief, workflow_title, workflow_run, workflow_stage, parent_task_id, audit)
+			INSERT INTO tasks (id, project_id, title, merge_title, branch, worktree_path, base_ref, port_base, color, status, created_by, created_at, todo_item, forked_from, forked_at, workflow_name, workflow_definition_id, workflow_brief, workflow_title, workflow_run, workflow_stage, parent_task_id, audit)
 			VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
-		`, t.Id, t.ProjectId, t.Title, mergeTitle, t.Branch, t.WorktreePath, t.BaseRef, t.PortBase, t.Color, string(t.Status), t.CreatedBy, formatTime(t.CreatedAt), todoItem, forkedFrom, forkedAt, workflowName, workflowDefinition, workflowBrief, workflowTitle, workflowRun, workflowStage, parentTaskID, audit)
+		`, t.Id, t.ProjectId, t.Title, mergeTitle, t.Branch, t.WorktreePath, t.BaseRef, t.PortBase, t.Color, string(t.Status), t.CreatedBy, formatTime(t.CreatedAt), todoItem, forkedFrom, forkedAt, workflowName, workflowDefinitionID, workflowBrief, workflowTitle, workflowRun, workflowStage, parentTaskID, audit)
 		if err != nil {
 			return err
 		}
@@ -3,7 +3,6 @@ package store
 import (
 	"context"
 	"database/sql"
-	"encoding/json"
 	"errors"
 	"fmt"

@@ -11,15 +10,16 @@ import (
 )

 type FrozenWorkflow struct {
-	Name       string
-	Definition api.WorkflowDefinition
-	Brief      string
-	Title      string
-	Audit      string
+	Name         string
+	DefinitionID string
+	Definition   api.WorkflowDefinition
+	Brief        string
+	Title        string
+	Audit        string
 }

 func (s *Store) ListWorkflows(ctx context.Context) ([]api.WorkflowSummary, error) {
-	rows, err := s.db.QueryContext(ctx, `SELECT name, coalesce(json_extract(definition, '$.description'), '') FROM workflows ORDER BY name`)
+	rows, err := s.db.QueryContext(ctx, `SELECT w.name, d.description FROM workflows w JOIN workflow_definitions d ON d.id = w.definition_id ORDER BY w.name`)
 	if err != nil {
 		return nil, err
 	}
@@ -38,26 +38,13 @@ func (s *Store) ListWorkflows(ctx context.Context) ([]api.WorkflowSummary, error
 }

 func (s *Store) GetWorkflow(ctx context.Context, name string) (api.WorkflowDefinition, error) {
-	def := ""
-	err := s.db.QueryRowContext(ctx, `SELECT definition FROM workflows WHERE name = ?`, name).Scan(&def)
-	if errors.Is(err, sql.ErrNoRows) {
-		return api.WorkflowDefinition{}, fmt.Errorf("%w: no workflow %q", ErrNotFound, name)
-	}
-	if err != nil {
-		return api.WorkflowDefinition{}, err
-	}
-	w := api.WorkflowDefinition{}
-	err = json.Unmarshal([]byte(def), &w)
-	return w, err
+	f, err := s.FreezeWorkflow(ctx, name)
+	return f.Definition, err
 }

 var ErrWorkflowExists = errors.New("workflow exists")

 func (s *Store) PutWorkflow(ctx context.Context, name, from string, def api.WorkflowDefinition, actor string) error {
-	b, err := json.Marshal(def)
-	if err != nil {
-		return err
-	}
 	return s.write(ctx, func(tx *sql.Tx, emit emitFunc) error {
 		if from != "" {
 			found := false
@@ -91,8 +78,12 @@ func (s *Store) PutWorkflow(ctx context.Context, name, from string, def api.Work
 				return err
 			}
 		}
-		_, err := tx.ExecContext(ctx, `INSERT INTO workflows (name, definition) VALUES (?, ?)
-			ON CONFLICT (name) DO UPDATE SET definition = excluded.definition`, name, string(b))
+		defID, err := insertDefinition(ctx, tx, def)
+		if err != nil {
+			return err
+		}
+		_, err = tx.ExecContext(ctx, `INSERT INTO workflows (name, definition_id) VALUES (?, ?)
+			ON CONFLICT (name) DO UPDATE SET definition_id = excluded.definition_id`, name, defID)
 		if err != nil {
 			return err
 		}
@@ -154,27 +145,33 @@ func (s *Store) DeleteWorkflow(ctx context.Context, name, actor string) error {
 }

 func (s *Store) FreezeWorkflow(ctx context.Context, name string) (FrozenWorkflow, error) {
-	def, err := s.GetWorkflow(ctx, name)
+	f := FrozenWorkflow{Name: name}
+	err := s.db.QueryRowContext(ctx, `SELECT definition_id FROM workflows WHERE name = ?`, name).Scan(&f.DefinitionID)
+	if errors.Is(err, sql.ErrNoRows) {
+		return FrozenWorkflow{}, fmt.Errorf("%w: no workflow %q", ErrNotFound, name)
+	}
 	if err != nil {
 		return FrozenWorkflow{}, err
 	}
-	return FrozenWorkflow{Name: name, Definition: def}, nil
+	f.Definition, err = loadDefinition(ctx, s.db, f.DefinitionID)
+	return f, err
 }

 func (s *Store) TaskWorkflow(ctx context.Context, taskID string) (FrozenWorkflow, error) {
 	f := FrozenWorkflow{}
-	def := ""
-	err := s.db.QueryRowContext(ctx, `SELECT workflow_name, workflow_definition, workflow_brief, workflow_title, audit FROM tasks WHERE id = ?`, taskID).Scan(&f.Name, &def, &f.Brief, &f.Title, &f.Audit)
+	defID := sql.NullString{}
+	err := s.db.QueryRowContext(ctx, `SELECT workflow_name, workflow_definition_id, workflow_brief, workflow_title, audit FROM tasks WHERE id = ?`, taskID).Scan(&f.Name, &defID, &f.Brief, &f.Title, &f.Audit)
 	if errors.Is(err, sql.ErrNoRows) {
 		return FrozenWorkflow{}, fmt.Errorf("%w: no task %s", ErrNotFound, taskID)
 	}
 	if err != nil {
 		return FrozenWorkflow{}, err
 	}
-	if f.Name == "" {
+	if !defID.Valid {
 		return FrozenWorkflow{}, fmt.Errorf("%w: task %s has no workflow", ErrNotFound, taskID)
 	}
-	err = json.Unmarshal([]byte(def), &f.Definition)
+	f.DefinitionID = defID.String
+	f.Definition, err = loadDefinition(ctx, s.db, f.DefinitionID)
 	return f, err
 }