Add recompute recovery acceptance coverage

internal/cli/recompute_execution_contract_test.go

@@ -0,0 +1,323 @@
+package cli
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"fmt"
+	"os"
+	"path/filepath"
+	"reflect"
+	"sort"
+	"strings"
+	"sync"
+	"testing"
+
+	"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
+	"gitea.maximumdirect.net/eric/notarius/internal/framework/pipeline"
+)
+
+func TestRecomputeStepRecoversThroughFilesystemCheckpoints(t *testing.T) {
+	tests := []struct {
+		name             string
+		invalidateOutput bool
+		wantCode         int
+	}{
+		{name: "accepted producer is hydrated", wantCode: 0},
+		{name: "invalid producer stops dependents", invalidateOutput: true, wantCode: 1},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			roots := newRecomputeTestRoots(t)
+			harness := newRecomputeTestHarness()
+			fresh := runRecomputeCommand(roots, harness.options(), false)
+			if fresh.code != 0 {
+				t.Fatalf("fresh run code=%d stderr=%q", fresh.code, fresh.stderr)
+			}
+			removeCheckpointLaneStage(t, roots.checkpoints, "extract", "first", "producer")
+			removeCheckpointLaneStage(t, roots.checkpoints, "merge", "first", "producer")
+			if tt.invalidateOutput {
+				path := findCheckpointFile(t, roots.checkpoints, "normalize", "first", "producer", "output.json")
+				if err := os.WriteFile(path, []byte("{"), 0o600); err != nil {
+					t.Fatal(err)
+				}
+			}
+			harness.resetCalls()
+
+			resumed := runRecomputeCommand(roots, harness.options(), true)
+			if resumed.code != tt.wantCode {
+				t.Fatalf("resumed code=%d stdout=%q stderr=%q", resumed.code, resumed.stdout, resumed.stderr)
+			}
+			events := readLatestCheckpointEvents(t, roots.debug)
+			if tt.invalidateOutput {
+				if harness.callsFor("test/extract/middle") != 0 || harness.callsFor("test/extract/dependent") != 0 {
+					t.Fatalf("dependent calls after invalid producer = %#v", harness.callsSnapshot())
+				}
+				if !strings.Contains(resumed.stderr, string(pipeline.CheckpointReasonDecodeFailed)) {
+					t.Fatalf("stderr=%q, want stable checkpoint reason", resumed.stderr)
+				}
+				assertNormalizeDecisionSequence(t, events, []checkpointDecisionExpectation{{"first", "producer", pipeline.CheckpointDecisionExecuted, pipeline.CheckpointReasonDecodeFailed}})
+				return
+			}
+
+			if got := harness.callsSnapshot(); !reflect.DeepEqual(got, map[string]int{"test/extract/dependent": 1, "test/extract/middle": 1}) {
+				t.Fatalf("resumed extractor calls = %#v", got)
+			}
+			outputPath := filepath.Join(latestChildDir(t, roots.output), "result.json")
+			data, err := os.ReadFile(outputPath)
+			if err != nil {
+				t.Fatal(err)
+			}
+			if string(data) != "[\"producer\",\"unrelated\",\"middle\",\"dependent\"]\n" {
+				t.Fatalf("ordered output = %q", data)
+			}
+			assertNormalizeDecisionSequence(t, events, []checkpointDecisionExpectation{
+				{"first", "producer", pipeline.CheckpointDecisionReused, pipeline.CheckpointReasonAcceptedArtifactReused},
+				{"first", "unrelated", pipeline.CheckpointDecisionReused, pipeline.CheckpointReasonReused},
+				{"second", "middle", pipeline.CheckpointDecisionForcedRecompute, pipeline.CheckpointReasonRecomputeStep},
+				{"third", "dependent", pipeline.CheckpointDecisionForcedRecompute, pipeline.CheckpointReasonRecomputeStep},
+			})
+		})
+	}
+}
+
+type checkpointDecisionExpectation struct {
+	step, lane string
+	action     pipeline.CheckpointDecisionCategory
+	reason     pipeline.CheckpointReasonCode
+}
+
+func assertNormalizeDecisionSequence(t *testing.T, events []pipeline.CheckpointEvent, want []checkpointDecisionExpectation) {
+	t.Helper()
+	var got []checkpointDecisionExpectation
+	for _, event := range events {
+		if event.Stage == string(pipeline.StageNormalize) {
+			got = append(got, checkpointDecisionExpectation{event.StepID, event.LaneID, event.Action, event.ReasonCode})
+		}
+	}
+	if !reflect.DeepEqual(got, want) {
+		t.Fatalf("normalize decisions = %#v, want %#v", got, want)
+	}
+}
+
+type recomputeTestHarness struct {
+	base  *stateTestHarness
+	mu    sync.Mutex
+	calls map[string]int
+}
+
+func newRecomputeTestHarness() *recomputeTestHarness {
+	return &recomputeTestHarness{base: newStateTestHarness(), calls: make(map[string]int)}
+}
+
+func (h *recomputeTestHarness) options() Options {
+	opts := h.base.options()
+	for _, key := range []string{"test/extract/producer", "test/extract/unrelated", "test/extract/middle", "test/extract/dependent"} {
+		moduleKey := key
+		spec := pipeline.ModuleSpec{
+			Key: moduleKey, Stage: pipeline.StageExtract, Requires: []string{"chunks"}, Provides: []string{"artifact"}, ArtifactKind: stateTestArtifactKind,
+			ReferenceSlots: []contracts.ReferenceSlot{{Name: "upstream", AcceptedMediaTypes: []string{"application/json"}, AcceptedArtifactKinds: []contracts.ArtifactKind{stateTestArtifactKind}}},
+		}
+		if err := pipeline.RegisterExtractor(opts.Registries.Extractors, spec, func() (contracts.Extractor[stateTestArtifact], error) {
+			return recomputeTestExtractor{key: moduleKey, harness: h}, nil
+		}); err != nil {
+			panic(err)
+		}
+	}
+	if err := opts.Registries.Outputs.RegisterWithSpec(pipeline.ModuleSpec{Key: "test/recompute-output", Stage: pipeline.StageOutput, Requires: []string{"normalized"}, Provides: []string{"output"}}, func() (contracts.OutputEncoder, error) {
+		return recomputeTestOutput{}, nil
+	}); err != nil {
+		panic(err)
+	}
+	opts.Catalog = catalogFromRegistries(opts.Registries)
+	return opts
+}
+
+func (h *recomputeTestHarness) record(key string) {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+	h.calls[key]++
+}
+
+func (h *recomputeTestHarness) resetCalls() {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+	h.calls = make(map[string]int)
+}
+
+func (h *recomputeTestHarness) callsFor(key string) int {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+	return h.calls[key]
+}
+
+func (h *recomputeTestHarness) callsSnapshot() map[string]int {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+	result := make(map[string]int, len(h.calls))
+	for key, value := range h.calls {
+		result[key] = value
+	}
+	return result
+}
+
+type recomputeTestExtractor struct {
+	key     string
+	harness *recomputeTestHarness
+}
+
+func (e recomputeTestExtractor) Key() string { return e.key }
+func (e recomputeTestExtractor) ReferenceSlots() []contracts.ReferenceSlot {
+	return []contracts.ReferenceSlot{{Name: "upstream", AcceptedMediaTypes: []string{"application/json"}, AcceptedArtifactKinds: []contracts.ArtifactKind{stateTestArtifactKind}}}
+}
+func (e recomputeTestExtractor) Extract(context.Context, contracts.TypedExtractionRequest) (contracts.TypedExtractionResult[stateTestArtifact], error) {
+	e.harness.record(e.key)
+	return contracts.TypedExtractionResult[stateTestArtifact]{Value: stateTestArtifact{Value: e.key}}, nil
+}
+
+type recomputeTestOutput struct{}
+
+func (recomputeTestOutput) Key() string { return "test/recompute-output" }
+func (recomputeTestOutput) Encode(_ context.Context, req contracts.OutputRequest) (contracts.OutputResult, error) {
+	lanes := make([]string, 0, len(req.NormalizeOutputs))
+	for _, output := range req.NormalizeOutputs {
+		lanes = append(lanes, output.LaneID)
+	}
+	data, err := json.Marshal(lanes)
+	if err != nil {
+		return contracts.OutputResult{}, err
+	}
+	return contracts.OutputResult{Files: []contracts.OutputFile{{Name: "result.json", Bytes: append(data, '\n')}}}, nil
+}
+
+func newRecomputeTestRoots(t *testing.T) stateTestRoots {
+	t.Helper()
+	roots := newStateTestRoots(t)
+	config := fmt.Sprintf(`version: 3
+output:
+  directory: %q
+cache:
+  chunk_plans:
+    directory: %q
+    mode: bypass
+  checkpoints:
+    enabled: true
+    directory: %q
+debug:
+  directory: %q
+pipelines:
+  sample:
+    input: test/input
+    chunk: test/chunk
+    steps:
+      - id: first
+        artifacts:
+          producer:
+            extract: test/extract/producer
+            merge: test/merge
+            normalize: test/normalize
+          unrelated:
+            extract: test/extract/unrelated
+            merge: test/merge
+            normalize: test/normalize
+      - id: second
+        references:
+          upstream:
+            artifact:
+              step: first
+              lane: producer
+        artifacts:
+          middle:
+            extract: test/extract/middle
+            merge: test/merge
+            normalize: test/normalize
+      - id: third
+        references:
+          upstream:
+            artifact:
+              step: second
+              lane: middle
+        artifacts:
+          dependent:
+            extract: test/extract/dependent
+            merge: test/merge
+            normalize: test/normalize
+    output: test/recompute-output
+`, roots.output, roots.plans, roots.checkpoints, roots.debug)
+	if err := os.WriteFile(roots.config, []byte(config), 0o600); err != nil {
+		t.Fatal(err)
+	}
+	return roots
+}
+
+func runRecomputeCommand(roots stateTestRoots, opts Options, recompute bool) stateTestResult {
+	args := []string{"run", "sample", "--config", roots.config, "--input", roots.input, "--chunk_cache", "bypass"}
+	if recompute {
+		args = append(args, "--resume", "--recompute-step", "second", "--debug")
+	}
+	var stdout, stderr bytes.Buffer
+	return stateTestResult{code: RunWithOptions(args, &stdout, &stderr, opts), stdout: stdout.String(), stderr: stderr.String()}
+}
+
+func removeCheckpointLaneStage(t *testing.T, root, stage, step, lane string) {
+	t.Helper()
+	dir := filepath.Dir(findCheckpointFile(t, root, stage, step, lane, "manifest.json"))
+	if err := os.RemoveAll(dir); err != nil {
+		t.Fatal(err)
+	}
+}
+
+func findCheckpointFile(t *testing.T, root, stage, step, lane, name string) string {
+	t.Helper()
+	want := filepath.Join(stage, step, lane, name)
+	var matches []string
+	err := filepath.WalkDir(root, func(path string, entry os.DirEntry, err error) error {
+		if err != nil {
+			return err
+		}
+		if !entry.IsDir() && strings.HasSuffix(path, want) {
+			matches = append(matches, path)
+		}
+		return nil
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(matches) != 1 {
+		t.Fatalf("checkpoint files ending in %q = %v", want, matches)
+	}
+	return matches[0]
+}
+
+func latestChildDir(t *testing.T, root string) string {
+	t.Helper()
+	entries, err := os.ReadDir(root)
+	if err != nil {
+		t.Fatal(err)
+	}
+	var dirs []string
+	for _, entry := range entries {
+		if entry.IsDir() {
+			dirs = append(dirs, filepath.Join(root, entry.Name()))
+		}
+	}
+	if len(dirs) == 0 {
+		t.Fatal("no child directory")
+	}
+	sort.Strings(dirs)
+	return dirs[len(dirs)-1]
+}
+
+func readLatestCheckpointEvents(t *testing.T, root string) []pipeline.CheckpointEvent {
+	t.Helper()
+	var events []pipeline.CheckpointEvent
+	data, err := os.ReadFile(filepath.Join(latestChildDir(t, root), "summary", "checkpoint-events.json"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := json.Unmarshal(data, &events); err != nil {
+		t.Fatal(err)
+	}
+	return events
+}

internal/cli/run_contract_test.go

@@ -60,6 +60,56 @@ func TestRunControlsRejectSyntaxWithoutAllocatingState(t *testing.T) {
 	}
 }
 
+func TestRecomputeStepCLIContract(t *testing.T) {
+	tests := []struct {
+		name       string
+		configure  func(*testing.T, stateTestRoots)
+		flags      []string
+		wantCode   int
+		wantOutput string
+		wantError  string
+	}{
+		{
+			name: "explicit step",
+			configure: func(t *testing.T, roots stateTestRoots) {
+				replaceStateTestConfigLine(t, roots.config, "    artifacts:\n      items:\n        extract: test/extract\n        merge: test/merge\n        normalize: test/normalize\n", "    steps:\n      - id: chosen\n        artifacts:\n          items:\n            extract: test/extract\n            merge: test/merge\n            normalize: test/normalize\n")
+			},
+			flags:      []string{"--resume", "--recompute-step", "chosen"},
+			wantCode:   0,
+			wantOutput: "outputs=1",
+		},
+		{name: "implicit default step", flags: []string{"--resume", "--recompute-step", "default"}, wantCode: 0, wantOutput: "outputs=1"},
+		{name: "repeated flag", flags: []string{"--resume", "--recompute-step", "default", "--recompute-step", "default"}, wantCode: 2, wantError: "specified only once"},
+		{name: "empty step", flags: []string{"--resume", "--recompute-step", ""}, wantCode: 2, wantError: "must not be empty"},
+		{name: "unknown step", flags: []string{"--resume", "--recompute-step", "missing"}, wantCode: 1, wantError: "unknown pipeline step"},
+		{name: "without resume", flags: []string{"--recompute-step", "default"}, wantCode: 2, wantError: "requires --resume"},
+		{
+			name: "checkpoint recording disabled",
+			configure: func(t *testing.T, roots stateTestRoots) {
+				replaceStateTestConfigLine(t, roots.config, "    enabled: true\n", "    enabled: false\n")
+			},
+			flags: []string{"--resume", "--recompute-step", "default"}, wantCode: 1, wantError: "cache.checkpoints.enabled",
+		},
+		{name: "with only", flags: []string{"--resume", "--recompute-step", "default", "--only", "items"}, wantCode: 2, wantError: "cannot be combined with --only"},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			roots := newStateTestRoots(t)
+			if tt.configure != nil {
+				tt.configure(t, roots)
+			}
+			args := []string{"run", "sample", "--config", roots.config, "--input", roots.input, "--chunk_cache", "bypass"}
+			args = append(args, tt.flags...)
+			var stdout, stderr bytes.Buffer
+			code := RunWithOptions(args, &stdout, &stderr, newStateTestHarness().options())
+			if code != tt.wantCode || (tt.wantOutput != "" && !strings.Contains(stdout.String(), tt.wantOutput)) || (tt.wantError != "" && !strings.Contains(stderr.String(), tt.wantError)) {
+				t.Fatalf("code=%d stdout=%q stderr=%q", code, stdout.String(), stderr.String())
+			}
+		})
+	}
+}
+
 func TestRunValidFailuresClassifyAndReportDebug(t *testing.T) {
 	tests := []struct {
 		name      string

internal/framework/pipeline/runner_accepted_checkpoint_test.go

@@ -14,10 +14,7 @@ type acceptedCheckpointLoader struct {
 	CheckpointLoader
 	accepted         map[string]NormalizeCheckpoint
 	acceptedDecision map[string]CheckpointDecision
-	acceptedCalls    map[string]int
+	extractDeps      map[string][]CheckpointFingerprint
-	extractDeps      [][]CheckpointFingerprint
-	mergeDeps        [][]CheckpointFingerprint
-	normalizeDeps    [][]CheckpointFingerprint
 }
 
 func newAcceptedCheckpointLoader() *acceptedCheckpointLoader {
@@ -25,26 +22,23 @@ func newAcceptedCheckpointLoader() *acceptedCheckpointLoader {
 		CheckpointLoader: NoopCheckpointLoader(),
 		accepted:         make(map[string]NormalizeCheckpoint),
 		acceptedDecision: make(map[string]CheckpointDecision),
-		acceptedCalls:    make(map[string]int),
+		extractDeps:      make(map[string][]CheckpointFingerprint),
 	}
 }
 
 func (l *acceptedCheckpointLoader) Enabled() bool { return true }
 func (l *acceptedCheckpointLoader) AcceptedNormalize(stepID, laneID, _ string) (NormalizeCheckpoint, CheckpointDecision) {
 	key := CheckpointLaneKey(stepID, laneID)
-	l.acceptedCalls[key]++
 	return l.accepted[key], l.acceptedDecision[key]
 }
-func (l *acceptedCheckpointLoader) Extract(_ string, _ string, dependencies []CheckpointFingerprint) (ExtractCheckpoint, CheckpointDecision) {
+func (l *acceptedCheckpointLoader) Extract(laneID string, _ string, dependencies []CheckpointFingerprint) (ExtractCheckpoint, CheckpointDecision) {
-	l.extractDeps = append(l.extractDeps, append([]CheckpointFingerprint(nil), dependencies...))
+	l.extractDeps[laneID] = append([]CheckpointFingerprint(nil), dependencies...)
 	return ExtractCheckpoint{}, NewCheckpointDecision(CheckpointDecisionExecuted, CheckpointReasonMissing, "checkpoint missing")
 }
-func (l *acceptedCheckpointLoader) Merge(_ string, _ string, dependencies []CheckpointFingerprint) (MergeCheckpoint, CheckpointDecision) {
+func (l *acceptedCheckpointLoader) Merge(_ string, _ string, _ []CheckpointFingerprint) (MergeCheckpoint, CheckpointDecision) {
-	l.mergeDeps = append(l.mergeDeps, append([]CheckpointFingerprint(nil), dependencies...))
 	return MergeCheckpoint{}, NewCheckpointDecision(CheckpointDecisionExecuted, CheckpointReasonMissing, "checkpoint missing")
 }
-func (l *acceptedCheckpointLoader) Normalize(_ string, _ string, dependencies []CheckpointFingerprint) (NormalizeCheckpoint, CheckpointDecision) {
+func (l *acceptedCheckpointLoader) Normalize(_ string, _ string, _ []CheckpointFingerprint) (NormalizeCheckpoint, CheckpointDecision) {
-	l.normalizeDeps = append(l.normalizeDeps, append([]CheckpointFingerprint(nil), dependencies...))
 	return NormalizeCheckpoint{}, NewCheckpointDecision(CheckpointDecisionExecuted, CheckpointReasonMissing, "checkpoint missing")
 }
 
@@ -103,9 +97,6 @@ func TestRunnerHydratesRequiredNormalizedArtifact(t *testing.T) {
 	if operationCalls != 0 || validatorCalls != 0 {
 		t.Fatalf("hydrated producer calls = operations %d validators %d, want zero", operationCalls, validatorCalls)
 	}
-	if loader.acceptedCalls[producerKey] != 1 || len(loader.extractDeps) != 1 {
-		t.Fatalf("loader calls = accepted %#v extract %d, want producer hydration and consumer execution only", loader.acceptedCalls, len(loader.extractDeps))
-	}
 	item := received.Slots["producer-output"].Items[0]
 	if string(item.Content) != string(stored.Artifact.Content) || item.Producer.StepID != producer.resolved.StepID || item.Producer.LaneID != producer.resolved.ID {
 		t.Fatalf("consumer generated reference = %#v, want exact hydrated producer bytes and identity", item)
@@ -136,8 +127,19 @@ func TestRunnerHydratesRequiredNormalizedArtifact(t *testing.T) {
 	if err != nil {
 		t.Fatalf("fresh Run() error = %v", err)
 	}
-	if len(freshLoader.extract) != 2 || len(loader.extractDeps) != 1 || !reflect.DeepEqual(freshLoader.extract[1], loader.extractDeps[0]) {
+	hydratedDependencies := loader.extractDeps[consumer.resolved.ID]
-		t.Fatalf("consumer dependencies differ: fresh %#v hydrated %#v", freshLoader.extract, loader.extractDeps)
+	if generatedFingerprintCount(hydratedDependencies) != 1 {
+		t.Fatalf("hydrated consumer dependencies = %#v, want generated producer fingerprint", hydratedDependencies)
+	}
+	var matchedFreshDependencies bool
+	for _, dependencies := range freshLoader.extract {
+		if reflect.DeepEqual(dependencies, hydratedDependencies) {
+			matchedFreshDependencies = true
+			break
+		}
+	}
+	if !matchedFreshDependencies {
+		t.Fatalf("consumer dependencies differ: fresh %#v hydrated %#v", freshLoader.extract, hydratedDependencies)
 	}
 	if !reflect.DeepEqual(freshOutput.Manifest.References, output.Manifest.References) {
 		t.Fatalf("generated provenance differs: fresh %#v hydrated %#v", freshOutput.Manifest.References, output.Manifest.References)
@@ -235,7 +237,9 @@ func TestForcedRequiredLaneExecutesInsteadOfHydrating(t *testing.T) {
 	producer.typed.normalize = func(context.Context, any, contracts.TypedNormalizeRequest[any]) (erasedTypedResult, error) {
 		return erasedTypedResult{Value: codecNotes{Items: []string{"fresh"}}}, nil
 	}
-	consumer.typed.extract = func(context.Context, any, contracts.TypedExtractionRequest) (erasedTypedResult, error) {
+	var consumerReferences contracts.ReferenceSet
+	consumer.typed.extract = func(_ context.Context, _ any, request contracts.TypedExtractionRequest) (erasedTypedResult, error) {
+		consumerReferences = CloneReferenceSet(request.References)
 		return erasedTypedResult{Value: codecScore{Value: 1}}, nil
 	}
 	loader := newAcceptedCheckpointLoader()
@@ -250,12 +254,18 @@ func TestForcedRequiredLaneExecutesInsteadOfHydrating(t *testing.T) {
 		ForcedLanes:          map[string]struct{}{producerKey: {}, consumerKey: {}},
 		RequireReusableLanes: map[string]struct{}{unrelatedKey: {}, producerKey: {}},
 	}
-	if _, err := New().Run(context.Background(), RunInput{Prepared: prepared, RawInput: []byte("input"), Checkpoint: loader, CheckpointPolicy: policy}); err != nil {
+	output, err := New().Run(context.Background(), RunInput{Prepared: prepared, RawInput: []byte("input"), Checkpoint: loader, CheckpointPolicy: policy})
+	if err != nil {
 		t.Fatalf("Run() error = %v", err)
 	}
-	if producerCalls == 0 || unrelatedCalls != 0 || loader.acceptedCalls[producerKey] != 0 || loader.acceptedCalls[unrelatedKey] != 1 {
+	if producerCalls == 0 || unrelatedCalls != 0 {
-		t.Fatalf("calls producer=%d unrelated=%d accepted=%#v, want forced producer execution and unrelated hydration", producerCalls, unrelatedCalls, loader.acceptedCalls)
+		t.Fatalf("calls producer=%d unrelated=%d, want forced producer execution and unrelated hydration", producerCalls, unrelatedCalls)
 	}
+	if got := string(consumerReferences.Slots["producer-output"].Items[0].Content); !strings.Contains(got, "fresh") || strings.Contains(got, "stale") {
+		t.Fatalf("forced producer reference = %q, want freshly executed output", got)
+	}
+	assertAcceptedNormalizeEvent(t, output.CheckpointEvents, unrelated.resolved.StepID, unrelated.resolved.ID, CheckpointDecisionReused, CheckpointReasonAcceptedArtifactReused)
+	assertAcceptedNormalizeEvent(t, output.CheckpointEvents, producer.resolved.StepID, producer.resolved.ID, CheckpointDecisionForcedRecompute, CheckpointReasonRecomputeStep)
 }
 
 func installGeneratedReferenceTarget(target *ResolvedReferenceTarget, stage ModuleStage, consumer *preparedLaneExecutor, producerStep, producerLane string) {