notarius/internal/framework/checkpoint/filesystem_test.go

package checkpoint

import (
	"bytes"
	"encoding/json"
	"os"
	"path/filepath"
	"strings"
	"testing"
	"unicode/utf8"

	"gitea.maximumdirect.net/eric/notarius/internal/core/source"
	"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
	"gitea.maximumdirect.net/eric/notarius/internal/framework/pipeline"
)

type filesystemCheckpointFixture struct {
	root         string
	identity     Identity
	loader       pipeline.CheckpointLoader
	doc          source.SourceDocument
	extract      pipeline.CheckpointArtifact
	merge        pipeline.CheckpointArtifact
	normalize    pipeline.CheckpointArtifact
	dependencies []pipeline.CheckpointFingerprint
	rejected     []contracts.RejectedOutput
}

func TestFilesystemCheckpointRoundTripsAllStages(t *testing.T) {
	fixture := seedFilesystemCheckpoints(t)

	// Recording owns its inputs. These mutations must not change the durable values.
	fixture.doc.Units[0].Text = "caller mutation"
	fixture.doc.Metadata["owner"] = "caller mutation"
	fixture.extract.Artifact.Content[0] = 'x'
	fixture.extract.Artifact.Metadata["content"] = "caller mutation"
	fixture.extract.Diagnostics[0].Diagnostic.Samples[0].Message = "caller mutation"
	fixture.merge.Artifact.Content[0] = 'x'
	fixture.normalize.Artifact.Content[0] = 'x'
	fixture.rejected[0].Message = "caller mutation"

	t.Run("source", func(t *testing.T) {
		got, decision := fixture.loader.Source("source-module")
		if !decision.Reused {
			t.Fatalf("source decision = %#v", decision)
		}
		if got.Document == nil || got.Document.ID != "document-1" || got.Document.Units[0].Text != "original source" || got.Document.Metadata["owner"] != "fixture" {
			t.Fatalf("source was not restored: %#v", got)
		}

		got.Document.Units[0].Text = "loaded mutation"
		got.Document.Metadata["owner"] = "loaded mutation"
		reloaded, decision := fixture.loader.Source("source-module")
		if !decision.Reused || reloaded.Document.Units[0].Text != "original source" || reloaded.Document.Metadata["owner"] != "fixture" {
			t.Fatalf("source reload changed after loaded mutation: %#v decision=%#v", reloaded, decision)
		}
	})

	t.Run("extract", func(t *testing.T) {
		got, decision := fixture.loader.Extract("lane-a", "extract-module", fixture.dependencies)
		if !decision.Reused || len(got.Outputs) != 1 || len(got.Rejected) != 1 || len(got.Outputs[0].Diagnostics) != 1 {
			t.Fatalf("extract result=%#v decision=%#v", got, decision)
		}
		output := got.Outputs[0]
		if !bytes.Equal(output.Artifact.Content, []byte(`{"spell":"fire"}`)) || output.Artifact.Kind != "spell" || output.Artifact.Schema.ID != "spell-schema" || output.Artifact.Schema.Version != "1" || output.Artifact.MediaType != "application/json" || output.Artifact.Metadata["chunk"] != "chunk-a" || output.ChunkRef.StartUnitID != 1 || output.Diagnostics[0].Diagnostic.ReasonCode != "normalized_record" || got.Rejected[0].ReasonCode != "invalid_source" {
			t.Fatalf("extract values were not restored: %#v", got)
		}
		manifest := readManifest[ExtractLaneManifest](t, filepath.Join(fixture.root, mustRelativePath(t, fixture.identity), "extract", "lane-a", "manifest.json"))
		if manifest.Status != StatusSucceededWithRejections {
			t.Fatalf("extract status = %q, want succeeded with rejections", manifest.Status)
		}

		got.Outputs[0].Artifact.Content[0] = 'y'
		reloaded, decision := fixture.loader.Extract("lane-a", "extract-module", fixture.dependencies)
		if !decision.Reused || !bytes.Equal(reloaded.Outputs[0].Artifact.Content, []byte(`{"spell":"fire"}`)) {
			t.Fatalf("extract reload changed after loaded mutation: %#v decision=%#v", reloaded, decision)
		}
	})

	for _, tt := range []struct {
		name string
		load func() (pipeline.CheckpointArtifact, pipeline.CheckpointDecision)
		want []byte
	}{
		{name: "merge", load: func() (pipeline.CheckpointArtifact, pipeline.CheckpointDecision) {
			got, decision := fixture.loader.Merge("lane-a", "merge-module", fixture.dependencies)
			return got.Output, decision
		}, want: []byte(`{"spells":["fire"]}`)},
		{name: "normalize", load: func() (pipeline.CheckpointArtifact, pipeline.CheckpointDecision) {
			got, decision := fixture.loader.Normalize("lane-a", "normalize-module", fixture.dependencies)
			return got.Output, decision
		}, want: []byte(`{"spells":["fire"],"normalized":true}`)},
	} {
		t.Run(tt.name, func(t *testing.T) {
			got, decision := tt.load()
			if !decision.Reused || !bytes.Equal(got.Artifact.Content, tt.want) || got.Artifact.Kind != "spell" || got.Artifact.Schema.ID != "spell-schema" || got.Artifact.Schema.Version != "1" || got.Artifact.Metadata["lane"] != "lane-a" || len(got.Diagnostics) != 1 || got.Diagnostics[0].Diagnostic.ReasonCode != "normalized_record" {
				t.Fatalf("%s result=%#v decision=%#v", tt.name, got, decision)
			}

			got.Artifact.Content[0] = 'z'
			reloaded, decision := tt.load()
			if !decision.Reused || !bytes.Equal(reloaded.Artifact.Content, tt.want) {
				t.Fatalf("%s reload changed after loaded mutation: %#v decision=%#v", tt.name, reloaded, decision)
			}
		})
	}

}

