Run each workflow stage as its own task, with fan-out and fan-in

2026-10-15 22:53:54 -07:00
parent f7a0ffcd4c
commit 2180a7e824
15 changed files with 739 additions and 408 deletions
@@ -357,6 +357,18 @@ UPDATE workflows SET definition = json_set(definition, '$.stages',
 UPDATE tasks SET workflow_definition = json_set(workflow_definition, '$.stages',
 	(SELECT json_group_array(json_remove(value, '$.vendor')) FROM (SELECT value FROM json_each(workflow_definition, '$.stages') ORDER BY key)))
 	WHERE workflow_definition != '';
+`}, {"workflow-stage-tasks", `
+ALTER TABLE tasks ADD COLUMN workflow_run TEXT NOT NULL DEFAULT '';
+ALTER TABLE tasks ADD COLUMN workflow_stage TEXT NOT NULL DEFAULT '';
+ALTER TABLE tasks ADD COLUMN workflow_title TEXT NOT NULL DEFAULT '';
+ALTER TABLE tasks ADD COLUMN stage_output TEXT NOT NULL DEFAULT '';
+ALTER TABLE tasks ADD COLUMN stage_commit TEXT NOT NULL DEFAULT '';
+UPDATE tasks SET workflow_run = id, workflow_title = title,
+	workflow_stage = coalesce(
+		(SELECT json_extract(tasks.workflow_definition, '$.stages[' || (s.workflow_stage - 1) || '].name') FROM agent_sessions s
+			WHERE s.task_id = tasks.id AND s.workflow_stage > 0 ORDER BY s.created_at DESC LIMIT 1),
+		json_extract(workflow_definition, '$.stages[0].name'))
+	WHERE workflow_name != '' AND workflow_run = '';
 `}}

 const (
@@ -626,6 +638,9 @@ func (s *Store) CreateTask(ctx context.Context, t api.Task, frozen *FrozenWorkfl
 	workflowName := ""
 	workflowDefinition := ""
 	workflowBrief := ""
+	workflowTitle := ""
+	workflowRun := ""
+	workflowStage := ""
 	if frozen != nil {
 		b, err := json.Marshal(frozen.Definition)
 		if err != nil {
@@ -634,6 +649,9 @@ func (s *Store) CreateTask(ctx context.Context, t api.Task, frozen *FrozenWorkfl
 		workflowName = frozen.Name
 		workflowDefinition = string(b)
 		workflowBrief = frozen.Brief
+		workflowTitle = frozen.Title
+		workflowRun = *t.WorkflowRun
+		workflowStage = *t.WorkflowStage
 		t.Workflow = &workflowName
 	}
 	todoItem := ""
@@ -672,9 +690,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)
-			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)
+			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)
+			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)
 		if err != nil {
 			return err
 		}
@@ -1017,15 +1035,11 @@ func (s *Store) CreateSession(ctx context.Context, sess api.AgentSession, projec
 		forkedFrom = *sess.ForkedFrom
 		forkedAt = *sess.ForkedAt
 	}
-	stage := 0
-	if sess.WorkflowStage != nil {
-		stage = *sess.WorkflowStage
-	}
 	return s.write(ctx, func(tx *sql.Tx, emit emitFunc) error {
 		_, err := tx.ExecContext(ctx, `
-			INSERT INTO agent_sessions (id, task_id, owner, vendor, model, vendor_session_id, state, role, created_by, created_at, forked_from, forked_at, workflow_stage)
-			VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
-		`, sess.Id, nullable(sess.TaskId), sess.Owner, string(sess.Vendor), sess.Model, sess.VendorSessionId, string(sess.State), string(sess.Role), sess.CreatedBy, formatTime(sess.CreatedAt), forkedFrom, forkedAt, stage)
+			INSERT INTO agent_sessions (id, task_id, owner, vendor, model, vendor_session_id, state, role, created_by, created_at, forked_from, forked_at)
+			VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+		`, sess.Id, nullable(sess.TaskId), sess.Owner, string(sess.Vendor), sess.Model, sess.VendorSessionId, string(sess.State), string(sess.Role), sess.CreatedBy, formatTime(sess.CreatedAt), forkedFrom, forkedAt)
 		if err != nil {
 			return err
 		}
