diff --git a/docs/internal/artifacts.md b/docs/internal/artifacts.md index 58c8a8f..63c10c2 100644 --- a/docs/internal/artifacts.md +++ b/docs/internal/artifacts.md @@ -68,7 +68,8 @@ Extraction sources (`narratio.extraction.*`): - use the shared typed bundle evidence inspection in `extraction_evidence.go`; - require a current successful extract record with the exact configured source, compatible contract and Notarius provenance, a confined regular durable - payload, and matching checksum; + payload, matching checksum, and the current resolved trimmed-transcript + identity; - remain unavailable unless catalog hydration receives valid evidence. Resume treats absent or obsolete evidence as a rerun decision and unsafe evidence as an error; and @@ -169,7 +170,8 @@ physical layout. - Registry and resolution: `internal/artifacts/artifact_resolver.go`, `internal/artifacts/catalog.go`, `internal/artifacts/transcripts.go`, `internal/artifacts/extraction_catalog.go`, - `internal/artifacts/extraction_evidence.go` + `internal/artifacts/extraction_evidence.go`, + `internal/artifacts/extraction_input.go` - Current state: `internal/artifacts/current_state.go` - Paths and keys: `internal/artifacts/paths.go`, `internal/artifacts/s3_keys.go` diff --git a/docs/internal/stage-extract.md b/docs/internal/stage-extract.md index 64e34cd..6913862 100644 --- a/docs/internal/stage-extract.md +++ b/docs/internal/stage-extract.md @@ -31,7 +31,10 @@ procedures belong in [Operations](../operations.md). Lane records retain checksum, contract, producer run ID, and Notarius system, run, pipeline, and lane provenance. Stage metadata retains the durable bundle root, receipt, diagnostic paths, rejection/warning summaries, producing -Narratio run ID, and invocation fingerprint. Validation completes before +Narratio run ID, the resolved trimmed-input identity, and invocation +fingerprint. The input identity binds the exact transcript bytes, canonical +source ID, producer stage/output/run identity, and resolution provenance. +Validation completes before promotion, so a rejected result cannot expose a partial durable bundle. Any executed extraction outcome that replaces a different effective outcome @@ -43,7 +46,10 @@ with no outputs is stable and does not repeatedly invalidate downstream stages. `internal/stage/extract_resume.go` permits a skip only when the existing stage record succeeded and still matches the current invocation fingerprint. The fingerprint covers the resolved executable and config paths, pipeline ID, -timeout, working directory, and sorted configured output contracts. +timeout, working directory, sorted configured output contracts, and the current +direct trimmed-transcript identity. The same identity is resolved again for +artifact evidence, so changing the current transcript bytes or producer +identity makes the prior extraction obsolete. The validator then checks the producing run identity, canonical immutable bundle root, path confinement and absence of symlink components, receipt diff --git a/docs/roadmap/implementation.md b/docs/roadmap/implementation.md index 4018a80..4661573 100644 --- a/docs/roadmap/implementation.md +++ b/docs/roadmap/implementation.md @@ -926,6 +926,8 @@ while keeping the external producer's transitive dependencies explicit. producer/source changes, resume/restart, force, unsafe root, over-limit lane result, and failure-before-publication tests. This is the primary TST-009 stage. +**Status:** Completed. + ## Stage 28 — Establish one effective artifact set and catalog bootstrap **Read first:** `audit-findings.md` lines 2456–2491 (COR-022), 3331–3362 diff --git a/internal/app/extract_lifecycle_test.go b/internal/app/extract_lifecycle_test.go index b7bc574..52fbbdb 100644 --- a/internal/app/extract_lifecycle_test.go +++ b/internal/app/extract_lifecycle_test.go @@ -252,6 +252,32 @@ func TestExtractLifecycleSkipsCurrentResumableResult(t *testing.T) { } } +func TestExtractLifecycleRerunsAfterDirectTranscriptChange(t *testing.T) { + cfg, env, runner := extractionLifecycleFixture(t, true) + analyzeRuns := 0 + plan := extractionLifecyclePlan(t, &analyzeRuns) + if _, err := executeStages(context.Background(), cfg, plan, RunOptions{Env: env}); err != nil { + t.Fatalf("first executeStages() error = %v", err) + } + + persisted := loadLifecycleManifest(t, cfg) + trimmed := persisted.Stages["trim"].Outputs[0].LocalPath + if err := os.WriteFile(trimmed, []byte(`{"segments":[{"id":"changed"}]}`), 0o644); err != nil { + t.Fatalf("WriteFile(trimmed transcript) error = %v", err) + } + if err := (&manifest.LocalStore{}).Save(context.Background(), manifestPathFor(cfg), persisted); err != nil { + t.Fatalf("Save(mutated) error = %v", err) + } + + rerun, err := executeStages(context.Background(), cfg, plan, RunOptions{Env: env}) + if err != nil { + t.Fatalf("rerun executeStages() error = %v", err) + } + if len(rerun.Executed) != 2 || len(rerun.Skipped) != 0 || len(runner.requests) != 2 || analyzeRuns != 2 { + t.Fatalf("rerun summary = %#v requests=%d analyze=%d", rerun, len(runner.requests), analyzeRuns) + } +} + func TestExtractLifecycleResumesAndRerunsObsoleteResults(t *testing.T) { for _, test := range []struct { name string diff --git a/internal/app/operator_helpers_test.go b/internal/app/operator_helpers_test.go index 6bcd3ee..7a4e014 100644 --- a/internal/app/operator_helpers_test.go +++ b/internal/app/operator_helpers_test.go @@ -1237,8 +1237,10 @@ func writeOperatorExtractionManifest(t *testing.T, workspaceRoot string) string bundleRoot := filepath.Join(paths.ArtifactsDir, "notarius", "extract-run-1") lanePath := filepath.Join(bundleRoot, "lanes", "encounters.json") indexPath := filepath.Join(bundleRoot, "index.json") + trimmedPath := filepath.Join(paths.Root, filepath.FromSlash(artifacts.TranscriptPathFinalTrimmed)) mustWriteTestFile(t, lanePath, `{"secret":"DO_NOT_PRINT"}`) mustWriteTestFile(t, indexPath, `{"lanes":[]}`) + mustWriteTestFile(t, trimmedPath, `{"segments":[]}`) laneChecksum, err := artifacts.SHA256File(lanePath) if err != nil { t.Fatal(err) @@ -1249,11 +1251,22 @@ func writeOperatorExtractionManifest(t *testing.T, workspaceRoot string) string } m := manifest.New("2026-05-03", time.Now().UTC()) m.Campaign = "sample-campaign" + m.Stages["trim"] = &manifest.StageRecord{ + Name: "trim", Status: manifest.StatusSucceeded, + Outputs: []manifest.ArtifactRecord{{ + Kind: artifactmodel.TranscriptOutputKindFinalTrimmed, LocalPath: trimmedPath, ProducerRunID: "trim-run-1", + }}, + } + 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": "extract-run-1", "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{ { diff --git a/internal/artifacts/extraction_catalog.go b/internal/artifacts/extraction_catalog.go index 29a4958..10b3a8e 100644 --- a/internal/artifacts/extraction_catalog.go +++ b/internal/artifacts/extraction_catalog.go @@ -31,7 +31,11 @@ func (c *ArtifactCatalog) HydrateExtractionArtifacts( if c == nil || m == nil || len(configured) == 0 { return } - proof := InspectExtractionEvidence(paths, m, configured) + input, err := ResolveExtractionInputIdentity(paths, m) + if err != nil { + return + } + proof := InspectExtractionEvidence(paths, m, configured, input) if proof.State != ExtractionEvidenceValid { return } diff --git a/internal/artifacts/extraction_catalog_test.go b/internal/artifacts/extraction_catalog_test.go index 1301556..fb263f7 100644 --- a/internal/artifacts/extraction_catalog_test.go +++ b/internal/artifacts/extraction_catalog_test.go @@ -79,9 +79,10 @@ func TestHydrateExtractionArtifactsAcceptsOnlyCompleteCurrentBundle(t *testing.T func TestInspectExtractionEvidenceClassifiesBundleStates(t *testing.T) { tests := []struct { - name string - state ExtractionEvidenceState - mutate func(*testing.T, *SessionPaths, *manifest.Manifest) + name string + state ExtractionEvidenceState + recomputeInputIdentity bool + mutate func(*testing.T, *SessionPaths, *manifest.Manifest) }{ {name: "valid", state: ExtractionEvidenceValid, mutate: func(_ *testing.T, _ *SessionPaths, _ *manifest.Manifest) {}}, {name: "absent", state: ExtractionEvidenceAbsent, mutate: func(_ *testing.T, _ *SessionPaths, m *manifest.Manifest) { @@ -90,6 +91,12 @@ func TestInspectExtractionEvidenceClassifiesBundleStates(t *testing.T) { {name: "obsolete version", state: ExtractionEvidenceObsolete, mutate: func(_ *testing.T, _ *SessionPaths, m *manifest.Manifest) { m.Stages["extract"].Outputs[0].Contract.SchemaVersion = "99" }}, + {name: "changed direct transcript", state: ExtractionEvidenceObsolete, recomputeInputIdentity: true, mutate: func(t *testing.T, _ *SessionPaths, m *manifest.Manifest) { + writeExtractionFixtureFile(t, m.Stages["trim"].Outputs[0].LocalPath, `{"segments":[{"id":"changed"}]}`) + }}, + {name: "changed direct producer", state: ExtractionEvidenceObsolete, recomputeInputIdentity: true, mutate: func(_ *testing.T, _ *SessionPaths, m *manifest.Manifest) { + m.Stages["trim"].Outputs[0].ProducerRunID = "trim-run-2" + }}, {name: "incomplete", state: ExtractionEvidenceObsolete, mutate: func(_ *testing.T, _ *SessionPaths, m *manifest.Manifest) { m.Stages["extract"].Outputs = m.Stages["extract"].Outputs[1:] }}, @@ -115,8 +122,18 @@ func TestInspectExtractionEvidenceClassifiesBundleStates(t *testing.T) { for _, test := range tests { t.Run(test.name, func(t *testing.T) { paths, currentManifest, definitions := validExtractionCatalogFixture(t) + input, err := ResolveExtractionInputIdentity(paths, currentManifest) + if err != nil { + t.Fatal(err) + } test.mutate(t, &paths, currentManifest) - proof := InspectExtractionEvidence(paths, currentManifest, definitions) + if test.recomputeInputIdentity { + input, err = ResolveExtractionInputIdentity(paths, currentManifest) + if err != nil { + t.Fatal(err) + } + } + proof := InspectExtractionEvidence(paths, currentManifest, definitions, input) if proof.State != test.state { t.Fatalf("proof = %#v, want %q", proof, test.state) } @@ -231,18 +248,31 @@ func validExtractionCatalogFixture(t *testing.T) (SessionPaths, *manifest.Manife t.Helper() root := t.TempDir() paths := SessionPaths{Root: root, ArtifactsDir: filepath.Join(root, "artifacts")} + trimmedPath := filepath.Join(paths.Root, filepath.FromSlash(TranscriptPathFinalTrimmed)) bundleRoot := filepath.Join(paths.ArtifactsDir, "notarius", "extract-run-1") lanePath := filepath.Join(bundleRoot, "lanes", "encounters.json") indexPath := filepath.Join(bundleRoot, "index.json") writeExtractionFixtureFile(t, lanePath, `{"encounters":[]}`) writeExtractionFixtureFile(t, indexPath, `{"lanes":[]}`) + writeExtractionFixtureFile(t, trimmedPath, `{"segments":[]}`) definitions := extractionFixtureDefinitions() m := manifest.New("session", fixtureTime) + m.Stages["trim"] = &manifest.StageRecord{ + Name: "trim", Status: manifest.StatusSucceeded, + Outputs: []manifest.ArtifactRecord{{ + Kind: TranscriptOutputKindFinalTrimmed, LocalPath: trimmedPath, ProducerRunID: "trim-run-1", + }}, + } + input, err := 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": "extract-run-1", "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"}, + extractionMetadataInput: input.Metadata(), }, Outputs: []manifest.ArtifactRecord{ { diff --git a/internal/artifacts/extraction_evidence.go b/internal/artifacts/extraction_evidence.go index 86eb7f8..8748d97 100644 --- a/internal/artifacts/extraction_evidence.go +++ b/internal/artifacts/extraction_evidence.go @@ -26,8 +26,13 @@ type ExtractionEvidence struct { Outputs map[string]string } -// InspectExtractionEvidence verifies structure, confinement, identity, contracts, and payload bytes. -func InspectExtractionEvidence(paths SessionPaths, m *manifest.Manifest, configured map[string]ExtractionArtifactDefinition) ExtractionEvidence { +// InspectExtractionEvidence verifies structure, confinement, identities, contracts, and payload bytes. +func InspectExtractionEvidence( + paths SessionPaths, + m *manifest.Manifest, + configured map[string]ExtractionArtifactDefinition, + input ExtractionInputIdentity, +) ExtractionEvidence { if m == nil || len(configured) == 0 { return ExtractionEvidence{State: ExtractionEvidenceAbsent, Reason: "extraction evidence is absent"} } @@ -35,6 +40,9 @@ func InspectExtractionEvidence(paths SessionPaths, m *manifest.Manifest, configu if r == nil || r.Name != extractStageName || r.Status != manifest.StatusSucceeded { return ExtractionEvidence{State: ExtractionEvidenceAbsent, Reason: "extract stage has no succeeded result"} } + if !extractionInputMatchesMetadata(r.Metadata, input) { + return ExtractionEvidence{State: ExtractionEvidenceObsolete, Reason: "extract result does not match the current trimmed transcript"} + } runID := extractionMetadataString(r.Metadata, extractionMetadataRun) if !safeExtractionPathSegment(runID) { return ExtractionEvidence{State: ExtractionEvidenceObsolete, Reason: "extract result has an invalid producing run ID"} diff --git a/internal/artifacts/extraction_input.go b/internal/artifacts/extraction_input.go new file mode 100644 index 0000000..d728a43 --- /dev/null +++ b/internal/artifacts/extraction_input.go @@ -0,0 +1,76 @@ +package artifacts + +import ( + "crypto/sha256" + "encoding/hex" + "fmt" + "path/filepath" + "strings" + + "gitea.maximumdirect.net/eric/narratio/internal/fileops" + "gitea.maximumdirect.net/eric/narratio/internal/manifest" +) + +const extractionMetadataInput = "direct_input" + +// ExtractionInputIdentity identifies the exact trimmed transcript used for extraction. +type ExtractionInputIdentity struct { + Path string `json:"-"` + Checksum string `json:"checksum"` + SourceID string `json:"source_id"` + ProducerStage string `json:"producer_stage"` + OutputKind string `json:"output_kind"` + ProducerRunID string `json:"producer_run_id"` + Provenance string `json:"provenance"` +} + +// ResolveExtractionInputIdentity resolves and hashes the direct transcript input. +func ResolveExtractionInputIdentity(paths SessionPaths, m *manifest.Manifest) (ExtractionInputIdentity, error) { + resolved, err := ResolveSessionArtifact(paths, m, ArtifactTranscriptFinalTrimmed) + if err != nil { + return ExtractionInputIdentity{}, err + } + path, err := filepath.Abs(resolved.Path) + if err != nil { + return ExtractionInputIdentity{}, fmt.Errorf("resolve trimmed transcript path: %w", err) + } + bytes, err := fileops.ReadRegularFile(path, MaxResolvedArtifactBytes) + if err != nil { + return ExtractionInputIdentity{}, fmt.Errorf("read trimmed transcript: %w", err) + } + digest := sha256.Sum256(bytes) + return ExtractionInputIdentity{ + Path: filepath.Clean(path), + Checksum: hex.EncodeToString(digest[:]), + SourceID: resolved.ID, + ProducerStage: resolved.ProducerStage, + OutputKind: resolved.OutputKind, + ProducerRunID: resolved.ProducerRunID, + Provenance: resolved.Provenance, + }, nil +} + +// Metadata returns the durable fields needed to compare this identity on reuse. +func (identity ExtractionInputIdentity) Metadata() map[string]any { + return map[string]any{ + "checksum": identity.Checksum, + "source_id": identity.SourceID, + "producer_stage": identity.ProducerStage, + "output_kind": identity.OutputKind, + "producer_run_id": identity.ProducerRunID, + "provenance": identity.Provenance, + } +} + +func extractionInputMatchesMetadata(metadata map[string]any, current ExtractionInputIdentity) bool { + stored, ok := metadata[extractionMetadataInput].(map[string]any) + if !ok || strings.TrimSpace(current.Checksum) == "" || strings.TrimSpace(current.SourceID) == "" { + return false + } + for key, want := range current.Metadata() { + if extractionMetadataString(stored, key) != want { + return false + } + } + return true +} diff --git a/internal/stage/analyze_test.go b/internal/stage/analyze_test.go index 2cf2788..74e0d10 100644 --- a/internal/stage/analyze_test.go +++ b/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 diff --git a/internal/stage/extract.go b/internal/stage/extract.go index 962d1f5..adce0b9 100644 --- a/internal/stage/extract.go +++ b/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 diff --git a/internal/stage/extract_resume.go b/internal/stage/extract_resume.go index 01e0964..74408a9 100644 --- a/internal/stage/extract_resume.go +++ b/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) } diff --git a/internal/stage/extract_test.go b/internal/stage/extract_test.go index 1c506ee..01bce55 100644 --- a/internal/stage/extract_test.go +++ b/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) diff --git a/internal/stage/publish_test.go b/internal/stage/publish_test.go index 2f4f72e..abc7703 100644 --- a/internal/stage/publish_test.go +++ b/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{ {