func TestFilesystemCheckpointRejectsMissingAndCorruptState(t *testing.T) {
	for _, stage := range checkpointStages() {
		t.Run(stage.name+" missing manifest", func(t *testing.T) {
			fixture := seedFilesystemCheckpoints(t)
			if err := os.Remove(stage.manifest(fixture)); err != nil {
				t.Fatal(err)
			}
			assertStageNotReused(t, stage, fixture, pipeline.CheckpointReasonMissing)
		})

		t.Run(stage.name+" malformed manifest", func(t *testing.T) {
			fixture := seedFilesystemCheckpoints(t)
			if err := os.WriteFile(stage.manifest(fixture), []byte("{"), 0o600); err != nil {
				t.Fatal(err)
			}
			assertStageNotReused(t, stage, fixture, pipeline.CheckpointReasonDecodeFailed)
		})
	}
}

func TestFilesystemCheckpointRejectsIncompatibleManifests(t *testing.T) {
	for _, tt := range []struct {
		name string
		edit func(map[string]any)
		want pipeline.CheckpointReasonCode
	}{
		{"v1 schema", func(m map[string]any) { m["workspace_schema_version"] = WorkspaceSchemaVersionV1 }, pipeline.CheckpointReasonWorkspaceSchemaIncompatible},
		{"v2 schema", func(m map[string]any) { m["workspace_schema_version"] = WorkspaceSchemaVersionV2 }, pipeline.CheckpointReasonWorkspaceSchemaIncompatible},
		{"v3 schema", func(m map[string]any) { m["workspace_schema_version"] = WorkspaceSchemaVersionV3 }, pipeline.CheckpointReasonWorkspaceSchemaIncompatible},
		{"unknown schema", func(m map[string]any) { m["workspace_schema_version"] = "notarius.workspace.future" }, pipeline.CheckpointReasonWorkspaceSchemaIncompatible},
		{"identity", func(m map[string]any) { m["metadata"].(map[string]any)["checkpoint_identity_digest"] = "sha256:other" }, pipeline.CheckpointReasonIdentityMismatch},
		{"stage", func(m map[string]any) { m["stage"] = string(StageMerge) }, pipeline.CheckpointReasonStageMismatch},
		{"lane", func(m map[string]any) { m["lane_id"] = "lane-other" }, pipeline.CheckpointReasonLaneMismatch},
		{"module", func(m map[string]any) { m["module_key"] = "module-other" }, pipeline.CheckpointReasonModuleMismatch},
		{"dependency", func(m map[string]any) {
			m["dependency_fingerprints"] = []map[string]string{{"name": "input", "value": "other"}}
		}, pipeline.CheckpointReasonDependencyMismatch},
	} {
		t.Run(tt.name, func(t *testing.T) {
			fixture := seedFilesystemCheckpoints(t)
			editManifest(t, checkpointStages()[1].manifest(fixture), tt.edit)
			assertStageNotReused(t, checkpointStages()[1], fixture, tt.want)
		})
	}
}

