Test real planning and execution parity

internal/app/run_control_test.go

@@ -3,11 +3,15 @@ package app
 import (
 	"context"
 	"fmt"
+	"io"
+	"log/slog"
+	"path/filepath"
 	"reflect"
 	"strings"
 	"testing"
 	"time"
 
+	"gitea.maximumdirect.net/eric/narratio/internal/config"
 	"gitea.maximumdirect.net/eric/narratio/internal/manifest"
 	"gitea.maximumdirect.net/eric/narratio/internal/stage"
 )
@@ -51,7 +55,7 @@ func TestDecideStageExecution(t *testing.T) {
 
 func resumeValidation(value stage.ResumeValidation) *stage.ResumeValidation { return &value }
 
-func TestPlanningAndExecutionTransitionParity(t *testing.T) {
+func TestPlannedStageModelMatchesStageExecution(t *testing.T) {
 	tests := []struct {
 		name       string
 		seed       func(*manifest.Manifest, time.Time)
@@ -64,8 +68,8 @@ func TestPlanningAndExecutionTransitionParity(t *testing.T) {
 		{
 			name: "ordinary skip",
 			seed: func(m *manifest.Manifest, at time.Time) {
-				m.MarkStageSucceeded("prepare", at, nil)
+				m.MarkStageSucceeded("extract", at, nil)
-				m.MarkStageSucceeded("transcribe", at, nil)
+				m.MarkStageSucceeded("analyze", at, nil)
 			},
 			wantAction: stageActionSkip,
 			wantStatus: manifest.StatusSucceeded,
@@ -73,8 +77,8 @@ func TestPlanningAndExecutionTransitionParity(t *testing.T) {
 		{
 			name: "non-resumable success",
 			seed: func(m *manifest.Manifest, at time.Time) {
-				m.MarkStageSucceeded("prepare", at, nil)
+				m.MarkStageSucceeded("extract", at, nil)
-				m.MarkStageSucceeded("transcribe", at, nil)
+				m.MarkStageSucceeded("analyze", at, nil)
 			},
 			resume:     resumeValidation(stage.NonResumable("output changed")),
 			result:     &stage.StageResult{},
@@ -84,8 +88,8 @@ func TestPlanningAndExecutionTransitionParity(t *testing.T) {
 		{
 			name: "forced replacement",
 			seed: func(m *manifest.Manifest, at time.Time) {
-				m.MarkStageSucceeded("prepare", at, nil)
+				m.MarkStageSucceeded("extract", at, nil)
-				m.MarkStageSucceeded("transcribe", at, nil)
+				m.MarkStageSucceeded("analyze", at, nil)
 			},
 			force:      true,
 			result:     &stage.StageResult{},
@@ -95,12 +99,12 @@ func TestPlanningAndExecutionTransitionParity(t *testing.T) {
 		{
 			name: "unchanged self-skip",
 			seed: func(m *manifest.Manifest, at time.Time) {
-				m.MarkStageSkipped("prepare", at, "disabled")
+				m.MarkStageSkipped("extract", at, "notarius_disabled")
-				m.MarkStageSucceeded("transcribe", at, nil)
+				m.MarkStageSucceeded("analyze", at, nil)
 			},
 			result: &stage.StageResult{
 				Disposition: stage.StageDispositionSkipped,
-				SkipReason:  "disabled",
+				SkipReason:  "notarius_disabled",
 			},
 			wantAction: stageActionRun,
 			wantStatus: manifest.StatusSucceeded,
@@ -108,12 +112,12 @@ func TestPlanningAndExecutionTransitionParity(t *testing.T) {
 		{
 			name: "changed self-skip",
 			seed: func(m *manifest.Manifest, at time.Time) {
-				m.MarkStageSkipped("prepare", at, "old reason")
+				m.MarkStageSkipped("extract", at, "integration_disabled")
-				m.MarkStageSucceeded("transcribe", at, nil)
+				m.MarkStageSucceeded("analyze", at, nil)
 			},
 			result: &stage.StageResult{
 				Disposition: stage.StageDispositionSkipped,
-				SkipReason:  "new reason",
+				SkipReason:  "notarius_disabled",
 			},
 			wantAction: stageActionRun,
 			wantStatus: manifest.StatusStale,
@@ -121,8 +125,8 @@ func TestPlanningAndExecutionTransitionParity(t *testing.T) {
 		{
 			name: "success replacing non-success",
 			seed: func(m *manifest.Manifest, at time.Time) {
-				m.MarkStageFailed("prepare", at, "failed")
+				m.MarkStageFailed("extract", at, "failed")
-				m.MarkStageSucceeded("transcribe", at, nil)
+				m.MarkStageSucceeded("analyze", at, nil)
 			},
 			result:     &stage.StageResult{},
 			wantAction: stageActionRun,
@@ -133,62 +137,85 @@ func TestPlanningAndExecutionTransitionParity(t *testing.T) {
 	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)
+			cfg := testConfig(t)
+			cfg.Pipeline.Notarius = &config.NotariusConfig{Enabled: test.result == nil || test.result.Disposition != stage.StageDispositionSkipped}
 			planned := manifest.New("2026-05-03", at)
 			executed := manifest.New("2026-05-03", at)
+			planned.Campaign = "sample-campaign"
+			executed.Campaign = "sample-campaign"
 			test.seed(planned, at)
 			test.seed(executed, at)
 
-			plannedAction := applyParityTransition(t, planned, test.force, test.resume, test.result, false, at.Add(time.Second))
+			plannedAction := decideStageExecution(recordedStageStatus(planned, "extract"), test.force, test.resume)
-			executedAction := applyParityTransition(t, executed, test.force, test.resume, test.result, true, at.Add(time.Second))
+			if _, err := applyPreExecutionDecision(planned, "extract", plannedAction, at.Add(time.Second)); err != nil {
-			if plannedAction != test.wantAction || executedAction != test.wantAction {
+				t.Fatalf("apply planning decision: %v", err)
-				t.Fatalf("actions = planned %q, executed %q; want %q", plannedAction, executedAction, test.wantAction)
 			}
-			if got := planned.Stages["transcribe"].Status; got != test.wantStatus {
+			selected := transitionParityStage{validation: stage.Resumable(), result: test.result}
+			if test.resume != nil {
+				selected.validation = *test.resume
+			}
+			if plannedAction.action == stageActionRun {
+				if err := modelPlannedStageRun(planned, selected, cfg, plannedAction, nil); err != nil {
+					t.Fatalf("modelPlannedStageRun() error = %v", err)
+				}
+			}
+
+			runtime := &invocationRuntime{
+				ctx:           context.Background(),
+				cfg:           cfg,
+				opts:          RunOptions{Force: test.force},
+				identity:      invocationIdentity{RunID: "20260503T120000Z-a1b2c3d4"},
+				env:           &stage.Env{Config: cfg, Logger: slog.New(slog.NewTextHandler(io.Discard, nil))},
+				manifestStore: &manifest.LocalStore{},
+				manifestPath:  filepath.Join(t.TempDir(), "manifest.json"),
+				manifest:      executed,
+				dependencies: invocationDependencies{
+					now: func() time.Time { return at.Add(time.Second) },
+				}.normalized(),
+			}
+			executedOutcome, err := executeSelectedStage(runtime, selected)
+			if err != nil {
+				t.Fatalf("executeSelectedStage() error = %v", err)
+			}
+			executedAction := stageActionRun
+			if executedOutcome.skipped && !executedOutcome.executed {
+				executedAction = stageActionSkip
+			}
+			if plannedAction.action != test.wantAction || executedAction != test.wantAction {
+				t.Fatalf("actions = planned %q, executed %q; want %q", plannedAction.action, executedAction, test.wantAction)
+			}
+			if got := planned.Stages["analyze"].Status; got != test.wantStatus {
 				t.Fatalf("planned dependent status = %q, want %q", got, test.wantStatus)
 			}
-			if got := executed.Stages["transcribe"].Status; got != test.wantStatus {
+			if got := executed.Stages["analyze"].Status; got != test.wantStatus {
 				t.Fatalf("executed dependent status = %q, want %q", got, test.wantStatus)
 			}
-			plannedError := planned.Stages["transcribe"].Error
+			if got, want := executed.Stages["extract"].Status, planned.Stages["extract"].Status; got != want {
-			executedError := executed.Stages["transcribe"].Error
+				t.Fatalf("selected stage status = %q, want planned status %q", got, want)
-			if !reflect.DeepEqual(plannedError, executedError) {
+			}
+			plannedError := planned.Stages["analyze"].Error
+			executedError := executed.Stages["analyze"].Error
+			if (plannedError == nil) != (executedError == nil) ||
+				(plannedError != nil && (plannedError.Code != executedError.Code || plannedError.Message != executedError.Message)) {
 				t.Fatalf("dependent errors differ: planned %#v, executed %#v", plannedError, executedError)
 			}
 		})
 	}
 }
 
-func applyParityTransition(
+type transitionParityStage struct {
-	t *testing.T,
+	validation stage.ResumeValidation
-	m *manifest.Manifest,
+	result     *stage.StageResult
-	force bool,
+}
-	resume *stage.ResumeValidation,
+
-	result *stage.StageResult,
+func (transitionParityStage) Name() string { return "extract" }
-	actual bool,
+
-	at time.Time,
+func (s transitionParityStage) Run(context.Context, *stage.Env, *manifest.Manifest) (*stage.StageResult, error) {
-) stageAction {
+	return s.result, nil
-	t.Helper()
+}
-	decision := decideStageExecution(recordedStageStatus(m, "prepare"), force, resume)
+
-	if _, err := applyPreExecutionDecision(m, "prepare", decision, at); err != nil {
+func (s transitionParityStage) ValidateResume(context.Context, *stage.Env, *manifest.Manifest) (stage.ResumeValidation, error) {
-		t.Fatalf("apply pre-execution decision: %v", err)
+	return s.validation, nil
-	}
-	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) {