narratio/internal/app/run_control.go

package app

import (
	"context"
	"fmt"
	"strings"
	"time"

	"gitea.maximumdirect.net/eric/narratio/internal/artifacts"
	"gitea.maximumdirect.net/eric/narratio/internal/config"
	"gitea.maximumdirect.net/eric/narratio/internal/manifest"
	"gitea.maximumdirect.net/eric/narratio/internal/stage"
)

type stageAction string

const (
	stageActionRun  stageAction = "run"
	stageActionSkip stageAction = "skip"
)

type stageDecision struct {
	Stage  stage.Stage
	Action stageAction
}

const (
	staleReasonForcedReplacement = "upstream stage was force-run"
	staleReasonChangedResult     = "upstream stage result changed"
	staleReasonFailure           = "upstream stage failed"
	staleReasonSelfSkip          = "upstream stage self-skipped"
	staleReasonNotResumable      = "upstream stage result was not resumable"
)

type priorStageOutcome struct {
	exists     bool
	status     manifest.StageStatus
	skipReason string
	outputs    int
}

func decideStageActions(stages []stage.Stage, m *manifest.Manifest, force bool) []stageDecision {
	out := make([]stageDecision, 0, len(stages))
	for _, s := range stages {
		out = append(out, stageDecision{
			Stage:  s,
			Action: decideStageAction(s, m, force),
		})
	}
	return out
}

func decideStageAction(s stage.Stage, m *manifest.Manifest, force bool) stageAction {
	// TODO: incorporate stale detection once checksum/input change tracking is implemented.
	if !force && stageSucceeded(m, s.Name()) {
		return stageActionSkip
	}
	return stageActionRun
}

func stageSucceeded(m *manifest.Manifest, name string) bool {
	if m == nil || m.Stages == nil {
		return false
	}
	sr := m.Stages[name]
	return sr != nil && sr.Status == manifest.StatusSucceeded
}

func capturePriorStageOutcome(m *manifest.Manifest, name string) priorStageOutcome {
	if m == nil || m.Stages == nil || m.Stages[name] == nil {
		return priorStageOutcome{}
	}
	record := m.Stages[name]
	outcome := priorStageOutcome{
		exists:  true,
		status:  record.Status,
		outputs: len(record.Outputs),
	}
	if record.Error != nil && record.Error.Code == "skipped" {
		outcome.skipReason = record.Error.Message
	}
	return outcome
}

func (o priorStageOutcome) isSameSelfSkip(reason string) bool {
	return o.exists &&
		o.status == manifest.StatusSkipped &&
		o.outputs == 0 &&
		o.skipReason == strings.TrimSpace(reason)
}

func loadManifestIfPresent(ctx context.Context, cfg *config.Config) (*manifest.Manifest, error) {
	path := artifacts.SessionManifestPathForCampaign(
		cfg.Pipeline.Workspace.Root,
		cfg.Session.Campaign,
		cfg.Session.SessionID,
	)
	exists, err := fileExists(path)
	if err != nil {
		return nil, fmt.Errorf("check manifest %q: %w", path, err)
	}
	if !exists {
		return nil, nil
	}
	store := &manifest.LocalStore{}
	m, err := store.Load(ctx, path)
	if err != nil {
		return nil, fmt.Errorf("load manifest %q: %w", path, err)
	}
	return m, nil
}

func canonicalStageNames() []string {
	all := stage.All()
	out := make([]string, 0, len(all))
	for _, s := range all {
		if s == nil {
			continue
		}
		out = append(out, s.Name())
	}
	return out
}

type invalidationRelation struct {
	canonical []string
	direct    map[string][]string
}

var canonicalInvalidationEdges = map[string][]string{
	"prepare":    {"transcribe"},
	"transcribe": {"merge"},
	"merge":      {"polish"},
	"polish":     {"normalize"},
	"normalize":  {"trim"},
	"trim":       {"render", "extract"},
	"render":     {"analyze"},
	"extract":    {"analyze"},
	"analyze":    {"publish"},
	"publish":    {"notify"},
	"notify":     {},
}