func TestFilesystemCheckpointRejectsNonTerminalStatuses(t *testing.T) {
	for _, status := range []StageStatus{StatusRunning, StatusFailed, StatusPending, StatusInvalidated} {
		t.Run(string(status), func(t *testing.T) {
			fixture := seedFilesystemCheckpoints(t)
			editManifest(t, checkpointStages()[1].manifest(fixture), func(m map[string]any) { m["status"] = string(status) })
			assertStageNotReused(t, checkpointStages()[1], fixture, pipeline.CheckpointReasonStatusNotReusable)
		})
	}
}

func TestFilesystemCheckpointRejectsIncompleteArtifactsAndContent(t *testing.T) {
	for _, tt := range []struct {
		name string
		edit func(map[string]any)
		want pipeline.CheckpointReasonCode
	}{
		{"artifact kind", func(m map[string]any) { m["outputs"].([]any)[0].(map[string]any)["artifact_kind"] = "" }, pipeline.CheckpointReasonArtifactCodecIncompatible},
		{"schema id", func(m map[string]any) { m["outputs"].([]any)[0].(map[string]any)["schema"].(map[string]any)["id"] = "" }, pipeline.CheckpointReasonArtifactCodecIncompatible},
		{"schema version", func(m map[string]any) {
			m["outputs"].([]any)[0].(map[string]any)["schema"].(map[string]any)["version"] = ""
		}, pipeline.CheckpointReasonArtifactCodecIncompatible},
		{"schema digest", func(m map[string]any) { m["outputs"].([]any)[0].(map[string]any)["schema_digest"] = "" }, pipeline.CheckpointReasonArtifactCodecIncompatible},
		{"base64", func(m map[string]any) {
			m["outputs"].([]any)[0].(map[string]any)["content"].(map[string]any)["content_base64"] = "%"
		}, pipeline.CheckpointReasonArtifactPayloadInvalid},
		{"content digest", func(m map[string]any) {
			m["outputs"].([]any)[0].(map[string]any)["content"].(map[string]any)["content_digest"] = "sha256:other"
		}, pipeline.CheckpointReasonArtifactDigestMismatch},
		{"diagnostics", func(m map[string]any) {
			m["outputs"].([]any)[0].(map[string]any)["diagnostics"] = []any{map[string]any{
				"diagnostic": map[string]any{"disposition": "warning", "category": "configuration", "reason_code": "invalid", "occurrence_count": float64(0)},
			}}
		}, pipeline.CheckpointReasonArtifactPayloadInvalid},
	} {
		t.Run(tt.name, func(t *testing.T) {
			fixture := seedFilesystemCheckpoints(t)
			editJSON(t, filepath.Join(fixture.root, mustRelativePath(t, fixture.identity), "extract", "lane-a", "outputs.json"), tt.edit)
			assertStageNotReused(t, checkpointStages()[1], fixture, tt.want)
		})
	}
}

