eric/narratio
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 15 Wiki Activity

Invalidate downstream results when stages are replaced

Browse Source
This commit is contained in:
Eric Rakestraw
2026-08-10 01:38:24 +00:00
parent df40cbec6e
commit 0d6f2dd0ce
7 changed files with 509 additions and 528 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
internal/app/analyze_artifacts_commands_test.go
View File

@@ -130,6 +130,7 @@ func TestRunArtifactsWithSucceededAnalyzeSkipsUnlessForced(t *testing.T) {
for _, stageName := range []string{"prepare", "transcribe", "merge", "polish", "normalize", "trim", "render", "analyze", "publish", "notify"} {
seed.MarkStageSucceeded(stageName, time.Date(2026, 5, 3, 10, 1, 0, 0, time.UTC), nil)
}
seed.MarkStageSkipped("extract", time.Date(2026, 5, 3, 10, 1, 0, 0, time.UTC), "notarius_disabled")
if err := store.Save(context.Background(), manifestPath, seed); err != nil {
t.Fatalf("save manifest: %v", err)
}

172
internal/app/extract_lifecycle_test.go
View File

@@ -3,6 +3,7 @@ package app
import (
"context"
"encoding/json"
"errors"
"fmt"
"os"
"path/filepath"
@@ -19,12 +20,17 @@ import (
)
type materializingNotariusRunner struct {
cfg *config.NotariusConfig
requests []notarius.RunRequest
cfg *config.NotariusConfig
requests []notarius.RunRequest
failuresRemaining int
}
func (r *materializingNotariusRunner) Run(_ context.Context, req notarius.RunRequest) (notarius.RunResult, error) {
r.requests = append(r.requests, req)
if r.failuresRemaining > 0 {
r.failuresRemaining--
return notarius.RunResult{}, errors.New("notarius execution failed")
}
externalRunID := fmt.Sprintf("notarius-run-%d", len(r.requests))
bundle := filepath.Join(req.OutputRoot, externalRunID)
lanesDir := filepath.Join(bundle, "lanes")
@@ -94,9 +100,121 @@ func TestExtractLifecycleDisabledThenEnabled(t *testing.T) {
}
}
func TestExtractLifecycleChangedOutcomeRerunsSucceededDownstream(t *testing.T) {
cfg, env, runner := extractionLifecycleFixture(t, false)
analyzeRuns := 0
plan := extractionLifecyclePlan(t, &analyzeRuns)
if _, err := executeStages(context.Background(), cfg, plan, RunOptions{Env: env}); err != nil {
t.Fatalf("disabled executeStages() error = %v", err)
}
if analyzeRuns != 1 || len(runner.requests) != 0 {
t.Fatalf("disabled run analyze=%d Notarius=%d, want 1 and 0", analyzeRuns, len(runner.requests))
}
cfg.Pipeline.Notarius.Enabled = true
summary, err := executeStages(context.Background(), cfg, plan, RunOptions{Env: env})
if err != nil {
t.Fatalf("enabled executeStages() error = %v", err)
}
if analyzeRuns != 2 || len(runner.requests) != 1 {
t.Fatalf("enabled run analyze=%d Notarius=%d, want 2 and 1", analyzeRuns, len(runner.requests))
}
if len(summary.Executed) != 2 || len(summary.Skipped) != 0 {
t.Fatalf("enabled summary = %#v, want extract and analyze executed", summary)
}
}
func TestExtractLifecycleFailureInvalidatesAndOrdinaryRetryRerunsDownstream(t *testing.T) {
cfg, env, runner := extractionLifecycleFixture(t, true)
runner.failuresRemaining = 1
markLifecycleStageSucceeded(t, cfg, "analyze")
analyzeRuns := 0
plan := extractionLifecyclePlan(t, &analyzeRuns)
if _, err := executeStages(context.Background(), cfg, plan, RunOptions{Env: env}); err == nil || !strings.Contains(err.Error(), "notarius execution failed") {
t.Fatalf("failed executeStages() error = %v", err)
}
failed := loadLifecycleManifest(t, cfg)
if failed.Stages["extract"].Status != manifest.StatusFailed || failed.Stages["analyze"].Status != manifest.StatusStale {
t.Fatalf("failed lifecycle extract=%#v analyze=%#v", failed.Stages["extract"], failed.Stages["analyze"])
}
if failed.Stages["analyze"].Error == nil || failed.Stages["analyze"].Error.Message != staleReasonFailure {
t.Fatalf("analyze stale reason = %#v, want %q", failed.Stages["analyze"].Error, staleReasonFailure)
}
summary, err := executeStages(context.Background(), cfg, plan, RunOptions{Env: env})
if err != nil {
t.Fatalf("retry executeStages() error = %v", err)
}
if analyzeRuns != 1 || len(runner.requests) != 2 || len(summary.Executed) != 2 {
t.Fatalf("retry analyze=%d Notarius=%d summary=%#v", analyzeRuns, len(runner.requests), summary)
}
}
func TestExtractLifecycleForcedSelfSkipInvalidatesDownstream(t *testing.T) {
cfg, env, _ := extractionLifecycleFixture(t, false)
markLifecycleStageSucceeded(t, cfg, "analyze")
plan, _ := BuildSingleStagePlan("extract")
if _, err := executeStages(context.Background(), cfg, plan, RunOptions{Env: env, Force: true}); err != nil {
t.Fatalf("executeStages() error = %v", err)
}
loaded := loadLifecycleManifest(t, cfg)
if loaded.Stages["extract"].Status != manifest.StatusSkipped || loaded.Stages["analyze"].Status != manifest.StatusStale {
t.Fatalf("forced self-skip extract=%#v analyze=%#v", loaded.Stages["extract"], loaded.Stages["analyze"])
}
if loaded.Stages["analyze"].Error == nil || loaded.Stages["analyze"].Error.Message != staleReasonForcedReplacement {
t.Fatalf("analyze stale reason = %#v, want %q", loaded.Stages["analyze"].Error, staleReasonForcedReplacement)
}
}
func TestExtractLifecycleForcedFailureInvalidatesDownstream(t *testing.T) {
cfg, env, runner := extractionLifecycleFixture(t, true)
runner.failuresRemaining = 1
markLifecycleStageSucceeded(t, cfg, "analyze")
plan, _ := BuildSingleStagePlan("extract")
if _, err := executeStages(context.Background(), cfg, plan, RunOptions{Env: env, Force: true}); err == nil {
t.Fatal("executeStages() error = nil, want forced extraction failure")
}
loaded := loadLifecycleManifest(t, cfg)
if loaded.Stages["extract"].Status != manifest.StatusFailed || loaded.Stages["analyze"].Status != manifest.StatusStale {
t.Fatalf("forced failure extract=%#v analyze=%#v", loaded.Stages["extract"], loaded.Stages["analyze"])
}
if loaded.Stages["analyze"].Error == nil || loaded.Stages["analyze"].Error.Message != staleReasonForcedReplacement {
t.Fatalf("analyze stale reason = %#v, want %q", loaded.Stages["analyze"].Error, staleReasonForcedReplacement)
}
}
func TestExtractLifecycleRepeatedSelfSkipPreservesSucceededDownstream(t *testing.T) {
cfg, env, runner := extractionLifecycleFixture(t, false)
analyzeRuns := 0
plan := extractionLifecyclePlan(t, &analyzeRuns)
if _, err := executeStages(context.Background(), cfg, plan, RunOptions{Env: env}); err != nil {
t.Fatalf("first executeStages() error = %v", err)
}
second, err := executeStages(context.Background(), cfg, plan, RunOptions{Env: env})
if err != nil {
t.Fatalf("second executeStages() error = %v", err)
}
if analyzeRuns != 1 || len(runner.requests) != 0 {
t.Fatalf("repeated disabled run analyze=%d Notarius=%d, want 1 and 0", analyzeRuns, len(runner.requests))
}
if len(second.Executed) != 1 || len(second.Skipped) != 2 {
t.Fatalf("second summary = %#v, want executed self-skip and skipped analyze", second)
}
loaded := loadLifecycleManifest(t, cfg)
if loaded.Stages["analyze"].Status != manifest.StatusSucceeded {
t.Fatalf("analyze = %#v, want succeeded", loaded.Stages["analyze"])
}
}
func TestExtractLifecycleSkipsCurrentResumableResult(t *testing.T) {
cfg, env, runner := extractionLifecycleFixture(t, true)
plan, _ := BuildSingleStagePlan("extract")
analyzeRuns := 0
plan := extractionLifecyclePlan(t, &analyzeRuns)
if _, err := executeStages(context.Background(), cfg, plan, RunOptions{Env: env}); err != nil {
t.Fatalf("first executeStages() error = %v", err)
}
@@ -104,8 +222,8 @@ func TestExtractLifecycleSkipsCurrentResumableResult(t *testing.T) {
if err != nil {
t.Fatalf("resume executeStages() error = %v", err)
}
if len(resumed.Executed) != 0 || len(resumed.Skipped) != 1 || resumed.Skipped[0] != "extract" || len(runner.requests) != 1 {
t.Fatalf("resume summary = %#v requests=%d", resumed, len(runner.requests))
if len(resumed.Executed) != 0 || len(resumed.Skipped) != 2 || len(runner.requests) != 1 || analyzeRuns != 1 {
t.Fatalf("resume summary = %#v requests=%d analyze=%d", resumed, len(runner.requests), analyzeRuns)
}
}
@@ -162,7 +280,8 @@ func TestExtractLifecycleResumesAndRerunsObsoleteResults(t *testing.T) {
func TestExtractLifecycleUnsafeResumeErrorPreservesSuccess(t *testing.T) {
cfg, env, runner := extractionLifecycleFixture(t, true)
plan, _ := BuildSingleStagePlan("extract")
analyzeRuns := 0
plan := extractionLifecyclePlan(t, &analyzeRuns)
first, err := executeStages(context.Background(), cfg, plan, RunOptions{Env: env})
if err != nil {
t.Fatalf("first executeStages() error = %v", err)
@@ -176,7 +295,10 @@ func TestExtractLifecycleUnsafeResumeErrorPreservesSuccess(t *testing.T) {
if err := store.Save(context.Background(), first.ManifestPath, persisted); err != nil {
t.Fatalf("Save() error = %v", err)
}
before, _ := json.Marshal(persisted.Stages["extract"])
before, _ := json.Marshal(map[string]*manifest.StageRecord{
"extract": persisted.Stages["extract"],
"analyze": persisted.Stages["analyze"],
})
if _, err := executeStages(context.Background(), cfg, plan, RunOptions{Env: env}); err == nil || !strings.Contains(err.Error(), "unsafe") {
t.Fatalf("executeStages() error = %v, want unsafe resume failure", err)
@@ -185,9 +307,39 @@ func TestExtractLifecycleUnsafeResumeErrorPreservesSuccess(t *testing.T) {
if err != nil {
t.Fatalf("Load(after) error = %v", err)
}
after, _ := json.Marshal(afterManifest.Stages["extract"])
if string(before) != string(after) || len(runner.requests) != 1 {
t.Fatalf("successful extract record changed: before=%s after=%s requests=%d", before, after, len(runner.requests))
after, _ := json.Marshal(map[string]*manifest.StageRecord{
"extract": afterManifest.Stages["extract"],
"analyze": afterManifest.Stages["analyze"],
})
if string(before) != string(after) || len(runner.requests) != 1 || analyzeRuns != 1 {
t.Fatalf("successful records changed: before=%s after=%s requests=%d analyze=%d", before, after, len(runner.requests), analyzeRuns)
}
}
func extractionLifecyclePlan(t *testing.T, analyzeRuns *int) []stage.Stage {
t.Helper()
plan, err := BuildSingleStagePlan("extract")
if err != nil {
t.Fatalf("BuildSingleStagePlan(extract) error = %v", err)
}
return append(plan, countingStage{name: "analyze", runs: analyzeRuns})
}
func loadLifecycleManifest(t *testing.T, cfg *config.Config) *manifest.Manifest {
t.Helper()
loaded, err := (&manifest.LocalStore{}).Load(context.Background(), manifestPathFor(cfg))
if err != nil {
t.Fatalf("Load() error = %v", err)
}
return loaded
}
func markLifecycleStageSucceeded(t *testing.T, cfg *config.Config, name string) {
t.Helper()
loaded := loadLifecycleManifest(t, cfg)
loaded.MarkStageSucceeded(name, time.Now().UTC(), nil)
if err := (&manifest.LocalStore{}).Save(context.Background(), manifestPathFor(cfg), loaded); err != nil {
t.Fatalf("Save() error = %v", err)
}
}

58
internal/app/run_control.go
View File

@@ -3,6 +3,7 @@ package app
import (
"context"
"fmt"
"strings"
"time"
"gitea.maximumdirect.net/eric/narratio/internal/artifacts"
@@ -23,22 +24,40 @@ type stageDecision struct {
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 {
action := stageActionRun
// TODO: incorporate stale detection once checksum/input change tracking is implemented.
if !force && stageSucceeded(m, s.Name()) {
action = stageActionSkip
}
out = append(out, stageDecision{
Stage: s,
Action: action,
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
@@ -47,6 +66,29 @@ func stageSucceeded(m *manifest.Manifest, name string) bool {
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,
@@ -91,10 +133,6 @@ func downstreamStageNames(stageName string) []string {
return nil
}
func invalidateDownstreamSucceededStages(m *manifest.Manifest, upstreamStage string, at time.Time) []string {
return invalidateDownstreamSucceededStagesWithReason(m, upstreamStage, at, "upstream stage rerun with force")
}
func invalidateDownstreamSucceededStagesWithReason(m *manifest.Manifest, upstreamStage string, at time.Time, reason string) []string {
if m == nil || m.Stages == nil {
return nil

8
internal/app/run_control_test.go
View File

@@ -43,7 +43,7 @@ func TestDownstreamStageNames(t *testing.T) {
}
}
func TestInvalidateDownstreamSucceededStages(t *testing.T) {
func TestInvalidateDownstreamSucceededStagesWithReason(t *testing.T) {
now := time.Now().UTC()
m := manifest.New("2026-05-03", now)
m.MarkStageSucceeded("prepare", now, nil)
@@ -58,10 +58,10 @@ func TestInvalidateDownstreamSucceededStages(t *testing.T) {
m.MarkStageSucceeded("publish", now, nil)
m.MarkStageSucceeded("notify", now, nil)
got := invalidateDownstreamSucceededStages(m, "polish", now.Add(1*time.Second))
got := invalidateDownstreamSucceededStagesWithReason(m, "polish", now.Add(1*time.Second), staleReasonChangedResult)
want := []string{"normalize", "trim", "extract", "render", "publish", "notify"}
if !reflect.DeepEqual(got, want) {
t.Fatalf("invalidateDownstreamSucceededStages() = %#v, want %#v", got, want)
t.Fatalf("invalidateDownstreamSucceededStagesWithReason() = %#v, want %#v", got, want)
}
for _, stageName := range want {
@@ -93,7 +93,7 @@ func TestExtractionPositionControlsForceInvalidation(t *testing.T) {
for _, name := range canonicalStageNames() {
m.MarkStageSucceeded(name, now, nil)
}
got := invalidateDownstreamSucceededStages(m, test.upstream, now.Add(time.Second))
got := invalidateDownstreamSucceededStagesWithReason(m, test.upstream, now.Add(time.Second), staleReasonForcedReplacement)
if !reflect.DeepEqual(got, test.want) {
t.Fatalf("invalidated = %#v, want %#v", got, test.want)
}

3
internal/app/run_stage_test.go
View File

@@ -59,6 +59,7 @@ func TestRunNoRemainingStagesRecordsSkippedStages(t *testing.T) {
for _, name := range []string{"prepare", "transcribe", "merge", "polish", "normalize", "trim", "render", "analyze", "publish", "notify"} {
m.MarkStageSucceeded(name, time.Date(2026, 5, 3, 10, 1, 0, 0, time.UTC), nil)
}
m.MarkStageSkipped("extract", time.Date(2026, 5, 3, 10, 1, 0, 0, time.UTC), "notarius_disabled")
if err := store.Save(context.Background(), manifestPath, m); err != nil {
t.Fatalf("save manifest: %v", err)
}
@@ -76,7 +77,7 @@ func TestRunNoRemainingStagesRecordsSkippedStages(t *testing.T) {
t.Fatalf("load migrated manifest: %v", err)
}
if loaded.Stages["extract"] == nil || loaded.Stages["extract"].Status != manifest.StatusSkipped {
t.Fatalf("legacy manifest extract record = %#v, want newly executed disabled skip", loaded.Stages["extract"])
t.Fatalf("extract record = %#v, want stable disabled skip", loaded.Stages["extract"])
}
}

18
internal/app/runner.go
View File

@@ -171,9 +171,7 @@ func executeStages(ctx context.Context, cfg *config.Config, stages []stage.Stage
for _, d := range decisions {
s := d.Stage
runNames = append(runNames, s.Name())
if d.Action == stageActionSkip && !stageSucceeded(m, s.Name()) {
d.Action = stageActionRun
}
d.Action = decideStageAction(s, m, opts.Force)
if d.Action == stageActionSkip {
if validator, ok := s.(stage.ResumeValidator); ok {
@@ -186,7 +184,7 @@ func executeStages(ctx context.Context, cfg *config.Config, stages []stage.Stage
staleAt := nowUTC()
m.MarkStageStale(s.Name(), staleAt, validation.Reason)
invalidateDownstreamSucceededStagesWithReason(
m, s.Name(), staleAt, "upstream stage result was not resumable",
m, s.Name(), staleAt, staleReasonNotResumable,
)
if err := env.ManifestStore.Save(ctx, manifestPath, m); err != nil {
return nil, fmt.Errorf("save manifest after resume validation for stage %q: %w", s.Name(), err)
@@ -209,6 +207,7 @@ func executeStages(ctx context.Context, cfg *config.Config, stages []stage.Stage
continue
}
executed = append(executed, s.Name())
priorOutcome := capturePriorStageOutcome(m, s.Name())
now := nowUTC()
runManifest.SetStageAction(s.Name(), manifest.RunStageActionRun, now)
@@ -217,6 +216,9 @@ func executeStages(ctx context.Context, cfg *config.Config, stages []stage.Stage
return nil, fmt.Errorf("save run manifest before stage %q: %w", s.Name(), err)
}
m.MarkStageRunning(s.Name(), now)
if opts.Force {
invalidateDownstreamSucceededStagesWithReason(m, s.Name(), now, staleReasonForcedReplacement)
}
env.Logger.Info("starting stage", "stage", s.Name())
if err := env.ManifestStore.Save(ctx, manifestPath, m); err != nil {
return nil, fmt.Errorf("save manifest before stage %q: %w", s.Name(), err)
@@ -230,6 +232,7 @@ func executeStages(ctx context.Context, cfg *config.Config, stages []stage.Stage
if err != nil {
failedAt := nowUTC()
m.MarkStageFailed(s.Name(), failedAt, err.Error())
invalidateDownstreamSucceededStagesWithReason(m, s.Name(), failedAt, staleReasonFailure)
if saveErr := env.ManifestStore.Save(ctx, manifestPath, m); saveErr != nil {
return nil, fmt.Errorf("stage %q failed (%v) and manifest save failed (%v)", s.Name(), err, saveErr)
}
@@ -247,6 +250,9 @@ func executeStages(ctx context.Context, cfg *config.Config, stages []stage.Stage
m.MarkStageSkipped(s.Name(), skippedAt, result.SkipReason)
clearStageResultDetails(m.Stages[s.Name()])
applyStageResultToManifest(m, s.Name(), result)
if !priorOutcome.isSameSelfSkip(result.SkipReason) {
invalidateDownstreamSucceededStagesWithReason(m, s.Name(), skippedAt, staleReasonSelfSkip)
}
if err := env.ManifestStore.Save(ctx, manifestPath, m); err != nil {
return nil, fmt.Errorf("save manifest after self-skip %q: %w", s.Name(), err)
}
@@ -265,8 +271,8 @@ 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 !priorOutcome.exists || priorOutcome.status != manifest.StatusSucceeded {
invalidateDownstreamSucceededStagesWithReason(m, s.Name(), succeededAt, staleReasonChangedResult)
}
if err := env.ManifestStore.Save(ctx, manifestPath, m); err != nil {