func newInvalidationRelation(registry []stage.Stage, direct map[string][]string) (*invalidationRelation, error) {
	canonical := make([]string, 0, len(registry))
	known := make(map[string]struct{}, len(registry))
	for index, candidate := range registry {
		if candidate == nil {
			return nil, fmt.Errorf("canonical stage registry entry %d is nil", index)
		}
		name := strings.TrimSpace(candidate.Name())
		if name == "" {
			return nil, fmt.Errorf("canonical stage registry entry %d has an empty name", index)
		}
		if _, duplicate := known[name]; duplicate {
			return nil, fmt.Errorf("canonical stage registry contains duplicate stage %q", name)
		}
		known[name] = struct{}{}
		canonical = append(canonical, name)
	}

	cloned := make(map[string][]string, len(direct))
	for source, targets := range direct {
		if _, ok := known[source]; !ok {
			return nil, fmt.Errorf("invalidation relation classifies unknown stage %q", source)
		}
		cloned[source] = []string{}
		seenTargets := make(map[string]struct{}, len(targets))
		for _, target := range targets {
			if _, ok := known[target]; !ok {
				return nil, fmt.Errorf("invalidation relation edge %q -> %q references an unknown stage", source, target)
			}
			if _, duplicate := seenTargets[target]; duplicate {
				return nil, fmt.Errorf("invalidation relation contains duplicate edge %q -> %q", source, target)
			}
			seenTargets[target] = struct{}{}
			cloned[source] = append(cloned[source], target)
		}
	}
	for _, name := range canonical {
		if _, classified := direct[name]; !classified {
			return nil, fmt.Errorf("invalidation relation is missing classification for stage %q", name)
		}
	}

	relation := &invalidationRelation{canonical: canonical, direct: cloned}
	visiting := make(map[string]bool, len(canonical))
	visited := make(map[string]bool, len(canonical))
	var visit func(string) error
	visit = func(name string) error {
		if visiting[name] {
			return fmt.Errorf("invalidation relation contains a cycle involving stage %q", name)
		}
		if visited[name] {
			return nil
		}
		visiting[name] = true
		for _, target := range relation.direct[name] {
			if err := visit(target); err != nil {
				return err
			}
		}
		visiting[name] = false
		visited[name] = true
		return nil
	}
	for _, name := range canonical {
		if err := visit(name); err != nil {
			return nil, err
		}
	}
	return relation, nil
}

func canonicalInvalidationRelation() (*invalidationRelation, error) {
	return newInvalidationRelation(stage.All(), canonicalInvalidationEdges)
}

func (r *invalidationRelation) Dependents(stageName string) ([]string, error) {
	if r == nil {
		return nil, fmt.Errorf("invalidation relation is nil")
	}
	if _, ok := r.direct[stageName]; !ok {
		return nil, fmt.Errorf("unknown stage %q in invalidation relation", stageName)
	}
	reachable := make(map[string]bool, len(r.canonical))
	var collect func(string)
	collect = func(name string) {
		for _, target := range r.direct[name] {
			if reachable[target] {
				continue
			}
			reachable[target] = true
			collect(target)
		}
	}
	collect(stageName)
	out := make([]string, 0, len(reachable))
	for _, name := range r.canonical {
		if reachable[name] {
			out = append(out, name)
		}
	}
	return out, nil
}

func dependentStageNames(stageName string) ([]string, error) {
	relation, err := canonicalInvalidationRelation()
	if err != nil {
		return nil, err
	}
	return relation.Dependents(stageName)
}

func invalidateDependentSucceededStagesWithReason(m *manifest.Manifest, upstreamStage string, at time.Time, reason string) ([]string, error) {
	dependents, err := dependentStageNames(upstreamStage)
	if err != nil {
		return nil, err
	}
	if m == nil || m.Stages == nil {
		return nil, nil
	}

	invalidated := make([]string, 0)
	for _, dependent := range dependents {
		sr := m.Stages[dependent]
		if sr == nil || sr.Status != manifest.StatusSucceeded {
			continue
		}
		m.MarkStageStale(dependent, at, reason)
		invalidated = append(invalidated, dependent)
	}
	return invalidated, nil
}