func TestFilesystemCheckpointRejectsSourceAndOutputDigestMismatches(t *testing.T) {
	t.Run("invalid source document", func(t *testing.T) {
		fixture := seedFilesystemCheckpoints(t)
		editJSON(t, filepath.Join(fixture.root, mustRelativePath(t, fixture.identity), "source", "source-document.json"), func(m map[string]any) {
			m["document"].(map[string]any)["units"].([]any)[0].(map[string]any)["text"] = ""
		})
		assertStageNotReused(t, checkpointStages()[0], fixture, pipeline.CheckpointReasonArtifactPayloadInvalid)
	})

	t.Run("source output digest", func(t *testing.T) {
		fixture := seedFilesystemCheckpoints(t)
		editJSON(t, filepath.Join(fixture.root, mustRelativePath(t, fixture.identity), "source", "source-document.json"), func(m map[string]any) {
			m["document"].(map[string]any)["digest"] = "sha256:other"
		})
		assertStageNotReused(t, checkpointStages()[0], fixture, pipeline.CheckpointReasonArtifactDigestMismatch)
	})

	for _, stage := range checkpointStages()[1:] {
		t.Run(stage.name+" output digest", func(t *testing.T) {
			fixture := seedFilesystemCheckpoints(t)
			editManifest(t, stage.manifest(fixture), func(m map[string]any) {
				m["output_digests"].([]any)[0].(map[string]any)["value"] = "sha256:other"
			})
			assertStageNotReused(t, stage, fixture, pipeline.CheckpointReasonArtifactDigestMismatch)
		})
	}
}

func TestFilesystemCheckpointReusesExtractWithRejections(t *testing.T) {
	fixture := seedFilesystemCheckpoints(t)
	result, decision := fixture.loader.Extract("lane-a", "extract-module", fixture.dependencies)
	if !decision.Reused || len(result.Rejected) != 1 || result.Rejected[0].Message != "source reference is invalid" {
		t.Fatalf("result=%#v decision=%#v", result, decision)
	}
}

func TestFilesystemCheckpointDependencyDecisionIsBoundedAndCategorized(t *testing.T) {
	fixture := seedFilesystemCheckpoints(t)
	_, decision := fixture.loader.Extract("lane-a", "extract-module", []pipeline.CheckpointFingerprint{{Name: "source", Value: "sha256:changed"}})
	if decision.Reused || decision.Category != "dependency_invalidated" || decision.ReasonCode != "dependency_mismatch" {
		t.Fatalf("dependency decision = %#v", decision)
	}
	if strings.Contains(decision.Reason, fixture.root) || strings.Contains(decision.Detail, fixture.root) {
		t.Fatalf("dependency decision leaked checkpoint path: %#v", decision)
	}
}

