From 1054b64d9f3153ceedcda1e1bafa5e6cdb21667d Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Mon, 18 May 2026 01:50:51 +0000 Subject: [PATCH] Implement minimal downstream invalidation after forced upstream reruns --- internal/app/resume_run_stage_test.go | 46 ++++++++++++++++++++++++ internal/app/run_control.go | 42 ++++++++++++++++++++++ internal/app/run_control_test.go | 46 ++++++++++++++++++++++++ internal/app/runner.go | 3 ++ internal/app/runner_test.go | 50 +++++++++++++++++++++++++++ internal/manifest/manifest.go | 9 +++++ 6 files changed, 196 insertions(+) diff --git a/internal/app/resume_run_stage_test.go b/internal/app/resume_run_stage_test.go index 9e24fed..04d3bd1 100644 --- a/internal/app/resume_run_stage_test.go +++ b/internal/app/resume_run_stage_test.go @@ -166,6 +166,52 @@ func TestRunStageSkipAndForce(t *testing.T) { } } +func TestRunStageForceMarksDownstreamStaleAndResumeContinuesFromStale(t *testing.T) { + workspaceRoot := t.TempDir() + pipelinePath, sessionPath := writeValidConfigFiles(t, workspaceRoot) + manifestPath := filepath.Join(workspaceRoot, "work", "sample-campaign", "2026-05-03", "manifest.json") + workRoot := filepath.Join(workspaceRoot, "work", "sample-campaign", "2026-05-03") + mustWriteTestFile(t, filepath.Join(workRoot, "transcripts", "merged.json"), `{"segments":[]}`) + mustWriteTestFile(t, filepath.Join(workRoot, "inputs", "glossary.yml"), "terms: []\n") + + store := &manifest.LocalStore{} + seed := manifest.New("2026-05-03", time.Date(2026, 5, 3, 10, 0, 0, 0, time.UTC)) + for _, name := range []string{"prepare", "transcribe", "merge", "polish", "normalize", "trim", "analyze", "archive", "notify"} { + seed.MarkStageSucceeded(name, time.Date(2026, 5, 3, 10, 1, 0, 0, time.UTC), nil) + } + if err := store.Save(context.Background(), manifestPath, seed); err != nil { + t.Fatalf("save manifest: %v", err) + } + + var out bytes.Buffer + err := RunStage(context.Background(), []string{"--config", pipelinePath, "--session", sessionPath, "--force", "polish"}, &out) + if err != nil { + t.Fatalf("RunStage(force) error = %v", err) + } + if !strings.Contains(out.String(), "stage=polish executed=1 skipped=0 force=true") { + t.Fatalf("output = %q, want forced polish rerun", out.String()) + } + + afterForce, err := store.Load(context.Background(), manifestPath) + if err != nil { + t.Fatalf("load manifest after force: %v", err) + } + for _, name := range []string{"normalize", "trim", "analyze", "archive", "notify"} { + if afterForce.Stages[name] == nil || afterForce.Stages[name].Status != manifest.StatusStale { + t.Fatalf("stage %q = %#v, want stale", name, afterForce.Stages[name]) + } + } + + out.Reset() + err = Resume(context.Background(), []string{"--config", pipelinePath, "--session", sessionPath}, &out) + if err != nil { + t.Fatalf("Resume() error = %v", err) + } + if !strings.Contains(out.String(), "executed=5 skipped=0") { + t.Fatalf("output = %q, want resume to execute normalize..notify", out.String()) + } +} + func TestRunStageTrimExecutes(t *testing.T) { workspaceRoot := t.TempDir() pipelinePath, sessionPath := writeValidConfigFiles(t, workspaceRoot) diff --git a/internal/app/run_control.go b/internal/app/run_control.go index e54288c..6e23ac0 100644 --- a/internal/app/run_control.go +++ b/internal/app/run_control.go @@ -1,6 +1,8 @@ package app import ( + "time" + "gitea.maximumdirect.net/eric/narratio/internal/manifest" "gitea.maximumdirect.net/eric/narratio/internal/stage" ) @@ -49,3 +51,43 @@ func firstNonSucceededIndex(stages []stage.Stage, m *manifest.Manifest) int { } return len(stages) } + +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 +} + +func downstreamStageNames(stageName string) []string { + names := canonicalStageNames() + for i, name := range names { + if name != stageName { + continue + } + return append([]string(nil), names[i+1:]...) + } + return nil +} + +func invalidateDownstreamSucceededStages(m *manifest.Manifest, upstreamStage string, at time.Time) []string { + if m == nil || m.Stages == nil { + return nil + } + + invalidated := make([]string, 0) + for _, downstream := range downstreamStageNames(upstreamStage) { + sr := m.Stages[downstream] + if sr == nil || sr.Status != manifest.StatusSucceeded { + continue + } + m.MarkStageStale(downstream, at, "upstream stage rerun with force") + invalidated = append(invalidated, downstream) + } + return invalidated +} diff --git a/internal/app/run_control_test.go b/internal/app/run_control_test.go index 79359ed..2c408bd 100644 --- a/internal/app/run_control_test.go +++ b/internal/app/run_control_test.go @@ -1,6 +1,7 @@ package app import ( + "reflect" "testing" "time" @@ -40,3 +41,48 @@ func TestDecideStageActions(t *testing.T) { t.Fatalf("forced prepare action = %q, want %q", forced[0].Action, stageActionRun) } } + +func TestDownstreamStageNames(t *testing.T) { + got := downstreamStageNames("polish") + want := []string{"normalize", "trim", "analyze", "archive", "notify"} + if !reflect.DeepEqual(got, want) { + t.Fatalf("downstreamStageNames(polish) = %#v, want %#v", got, want) + } + + missing := downstreamStageNames("unknown") + if len(missing) != 0 { + t.Fatalf("downstreamStageNames(unknown) = %#v, want empty", missing) + } +} + +func TestInvalidateDownstreamSucceededStages(t *testing.T) { + now := time.Now().UTC() + m := manifest.New("2026-05-03", now) + m.MarkStageSucceeded("prepare", now, nil) + m.MarkStageSucceeded("transcribe", now, nil) + m.MarkStageSucceeded("merge", now, nil) + m.MarkStageSucceeded("polish", now, nil) + m.MarkStageSucceeded("normalize", now, nil) + m.MarkStageSucceeded("trim", now, nil) + m.MarkStageFailed("analyze", now, "analysis failed") + m.MarkStageSucceeded("archive", now, nil) + m.MarkStageSucceeded("notify", now, nil) + + got := invalidateDownstreamSucceededStages(m, "polish", now.Add(1*time.Second)) + want := []string{"normalize", "trim", "archive", "notify"} + if !reflect.DeepEqual(got, want) { + t.Fatalf("invalidateDownstreamSucceededStages() = %#v, want %#v", got, want) + } + + for _, stageName := range want { + if m.Stages[stageName].Status != manifest.StatusStale { + t.Fatalf("%s status = %q, want stale", stageName, m.Stages[stageName].Status) + } + } + if m.Stages["analyze"].Status != manifest.StatusFailed { + t.Fatalf("analyze status = %q, want failed", m.Stages["analyze"].Status) + } + if m.Stages["prepare"].Status != manifest.StatusSucceeded { + t.Fatalf("prepare status = %q, want succeeded", m.Stages["prepare"].Status) + } +} diff --git a/internal/app/runner.go b/internal/app/runner.go index abe3b8e..b494588 100644 --- a/internal/app/runner.go +++ b/internal/app/runner.go @@ -206,6 +206,9 @@ func executeStages(ctx context.Context, cfg *config.Config, stages []stage.Stage succeededAt := nowUTC() m.MarkStageSucceeded(s.Name(), succeededAt, outputs) applyStageResultToManifest(m, s.Name(), result) + if opts.Force { + invalidateDownstreamSucceededStages(m, s.Name(), succeededAt) + } if err := env.ManifestStore.Save(ctx, manifestPath, m); err != nil { return nil, fmt.Errorf("save manifest after stage %q: %w", s.Name(), err) diff --git a/internal/app/runner_test.go b/internal/app/runner_test.go index 0641ce6..80b53d5 100644 --- a/internal/app/runner_test.go +++ b/internal/app/runner_test.go @@ -235,6 +235,56 @@ func TestExecuteStagesForceRerunsSucceeded(t *testing.T) { } } +func TestExecuteStagesForceSuccessInvalidatesDownstreamSucceededStages(t *testing.T) { + cfg := testConfig(t) + manifestPath := manifestPathFor(cfg) + store := &manifest.LocalStore{} + + existing := manifest.New(cfg.Session.SessionID, time.Date(2026, 5, 3, 1, 0, 0, 0, time.UTC)) + for _, stageName := range []string{"prepare", "transcribe", "merge", "polish", "normalize", "trim", "archive", "notify"} { + existing.MarkStageSucceeded(stageName, time.Date(2026, 5, 3, 1, 1, 0, 0, time.UTC), nil) + } + existing.MarkStageFailed("analyze", time.Date(2026, 5, 3, 1, 1, 0, 0, time.UTC), "previous analyze failure") + if err := os.MkdirAll(filepath.Dir(manifestPath), 0o755); err != nil { + t.Fatalf("MkdirAll() error = %v", err) + } + if err := store.Save(context.Background(), manifestPath, existing); err != nil { + t.Fatalf("Save manifest error = %v", err) + } + + runs := 0 + stageToRun := countingStage{name: "polish", runs: &runs} + summary, err := executeStages(context.Background(), cfg, []stage.Stage{stageToRun}, RunOptions{Force: true}) + if err != nil { + t.Fatalf("executeStages() error = %v", err) + } + if runs != 1 { + t.Fatalf("runs = %d, want 1 with force", runs) + } + if len(summary.Executed) != 1 || summary.Executed[0] != "polish" || len(summary.Skipped) != 0 { + t.Fatalf("summary = %#v, want executed polish", summary) + } + + loaded, err := store.Load(context.Background(), manifestPath) + if err != nil { + t.Fatalf("Load manifest error = %v", err) + } + if loaded.Stages["polish"] == nil || loaded.Stages["polish"].Status != manifest.StatusSucceeded { + t.Fatalf("polish status = %#v, want succeeded", loaded.Stages["polish"]) + } + for _, stageName := range []string{"normalize", "trim", "archive", "notify"} { + if loaded.Stages[stageName] == nil || loaded.Stages[stageName].Status != manifest.StatusStale { + t.Fatalf("%s status = %#v, want stale", stageName, loaded.Stages[stageName]) + } + } + if loaded.Stages["analyze"] == nil || loaded.Stages["analyze"].Status != manifest.StatusFailed { + t.Fatalf("analyze status = %#v, want preserved failed", loaded.Stages["analyze"]) + } + if loaded.Stages["transcribe"] == nil || loaded.Stages["transcribe"].Status != manifest.StatusSucceeded { + t.Fatalf("transcribe status = %#v, want preserved succeeded", loaded.Stages["transcribe"]) + } +} + func TestExecuteStagesFailureUpdatesManifest(t *testing.T) { cfg := testConfig(t) diff --git a/internal/manifest/manifest.go b/internal/manifest/manifest.go index 28c3634..fae84f9 100644 --- a/internal/manifest/manifest.go +++ b/internal/manifest/manifest.go @@ -122,6 +122,15 @@ func (m *Manifest) MarkStageSkipped(name string, at time.Time, reason string) { m.UpdatedAt = at } +// MarkStageStale marks a stage as stale so it is not skipped as idempotently complete. +func (m *Manifest) MarkStageStale(name string, at time.Time, reason string) { + s := m.ensureStage(name, at) + s.Status = StatusStale + s.Error = &ErrorRecord{Message: strings.TrimSpace(reason), Code: "stale", At: timePtr(at)} + s.UpdatedAt = at + m.UpdatedAt = at +} + func (m *Manifest) ensureStage(name string, at time.Time) *StageRecord { if m.Stages == nil { m.Stages = map[string]*StageRecord{}