Integrate extraction lifecycle and resume validation

internal/stage/extract_resume.go

@@ -0,0 +1,250 @@
+package stage
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"path/filepath"
+	"strings"
+	"time"
+
+	"gitea.maximumdirect.net/eric/narratio/internal/artifactmodel"
+	"gitea.maximumdirect.net/eric/narratio/internal/artifacts"
+	"gitea.maximumdirect.net/eric/narratio/internal/config"
+	"gitea.maximumdirect.net/eric/narratio/internal/manifest"
+	"gitea.maximumdirect.net/eric/narratio/internal/pathsafe"
+)
+
+func (extractStage) ValidateResume(_ context.Context, env *Env, m *manifest.Manifest) (ResumeValidation, error) {
+	if env == nil || env.Config == nil || env.Config.Pipeline == nil || env.Config.Session == nil {
+		return ResumeValidation{}, fmt.Errorf("extract resume: resolved stage environment config is required")
+	}
+	cfg := env.Config.Pipeline.Notarius
+	if cfg == nil || !cfg.Enabled {
+		return NonResumable("Notarius extraction is disabled"), nil
+	}
+	if m == nil {
+		return ResumeValidation{}, fmt.Errorf("extract resume: session manifest is required")
+	}
+	record := m.Stages[(extractStage{}).Name()]
+	if record == nil || record.Status != manifest.StatusSucceeded {
+		return NonResumable("extract stage has no succeeded result"), nil
+	}
+	if record.Name != (extractStage{}).Name() {
+		return NonResumable("extract stage record identity is inconsistent"), nil
+	}
+
+	producerRunID := metadataString(record.Metadata, "narratio_run_id")
+	if producerRunID == "" {
+		return NonResumable("extract result is missing its producing run ID"), nil
+	}
+	if !safePathSegment(producerRunID) {
+		return NonResumable("extract result has an invalid producing run ID"), nil
+	}
+
+	timeout, err := time.ParseDuration(strings.TrimSpace(cfg.Timeout))
+	if err != nil || timeout <= 0 {
+		return ResumeValidation{}, fmt.Errorf("extract resume: invalid Notarius timeout %q", cfg.Timeout)
+	}
+	resolvedBinary, err := resolveExecutable(cfg.Binary)
+	if err != nil {
+		return ResumeValidation{}, fmt.Errorf("extract resume: resolve Notarius binary: %w", err)
+	}
+	configPath, err := absolutePath(cfg.ConfigPath)
+	if err != nil {
+		return ResumeValidation{}, fmt.Errorf("extract resume: resolve Notarius config path: %w", err)
+	}
+	workingDirectory, err := absolutePath(cfg.WorkingDirectory)
+	if err != nil {
+		return ResumeValidation{}, fmt.Errorf("extract resume: resolve Notarius working directory: %w", err)
+	}
+	fingerprint, err := extractionFingerprint(resolvedBinary, configPath, cfg, timeout, workingDirectory)
+	if err != nil {
+		return ResumeValidation{}, fmt.Errorf("extract resume: build configuration fingerprint: %w", err)
+	}
+	if metadataString(record.Metadata, "configuration_fingerprint") != fingerprint {
+		return NonResumable("Notarius invocation contract changed"), nil
+	}
+
+	sessionID := strings.TrimSpace(m.SessionID)
+	if sessionID == "" {
+		sessionID = strings.TrimSpace(env.Config.Session.SessionID)
+	}
+	campaign := strings.TrimSpace(m.Campaign)
+	if campaign == "" {
+		campaign = strings.TrimSpace(env.Config.Session.Campaign)
+	}
+	if sessionID == "" || campaign == "" {
+		return ResumeValidation{}, fmt.Errorf("extract resume: session ID and campaign are required")
+	}
+	paths := sessionPathsForEnv(env, sessionID)
+	workspaceRoot := strings.TrimSpace(paths.WorkspaceRoot)
+	if workspaceRoot == "" {
+		workspaceRoot = env.Config.Pipeline.Workspace.Root
+	}
+	bundleRoot, err := absolutePath(artifacts.SessionNotariusBundleDirForCampaign(workspaceRoot, campaign, sessionID, producerRunID))
+	if err != nil {
+		return ResumeValidation{}, fmt.Errorf("extract resume: resolve durable bundle path: %w", err)
+	}
+	storedBundleRoot := metadataString(record.Metadata, "bundle_root")
+	if storedBundleRoot == "" || !filepath.IsAbs(storedBundleRoot) || filepath.Clean(storedBundleRoot) != bundleRoot {
+		return NonResumable("extract result does not identify its canonical immutable bundle"), nil
+	}
+	bundleRelative, err := pathsafe.SlashRelativeFromRoot(paths.Root, bundleRoot)
+	if err != nil {
+		return ResumeValidation{}, fmt.Errorf("extract resume: immutable bundle path is unsafe: %w", err)
+	}
+	if err := rejectSymlinkComponents(paths.Root, bundleRelative); err != nil {
+		return ResumeValidation{}, err
+	}
+	bundleInfo, err := os.Lstat(bundleRoot)
+	if os.IsNotExist(err) {
+		return NonResumable("immutable Notarius bundle is missing"), nil
+	}
+	if err != nil {
+		return ResumeValidation{}, fmt.Errorf("extract resume: inspect immutable bundle: %w", err)
+	}
+	if bundleInfo.Mode()&os.ModeSymlink != 0 {
+		return ResumeValidation{}, fmt.Errorf("extract resume: immutable bundle must not be a symlink")
+	}
+	if !bundleInfo.IsDir() {
+		return NonResumable("immutable Notarius bundle is not a directory"), nil
+	}
+
+	receiptRunID, receiptPipelineID := receiptIdentity(record.Metadata)
+	if receiptRunID == "" || receiptPipelineID != cfg.PipelineID {
+		return NonResumable("extract result has incompatible Notarius receipt identity"), nil
+	}
+
+	expectedSources := make(map[string]config.NotariusOutputConfig, len(cfg.Outputs))
+	for key, output := range cfg.Outputs {
+		expectedSources[artifacts.ExtractionArtifactSourceID(key)] = output
+	}
+	seenSources := make(map[string]struct{}, len(expectedSources))
+	indexSeen := false
+	for _, output := range record.Outputs {
+		if output.ProducerRunID != producerRunID {
+			return NonResumable("extract output producer identity is inconsistent"), nil
+		}
+		if output.SourceID == "" {
+			if indexSeen || output.Kind != extractIndexOutputKind {
+				return NonResumable("extract result has an unexpected non-selectable output"), nil
+			}
+			indexSeen = true
+			expectedIndex := filepath.Join(bundleRoot, "index.json")
+			if filepath.Clean(output.LocalPath) != expectedIndex {
+				return NonResumable("extract index path is not canonical"), nil
+			}
+			validation, err := validateResumePayload(bundleRoot, output.LocalPath, output.Checksum)
+			if err != nil || !validation.Resumable {
+				return validation, err
+			}
+			continue
+		}
+
+		expected, ok := expectedSources[output.SourceID]
+		if !ok || output.Kind != extractLaneOutputKind {
+			return NonResumable("extract result source set differs from current configuration"), nil
+		}
+		if _, duplicate := seenSources[output.SourceID]; duplicate {
+			return NonResumable("extract result contains a duplicate configured source"), nil
+		}
+		seenSources[output.SourceID] = struct{}{}
+		if !compatibleExtractionContract(output.Contract, expected) {
+			return NonResumable("extract output contract is incompatible with current configuration"), nil
+		}
+		if !compatibleExtractionProvenance(output.ExternalProvenance, receiptRunID, receiptPipelineID, expected.LaneID) {
+			return NonResumable("extract output has incompatible Notarius provenance"), nil
+		}
+		validation, err := validateResumePayload(bundleRoot, output.LocalPath, output.Checksum)
+		if err != nil || !validation.Resumable {
+			return validation, err
+		}
+	}
+	if !indexSeen {
+		return NonResumable("extract result is missing its canonical index"), nil
+	}
+	if len(seenSources) != len(expectedSources) || len(record.Outputs) != len(expectedSources)+1 {
+		return NonResumable("extract result is missing configured sources"), nil
+	}
+	return Resumable(), nil
+}
+
+func validateResumePayload(bundleRoot, path, checksum string) (ResumeValidation, error) {
+	if !filepath.IsAbs(path) {
+		return ResumeValidation{}, fmt.Errorf("extract resume: persisted output path must be absolute")
+	}
+	relative, err := pathsafe.SlashRelativeFromRoot(bundleRoot, path)
+	if err != nil {
+		return ResumeValidation{}, fmt.Errorf("extract resume: persisted output path is unsafe: %w", err)
+	}
+	if err := rejectSymlinkComponents(bundleRoot, relative); err != nil {
+		return ResumeValidation{}, err
+	}
+	info, err := os.Lstat(path)
+	if os.IsNotExist(err) {
+		return NonResumable("extract output is missing"), nil
+	}
+	if err != nil {
+		return ResumeValidation{}, fmt.Errorf("extract resume: inspect output %q: %w", path, err)
+	}
+	if info.Mode()&os.ModeSymlink != 0 || !info.Mode().IsRegular() {
+		return ResumeValidation{}, fmt.Errorf("extract resume: output %q must be a regular file without symlinks", path)
+	}
+	if strings.TrimSpace(checksum) == "" {
+		return NonResumable("extract output is missing its checksum"), nil
+	}
+	actual, err := artifacts.SHA256File(path)
+	if err != nil {
+		return ResumeValidation{}, fmt.Errorf("extract resume: checksum output %q: %w", path, err)
+	}
+	if actual != checksum {
+		return NonResumable("extract output checksum does not match durable bytes"), nil
+	}
+	return Resumable(), nil
+}
+
+func rejectSymlinkComponents(root, slashRelative string) error {
+	current := filepath.Clean(root)
+	parts := strings.Split(filepath.FromSlash(slashRelative), string(filepath.Separator))
+	for _, part := range parts[:len(parts)-1] {
+		current = filepath.Join(current, part)
+		info, err := os.Lstat(current)
+		if os.IsNotExist(err) {
+			return nil
+		}
+		if err != nil {
+			return fmt.Errorf("extract resume: inspect output directory %q: %w", current, err)
+		}
+		if info.Mode()&os.ModeSymlink != 0 {
+			return fmt.Errorf("extract resume: output directory %q must not be a symlink", current)
+		}
+	}
+	return nil
+}
+
+func compatibleExtractionContract(got *artifactmodel.ContractMetadata, want config.NotariusOutputConfig) bool {
+	return got != nil && got.MediaType == want.MediaType && got.SchemaID == want.SchemaID &&
+		got.SchemaVersion == want.SchemaVersion && (want.ModuleKey == "" || got.ModuleKey == want.ModuleKey)
+}
+
+func compatibleExtractionProvenance(got *artifactmodel.ExternalProvenance, runID, pipelineID, laneID string) bool {
+	return got != nil && got.System == "notarius" && got.RunID == runID &&
+		got.PipelineID == pipelineID && got.ArtifactID == laneID
+}
+
+func metadataString(metadata map[string]any, key string) string {
+	if metadata == nil {
+		return ""
+	}
+	value, _ := metadata[key].(string)
+	return strings.TrimSpace(value)
+}
+
+func receiptIdentity(metadata map[string]any) (string, string) {
+	if metadata == nil {
+		return "", ""
+	}
+	receipt, _ := metadata["receipt"].(map[string]any)
+	return metadataString(receipt, "run_id"), metadataString(receipt, "pipeline_id")
+}