func TestFilesystemCheckpointDecisionFamiliesAreStableAndSafe(t *testing.T) {
	const secretSentinel = "do-not-expose-checkpoint-secret"
	tests := []struct {
		name     string
		prepare  func(*testing.T, filesystemCheckpointFixture) pipeline.CheckpointDecision
		category pipeline.CheckpointDecisionCategory
		code     pipeline.CheckpointReasonCode
	}{
		{"loading disabled", func(t *testing.T, _ filesystemCheckpointFixture) pipeline.CheckpointDecision {
			_, decision := pipeline.NoopCheckpointLoader().Source("source")
			return decision
		}, pipeline.CheckpointDecisionExecuted, pipeline.CheckpointReasonLoadingDisabled},
		{"checkpoint unavailable", func(t *testing.T, fixture filesystemCheckpointFixture) pipeline.CheckpointDecision {
			if err := os.Remove(checkpointStages()[1].manifest(fixture)); err != nil {
				t.Fatal(err)
			}
			return checkpointStages()[1].load(fixture)
		}, pipeline.CheckpointDecisionExecuted, pipeline.CheckpointReasonMissing},
		{"workspace identity", func(t *testing.T, fixture filesystemCheckpointFixture) pipeline.CheckpointDecision {
			editManifest(t, checkpointStages()[1].manifest(fixture), func(m map[string]any) {
				m["workspace_schema_version"] = secretSentinel
			})
			return checkpointStages()[1].load(fixture)
		}, pipeline.CheckpointDecisionExecuted, pipeline.CheckpointReasonWorkspaceSchemaIncompatible},
		{"manifest scope", func(t *testing.T, fixture filesystemCheckpointFixture) pipeline.CheckpointDecision {
			editManifest(t, checkpointStages()[1].manifest(fixture), func(m map[string]any) { m["module_key"] = secretSentinel })
			return checkpointStages()[1].load(fixture)
		}, pipeline.CheckpointDecisionExecuted, pipeline.CheckpointReasonModuleMismatch},
		{"dependency invalidated", func(t *testing.T, fixture filesystemCheckpointFixture) pipeline.CheckpointDecision {
			_, decision := fixture.loader.Extract("lane-a", "extract-module", []pipeline.CheckpointFingerprint{{Name: "input", Value: secretSentinel}})
			return decision
		}, pipeline.CheckpointDecisionDependencyInvalidated, pipeline.CheckpointReasonDependencyMismatch},
		{"artifact payload", func(t *testing.T, fixture filesystemCheckpointFixture) pipeline.CheckpointDecision {
			editJSON(t, filepath.Join(fixture.root, mustRelativePath(t, fixture.identity), "extract", "lane-a", "outputs.json"), func(m map[string]any) {
				m["outputs"].([]any)[0].(map[string]any)["artifact_kind"] = ""
				m["outputs"].([]any)[0].(map[string]any)["source_id"] = secretSentinel
			})
			return checkpointStages()[1].load(fixture)
		}, pipeline.CheckpointDecisionExecuted, pipeline.CheckpointReasonArtifactCodecIncompatible},
		{"checkpoint reused", func(t *testing.T, fixture filesystemCheckpointFixture) pipeline.CheckpointDecision {
			return checkpointStages()[1].load(fixture)
		}, pipeline.CheckpointDecisionReused, pipeline.CheckpointReasonReused},
	}
	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			fixture := seedFilesystemCheckpoints(t)
			decision := test.prepare(t, fixture)
			if decision.Category != test.category || decision.ReasonCode != test.code {
				t.Fatalf("decision = %#v, want category %q and code %q", decision, test.category, test.code)
			}
			if len([]byte(decision.Detail)) > 512 || !utf8.ValidString(decision.Detail) {
				t.Fatalf("decision detail is not bounded valid UTF-8: %#v", decision)
			}
			for _, forbidden := range []string{secretSentinel, fixture.root} {
				if strings.Contains(decision.Detail, forbidden) || strings.Contains(decision.Reason, forbidden) {
					t.Fatalf("decision leaked %q: %#v", forbidden, decision)
				}
			}
		})
	}
}

func TestFilesystemLoaderReadsAcceptedNormalizeWithoutStageDependencies(t *testing.T) {
	fixture := seedAcceptedNormalizeCheckpoint(t)
	identityRoot := filepath.Join(fixture.root, mustRelativePath(t, fixture.identity))
	for _, stage := range []string{"extract", "merge"} {
		if _, err := os.Stat(filepath.Join(identityRoot, laneManifestPath(stage, "step-1", "lane-a"))); !os.IsNotExist(err) {
			t.Fatalf("%s checkpoint stat error = %v, want absent prerequisite", stage, err)
		}
	}
	checkpoint, decision := fixture.loader.AcceptedNormalize("step-1", "lane-a", "normalize-module")
	if !decision.Reused || decision.Category != pipeline.CheckpointDecisionReused || decision.ReasonCode != pipeline.CheckpointReasonAcceptedArtifactReused {
		t.Fatalf("accepted normalize decision = %#v", decision)
	}
	if checkpoint.Output.Artifact.Content == nil || string(checkpoint.Output.Artifact.Content) != string(fixture.normalize.Artifact.Content) {
		t.Fatalf("accepted normalize output = %#v, want recorded artifact", checkpoint.Output)
	}
}

