Add Workflow tab and run-progress API for workflow runs

2026-10-16 11:01:53 -07:00
parent eb0c80a645
commit ec169583ff
19 changed files with 1285 additions and 209 deletions
@@ -51,6 +51,151 @@ func (m *Manager) TaskWorkflow(ctx context.Context, taskID string) (api.TaskWork
 	return api.TaskWorkflow{Name: frozen.Name, Definition: string(file)}, nil
 }

+func (m *Manager) WorkflowRun(ctx context.Context, taskID string) (api.WorkflowRun, error) {
+	frozen, err := m.store.TaskWorkflow(ctx, taskID)
+	if err != nil {
+		return api.WorkflowRun{}, err
+	}
+	task, err := m.store.GetTask(ctx, taskID)
+	if err != nil {
+		return api.WorkflowRun{}, err
+	}
+	tasks, err := m.store.RunTasks(ctx, *task.WorkflowRun)
+	if err != nil {
+		return api.WorkflowRun{}, err
+	}
+	events, err := m.store.RunEvents(ctx, *task.WorkflowRun)
+	if err != nil {
+		return api.WorkflowRun{}, err
+	}
+
+	run := api.WorkflowRun{Name: frozen.Name, Run: *task.WorkflowRun, Title: frozen.Title, Brief: frozen.Brief, Definition: frozen.Definition, Stages: []api.WorkflowRunStage{}}
+	after := map[string][]string{}
+	discardedBy := map[string]string{}
+	for _, ev := range events {
+		switch ev.Kind {
+		case "task.workflow_stage":
+			ids := []string{}
+			list, _ := ev.Payload["after"].([]any)
+			for _, v := range list {
+				id, _ := v.(string)
+				ids = append(ids, id)
+			}
+			after[*ev.TaskId] = ids
+		case "task.discarded":
+			discardedBy[*ev.TaskId] = ev.Actor
+		case "task.workflow_finished":
+			run.Finished = true
+		}
+	}
+
+	byID := map[string]api.Task{}
+	for _, t := range tasks {
+		byID[t.Id] = t
+	}
+	var inputs func(id string) []api.StageInput
+	inputs = func(id string) []api.StageInput {
+		ids, found := after[id]
+		if !found {
+			t := byID[id]
+			if t.ForkedFrom == nil {
+				return []api.StageInput{}
+			}
+			return inputs(*t.ForkedFrom)
+		}
+		in := []api.StageInput{}
+		for _, a := range ids {
+			stage := ""
+			if t, found := byID[a]; found {
+				stage = *t.WorkflowStage
+			}
+			in = append(in, api.StageInput{Stage: stage, TaskId: a})
+		}
+		return in
+	}
+
+	attempts := map[string][]api.StageAttempt{}
+	done := map[string]bool{}
+	for _, t := range tasks {
+		output, commit, err := m.store.StageResult(ctx, t.Id)
+		if err != nil {
+			return api.WorkflowRun{}, err
+		}
+		running := false
+		if t.Status == api.TaskStatusActive {
+			sessions, err := m.store.ListSessions(ctx, t.Id)
+			if err != nil {
+				return api.WorkflowRun{}, err
+			}
+			for _, s := range sessions {
+				running = running || s.State == api.AgentSessionStateRunning
+			}
+		}
+		a := api.StageAttempt{
+			TaskId:     t.Id,
+			Title:      t.Title,
+			State:      attemptState(t, output != "", running),
+			Discarding: t.Discarding,
+			ForkedFrom: t.ForkedFrom,
+			Inputs:     inputs(t.Id),
+		}
+		if output != "" {
+			a.Output = &output
+			a.Commit = &commit
+			done[*t.WorkflowStage] = true
+		}
+		if by, found := discardedBy[t.Id]; found && t.Status == api.TaskStatusDiscarded {
+			a.DiscardedBy = &by
+		}
+		attempts[*t.WorkflowStage] = append(attempts[*t.WorkflowStage], a)
+	}
+
+	for _, s := range frozen.Definition.Stages {
+		stage := api.WorkflowRunStage{
+			Name:      s.Name,
+			After:     workflow.After(frozen.Definition, s.Name),
+			State:     api.StageStateWaiting,
+			WaitingOn: []string{},
+			Attempts:  attempts[s.Name],
+		}
+		if stage.Attempts == nil {
+			stage.Attempts = []api.StageAttempt{}
+		}
+		for _, a := range stage.After {
+			if !done[a] {
+				stage.WaitingOn = append(stage.WaitingOn, a)
+			}
+		}
+		if len(stage.Attempts) > 0 {
+			deciding := stage.Attempts[len(stage.Attempts)-1]
+			for _, a := range stage.Attempts {
+				if a.Output != nil {
+					deciding = a
+				}
+			}
+			stage.State = api.WorkflowRunStageState(deciding.State)
+		}
+		run.Stages = append(run.Stages, stage)
+	}
+	return run, nil
+}
+
+func attemptState(t api.Task, finished, running bool) api.StageAttemptState {
+	switch {
+	case t.Status == api.TaskStatusMerged:
+		return api.AttemptStateMerged
+	case t.Status == api.TaskStatusDiscarded && finished:
+		return api.AttemptStateCleanedUp
+	case t.Status == api.TaskStatusDiscarded:
+		return api.AttemptStateDiscarded
+	case finished:
+		return api.AttemptStateFinished
+	case running:
+		return api.AttemptStateRunning
+	}
+	return api.AttemptStateIdle
+}
+
 var errModelWithWorkflow = fmt.Errorf("%w: model can't be combined with workflow: each stage names its own model", ErrInvalid)

 type parentStage struct {