Bind extraction reuse to transcript identity

internal/stage/analyze_test.go

@@ -1491,6 +1491,10 @@ func configureAnalyzeExtractionFixture(t *testing.T, env *Env, m *manifest.Manif
 	if err != nil {
 		t.Fatal(err)
 	}
+	input, err := artifacts.ResolveExtractionInputIdentity(paths, m)
+	if err != nil {
+		t.Fatal(err)
+	}
 	outputs = append(outputs, manifest.ArtifactRecord{
 		Kind: "notarius_index", LocalPath: indexPath, ProducerRunID: producerRunID, Checksum: indexChecksum,
 	})
@@ -1498,7 +1502,8 @@ func configureAnalyzeExtractionFixture(t *testing.T, env *Env, m *manifest.Manif
 		Name: "extract", Status: manifest.StatusSucceeded, Outputs: outputs,
 		Metadata: map[string]any{
 			"narratio_run_id": producerRunID, "bundle_root": bundleRoot,
-			"receipt": map[string]any{"run_id": "notarius-run-1", "pipeline_id": "campaign.extract"},
+			"receipt":      map[string]any{"run_id": "notarius-run-1", "pipeline_id": "campaign.extract"},
+			"direct_input": input.Metadata(),
 		},
 	}
 	return outputPaths

internal/stage/extract.go

@@ -91,9 +91,9 @@ func (extractStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*S
 	}
 
 	paths := sessionPathsForEnv(env, sessionID)
-	input, err := artifacts.ResolveSessionArtifact(paths, m, artifacts.ArtifactTranscriptFinalTrimmed)
+	input, err := artifacts.ResolveExtractionInputIdentity(paths, m)
 	if err != nil {
-		return nil, fmt.Errorf("extract: resolve final-trimmed transcript: %w", err)
+		return nil, fmt.Errorf("extract: resolve final-trimmed transcript identity: %w", err)
 	}
 
 	timeout, err := time.ParseDuration(strings.TrimSpace(notariusConfig.Timeout))
@@ -108,10 +108,7 @@ func (extractStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*S
 	if err != nil {
 		return nil, fmt.Errorf("extract: resolve notarius config path: %w", err)
 	}
-	inputPath, err := absolutePath(input.Path)
-	if err != nil {
-		return nil, fmt.Errorf("extract: resolve transcript input path: %w", err)
-	}
+	inputPath := input.Path
 	workingDirectory, err := absolutePath(notariusConfig.WorkingDirectory)
 	if err != nil {
 		return nil, fmt.Errorf("extract: resolve notarius working directory: %w", err)
@@ -143,7 +140,7 @@ func (extractStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*S
 		}
 	}
 
-	fingerprint, err := extractionFingerprint(resolvedBinary, configPath, notariusConfig, timeout, workingDirectory)
+	fingerprint, err := extractionFingerprint(resolvedBinary, configPath, notariusConfig, timeout, workingDirectory, input)
 	if err != nil {
 		return nil, fmt.Errorf("extract: build configuration fingerprint: %w", err)
 	}
@@ -257,6 +254,7 @@ func (extractStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*S
 		"warnings_path":             promotedWarningsPath,
 		"narratio_run_id":           runID,
 		"configuration_fingerprint": fingerprint,
+		"direct_input":              input.Metadata(),
 		"receipt": map[string]any{
 			"run_id": adapterResult.Receipt.RunID, "pipeline_id": adapterResult.Receipt.PipelineID,
 			"normalized_output_count": adapterResult.Receipt.NormalizedOutputCount,
@@ -370,12 +368,13 @@ type fingerprintOutput struct {
 }
 
 type fingerprintDocument struct {
-	Binary           string              `json:"binary"`
-	ConfigPath       string              `json:"config_path"`
-	PipelineID       string              `json:"pipeline_id"`
-	Timeout          string              `json:"timeout"`
-	WorkingDirectory string              `json:"working_directory"`
-	Outputs          []fingerprintOutput `json:"outputs"`
+	Binary           string                            `json:"binary"`
+	ConfigPath       string                            `json:"config_path"`
+	PipelineID       string                            `json:"pipeline_id"`
+	Timeout          string                            `json:"timeout"`
+	WorkingDirectory string                            `json:"working_directory"`
+	Input            artifacts.ExtractionInputIdentity `json:"input"`
+	Outputs          []fingerprintOutput               `json:"outputs"`
 }
 
 func extractionFingerprint(
@@ -383,6 +382,7 @@ func extractionFingerprint(
 	cfg *config.NotariusConfig,
 	timeout time.Duration,
 	workingDirectory string,
+	input artifacts.ExtractionInputIdentity,
 ) (string, error) {
 	keys := make([]string, 0, len(cfg.Outputs))
 	for key := range cfg.Outputs {
@@ -399,7 +399,7 @@ func extractionFingerprint(
 	}
 	payload, err := json.Marshal(fingerprintDocument{
 		Binary: binary, ConfigPath: configPath, PipelineID: cfg.PipelineID,
-		Timeout: timeout.String(), WorkingDirectory: workingDirectory, Outputs: outputs,
+		Timeout: timeout.String(), WorkingDirectory: workingDirectory, Input: input, Outputs: outputs,
 	})
 	if err != nil {
 		return "", err

internal/stage/extract_resume.go

@@ -37,15 +37,6 @@ func (extractStage) ValidateResume(_ context.Context, env *Env, m *manifest.Mani
 	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)
-	}
-	record := m.Stages[(extractStage{}).Name()]
-	if record == nil || 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)
@@ -54,6 +45,18 @@ func (extractStage) ValidateResume(_ context.Context, env *Env, m *manifest.Mani
 		return ResumeValidation{}, fmt.Errorf("extract resume: session ID is required")
 	}
 	paths := sessionPathsForEnv(env, sessionID)
+	input, err := artifacts.ResolveExtractionInputIdentity(paths, m)
+	if err != nil {
+		return ResumeValidation{}, fmt.Errorf("extract resume: resolve final-trimmed transcript identity: %w", err)
+	}
+	fingerprint, err := extractionFingerprint(resolvedBinary, configPath, cfg, timeout, workingDirectory, input)
+	if err != nil {
+		return ResumeValidation{}, fmt.Errorf("extract resume: build configuration fingerprint: %w", err)
+	}
+	record := m.Stages[(extractStage{}).Name()]
+	if record == nil || metadataString(record.Metadata, "configuration_fingerprint") != fingerprint {
+		return NonResumable("Notarius invocation contract changed"), nil
+	}
 	definitions := make(map[string]artifacts.ExtractionArtifactDefinition, len(cfg.Outputs))
 	for key, output := range cfg.Outputs {
 		definitions[key] = artifacts.ExtractionArtifactDefinition{
@@ -61,7 +64,7 @@ func (extractStage) ValidateResume(_ context.Context, env *Env, m *manifest.Mani
 			SchemaID: output.SchemaID, SchemaVersion: output.SchemaVersion, ModuleKey: output.ModuleKey,
 		}
 	}
-	proof := artifacts.InspectExtractionEvidence(paths, m, definitions)
+	proof := artifacts.InspectExtractionEvidence(paths, m, definitions, input)
 	if proof.State == artifacts.ExtractionEvidenceUnsafe {
 		return ResumeValidation{}, fmt.Errorf("extract resume: %s", proof.Reason)
 	}

internal/stage/extract_test.go

@@ -269,6 +269,11 @@ func TestExtractStageAdapterAndPromotionFailuresReturnNoOutputs(t *testing.T) {
 }
 
 func TestExtractionFingerprintIsIndependentOfOutputMapOrder(t *testing.T) {
+	input := artifacts.ExtractionInputIdentity{
+		Checksum: "checksum", SourceID: artifacts.ArtifactTranscriptFinalTrimmed,
+		ProducerStage: "trim", OutputKind: artifacts.TranscriptOutputKindFinalTrimmed,
+		ProducerRunID: "trim-run", Provenance: "manifest.trim.outputs",
+	}
 	first := &config.NotariusConfig{PipelineID: "pipeline", Outputs: map[string]config.NotariusOutputConfig{
 		"zeta":  {LaneID: "z", MediaType: "application/json", SchemaID: "z", SchemaVersion: "v1"},
 		"alpha": {LaneID: "a", MediaType: "application/json", SchemaID: "a", SchemaVersion: "v1"},
@@ -276,17 +281,39 @@ func TestExtractionFingerprintIsIndependentOfOutputMapOrder(t *testing.T) {
 	second := &config.NotariusConfig{PipelineID: "pipeline", Outputs: map[string]config.NotariusOutputConfig{
 		"alpha": first.Outputs["alpha"], "zeta": first.Outputs["zeta"],
 	}}
-	one, err := extractionFingerprint("/bin/notarius", "/etc/notarius.yml", first, time.Minute, "/work")
+	one, err := extractionFingerprint("/bin/notarius", "/etc/notarius.yml", first, time.Minute, "/work", input)
 	if err != nil {
 		t.Fatalf("extractionFingerprint(first) error = %v", err)
 	}
-	two, err := extractionFingerprint("/bin/notarius", "/etc/notarius.yml", second, time.Minute, "/work")
+	two, err := extractionFingerprint("/bin/notarius", "/etc/notarius.yml", second, time.Minute, "/work", input)
 	if err != nil {
 		t.Fatalf("extractionFingerprint(second) error = %v", err)
 	}
 	if one != two {
 		t.Fatalf("fingerprints differ: %q != %q", one, two)
 	}
+	for _, test := range []struct {
+		name   string
+		mutate func(*artifacts.ExtractionInputIdentity)
+	}{
+		{name: "checksum", mutate: func(identity *artifacts.ExtractionInputIdentity) { identity.Checksum = "different-checksum" }},
+		{name: "source", mutate: func(identity *artifacts.ExtractionInputIdentity) {
+			identity.SourceID = artifacts.ArtifactTranscriptFinal
+		}},
+		{name: "producer", mutate: func(identity *artifacts.ExtractionInputIdentity) { identity.ProducerRunID = "different-trim-run" }},
+	} {
+		t.Run(test.name, func(t *testing.T) {
+			changedInput := input
+			test.mutate(&changedInput)
+			changed, err := extractionFingerprint("/bin/notarius", "/etc/notarius.yml", second, time.Minute, "/work", changedInput)
+			if err != nil {
+				t.Fatalf("extractionFingerprint(changed input) error = %v", err)
+			}
+			if one == changed {
+				t.Fatalf("fingerprint did not change with %s identity", test.name)
+			}
+		})
+	}
 }
 
 func TestExtractStageResumeValidationAcceptsCurrentImmutableResult(t *testing.T) {
@@ -376,6 +403,14 @@ func TestExtractStageResumeValidationRequiresFreshEvidence(t *testing.T) {
 		{name: "no succeeded extract record", mutate: func(_ *testing.T, _ *Env, m *manifest.Manifest) {
 			m.Stages["extract"].Status = manifest.StatusFailed
 		}},
+		{name: "changed direct transcript", mutate: func(t *testing.T, _ *Env, m *manifest.Manifest) {
+			if err := os.WriteFile(m.Stages["trim"].Outputs[0].LocalPath, []byte(`{"segments":[{"id":"changed"}]}`), 0o644); err != nil {
+				t.Fatal(err)
+			}
+		}},
+		{name: "changed direct producer", mutate: func(_ *testing.T, _ *Env, m *manifest.Manifest) {
+			m.Stages["trim"].Outputs[0].ProducerRunID = "trim-run-2"
+		}},
 		{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)

internal/stage/publish_test.go

@@ -1126,11 +1126,16 @@ func configurePublishExtractionFixture(t *testing.T, env *Env, m *manifest.Manif
 	if err != nil {
 		t.Fatal(err)
 	}
+	input, err := artifacts.ResolveExtractionInputIdentity(paths, m)
+	if err != nil {
+		t.Fatal(err)
+	}
 	m.Stages["extract"] = &manifest.StageRecord{
 		Name: "extract", Status: manifest.StatusSucceeded,
 		Metadata: map[string]any{
 			"narratio_run_id": producerRunID, "bundle_root": bundleRoot,
-			"receipt": map[string]any{"run_id": "notarius-run-1", "pipeline_id": "campaign.extract"},
+			"receipt":      map[string]any{"run_id": "notarius-run-1", "pipeline_id": "campaign.extract"},
+			"direct_input": input.Metadata(),
 		},
 		Outputs: []manifest.ArtifactRecord{
 			{