func TestFilesystemLoaderRejectsInvalidAcceptedNormalize(t *testing.T) {
	tests := []struct {
		name   string
		mutate func(*testing.T, *filesystemCheckpointFixture)
		code   pipeline.CheckpointReasonCode
	}{
		{"missing", func(t *testing.T, fixture *filesystemCheckpointFixture) {
			if err := os.Remove(acceptedNormalizeManifest(t, *fixture)); err != nil {
				t.Fatal(err)
			}
		}, pipeline.CheckpointReasonMissing},
		{"rejected status", func(t *testing.T, fixture *filesystemCheckpointFixture) {
			editManifest(t, acceptedNormalizeManifest(t, *fixture), func(m map[string]any) { m["status"] = string(StatusSucceededWithRejections) })
		}, pipeline.CheckpointReasonStatusNotReusable},
		{"corrupt payload", func(t *testing.T, fixture *filesystemCheckpointFixture) {
			if err := os.WriteFile(acceptedNormalizePayload(t, *fixture), []byte("{"), 0o600); err != nil {
				t.Fatal(err)
			}
		}, pipeline.CheckpointReasonDecodeFailed},
		{"wrong codec identity", func(t *testing.T, fixture *filesystemCheckpointFixture) {
			editJSON(t, acceptedNormalizePayload(t, *fixture), func(m map[string]any) { m["output"].(map[string]any)["artifact_kind"] = "" })
		}, pipeline.CheckpointReasonArtifactCodecIncompatible},
		{"wrong content digest", func(t *testing.T, fixture *filesystemCheckpointFixture) {
			editJSON(t, acceptedNormalizePayload(t, *fixture), func(m map[string]any) {
				m["output"].(map[string]any)["content"].(map[string]any)["content_digest"] = "sha256:wrong"
			})
		}, pipeline.CheckpointReasonArtifactDigestMismatch},
		{"unverifiable identity", func(t *testing.T, fixture *filesystemCheckpointFixture) {
			loader := fixture.loader.(*FilesystemLoader)
			fixture.loader = &FilesystemLoader{root: loader.root}
		}, pipeline.CheckpointReasonIdentityMismatch},
	}
	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			fixture := seedAcceptedNormalizeCheckpoint(t)
			test.mutate(t, &fixture)
			_, decision := fixture.loader.AcceptedNormalize("step-1", "lane-a", "normalize-module")
			if decision.Reused || decision.ReasonCode != test.code {
				t.Fatalf("accepted normalize decision = %#v, want %q", decision, test.code)
			}
			if strings.Contains(decision.Detail, fixture.root) || strings.Contains(decision.Detail, string(fixture.normalize.Artifact.Content)) {
				t.Fatalf("accepted normalize decision leaked path or content: %#v", decision)
			}
			if _, err := os.Stat(acceptedNormalizePayload(t, fixture)); err != nil {
				t.Fatalf("accepted normalize payload was removed after rejection: %v", err)
			}
		})
	}
}

func seedAcceptedNormalizeCheckpoint(t *testing.T) filesystemCheckpointFixture {
	t.Helper()
	root := t.TempDir()
	identity := testIdentity(t)
	artifact := checkpointArtifact("normalize-module", `{"accepted":true}`)
	recorder, err := NewFilesystemRecorder(root, identity)
	if err != nil {
		t.Fatal(err)
	}
	stepRecorder := recorder.(pipeline.StepCheckpointRecorder)
	if err := stepRecorder.NormalizeSucceededForStep("step-1", "lane-a", "normalize-module", []pipeline.CheckpointFingerprint{{Name: "merge", Value: "sha256:unavailable"}}, artifact); err != nil {
		t.Fatal(err)
	}
	loader, err := NewFilesystemLoader(root, identity)
	if err != nil {
		t.Fatal(err)
	}
	return filesystemCheckpointFixture{root: root, identity: identity, loader: loader, normalize: artifact}
}

