From ca1b694aa9f7b37e393cc31cb6969979d5160b95 Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Fri, 4 Sep 2026 01:01:46 +0000 Subject: [PATCH] Unify pre-execution stage decisions --- docs/roadmap/implementation.md | 2 +- internal/app/plan.go | 65 ++++------- internal/app/run_control.go | 88 +++++++++++++-- internal/app/run_control_test.go | 185 +++++++++++++++++++++++++++++-- internal/app/runner.go | 62 ++++------- 5 files changed, 297 insertions(+), 105 deletions(-) diff --git a/docs/roadmap/implementation.md b/docs/roadmap/implementation.md index a66fcfb..3376435 100644 --- a/docs/roadmap/implementation.md +++ b/docs/roadmap/implementation.md @@ -121,7 +121,7 @@ During every stage: ## Stage 1 — Unify Pre-Execution Decisions -**Status: Pending** +**Status: Completed** ### Goal diff --git a/internal/app/plan.go b/internal/app/plan.go index 598b5a9..d98e18f 100644 --- a/internal/app/plan.go +++ b/internal/app/plan.go @@ -73,26 +73,20 @@ func Plan(ctx context.Context, args []string, out io.Writer) error { if err != nil { return fmt.Errorf("plan: fingerprint semantic configuration for stage %q: %w", selectedStage.Name(), err) } - action := decideStageAction(selectedStage, model, request.Force) + decision := decideStageExecution(recordedStageStatus(model, selectedStage.Name()), request.Force, nil) var validation *stage.ResumeValidation - if action == stageActionSkip { + if decision.action == stageActionSkip { checked, validationErr := evaluateStageResume(ctx, selectedStage, stageEnv, model, semanticConfig) if validationErr != nil { return fmt.Errorf("plan: validate resume for stage %q: %w", selectedStage.Name(), validationErr) } validation = checked - if checked != nil && !checked.Resumable { - at := time.Now().UTC() - model.MarkStageStale(selectedStage.Name(), at, checked.Reason) - if _, invalidationErr := invalidateDependentSucceededStagesWithReason( - model, selectedStage.Name(), at, staleReasonNotResumable, - ); invalidationErr != nil { - return fmt.Errorf("plan: model resume invalidation for stage %q: %w", selectedStage.Name(), invalidationErr) - } - action = stageActionRun - } + decision = decideStageExecution(recordedStageStatus(model, selectedStage.Name()), request.Force, checked) } - if action == stageActionRun && selectedStage.Name() == "analyze" { + if _, err := applyPreExecutionDecision(model, selectedStage.Name(), decision, time.Now().UTC()); err != nil { + return fmt.Errorf("plan: model resume invalidation for stage %q: %w", selectedStage.Name(), err) + } + if decision.action == stageActionRun && selectedStage.Name() == "analyze" { if validator, ok := selectedStage.(stage.ResumeValidator); ok { checked, validationErr := validator.ValidateResume(ctx, stageEnv, model) if validationErr != nil { @@ -102,12 +96,12 @@ func Plan(ctx context.Context, args []string, out io.Writer) error { validation = &checked } } - if action == stageActionRun { + if decision.action == stageActionRun { runCount++ } else { skipCount++ } - if _, err := fmt.Fprintf(out, "%s: %s\n", selectedStage.Name(), action); err != nil { + if _, err := fmt.Fprintf(out, "%s: %s\n", selectedStage.Name(), decision.action); err != nil { return err } if validation != nil && validation.Analyze != nil { @@ -115,8 +109,8 @@ func Plan(ctx context.Context, args []string, out io.Writer) error { return err } } - if action == stageActionRun { - if err := modelPlannedStageRun(model, selectedStage, cfg, request.Force, semanticConfig); err != nil { + if decision.action == stageActionRun { + if err := modelPlannedStageRun(model, selectedStage, cfg, decision, semanticConfig); err != nil { return fmt.Errorf("plan: model stage %q: %w", selectedStage.Name(), err) } } @@ -149,40 +143,23 @@ func modelPlannedStageRun( model *manifest.Manifest, selectedStage stage.Stage, cfg *config.Config, - force bool, + decision preExecutionDecision, semanticConfig *manifest.SemanticConfigFingerprint, ) error { prior := capturePriorStageOutcome(model, selectedStage.Name()) at := time.Now().UTC() - model.MarkStageRunning(selectedStage.Name(), at) - if force { - if _, err := invalidateDependentSucceededStagesWithReason( - model, selectedStage.Name(), at, staleReasonForcedReplacement, - ); err != nil { - return err - } + if _, err := beginStageAttempt(model, selectedStage.Name(), decision, at); err != nil { + return err } if reason := plannedSelfSkipReason(selectedStage.Name(), cfg); reason != "" { - model.MarkStageSkipped(selectedStage.Name(), at, reason) - setSessionStageSemanticConfig(model, selectedStage.Name(), semanticConfig) - if !prior.isSameSelfSkip(reason) { - _, err := invalidateDependentSucceededStagesWithReason( - model, selectedStage.Name(), at, staleReasonSelfSkip, - ) - return err - } - return nil + _, err := applyCompletedStageOutcome(model, selectedStage.Name(), &stage.StageResult{ + Disposition: stage.StageDispositionSkipped, + SkipReason: reason, + }, nil, semanticConfig, prior, at) + return err } - model.MarkStageSucceeded(selectedStage.Name(), at, nil) - setSessionStageSemanticConfig(model, selectedStage.Name(), semanticConfig) - if !prior.exists || prior.status != manifest.StatusSucceeded { - if _, err := invalidateDependentSucceededStagesWithReason( - model, selectedStage.Name(), at, staleReasonChangedResult, - ); err != nil { - return err - } - } - return nil + _, err := applyCompletedStageOutcome(model, selectedStage.Name(), &stage.StageResult{}, nil, semanticConfig, prior, at) + return err } func plannedSelfSkipReason(stageName string, cfg *config.Config) string { diff --git a/internal/app/run_control.go b/internal/app/run_control.go index bb92c04..000f004 100644 --- a/internal/app/run_control.go +++ b/internal/app/run_control.go @@ -34,21 +34,89 @@ type priorStageOutcome struct { outputs int } -func decideStageAction(s stage.Stage, m *manifest.Manifest, force bool) stageAction { - // A succeeded aggregate record is the initial skip candidate. The runner and - // planner then let stage-owned resume validation refine that decision. - if !force && stageSucceeded(m, s.Name()) { - return stageActionSkip - } - return stageActionRun +type preExecutionDecision struct { + action stageAction + markSelectedStale bool + selectedStaleReason string + invalidationReason string + invalidateWhenStarted bool } -func stageSucceeded(m *manifest.Manifest, name string) bool { +func decideStageExecution(status manifest.StageStatus, force bool, resume *stage.ResumeValidation) preExecutionDecision { + if force { + return preExecutionDecision{ + action: stageActionRun, + invalidationReason: staleReasonForcedReplacement, + invalidateWhenStarted: true, + } + } + if status != manifest.StatusSucceeded { + return preExecutionDecision{action: stageActionRun} + } + if resume != nil && !resume.Resumable { + return preExecutionDecision{ + action: stageActionRun, + markSelectedStale: true, + selectedStaleReason: resume.Reason, + invalidationReason: staleReasonNotResumable, + } + } + return preExecutionDecision{action: stageActionSkip} +} + +func recordedStageStatus(m *manifest.Manifest, name string) manifest.StageStatus { if m == nil || m.Stages == nil { - return false + return "" } sr := m.Stages[name] - return sr != nil && sr.Status == manifest.StatusSucceeded + if sr == nil { + return "" + } + return sr.Status +} + +func applyPreExecutionDecision(m *manifest.Manifest, name string, decision preExecutionDecision, at time.Time) ([]string, error) { + if !decision.markSelectedStale { + return nil, nil + } + m.MarkStageStale(name, at, decision.selectedStaleReason) + return invalidateDependentSucceededStagesWithReason(m, name, at, decision.invalidationReason) +} + +func beginStageAttempt(m *manifest.Manifest, name string, decision preExecutionDecision, at time.Time) ([]string, error) { + m.MarkStageRunning(name, at) + if !decision.invalidateWhenStarted { + return nil, nil + } + return invalidateDependentSucceededStagesWithReason(m, name, at, decision.invalidationReason) +} + +func applyCompletedStageOutcome( + m *manifest.Manifest, + name string, + result *stage.StageResult, + outputs []manifest.ArtifactRecord, + semanticConfig *manifest.SemanticConfigFingerprint, + prior priorStageOutcome, + at time.Time, +) ([]string, error) { + if result != nil && result.Disposition == stage.StageDispositionSkipped { + m.MarkStageSkipped(name, at, result.SkipReason) + applyStageResultToManifest(m, name, result) + setSessionStageSemanticConfig(m, name, semanticConfig) + if prior.isSameSelfSkip(result.SkipReason) { + return nil, nil + } + return invalidateDependentSucceededStagesWithReason(m, name, at, staleReasonSelfSkip) + } + + m.MarkStageSucceeded(name, at, outputs) + applyStageResultToManifest(m, name, result) + setSessionStageSemanticConfig(m, name, semanticConfig) + if prior.exists && prior.status == manifest.StatusSucceeded { + return nil, nil + } + return invalidateDependentSucceededStagesWithReason(m, name, at, staleReasonChangedResult) } func capturePriorStageOutcome(m *manifest.Manifest, name string) priorStageOutcome { diff --git a/internal/app/run_control_test.go b/internal/app/run_control_test.go index 0e0e4dc..68c4fd7 100644 --- a/internal/app/run_control_test.go +++ b/internal/app/run_control_test.go @@ -2,6 +2,7 @@ package app import ( "context" + "fmt" "reflect" "strings" "testing" @@ -11,23 +12,185 @@ import ( "gitea.maximumdirect.net/eric/narratio/internal/stage" ) -func TestDecideStageAction(t *testing.T) { - stages := BuildFullPlan()[:2] - m := manifest.New("2026-05-03", time.Now().UTC()) - m.MarkStageSucceeded("prepare", time.Now().UTC(), nil) - - if got := decideStageAction(stages[0], m, false); got != stageActionSkip { - t.Fatalf("prepare action = %q, want %q", got, stageActionSkip) +func TestDecideStageExecution(t *testing.T) { + statuses := []manifest.StageStatus{"", manifest.StatusSucceeded, manifest.StatusFailed, manifest.StatusStale, manifest.StatusSkipped} + validations := []struct { + name string + value *stage.ResumeValidation + }{ + {name: "none"}, + {name: "resumable", value: resumeValidation(stage.Resumable())}, + {name: "not resumable", value: resumeValidation(stage.NonResumable("output changed"))}, } - if got := decideStageAction(stages[1], m, false); got != stageActionRun { - t.Fatalf("transcribe action = %q, want %q", got, stageActionRun) + for _, status := range statuses { + for _, force := range []bool{false, true} { + for _, validation := range validations { + name := fmt.Sprintf("status=%s/force=%t/resume=%s", status, force, validation.name) + t.Run(name, func(t *testing.T) { + got := decideStageExecution(status, force, validation.value) + want := preExecutionDecision{action: stageActionRun} + switch { + case force: + want.invalidationReason = staleReasonForcedReplacement + want.invalidateWhenStarted = true + case status == manifest.StatusSucceeded && validation.value != nil && !validation.value.Resumable: + want.markSelectedStale = true + want.selectedStaleReason = "output changed" + want.invalidationReason = staleReasonNotResumable + case status == manifest.StatusSucceeded: + want.action = stageActionSkip + } + if !reflect.DeepEqual(got, want) { + t.Fatalf("decision = %#v, want %#v", got, want) + } + }) + } + } + } +} + +func resumeValidation(value stage.ResumeValidation) *stage.ResumeValidation { return &value } + +func TestPlanningAndExecutionTransitionParity(t *testing.T) { + tests := []struct { + name string + seed func(*manifest.Manifest, time.Time) + force bool + resume *stage.ResumeValidation + result *stage.StageResult + wantAction stageAction + wantStatus manifest.StageStatus + }{ + { + name: "ordinary skip", + seed: func(m *manifest.Manifest, at time.Time) { + m.MarkStageSucceeded("prepare", at, nil) + m.MarkStageSucceeded("transcribe", at, nil) + }, + wantAction: stageActionSkip, + wantStatus: manifest.StatusSucceeded, + }, + { + name: "non-resumable success", + seed: func(m *manifest.Manifest, at time.Time) { + m.MarkStageSucceeded("prepare", at, nil) + m.MarkStageSucceeded("transcribe", at, nil) + }, + resume: resumeValidation(stage.NonResumable("output changed")), + result: &stage.StageResult{}, + wantAction: stageActionRun, + wantStatus: manifest.StatusStale, + }, + { + name: "forced replacement", + seed: func(m *manifest.Manifest, at time.Time) { + m.MarkStageSucceeded("prepare", at, nil) + m.MarkStageSucceeded("transcribe", at, nil) + }, + force: true, + result: &stage.StageResult{}, + wantAction: stageActionRun, + wantStatus: manifest.StatusStale, + }, + { + name: "unchanged self-skip", + seed: func(m *manifest.Manifest, at time.Time) { + m.MarkStageSkipped("prepare", at, "disabled") + m.MarkStageSucceeded("transcribe", at, nil) + }, + result: &stage.StageResult{ + Disposition: stage.StageDispositionSkipped, + SkipReason: "disabled", + }, + wantAction: stageActionRun, + wantStatus: manifest.StatusSucceeded, + }, + { + name: "changed self-skip", + seed: func(m *manifest.Manifest, at time.Time) { + m.MarkStageSkipped("prepare", at, "old reason") + m.MarkStageSucceeded("transcribe", at, nil) + }, + result: &stage.StageResult{ + Disposition: stage.StageDispositionSkipped, + SkipReason: "new reason", + }, + wantAction: stageActionRun, + wantStatus: manifest.StatusStale, + }, + { + name: "success replacing non-success", + seed: func(m *manifest.Manifest, at time.Time) { + m.MarkStageFailed("prepare", at, "failed") + m.MarkStageSucceeded("transcribe", at, nil) + }, + result: &stage.StageResult{}, + wantAction: stageActionRun, + wantStatus: manifest.StatusStale, + }, } - if got := decideStageAction(stages[0], m, true); got != stageActionRun { - t.Fatalf("forced prepare action = %q, want %q", got, stageActionRun) + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + at := time.Date(2026, time.May, 3, 12, 0, 0, 0, time.UTC) + planned := manifest.New("2026-05-03", at) + executed := manifest.New("2026-05-03", at) + test.seed(planned, at) + test.seed(executed, at) + + plannedAction := applyParityTransition(t, planned, test.force, test.resume, test.result, false, at.Add(time.Second)) + executedAction := applyParityTransition(t, executed, test.force, test.resume, test.result, true, at.Add(time.Second)) + if plannedAction != test.wantAction || executedAction != test.wantAction { + t.Fatalf("actions = planned %q, executed %q; want %q", plannedAction, executedAction, test.wantAction) + } + if got := planned.Stages["transcribe"].Status; got != test.wantStatus { + t.Fatalf("planned dependent status = %q, want %q", got, test.wantStatus) + } + if got := executed.Stages["transcribe"].Status; got != test.wantStatus { + t.Fatalf("executed dependent status = %q, want %q", got, test.wantStatus) + } + plannedError := planned.Stages["transcribe"].Error + executedError := executed.Stages["transcribe"].Error + if !reflect.DeepEqual(plannedError, executedError) { + t.Fatalf("dependent errors differ: planned %#v, executed %#v", plannedError, executedError) + } + }) } } +func applyParityTransition( + t *testing.T, + m *manifest.Manifest, + force bool, + resume *stage.ResumeValidation, + result *stage.StageResult, + actual bool, + at time.Time, +) stageAction { + t.Helper() + decision := decideStageExecution(recordedStageStatus(m, "prepare"), force, resume) + if _, err := applyPreExecutionDecision(m, "prepare", decision, at); err != nil { + t.Fatalf("apply pre-execution decision: %v", err) + } + if decision.action == stageActionSkip { + return decision.action + } + prior := capturePriorStageOutcome(m, "prepare") + if _, err := beginStageAttempt(m, "prepare", decision, at); err != nil { + t.Fatalf("begin attempt: %v", err) + } + if result != nil { + modeled := *result + if actual { + modeled.Logs = []string{"run.log"} + } + if _, err := applyCompletedStageOutcome(m, "prepare", &modeled, nil, nil, prior, at); err != nil { + t.Fatalf("apply completed outcome: %v", err) + } + } + return decision.action +} + func TestInvalidationDependents(t *testing.T) { tests := []struct { stage string diff --git a/internal/app/runner.go b/internal/app/runner.go index e05288c..bc189e1 100644 --- a/internal/app/runner.go +++ b/internal/app/runner.go @@ -244,9 +244,9 @@ func executePlan(ctx context.Context, cfg *config.Config, plan BoundedPlan, opts fmt.Errorf("fingerprint semantic configuration for stage %q: %w", s.Name(), err), ) } - action := decideStageAction(s, m, opts.Force) + decision := decideStageExecution(recordedStageStatus(m, s.Name()), opts.Force, nil) - if action == stageActionSkip { + if decision.action == stageActionSkip { validation, err := evaluateStageResume(ctx, s, stageEnv, m, semanticConfig) if err != nil { return nil, persistSessionTerminalFailure( @@ -254,12 +254,10 @@ func executePlan(ctx context.Context, cfg *config.Config, plan BoundedPlan, opts fmt.Errorf("validate resume for stage %q: %w", s.Name(), err), ) } - if validation != nil && !validation.Resumable { + decision = decideStageExecution(recordedStageStatus(m, s.Name()), opts.Force, validation) + if decision.markSelectedStale { staleAt := nowUTC() - m.MarkStageStale(s.Name(), staleAt, validation.Reason) - if _, err := invalidateDependentSucceededStagesWithReason( - m, s.Name(), staleAt, staleReasonNotResumable, - ); err != nil { + if _, err := applyPreExecutionDecision(m, s.Name(), decision, staleAt); err != nil { return nil, persistSessionTerminalFailure( ctx, env.ManifestStore, manifestPath, m, fmt.Errorf("invalidate dependents after resume validation for stage %q: %w", s.Name(), err), @@ -271,12 +269,11 @@ func executePlan(ctx context.Context, cfg *config.Config, plan BoundedPlan, opts fmt.Errorf("save manifest after resume validation for stage %q: %w", s.Name(), err), ) } - env.Logger.Info("stage result is not resumable", "stage", s.Name(), "reason", validation.Reason) - action = stageActionRun + env.Logger.Info("stage result is not resumable", "stage", s.Name(), "reason", decision.selectedStaleReason) } } - if action == stageActionSkip { + if decision.action == stageActionSkip { skipped = append(skipped, s.Name()) env.Logger.Info("skipping stage", "stage", s.Name(), "reason", "already_succeeded", "force", opts.Force) continue @@ -286,14 +283,11 @@ func executePlan(ctx context.Context, cfg *config.Config, plan BoundedPlan, opts priorAnalyzeState := captureAnalyzeState(m, s.Name()) now := nowUTC() - m.MarkStageRunning(s.Name(), now) - if opts.Force { - if _, err := invalidateDependentSucceededStagesWithReason(m, s.Name(), now, staleReasonForcedReplacement); err != nil { - return nil, persistSessionTerminalFailure( - ctx, env.ManifestStore, manifestPath, m, - fmt.Errorf("invalidate dependents before forced stage %q: %w", s.Name(), err), - ) - } + if _, err := beginStageAttempt(m, s.Name(), decision, now); err != nil { + return nil, persistSessionTerminalFailure( + ctx, env.ManifestStore, manifestPath, m, + fmt.Errorf("invalidate dependents before forced stage %q: %w", s.Name(), err), + ) } env.Logger.Info("starting stage", "stage", s.Name()) if err := env.ManifestStore.Save(ctx, manifestPath, m); err != nil { @@ -340,16 +334,11 @@ func executePlan(ctx context.Context, cfg *config.Config, plan BoundedPlan, opts if result != nil && result.Disposition == stage.StageDispositionSkipped { skipped = append(skipped, s.Name()) skippedAt := nowUTC() - m.MarkStageSkipped(s.Name(), skippedAt, result.SkipReason) - applyStageResultToManifest(m, s.Name(), result) - setSessionStageSemanticConfig(m, s.Name(), semanticConfig) - if !priorOutcome.isSameSelfSkip(result.SkipReason) { - if _, err := invalidateDependentSucceededStagesWithReason(m, s.Name(), skippedAt, staleReasonSelfSkip); err != nil { - return nil, persistSessionTerminalFailure( - ctx, env.ManifestStore, manifestPath, m, - fmt.Errorf("invalidate dependents after stage %q self-skip: %w", s.Name(), err), - ) - } + if _, err := applyCompletedStageOutcome(m, s.Name(), result, nil, semanticConfig, priorOutcome, skippedAt); err != nil { + return nil, persistSessionTerminalFailure( + ctx, env.ManifestStore, manifestPath, m, + fmt.Errorf("invalidate dependents after stage %q self-skip: %w", s.Name(), err), + ) } if err := env.ManifestStore.Save(ctx, manifestPath, m); err != nil { return nil, persistSessionTerminalFailure( @@ -367,18 +356,13 @@ func executePlan(ctx context.Context, cfg *config.Config, plan BoundedPlan, opts sessionOutputs = analyzeProjectionOutputs(analyzeProjection.session) } succeededAt := nowUTC() - m.MarkStageSucceeded(s.Name(), succeededAt, sessionOutputs) - applyAnalyzeProjection(m, analyzeProjection) - applyStageResultToManifest(m, s.Name(), result) - setSessionStageSemanticConfig(m, s.Name(), semanticConfig) - if !priorOutcome.exists || priorOutcome.status != manifest.StatusSucceeded { - if _, err := invalidateDependentSucceededStagesWithReason(m, s.Name(), succeededAt, staleReasonChangedResult); err != nil { - return nil, persistSessionTerminalFailure( - ctx, env.ManifestStore, manifestPath, m, - fmt.Errorf("invalidate dependents after changed stage %q result: %w", s.Name(), err), - ) - } + if _, err := applyCompletedStageOutcome(m, s.Name(), result, sessionOutputs, semanticConfig, priorOutcome, succeededAt); err != nil { + return nil, persistSessionTerminalFailure( + ctx, env.ManifestStore, manifestPath, m, + fmt.Errorf("invalidate dependents after changed stage %q result: %w", s.Name(), err), + ) } + applyAnalyzeProjection(m, analyzeProjection) if err := env.ManifestStore.Save(ctx, manifestPath, m); err != nil { operationErr := fmt.Errorf("save manifest after stage %q: %w", s.Name(), err)