@@ -1264,9 +1278,9 @@ func (s *Store) write(ctx context.Context, fn func(tx *sql.Tx, emit emitFunc) er

 const projectColumns = `id, name, repo_path, default_branch, remote, created_by, created_at`

-const taskColumns = `id, project_id, title, merge_title, branch, worktree_path, base_ref, port_base, color, status, created_by, created_at, closed_at, merge_commit, todo_item, forked_from, forked_at, merge_queued_at, workflow_name`
+const taskColumns = `id, project_id, title, merge_title, branch, worktree_path, base_ref, port_base, color, status, created_by, created_at, closed_at, merge_commit, todo_item, forked_from, forked_at, merge_queued_at, workflow_name, workflow_run, workflow_stage`

-const sessionColumns = `id, coalesce(task_id, ''), owner, vendor, model, vendor_session_id, state, role, created_by, created_at, forked_from, forked_at, workflow_stage`
+const sessionColumns = `id, coalesce(task_id, ''), owner, vendor, model, vendor_session_id, state, role, created_by, created_at, forked_from, forked_at`

 type scanner interface {
 	Scan(dest ...any) error
@@ -1294,12 +1308,16 @@ func scanTask(row scanner) (api.Task, error) {
 	forkedAt := 0
 	mergeQueuedAt := ""
 	workflowName := ""
-	err := row.Scan(&t.Id, &t.ProjectId, &t.Title, &t.MergeTitle, &t.Branch, &t.WorktreePath, &t.BaseRef, &t.PortBase, &t.Color, &status, &t.CreatedBy, &createdAt, &closedAt, &mergeCommit, &todoItem, &forkedFrom, &forkedAt, &mergeQueuedAt, &workflowName)
+	workflowRun := ""
+	workflowStage := ""
+	err := row.Scan(&t.Id, &t.ProjectId, &t.Title, &t.MergeTitle, &t.Branch, &t.WorktreePath, &t.BaseRef, &t.PortBase, &t.Color, &status, &t.CreatedBy, &createdAt, &closedAt, &mergeCommit, &todoItem, &forkedFrom, &forkedAt, &mergeQueuedAt, &workflowName, &workflowRun, &workflowStage)
 	if err != nil {
 		return t, err
 	}
 	if workflowName != "" {
 		t.Workflow = &workflowName
+		t.WorkflowRun = &workflowRun
+		t.WorkflowStage = &workflowStage
 	}
 	if t.MergeTitle == "" {
 		t.MergeTitle = t.Title
@@ -1344,14 +1362,10 @@ func scanSession(row scanner) (api.AgentSession, error) {
 	createdAt := ""
 	forkedFrom := ""
 	forkedAt := ""
-	stage := 0
-	err := row.Scan(&sess.Id, &sess.TaskId, &sess.Owner, &vendor, &sess.Model, &sess.VendorSessionId, &state, &role, &sess.CreatedBy, &createdAt, &forkedFrom, &forkedAt, &stage)
+	err := row.Scan(&sess.Id, &sess.TaskId, &sess.Owner, &vendor, &sess.Model, &sess.VendorSessionId, &state, &role, &sess.CreatedBy, &createdAt, &forkedFrom, &forkedAt)
 	if err != nil {
 		return sess, err
 	}
-	if stage > 0 {
-		sess.WorkflowStage = &stage
-	}
 	if forkedFrom != "" {
 		sess.ForkedFrom = &forkedFrom
 		sess.ForkedAt = &forkedAt
@@ -14,6 +14,7 @@ type FrozenWorkflow struct {
 	Name       string
 	Definition api.WorkflowDefinition
 	Brief      string
+	Title      string
 }

 func (s *Store) ListWorkflows(ctx context.Context) ([]api.WorkflowSummary, error) {
@@ -92,7 +93,7 @@ func (s *Store) FreezeWorkflow(ctx context.Context, name string) (FrozenWorkflow
 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 FROM tasks WHERE id = ?`, taskID).Scan(&f.Name, &def, &f.Brief)
+	err := s.db.QueryRowContext(ctx, `SELECT workflow_name, workflow_definition, workflow_brief, workflow_title FROM tasks WHERE id = ?`, taskID).Scan(&f.Name, &def, &f.Brief, &f.Title)
 	if errors.Is(err, sql.ErrNoRows) {
 		return FrozenWorkflow{}, fmt.Errorf("%w: no task %s", ErrNotFound, taskID)
 	}
@@ -106,6 +107,24 @@ func (s *Store) TaskWorkflow(ctx context.Context, taskID string) (FrozenWorkflow
 	return f, err
 }

+func (s *Store) RunTasks(ctx context.Context, run string) ([]api.Task, error) {
+	return s.queryTasks(ctx, `SELECT `+taskColumns+` FROM tasks WHERE workflow_run = ? ORDER BY created_at`, run)
+}
+
+func (s *Store) RecordStage(ctx context.Context, taskID, output, commit string) error {
+	return s.write(ctx, func(tx *sql.Tx, emit emitFunc) error {
+		_, err := tx.ExecContext(ctx, `UPDATE tasks SET stage_output = ?, stage_commit = ? WHERE id = ?`, output, commit, taskID)
+		return err
+	})
+}
+
+func (s *Store) StageResult(ctx context.Context, taskID string) (string, string, error) {
+	output := ""
+	commit := ""
+	err := s.db.QueryRowContext(ctx, `SELECT stage_output, stage_commit FROM tasks WHERE id = ?`, taskID).Scan(&output, &commit)
+	return output, commit, err
+}
+
 func (s *Store) SetStageOutput(ctx context.Context, sessionID, output string) error {
 	return s.write(ctx, func(tx *sql.Tx, emit emitFunc) error {
 		_, err := tx.ExecContext(ctx, `UPDATE agent_sessions SET stage_output = ? WHERE id = ?`, output, sessionID)
@@ -186,8 +205,8 @@ func (s *Store) FinishWorkflow(ctx context.Context, task api.Task, sessionID, wo
 	})
 }

-func (s *Store) WorkflowFinished(ctx context.Context, taskID string) (bool, error) {
+func (s *Store) WorkflowFinished(ctx context.Context, run string) (bool, error) {
 	finished := false
-	err := s.db.QueryRowContext(ctx, `SELECT EXISTS (SELECT 1 FROM events WHERE task_id = ? AND kind = 'task.workflow_finished')`, taskID).Scan(&finished)
+	err := s.db.QueryRowContext(ctx, `SELECT EXISTS (SELECT 1 FROM events e JOIN tasks t ON t.id = e.task_id WHERE t.workflow_run = ? AND e.kind = 'task.workflow_finished')`, run).Scan(&finished)
 	return finished, err
 }