func acceptedNormalizeManifest(t *testing.T, fixture filesystemCheckpointFixture) string {
	t.Helper()
	return filepath.Join(fixture.root, mustRelativePath(t, fixture.identity), laneManifestPath("normalize", "step-1", "lane-a"))
}

func acceptedNormalizePayload(t *testing.T, fixture filesystemCheckpointFixture) string {
	t.Helper()
	return filepath.Join(fixture.root, mustRelativePath(t, fixture.identity), lanePayloadPath("normalize", "step-1", "lane-a", "output.json"))
}

type checkpointStage struct {
	name     string
	manifest func(filesystemCheckpointFixture) string
	load     func(filesystemCheckpointFixture) pipeline.CheckpointDecision
}

func checkpointStages() []checkpointStage {
	return []checkpointStage{
		{
			name: "source",
			manifest: func(f filesystemCheckpointFixture) string {
				return filepath.Join(f.root, mustRelativePathForTest(f.identity), "source", "manifest.json")
			},
			load: func(f filesystemCheckpointFixture) pipeline.CheckpointDecision {
				_, d := f.loader.Source("source-module")
				return d
			},
		},
		{
			name: "extract",
			manifest: func(f filesystemCheckpointFixture) string {
				return filepath.Join(f.root, mustRelativePathForTest(f.identity), "extract", "lane-a", "manifest.json")
			},
			load: func(f filesystemCheckpointFixture) pipeline.CheckpointDecision {
				_, d := f.loader.Extract("lane-a", "extract-module", f.dependencies)
				return d
			},
		},
		{
			name: "merge",
			manifest: func(f filesystemCheckpointFixture) string {
				return filepath.Join(f.root, mustRelativePathForTest(f.identity), "merge", "lane-a", "manifest.json")
			},
			load: func(f filesystemCheckpointFixture) pipeline.CheckpointDecision {
				_, d := f.loader.Merge("lane-a", "merge-module", f.dependencies)
				return d
			},
		},
		{
			name: "normalize",
			manifest: func(f filesystemCheckpointFixture) string {
				return filepath.Join(f.root, mustRelativePathForTest(f.identity), "normalize", "lane-a", "manifest.json")
			},
			load: func(f filesystemCheckpointFixture) pipeline.CheckpointDecision {
				_, d := f.loader.Normalize("lane-a", "normalize-module", f.dependencies)
				return d
			},
		},
	}
}

func seedFilesystemCheckpoints(t *testing.T) filesystemCheckpointFixture {
	t.Helper()
	root := t.TempDir()
	identity, err := NewIdentity(representativeIdentityInput())
	if err != nil {
		t.Fatal(err)
	}
	recorder, err := NewFilesystemRecorder(root, identity)
	if err != nil {
		t.Fatal(err)
	}
	fixture := filesystemCheckpointFixture{
		root:         root,
		identity:     identity,
		doc:          checkpointDocument(),
		extract:      checkpointArtifact("extract", `{"spell":"fire"}`),
		merge:        checkpointArtifact("merge", `{"spells":["fire"]}`),
		normalize:    checkpointArtifact("normalize", `{"spells":["fire"],"normalized":true}`),
		dependencies: []pipeline.CheckpointFingerprint{{Name: "source", Value: "sha256:source"}, {Name: "chunk-plan", Value: "sha256:plan"}},
		rejected:     []contracts.RejectedOutput{{Stage: "extract", LaneID: "lane-a", ModuleKey: "extract-module", ChunkID: "chunk-a", ValidatorName: "source_refs", ReasonCode: "invalid_source", Message: "source reference is invalid", AttemptCount: 1}},
	}
	if err := recorder.SourceSucceeded("source-module", &fixture.doc); err != nil {
		t.Fatal(err)
	}
	if err := recorder.ExtractSucceeded("lane-a", "extract-module", fixture.dependencies, []pipeline.CheckpointArtifact{fixture.extract}, fixture.rejected); err != nil {
		t.Fatal(err)
	}
	if err := recorder.MergeSucceeded("lane-a", "merge-module", fixture.dependencies, fixture.merge); err != nil {
		t.Fatal(err)
	}
	if err := recorder.NormalizeSucceeded("lane-a", "normalize-module", fixture.dependencies, fixture.normalize); err != nil {
		t.Fatal(err)
	}
	fixture.loader, err = NewFilesystemLoader(root, identity)
	if err != nil {
		t.Fatal(err)
	}
	return fixture
}