internal/stage/extract_test.go

@@ -47,14 +47,6 @@ func TestExtractStageDisabledReturnsExplicitSkip(t *testing.T) {
 	}
 }
 
-func TestExtractStageIsNotInCanonicalPlansYet(t *testing.T) {
-	for _, candidate := range All() {
-		if candidate.Name() == "extract" {
-			t.Fatal("extract must remain directly executable until lifecycle integration is implemented")
-		}
-	}
-}
-
 func TestExtractStageResolvesManifestInputAndBuildsExactRequest(t *testing.T) {
 	env, m, fake := setupExtractEnv(t)
 	result, err := (extractStage{}).Run(context.Background(), env, m)
@@ -294,6 +286,110 @@ func TestExtractionFingerprintIsIndependentOfOutputMapOrder(t *testing.T) {
 	}
 }
 
+func TestExtractStageResumeValidationAcceptsCurrentImmutableResult(t *testing.T) {
+	env, m, _ := setupExtractEnv(t)
+	seedSucceededExtractResult(t, env, m)
+	m.RunID = "20260810T020304Z-fedcba98"
+
+	validation, err := (extractStage{}).ValidateResume(context.Background(), env, m)
+	if err != nil {
+		t.Fatalf("ValidateResume() error = %v", err)
+	}
+	if !validation.Resumable || validation.Reason != "" {
+		t.Fatalf("validation = %#v, want resumable", validation)
+	}
+}
+
+func TestExtractStageResumeValidationRejectsObsoleteResults(t *testing.T) {
+	tests := []struct {
+		name   string
+		mutate func(*testing.T, *Env, *manifest.Manifest)
+	}{
+		{name: "disabled", mutate: func(_ *testing.T, env *Env, _ *manifest.Manifest) {
+			env.Config.Pipeline.Notarius.Enabled = false
+		}},
+		{name: "config changed", mutate: func(_ *testing.T, env *Env, _ *manifest.Manifest) {
+			env.Config.Pipeline.Notarius.PipelineID = "changed"
+		}},
+		{name: "missing lane", mutate: func(t *testing.T, _ *Env, m *manifest.Manifest) {
+			if err := os.Remove(m.Stages["extract"].Outputs[0].LocalPath); err != nil {
+				t.Fatalf("Remove(lane) error = %v", err)
+			}
+		}},
+		{name: "tampered lane", mutate: func(t *testing.T, _ *Env, m *manifest.Manifest) {
+			if err := os.WriteFile(m.Stages["extract"].Outputs[0].LocalPath, []byte(`{"tampered":true}`), 0o644); err != nil {
+				t.Fatalf("WriteFile(lane) error = %v", err)
+			}
+		}},
+		{name: "incompatible contract", mutate: func(_ *testing.T, _ *Env, m *manifest.Manifest) {
+			m.Stages["extract"].Outputs[0].Contract.SchemaVersion = "v2"
+		}},
+		{name: "missing source", mutate: func(_ *testing.T, _ *Env, m *manifest.Manifest) {
+			m.Stages["extract"].Outputs = m.Stages["extract"].Outputs[1:]
+		}},
+		{name: "producer mismatch", mutate: func(_ *testing.T, _ *Env, m *manifest.Manifest) {
+			m.Stages["extract"].Outputs[0].ProducerRunID = "different-run"
+		}},
+		{name: "provenance mismatch", mutate: func(_ *testing.T, _ *Env, m *manifest.Manifest) {
+			m.Stages["extract"].Outputs[0].ExternalProvenance.RunID = "different-run"
+		}},
+		{name: "missing index", mutate: func(t *testing.T, _ *Env, m *manifest.Manifest) {
+			index := m.Stages["extract"].Outputs[1]
+			if err := os.Remove(index.LocalPath); err != nil {
+				t.Fatalf("Remove(index) error = %v", err)
+			}
+		}},
+	}
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			env, m, _ := setupExtractEnv(t)
+			seedSucceededExtractResult(t, env, m)
+			test.mutate(t, env, m)
+
+			validation, err := (extractStage{}).ValidateResume(context.Background(), env, m)
+			if err != nil {
+				t.Fatalf("ValidateResume() error = %v", err)
+			}
+			if validation.Resumable || validation.Reason == "" || len(validation.Reason) > maxResumeReasonLength {
+				t.Fatalf("validation = %#v, want bounded non-resumable result", validation)
+			}
+		})
+	}
+}
+
+func TestExtractStageResumeValidationRejectsUnsafePathWithError(t *testing.T) {
+	env, m, _ := setupExtractEnv(t)
+	seedSucceededExtractResult(t, env, m)
+	prior := *m.Stages["extract"]
+	m.Stages["extract"].Outputs[0].LocalPath = filepath.Join(env.Config.Pipeline.Workspace.Root, "outside.json")
+
+	if _, err := (extractStage{}).ValidateResume(context.Background(), env, m); err == nil || !strings.Contains(err.Error(), "unsafe") {
+		t.Fatalf("ValidateResume() error = %v, want unsafe path error", err)
+	}
+	if m.Stages["extract"].Status != prior.Status || m.Stages["extract"].Error != prior.Error {
+		t.Fatalf("validation mutated stage record: %#v", m.Stages["extract"])
+	}
+}
+
+func seedSucceededExtractResult(t *testing.T, env *Env, m *manifest.Manifest) {
+	t.Helper()
+	producerRunID := m.RunID
+	result, err := (extractStage{}).Run(context.Background(), env, m)
+	if err != nil {
+		t.Fatalf("Run() error = %v", err)
+	}
+	records := make([]manifest.ArtifactRecord, 0, len(result.Outputs))
+	for _, output := range result.Outputs {
+		records = append(records, manifest.ArtifactRecord{
+			Kind: output.Kind, SourceID: output.SourceID, LocalPath: output.AbsolutePath,
+			Contract: output.Contract, ExternalProvenance: output.ExternalProvenance,
+			ProducerRunID: producerRunID, Checksum: output.Checksum,
+		})
+	}
+	m.MarkStageSucceeded("extract", time.Now().UTC(), records)
+	m.Stages["extract"].Metadata = result.Metadata
+}
+
 type extractFixture struct {
 	workspace string
 	campaign  string

internal/stage/placeholders.go

@@ -74,6 +74,7 @@ func All() []Stage {
 		polishStage{},
 		normalizeStage{},
 		trimStage{},
+		extractStage{},
 		renderStage{},
 		analyzeStage{},
 		publishStage{},

internal/stage/placeholders_test.go

@@ -113,7 +113,7 @@ func TestStagesReturnExpectedMetadata(t *testing.T) {
 	}
 
 	m := manifest.New("2026-05-03", time.Now().UTC())
-	for _, name := range []string{"prepare", "transcribe", "merge", "polish", "normalize", "trim", "render", "analyze"} {
+	for _, name := range []string{"prepare", "transcribe", "merge", "polish", "normalize", "trim", "extract", "render", "analyze"} {
 		m.MarkStageSucceeded(name, time.Now().UTC(), nil)
 	}
 	m.RunID = "20260516T000000Z-abcdef12"
@@ -206,6 +206,12 @@ func TestStagesReturnExpectedMetadata(t *testing.T) {
 			}
 			continue
 		}
+		if s.Name() == "extract" {
+			if result.Disposition != StageDispositionSkipped || result.SkipReason != extractSkipReason {
+				t.Fatalf("extract result = %#v, want disabled self-skip", result)
+			}
+			continue
+		}
 		if s.Name() == "render" {
 			if result.Metadata["stage"] != "render" {
 				t.Fatalf("render metadata = %#v, want stage=render", result.Metadata)

internal/stage/stage.go

@@ -3,6 +3,8 @@ package stage
 import (
 	"context"
 	"log/slog"
+	"strings"
+	"unicode/utf8"
 
 	"gitea.maximumdirect.net/eric/narratio/internal/adapters/audita"
 	"gitea.maximumdirect.net/eric/narratio/internal/adapters/notarius"
@@ -46,6 +48,48 @@ type Stage interface {
 	Run(ctx context.Context, env *Env, m *manifest.Manifest) (*StageResult, error)
 }
 
+const maxResumeReasonLength = 512
+
+// ResumeValidation reports whether a previously succeeded stage can be reused.
+type ResumeValidation struct {
+	Resumable bool
+	Reason    string
+}
+
+// Normalized returns a result with a bounded reason and no reason on success.
+func (r ResumeValidation) Normalized() ResumeValidation {
+	if r.Resumable {
+		return Resumable()
+	}
+	return NonResumable(r.Reason)
+}
+
+// ResumeValidator is implemented by stages that validate persisted success before reuse.
+type ResumeValidator interface {
+	ValidateResume(ctx context.Context, env *Env, m *manifest.Manifest) (ResumeValidation, error)
+}
+
+// Resumable reports a successful resume validation.
+func Resumable() ResumeValidation {
+	return ResumeValidation{Resumable: true}
+}
+
+// NonResumable reports a bounded reason that persisted success must be rerun.
+func NonResumable(reason string) ResumeValidation {
+	reason = strings.TrimSpace(reason)
+	if reason == "" {
+		reason = "persisted stage result is not reusable"
+	}
+	if len(reason) > maxResumeReasonLength {
+		cutoff := maxResumeReasonLength
+		for cutoff > 0 && !utf8.ValidString(reason[:cutoff]) {
+			cutoff--
+		}
+		reason = reason[:cutoff]
+	}
+	return ResumeValidation{Reason: reason}
+}
+
 // StageDisposition describes the outcome of a stage that returned without an error.
 type StageDisposition string