byre/internal/store/migrateworkflows.go

package store

import (
	"database/sql"
	"encoding/json"
	"fmt"
	"maps"
	"slices"

	"g.fc.run/ai/byre/internal/api"
	"g.fc.run/ai/byre/internal/workflow"
)

type oldStage struct {
	Name   string            `json:"name"`
	After  *[]string         `json:"after"`
	Base   *string           `json:"base"`
	Model  string            `json:"model"`
	Prompt string            `json:"prompt"`
	Agents *[]api.TeamMember `json:"agents"`
}

type oldDefinition struct {
	Description string           `json:"description"`
	Stages      []oldStage       `json:"stages"`
	Steps       *json.RawMessage `json:"steps"`
}

func migrateWorkflowSteps(tx *sql.Tx) error {
	err := convertColumn(tx, "workflows", "name", "definition")
	if err != nil {
		return err
	}
	return convertColumn(tx, "tasks", "id", "workflow_definition")
}

func convertColumn(tx *sql.Tx, table, key, column string) error {
	rows, err := tx.Query(fmt.Sprintf(`SELECT %s, %s FROM %s WHERE %s != ''`, key, column, table, column))
	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
	}

	for _, r := range all {
		old := oldDefinition{}
		err = json.Unmarshal([]byte(r.def), &old)
		if err != nil {
			return fmt.Errorf("%s %s: %w", table, r.key, err)
		}
		if old.Steps != nil {
			continue
		}
		def, err := toSteps(old)
		if err != nil {
			return fmt.Errorf("%s %s: %w", table, r.key, err)
		}
		j, err := json.Marshal(def)
		if err != nil {
			return err
		}
		_, err = tx.Exec(fmt.Sprintf(`UPDATE %s SET %s = ? WHERE %s = ?`, table, column, key), string(j), r.key)
		if err != nil {
			return err
		}
	}
	return nil
}

func toSteps(old oldDefinition) (api.WorkflowDefinition, error) {
	after := map[string][]string{}
	byName := map[string]oldStage{}
	for i, s := range old.Stages {
		if s.Base != nil && *s.Base != "parent" {
			return api.WorkflowDefinition{}, fmt.Errorf("stage %q has base %q, which steps can't express", s.Name, *s.Base)
		}
		byName[s.Name] = s
		switch {
		case s.After != nil:
			after[s.Name] = *s.After
		case i == 0:
			after[s.Name] = []string{}
		default:
			after[s.Name] = []string{old.Stages[i-1].Name}
		}
	}

	pending := func(names []string, done map[string]bool) []string {
		out := []string{}
		for _, n := range names {
			if !done[n] {
				out = append(out, n)
			}
		}
		return out
	}
	var closure func(name string, done map[string]bool) map[string]bool
	closure = func(name string, done map[string]bool) map[string]bool {
		c := map[string]bool{name: true}
		for _, a := range pending(after[name], done) {
			for n := range closure(a, done) {
				c[n] = true
			}
		}
		return c
	}
	sequence := func(steps ...api.WorkflowStep) api.WorkflowStep {
		order := []api.WorkflowStep{}
		for _, s := range steps {
			if s.Order != nil {
				order = append(order, *s.Order...)
				continue
			}
			order = append(order, s)
		}
		return api.WorkflowStep{Order: &order}
	}

	var tree func(ends []string, done map[string]bool) (api.WorkflowStep, error)
	tree = func(ends []string, done map[string]bool) (api.WorkflowStep, error) {
		groups := [][]string{}
		for _, e := range ends {
			merged := []string{e}
			kept := [][]string{}
			for _, g := range groups {
				if slices.ContainsFunc(g, func(other string) bool { return shareAny(closure(e, done), closure(other, done)) }) {
					merged = append(g, merged...)
					continue
				}
				kept = append(kept, g)
			}
			groups = append(kept, merged)
		}
		if len(groups) > 1 {
			branches := []api.WorkflowStep{}
			for _, g := range groups {
				b, err := tree(g, done)
				if err != nil {
					return api.WorkflowStep{}, err
				}
				branches = append(branches, b)
			}
			return api.WorkflowStep{Parallel: &branches}, nil
		}

		if len(ends) > 1 {
			common := closure(ends[0], done)
			for _, e := range ends[1:] {
				c := closure(e, done)
				for n := range common {
					if !c[n] {
						delete(common, n)
					}
				}
			}
			heads := []string{}
			for n := range common {
				later := false
				for m := range common {
					later = later || (m != n && closure(m, done)[n])
				}
				if !later {
					heads = append(heads, n)
				}
			}
			if len(heads) == 0 {
				return api.WorkflowStep{}, fmt.Errorf("stages %v overlap in earlier stages without sharing all of them, which steps can't express", ends)
			}
			slices.Sort(heads)
			prefix, err := tree(heads, done)
			if err != nil {
				return api.WorkflowStep{}, err
			}
			next := maps.Clone(done)
			for n := range common {
				next[n] = true
			}
			rest, err := tree(ends, next)
			if err != nil {
				return api.WorkflowStep{}, err
			}
			return sequence(prefix, rest), nil
		}

		s := byName[ends[0]]
		stage := api.WorkflowStep{Stage: s.Name, Model: s.Model, Prompt: s.Prompt, Agents: s.Agents}
		preds := pending(after[s.Name], done)
		if len(preds) == 0 {
			return stage, nil
		}
		before, err := tree(preds, done)
		if err != nil {
			return api.WorkflowStep{}, err
		}
		return sequence(before, stage), nil
	}

	finals := []string{}
	for _, s := range old.Stages {
		next := false
		for _, a := range after {
			next = next || slices.Contains(a, s.Name)
		}
		if !next {
			finals = append(finals, s.Name)
		}
	}
	top, err := tree(finals, map[string]bool{})
	if err != nil {
		return api.WorkflowDefinition{}, err
	}
	steps := []api.WorkflowStep{top}
	if top.Order != nil {
		steps = *top.Order
	}
	def := api.WorkflowDefinition{Description: old.Description, Steps: steps}

	if len(workflow.Stages(def)) != len(old.Stages) {
		return api.WorkflowDefinition{}, fmt.Errorf("converted to %d stages from %d", len(workflow.Stages(def)), len(old.Stages))
	}
	for _, s := range old.Stages {
		got := workflow.After(def, s.Name)
		want := after[s.Name]
		if !slices.Equal(sorted(got), sorted(want)) {
			return api.WorkflowDefinition{}, fmt.Errorf("stage %q converted to come after %v, not %v", s.Name, got, want)
		}
	}
	return def, nil
}

func shareAny(a, b map[string]bool) bool {
	for n := range a {
		if b[n] {
			return true
		}
	}
	return false
}

func sorted(names []string) []string {
	out := slices.Clone(names)
	slices.Sort(out)
	return out
}