func checkpointDocument() source.SourceDocument {
	return source.SourceDocument{
		ID: "document-1", Kind: "transcript", Format: "text", Digest: "sha256:document", Metadata: map[string]any{"owner": "fixture", "number": float64(1)},
		Units: []source.SourceUnit{{ID: 1, Kind: "line", Text: "original source", Ref: source.SourceRef{SourceID: "document-1", StartUnitID: 1, EndUnitID: 1}, Metadata: map[string]any{"speaker": "narrator"}}},
	}
}

func checkpointArtifact(module, content string) pipeline.CheckpointArtifact {
	return pipeline.CheckpointArtifact{
		LaneID: "lane-a", ModuleKey: module, SourceID: "document-1", ChunkID: "chunk-a", ChunkIndex: 0,
		ChunkRef: source.SourceRef{SourceID: "document-1", StartUnitID: 1, EndUnitID: 1}, SchemaDigest: "sha256:schema",
		Artifact:    contracts.SerializedArtifact{Kind: "spell", Schema: contracts.ArtifactSchema{ID: "spell-schema", Name: "Spell", Version: "1", JSONSchema: []byte(`{"type":"object"}`)}, MediaType: "application/json", Content: []byte(content), Metadata: map[string]any{"chunk": "chunk-a", "lane": "lane-a"}},
		Diagnostics: []pipeline.CheckpointDiagnostic{{Diagnostic: contracts.ProducerDiagnostic{Disposition: contracts.DiagnosticDispositionObservation, Category: contracts.DiagnosticCategoryNormalization, ReasonCode: "normalized_record", OccurrenceCount: 1, Samples: []contracts.DiagnosticSample{{Scope: "fixture", Message: "record normalized"}}}}},
	}
}

func assertStageNotReused(t *testing.T, stage checkpointStage, fixture filesystemCheckpointFixture, want pipeline.CheckpointReasonCode) {
	t.Helper()
	decision := stage.load(fixture)
	if decision.Reused || decision.ReasonCode != want {
		t.Fatalf("%s decision=%#v, want non-reused reason code %q", stage.name, decision, want)
	}
}

func editManifest(t *testing.T, path string, edit func(map[string]any)) {
	t.Helper()
	editJSON(t, path, edit)
}

func editJSON(t *testing.T, path string, edit func(map[string]any)) {
	t.Helper()
	data, err := os.ReadFile(path)
	if err != nil {
		t.Fatal(err)
	}
	var value map[string]any
	if err := json.Unmarshal(data, &value); err != nil {
		t.Fatal(err)
	}
	edit(value)
	data, err = json.Marshal(value)
	if err != nil {
		t.Fatal(err)
	}
	if err := os.WriteFile(path, data, 0o600); err != nil {
		t.Fatal(err)
	}
}

func readManifest[T any](t *testing.T, path string) T {
	t.Helper()
	data, err := os.ReadFile(path)
	if err != nil {
		t.Fatal(err)
	}
	var value T
	if err := json.Unmarshal(data, &value); err != nil {
		t.Fatal(err)
	}
	return value
}

func mustRelativePath(t *testing.T, identity Identity) string {
	t.Helper()
	return mustRelativePathForTest(identity)
}

func mustRelativePathForTest(identity Identity) string {
	path, err := identity.RelativePath()
	if err != nil {
		panic(err)
	}
	return path
}