From 39af7d4f3cf3dc62e803869a49740ac8f525c24d Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Mon, 10 Aug 2026 00:24:02 +0000 Subject: [PATCH] Integrate extraction artifacts into analysis catalog --- docs/roadmap/implementation.md | 2 +- internal/artifacts/artifact_resolver.go | 15 +- internal/artifacts/catalog.go | 84 ++++++- internal/artifacts/extraction_catalog.go | 180 ++++++++++++++ internal/artifacts/extraction_catalog_test.go | 222 ++++++++++++++++++ internal/stage/analyze.go | 25 +- internal/stage/analyze_test.go | 145 ++++++++++++ internal/stage/runtime_catalog.go | 24 ++ 8 files changed, 690 insertions(+), 7 deletions(-) create mode 100644 internal/artifacts/extraction_catalog.go create mode 100644 internal/artifacts/extraction_catalog_test.go create mode 100644 internal/stage/runtime_catalog.go diff --git a/docs/roadmap/implementation.md b/docs/roadmap/implementation.md index 315dbe0..032a902 100644 --- a/docs/roadmap/implementation.md +++ b/docs/roadmap/implementation.md @@ -22,7 +22,7 @@ implementation sequence. | Stage 4 | Complete | | Stage 5 | Complete | | Stage 6 | Complete | -| Stage 7 | Not started | +| Stage 7 | Complete | | Stage 8 | Not started | | Stage 9 | Not started | diff --git a/internal/artifacts/artifact_resolver.go b/internal/artifacts/artifact_resolver.go index 81cdf47..04103f7 100644 --- a/internal/artifacts/artifact_resolver.go +++ b/internal/artifacts/artifact_resolver.go @@ -220,17 +220,19 @@ func ResolveSessionArtifact(paths SessionPaths, m *manifest.Manifest, source str } // ResolveSessionArtifactWithCatalog resolves built-in sources using existing rules and resolves -// configured narratio.artifact. sources through runtime catalog availability. +// configured artifact and extraction sources through runtime catalog availability. func ResolveSessionArtifactWithCatalog(paths SessionPaths, m *manifest.Manifest, source string, catalog *ArtifactCatalog) (ResolvedSessionArtifact, error) { normalized := strings.TrimSpace(source) if IsPreviousSessionArtifactSource(normalized) { return ResolvePreviousSessionArtifactWithCatalog(paths, m, normalized, catalog) } - if !IsConfiguredArtifactSource(normalized) { + configuredSource := IsConfiguredArtifactSource(normalized) + extractionSource := IsExtractionArtifactSource(normalized) + if !configuredSource && !extractionSource { return ResolveSessionArtifact(paths, m, normalized) } if catalog == nil { - return ResolvedSessionArtifact{}, fmt.Errorf("configured artifact source %q requires runtime artifact catalog", source) + return ResolvedSessionArtifact{}, fmt.Errorf("catalog-backed artifact source %q requires runtime artifact catalog", source) } entry, ok := catalog.Lookup(normalized) if !ok { @@ -239,7 +241,11 @@ func ResolveSessionArtifactWithCatalog(paths SessionPaths, m *manifest.Manifest, if !entry.Available { return ResolvedSessionArtifact{}, &SessionArtifactNotFoundError{ArtifactID: normalized} } - if err := validateResolvedContent(entry.Path, contentText); err != nil { + contentKind := contentText + if extractionSource { + contentKind = contentJSON + } + if err := validateResolvedContent(entry.Path, contentKind); err != nil { return ResolvedSessionArtifact{}, fmt.Errorf("validate %q: %w", normalized, err) } return ResolvedSessionArtifact{ @@ -247,6 +253,7 @@ func ResolveSessionArtifactWithCatalog(paths SessionPaths, m *manifest.Manifest, Path: filepath.Clean(entry.Path), ProducerStage: entry.ProducerStage, OutputKind: entry.OutputKind, + ProducerRunID: entry.ProducerRunID, Provenance: entry.Provenance, }, nil } diff --git a/internal/artifacts/catalog.go b/internal/artifacts/catalog.go index 126bd68..d1bbf94 100644 --- a/internal/artifacts/catalog.go +++ b/internal/artifacts/catalog.go @@ -11,6 +11,7 @@ import ( const ( ArtifactProvenanceGeneratedCurrentAnalyzeRun = "generated.current_analyze_run" ArtifactProvenanceDisabledFromDisk = "filesystem.disabled_artifact_output" + ArtifactProvenanceCurrentExtractManifest = "manifest.current_extract_run" ) // ConfiguredArtifactDefinition describes one configured analyze artifact. @@ -19,10 +20,21 @@ type ConfiguredArtifactDefinition struct { OutputPath string } +// ExtractionArtifactDefinition describes one configured Notarius output lane. +type ExtractionArtifactDefinition struct { + LaneID string + PipelineID string + MediaType string + SchemaID string + SchemaVersion string + ModuleKey string +} + // CatalogEntry is one runtime catalog entry resolved by source ID. type CatalogEntry struct { SourceID string ConfiguredKey string + ExtractionKey string CanonicalRelPath string ProducerStage string OutputKind string @@ -31,12 +43,14 @@ type CatalogEntry struct { Available bool Path string Provenance string + ProducerRunID string } -// ArtifactCatalog tracks built-in and configured artifact definitions and runtime state. +// ArtifactCatalog tracks built-in, configured, and extraction artifact definitions and runtime state. type ArtifactCatalog struct { entries map[string]CatalogEntry configuredIndex map[string]string + extractionIndex map[string]string } // NewArtifactCatalog returns an empty runtime artifact catalog. @@ -44,9 +58,41 @@ func NewArtifactCatalog() *ArtifactCatalog { return &ArtifactCatalog{ entries: map[string]CatalogEntry{}, configuredIndex: map[string]string{}, + extractionIndex: map[string]string{}, } } +// RegisterExtractionArtifacts registers the configured Notarius output lanes. +func (c *ArtifactCatalog) RegisterExtractionArtifacts(configured map[string]ExtractionArtifactDefinition) error { + keys := make([]string, 0, len(configured)) + for key := range configured { + keys = append(keys, key) + } + sort.Strings(keys) + + for _, key := range keys { + trimmed := strings.TrimSpace(key) + if trimmed == "" { + return fmt.Errorf("extraction artifact keys must be non-empty") + } + if _, exists := c.extractionIndex[trimmed]; exists { + return fmt.Errorf("duplicate extraction artifact key %q", trimmed) + } + sourceID := ExtractionArtifactSourceID(trimmed) + if err := c.addEntry(CatalogEntry{ + SourceID: sourceID, + ExtractionKey: trimmed, + ProducerStage: "extract", + OutputKind: "notarius_lane", + Planned: true, + }); err != nil { + return fmt.Errorf("register extraction artifact %q: %w", trimmed, err) + } + c.extractionIndex[trimmed] = sourceID + } + return nil +} + // ConfiguredArtifactSourceID converts a configured artifact key into canonical source ID. func ConfiguredArtifactSourceID(key string) string { return artifactpolicy.ConfiguredSourceID(key) @@ -151,6 +197,15 @@ func (c *ArtifactCatalog) SourceIDForConfiguredKey(key string) (string, bool) { return sourceID, ok } +// SourceIDForExtractionKey returns the canonical source ID for one Notarius output key. +func (c *ArtifactCatalog) SourceIDForExtractionKey(key string) (string, bool) { + if c == nil { + return "", false + } + sourceID, ok := c.extractionIndex[strings.TrimSpace(key)] + return sourceID, ok +} + // ListConfigured returns configured entries sorted by configured key. func (c *ArtifactCatalog) ListConfigured() []CatalogEntry { if c == nil || len(c.configuredIndex) == 0 { @@ -169,6 +224,23 @@ func (c *ArtifactCatalog) ListConfigured() []CatalogEntry { return out } +// ListExtraction returns extraction entries sorted by configured output key. +func (c *ArtifactCatalog) ListExtraction() []CatalogEntry { + if c == nil || len(c.extractionIndex) == 0 { + return nil + } + keys := make([]string, 0, len(c.extractionIndex)) + for key := range c.extractionIndex { + keys = append(keys, key) + } + sort.Strings(keys) + out := make([]CatalogEntry, 0, len(keys)) + for _, key := range keys { + out = append(out, c.entries[c.extractionIndex[key]]) + } + return out +} + // MarkAvailableGenerated marks one source as available in current analyze execution. func (c *ArtifactCatalog) MarkAvailableGenerated(sourceID, path string) error { return c.markAvailable(sourceID, path, ArtifactProvenanceGeneratedCurrentAnalyzeRun) @@ -179,6 +251,16 @@ func (c *ArtifactCatalog) MarkAvailableFromDisk(sourceID, path string) error { return c.markAvailable(sourceID, path, ArtifactProvenanceDisabledFromDisk) } +func (c *ArtifactCatalog) markAvailableFromExtractManifest(sourceID, path, producerRunID string) error { + if err := c.markAvailable(sourceID, path, ArtifactProvenanceCurrentExtractManifest); err != nil { + return err + } + entry := c.entries[strings.TrimSpace(sourceID)] + entry.ProducerRunID = strings.TrimSpace(producerRunID) + c.entries[entry.SourceID] = entry + return nil +} + func (c *ArtifactCatalog) markAvailable(sourceID, path, provenance string) error { if c == nil { return fmt.Errorf("artifact catalog is nil") diff --git a/internal/artifacts/extraction_catalog.go b/internal/artifacts/extraction_catalog.go new file mode 100644 index 0000000..ca225ce --- /dev/null +++ b/internal/artifacts/extraction_catalog.go @@ -0,0 +1,180 @@ +package artifacts + +import ( + "encoding/json" + "os" + "path/filepath" + "strings" + + "gitea.maximumdirect.net/eric/narratio/internal/artifactmodel" + "gitea.maximumdirect.net/eric/narratio/internal/manifest" +) + +const ( + extractStageName = "extract" + extractionLaneKind = "notarius_lane" + extractionIndexKind = "notarius_index" + extractionMetadataRun = "narratio_run_id" + extractionMetadataRoot = "bundle_root" +) + +type hydratedExtraction struct { + sourceID string + path string +} + +// HydrateExtractionArtifacts marks extraction sources available only when the current +// manifest contains one complete, internally consistent, succeeded extraction bundle. +// Invalid, stale, incomplete, or unsafe records leave every extraction source unavailable. +func (c *ArtifactCatalog) HydrateExtractionArtifacts( + paths SessionPaths, + m *manifest.Manifest, + configured map[string]ExtractionArtifactDefinition, +) { + if c == nil || m == nil || len(configured) == 0 { + return + } + record := m.Stages[extractStageName] + if record == nil || record.Name != extractStageName || record.Status != manifest.StatusSucceeded { + return + } + producerRunID := extractionMetadataString(record.Metadata, extractionMetadataRun) + if !safeExtractionPathSegment(producerRunID) { + return + } + bundleRoot := filepath.Clean(filepath.Join(paths.ArtifactsDir, "notarius", producerRunID)) + if !filepath.IsAbs(bundleRoot) || extractionMetadataString(record.Metadata, extractionMetadataRoot) != bundleRoot { + return + } + if !safeExistingExtractionDirectory(paths.Root, bundleRoot) { + return + } + receiptRunID, receiptPipelineID := extractionReceiptIdentity(record.Metadata) + if receiptRunID == "" || receiptPipelineID == "" { + return + } + + expected := make(map[string]ExtractionArtifactDefinition, len(configured)) + for key, definition := range configured { + sourceID, ok := c.SourceIDForExtractionKey(key) + if !ok { + return + } + expected[sourceID] = definition + } + + seen := make(map[string]struct{}, len(expected)) + hydrated := make([]hydratedExtraction, 0, len(expected)) + indexSeen := false + for _, output := range record.Outputs { + if strings.TrimSpace(output.ProducerRunID) != producerRunID { + return + } + if output.SourceID == "" { + if indexSeen || output.Kind != extractionIndexKind || filepath.Clean(output.LocalPath) != filepath.Join(bundleRoot, "index.json") || + !validExtractionPayload(bundleRoot, output.LocalPath, output.Checksum) { + return + } + indexSeen = true + continue + } + + definition, ok := expected[output.SourceID] + if !ok || output.Kind != extractionLaneKind { + return + } + if _, duplicate := seen[output.SourceID]; duplicate { + return + } + if !compatibleCatalogExtractionContract(output.Contract, definition) || + !compatibleCatalogExtractionProvenance(output.ExternalProvenance, receiptRunID, receiptPipelineID, definition) || + !validExtractionPayload(bundleRoot, output.LocalPath, output.Checksum) { + return + } + seen[output.SourceID] = struct{}{} + hydrated = append(hydrated, hydratedExtraction{sourceID: output.SourceID, path: output.LocalPath}) + } + if !indexSeen || len(seen) != len(expected) || len(record.Outputs) != len(expected)+1 { + return + } + for _, item := range hydrated { + _ = c.markAvailableFromExtractManifest(item.sourceID, item.path, producerRunID) + } +} + +func compatibleCatalogExtractionContract(got *artifactmodel.ContractMetadata, want ExtractionArtifactDefinition) bool { + return got != nil && got.MediaType == want.MediaType && got.SchemaID == want.SchemaID && + got.SchemaVersion == want.SchemaVersion && (want.ModuleKey == "" || got.ModuleKey == want.ModuleKey) +} + +func compatibleCatalogExtractionProvenance( + got *artifactmodel.ExternalProvenance, + runID, pipelineID string, + want ExtractionArtifactDefinition, +) bool { + return got != nil && got.System == "notarius" && got.RunID == runID && got.PipelineID == pipelineID && + pipelineID == strings.TrimSpace(want.PipelineID) && got.ArtifactID == want.LaneID +} + +func validExtractionPayload(bundleRoot, path, checksum string) bool { + if !filepath.IsAbs(path) || !pathWithinExtractionRoot(bundleRoot, path) || strings.TrimSpace(checksum) == "" { + return false + } + info, err := os.Lstat(path) + if err != nil || !info.Mode().IsRegular() || info.Mode()&os.ModeSymlink != 0 { + return false + } + if !safeExtractionComponents(bundleRoot, path) { + return false + } + actual, err := SHA256File(path) + if err != nil || actual != checksum { + return false + } + body, err := os.ReadFile(path) + return err == nil && json.Valid(body) +} + +func safeExistingExtractionDirectory(sessionRoot, bundleRoot string) bool { + if !pathWithinExtractionRoot(sessionRoot, bundleRoot) || !safeExtractionComponents(sessionRoot, bundleRoot) { + return false + } + info, err := os.Lstat(bundleRoot) + return err == nil && info.IsDir() && info.Mode()&os.ModeSymlink == 0 +} + +func safeExtractionComponents(root, target string) bool { + relative, err := filepath.Rel(filepath.Clean(root), filepath.Clean(target)) + if err != nil || relative == "." || relative == ".." || strings.HasPrefix(relative, ".."+string(filepath.Separator)) { + return false + } + current := filepath.Clean(root) + for _, part := range strings.Split(relative, string(filepath.Separator)) { + current = filepath.Join(current, part) + info, err := os.Lstat(current) + if err != nil || info.Mode()&os.ModeSymlink != 0 { + return false + } + } + return true +} + +func pathWithinExtractionRoot(root, target string) bool { + relative, err := filepath.Rel(filepath.Clean(root), filepath.Clean(target)) + return err == nil && relative != "." && relative != ".." && !strings.HasPrefix(relative, ".."+string(filepath.Separator)) +} + +func safeExtractionPathSegment(value string) bool { + return value != "" && value != "." && value != ".." && filepath.Base(value) == value && + !strings.ContainsAny(value, `/\\`) +} + +func extractionMetadataString(metadata map[string]any, key string) string { + value, _ := metadata[key].(string) + return strings.TrimSpace(value) +} + +func extractionReceiptIdentity(metadata map[string]any) (string, string) { + receipt, _ := metadata["receipt"].(map[string]any) + return extractionMetadataString(receipt, "run_id"), extractionMetadataString(receipt, "pipeline_id") +} diff --git a/internal/artifacts/extraction_catalog_test.go b/internal/artifacts/extraction_catalog_test.go new file mode 100644 index 0000000..e017a9e --- /dev/null +++ b/internal/artifacts/extraction_catalog_test.go @@ -0,0 +1,222 @@ +package artifacts + +import ( + "errors" + "os" + "path/filepath" + "testing" + "time" + + "gitea.maximumdirect.net/eric/narratio/internal/artifactmodel" + "gitea.maximumdirect.net/eric/narratio/internal/manifest" +) + +func TestArtifactCatalogRegistersExtractionArtifactsDeterministically(t *testing.T) { + catalog := NewArtifactCatalog() + if err := catalog.RegisterConfiguredArtifacts(map[string]ConfiguredArtifactDefinition{ + "summary": {Enabled: true}, + }, nil); err != nil { + t.Fatal(err) + } + if err := catalog.RegisterExtractionArtifacts(map[string]ExtractionArtifactDefinition{ + "zeta": {LaneID: "zeta"}, + "summary": {LaneID: "summary"}, + "alpha": {LaneID: "alpha"}, + }); err != nil { + t.Fatalf("RegisterExtractionArtifacts() error = %v", err) + } + + entries := catalog.ListExtraction() + if len(entries) != 3 || entries[0].ExtractionKey != "alpha" || entries[1].ExtractionKey != "summary" || entries[2].ExtractionKey != "zeta" { + t.Fatalf("ListExtraction() = %#v, want alpha, summary, zeta", entries) + } + extractionID, ok := catalog.SourceIDForExtractionKey("summary") + if !ok || extractionID != "narratio.extraction.summary" { + t.Fatalf("SourceIDForExtractionKey(summary) = %q, %v", extractionID, ok) + } + configuredID, _ := catalog.SourceIDForConfiguredKey("summary") + if configuredID == extractionID { + t.Fatalf("configured and extraction source families collided at %q", extractionID) + } + entry, ok := catalog.Lookup(extractionID) + if !ok || entry.ProducerStage != "extract" || entry.OutputKind != "notarius_lane" || !entry.Planned { + t.Fatalf("Lookup(%q) = %#v, %v", extractionID, entry, ok) + } +} + +func TestArtifactCatalogRejectsDuplicateExtractionRegistration(t *testing.T) { + catalog := NewArtifactCatalog() + definitions := map[string]ExtractionArtifactDefinition{"summary": {LaneID: "summary"}} + if err := catalog.RegisterExtractionArtifacts(definitions); err != nil { + t.Fatal(err) + } + if err := catalog.RegisterExtractionArtifacts(definitions); err == nil { + t.Fatal("second RegisterExtractionArtifacts() error = nil, want collision error") + } +} + +func TestHydrateExtractionArtifactsAcceptsOnlyCompleteCurrentBundle(t *testing.T) { + paths, currentManifest, definitions := validExtractionCatalogFixture(t) + catalog := registeredExtractionCatalog(t, definitions) + catalog.HydrateExtractionArtifacts(paths, currentManifest, definitions) + + entry, ok := catalog.Lookup(ExtractionArtifactSourceID("encounters")) + if !ok || !entry.Available { + t.Fatalf("hydrated entry = %#v, %v; want available", entry, ok) + } + if entry.Provenance != ArtifactProvenanceCurrentExtractManifest || entry.ProducerRunID != "extract-run-1" { + t.Fatalf("hydrated provenance = %#v", entry) + } + resolved, err := ResolveSessionArtifactWithCatalog(paths, currentManifest, entry.SourceID, catalog) + if err != nil { + t.Fatalf("ResolveSessionArtifactWithCatalog() error = %v", err) + } + if resolved.Path != entry.Path || resolved.ProducerRunID != "extract-run-1" { + t.Fatalf("resolved = %#v", resolved) + } +} + +func TestHydrateExtractionArtifactsRejectsUntrustedManifestState(t *testing.T) { + tests := []struct { + name string + mutate func(t *testing.T, paths SessionPaths, m *manifest.Manifest) + }{ + {name: "missing stage", mutate: func(_ *testing.T, _ SessionPaths, m *manifest.Manifest) { delete(m.Stages, "extract") }}, + {name: "skipped", mutate: setExtractionStatus(manifest.StatusSkipped)}, + {name: "failed", mutate: setExtractionStatus(manifest.StatusFailed)}, + {name: "stale", mutate: setExtractionStatus(manifest.StatusStale)}, + {name: "interrupted", mutate: setExtractionStatus(manifest.StatusInterrupted)}, + {name: "missing source", mutate: func(_ *testing.T, _ SessionPaths, m *manifest.Manifest) { + m.Stages["extract"].Outputs = m.Stages["extract"].Outputs[1:] + }}, + {name: "inconsistent producer identity", mutate: func(_ *testing.T, _ SessionPaths, m *manifest.Manifest) { + m.Stages["extract"].Outputs[0].ProducerRunID = "another-run" + }}, + {name: "incompatible contract", mutate: func(_ *testing.T, _ SessionPaths, m *manifest.Manifest) { + m.Stages["extract"].Outputs[0].Contract.SchemaVersion = "99" + }}, + {name: "incompatible provenance", mutate: func(_ *testing.T, _ SessionPaths, m *manifest.Manifest) { + m.Stages["extract"].Outputs[0].ExternalProvenance.RunID = "another-notarius-run" + }}, + {name: "missing file", mutate: func(t *testing.T, _ SessionPaths, m *manifest.Manifest) { + if err := os.Remove(m.Stages["extract"].Outputs[0].LocalPath); err != nil { + t.Fatal(err) + } + }}, + {name: "tampered checksum", mutate: func(t *testing.T, _ SessionPaths, m *manifest.Manifest) { + if err := os.WriteFile(m.Stages["extract"].Outputs[0].LocalPath, []byte(`{"tampered":true}`), 0o644); err != nil { + t.Fatal(err) + } + }}, + {name: "unsafe path", mutate: func(t *testing.T, paths SessionPaths, m *manifest.Manifest) { + outside := filepath.Join(paths.Root, "incidental.json") + writeExtractionFixtureFile(t, outside, `{"incidental":true}`) + m.Stages["extract"].Outputs[0].LocalPath = outside + m.Stages["extract"].Outputs[0].Checksum = extractionFixtureChecksum(t, outside) + }}, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + paths, currentManifest, definitions := validExtractionCatalogFixture(t) + test.mutate(t, paths, currentManifest) + catalog := registeredExtractionCatalog(t, definitions) + catalog.HydrateExtractionArtifacts(paths, currentManifest, definitions) + entry, _ := catalog.Lookup(ExtractionArtifactSourceID("encounters")) + if entry.Available { + t.Fatalf("entry became available from %s manifest", test.name) + } + }) + } +} + +func TestResolveExtractionArtifactNeverDiscoversIncidentalBundleFile(t *testing.T) { + root := t.TempDir() + paths := SessionPaths{Root: root, ArtifactsDir: filepath.Join(root, "artifacts")} + incidental := filepath.Join(paths.ArtifactsDir, "notarius", "incidental", "lanes", "encounters.json") + writeExtractionFixtureFile(t, incidental, `{"encounters":[]}`) + definitions := extractionFixtureDefinitions() + catalog := registeredExtractionCatalog(t, definitions) + + _, err := ResolveSessionArtifactWithCatalog(paths, manifest.New("session", fixtureTime), ExtractionArtifactSourceID("encounters"), catalog) + if err == nil || !errors.Is(err, ErrSessionArtifactNotFound) { + t.Fatalf("ResolveSessionArtifactWithCatalog() error = %v, want not found", err) + } +} + +var fixtureTime = mustFixtureTime() + +func mustFixtureTime() (value time.Time) { + return time.Date(2026, 8, 10, 0, 0, 0, 0, time.UTC) +} + +func setExtractionStatus(status manifest.StageStatus) func(*testing.T, SessionPaths, *manifest.Manifest) { + return func(_ *testing.T, _ SessionPaths, m *manifest.Manifest) { m.Stages["extract"].Status = status } +} + +func extractionFixtureDefinitions() map[string]ExtractionArtifactDefinition { + return map[string]ExtractionArtifactDefinition{ + "encounters": { + LaneID: "encounters", PipelineID: "campaign.extract", MediaType: "application/json", + SchemaID: "encounters", SchemaVersion: "1", ModuleKey: "encounters", + }, + } +} + +func registeredExtractionCatalog(t *testing.T, definitions map[string]ExtractionArtifactDefinition) *ArtifactCatalog { + t.Helper() + catalog := NewArtifactCatalog() + if err := catalog.RegisterExtractionArtifacts(definitions); err != nil { + t.Fatal(err) + } + return catalog +} + +func validExtractionCatalogFixture(t *testing.T) (SessionPaths, *manifest.Manifest, map[string]ExtractionArtifactDefinition) { + t.Helper() + root := t.TempDir() + paths := SessionPaths{Root: root, ArtifactsDir: filepath.Join(root, "artifacts")} + 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":[]}`) + definitions := extractionFixtureDefinitions() + m := manifest.New("session", fixtureTime) + 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"}, + }, + Outputs: []manifest.ArtifactRecord{ + { + Kind: "notarius_lane", SourceID: ExtractionArtifactSourceID("encounters"), LocalPath: lanePath, + ProducerRunID: "extract-run-1", Checksum: extractionFixtureChecksum(t, lanePath), + Contract: &artifactmodel.ContractMetadata{MediaType: "application/json", SchemaID: "encounters", SchemaVersion: "1", ModuleKey: "encounters"}, + ExternalProvenance: &artifactmodel.ExternalProvenance{System: "notarius", RunID: "notarius-run-1", PipelineID: "campaign.extract", ArtifactID: "encounters"}, + }, + {Kind: "notarius_index", LocalPath: indexPath, ProducerRunID: "extract-run-1", Checksum: extractionFixtureChecksum(t, indexPath)}, + }, + } + return paths, m, definitions +} + +func writeExtractionFixtureFile(t *testing.T, path, body string) { + t.Helper() + if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(path, []byte(body), 0o644); err != nil { + t.Fatal(err) + } +} + +func extractionFixtureChecksum(t *testing.T, path string) string { + t.Helper() + checksum, err := SHA256File(path) + if err != nil { + t.Fatal(err) + } + return checksum +} diff --git a/internal/stage/analyze.go b/internal/stage/analyze.go index d43ffba..2bc4378 100644 --- a/internal/stage/analyze.go +++ b/internal/stage/analyze.go @@ -84,7 +84,13 @@ func (analyzeStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*S }}, nil } - runtimeCatalog, err := buildAnalyzeRuntimeArtifactCatalog(paths, env.Config.Pipeline.Scriptorium, env.SelectedArtifactKeys) + runtimeCatalog, err := buildAnalyzeRuntimeArtifactCatalog( + paths, + m, + env.Config.Pipeline.Scriptorium, + env.Config.Pipeline.Notarius, + env.SelectedArtifactKeys, + ) if err != nil { return nil, fmt.Errorf("analyze: build runtime artifact catalog: %w", err) } @@ -668,6 +674,16 @@ func resolveScriptoriumInput( return resolved.Path, true, ©, nil } if errors.Is(err, artifacts.ErrSessionArtifactNotFound) { + if descriptor.Source.Kind == artifactpolicy.SourceKindExtraction { + if inputCfg.Required { + return "", false, nil, fmt.Errorf( + "required extraction source %q is unavailable; enable and configure pipeline.notarius output %q, then rerun extract with --force", + source, + descriptor.Source.ConfiguredKey, + ) + } + return "", false, nil, nil + } if descriptor.Source.Kind == artifactpolicy.SourceKindConfiguredArtifact { if inputCfg.Required { return "", false, nil, fmt.Errorf("configured artifact source %q is unavailable", source) @@ -727,13 +743,20 @@ func preparedStableInputFilename(sourceID string) (string, bool) { func buildAnalyzeRuntimeArtifactCatalog( paths artifacts.SessionPaths, + m *manifest.Manifest, scriptoriumCfg *config.ScriptoriumConfig, + notariusCfg *config.NotariusConfig, selectedArtifacts []string, ) (*artifacts.ArtifactCatalog, error) { catalog := artifacts.NewArtifactCatalog() if err := catalog.RegisterBuiltIns(); err != nil { return nil, err } + extractionDefinitions := configuredExtractionDefinitions(notariusCfg) + if err := catalog.RegisterExtractionArtifacts(extractionDefinitions); err != nil { + return nil, err + } + catalog.HydrateExtractionArtifacts(paths, m, extractionDefinitions) if scriptoriumCfg == nil { return catalog, nil } diff --git a/internal/stage/analyze_test.go b/internal/stage/analyze_test.go index 3a83441..fa7cb68 100644 --- a/internal/stage/analyze_test.go +++ b/internal/stage/analyze_test.go @@ -5,12 +5,14 @@ import ( "errors" "os" "path/filepath" + "sort" "strings" "testing" "time" "gitea.maximumdirect.net/eric/narratio/internal/adapters/scriptorium" "gitea.maximumdirect.net/eric/narratio/internal/adapters/storage" + "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" @@ -1358,6 +1360,149 @@ func TestAnalyzeSkipsWhenArtifactMapEmpty(t *testing.T) { } } +func TestAnalyzePassesOnlyExplicitExtractionInputsToSelectedArtifact(t *testing.T) { + env, m, fake := setupAnalyzeEnv(t) + paths := sessionPathsForEnv(env, m.SessionID) + writeAnalyzeFile(t, filepath.Join(paths.TranscriptsDir, "final.trimmed.json"), `{"segments":[]}`) + extractionPaths := configureAnalyzeExtractionFixture(t, env, m) + + artifact := env.Config.Pipeline.Scriptorium.Artifacts["session_recap"] + artifact.Inputs["encounters"] = config.ScriptoriumInputConfig{ + Source: artifacts.ExtractionArtifactSourceID("encounters"), Required: true, + } + env.Config.Pipeline.Scriptorium.Artifacts["session_recap"] = artifact + env.SelectedArtifactKeys = []string{"session_recap"} + tracker := &analyzeObjectStoreTracker{} + env.ObjectStore = tracker + + if _, err := (analyzeStage{}).Run(context.Background(), env, m); err != nil { + t.Fatalf("Run() error = %v", err) + } + if len(fake.RunRequests) != 1 { + t.Fatalf("run requests = %d, want 1", len(fake.RunRequests)) + } + inputs := fake.RunRequests[0].InputPaths + if inputs["encounters"] != extractionPaths["encounters"] { + t.Fatalf("encounters input = %q, want %q", inputs["encounters"], extractionPaths["encounters"]) + } + if _, exists := inputs["characters"]; exists { + t.Fatalf("unconfigured extraction input was added: %#v", inputs) + } + if len(inputs) != 2 { + t.Fatalf("input paths = %#v, want transcript plus explicit encounters", inputs) + } + if tracker.called { + t.Fatal("analyze extraction resolution called the object store") + } +} + +func TestAnalyzeRequiredUnavailableExtractionFailsWithGuidance(t *testing.T) { + env, m, fake := setupAnalyzeEnv(t) + paths := sessionPathsForEnv(env, m.SessionID) + writeAnalyzeFile(t, filepath.Join(paths.TranscriptsDir, "final.trimmed.json"), `{"segments":[]}`) + configureAnalyzeExtraction(t, env) + artifact := env.Config.Pipeline.Scriptorium.Artifacts["session_recap"] + artifact.Inputs["encounters"] = config.ScriptoriumInputConfig{ + Source: artifacts.ExtractionArtifactSourceID("encounters"), Required: true, + } + env.Config.Pipeline.Scriptorium.Artifacts["session_recap"] = artifact + + _, err := (analyzeStage{}).Run(context.Background(), env, m) + if err == nil || !strings.Contains(err.Error(), "pipeline.notarius output \"encounters\"") || !strings.Contains(err.Error(), "rerun extract with --force") { + t.Fatalf("Run() error = %v, want actionable extraction guidance", err) + } + if len(fake.RunRequests) != 0 { + t.Fatalf("run requests = %d, want 0", len(fake.RunRequests)) + } +} + +func TestAnalyzeOmitsOptionalUnavailableExtraction(t *testing.T) { + env, m, fake := setupAnalyzeEnv(t) + paths := sessionPathsForEnv(env, m.SessionID) + writeAnalyzeFile(t, filepath.Join(paths.TranscriptsDir, "final.trimmed.json"), `{"segments":[]}`) + configureAnalyzeExtraction(t, env) + artifact := env.Config.Pipeline.Scriptorium.Artifacts["session_recap"] + artifact.Inputs["encounters"] = config.ScriptoriumInputConfig{ + Source: artifacts.ExtractionArtifactSourceID("encounters"), Required: false, + } + env.Config.Pipeline.Scriptorium.Artifacts["session_recap"] = artifact + + if _, err := (analyzeStage{}).Run(context.Background(), env, m); err != nil { + t.Fatalf("Run() error = %v", err) + } + if len(fake.RunRequests) != 1 { + t.Fatalf("run requests = %d, want 1", len(fake.RunRequests)) + } + if _, exists := fake.RunRequests[0].InputPaths["encounters"]; exists { + t.Fatalf("optional unavailable extraction was passed: %#v", fake.RunRequests[0].InputPaths) + } +} + +func configureAnalyzeExtraction(t *testing.T, env *Env) { + t.Helper() + env.Config.Pipeline.Notarius = &config.NotariusConfig{ + Enabled: true, PipelineID: "campaign.extract", + Outputs: map[string]config.NotariusOutputConfig{ + "characters": {LaneID: "characters", MediaType: "application/json", SchemaID: "characters", SchemaVersion: "1"}, + "encounters": {LaneID: "encounters", MediaType: "application/json", SchemaID: "encounters", SchemaVersion: "1", ModuleKey: "encounters"}, + }, + } +} + +func configureAnalyzeExtractionFixture(t *testing.T, env *Env, m *manifest.Manifest) map[string]string { + t.Helper() + configureAnalyzeExtraction(t, env) + paths := sessionPathsForEnv(env, m.SessionID) + producerRunID := "extract-run-1" + bundleRoot := filepath.Join(paths.ArtifactsDir, "notarius", producerRunID) + keys := make([]string, 0, len(env.Config.Pipeline.Notarius.Outputs)) + for key := range env.Config.Pipeline.Notarius.Outputs { + keys = append(keys, key) + } + sort.Strings(keys) + + outputPaths := make(map[string]string, len(keys)) + outputs := make([]manifest.ArtifactRecord, 0, len(keys)+1) + for _, key := range keys { + definition := env.Config.Pipeline.Notarius.Outputs[key] + outputPath := filepath.Join(bundleRoot, "lanes", key+".json") + writeAnalyzeFile(t, outputPath, `{"items":[]}`) + checksum, err := artifacts.SHA256File(outputPath) + if err != nil { + t.Fatal(err) + } + outputPaths[key] = outputPath + outputs = append(outputs, manifest.ArtifactRecord{ + Kind: "notarius_lane", SourceID: artifacts.ExtractionArtifactSourceID(key), LocalPath: outputPath, + ProducerRunID: producerRunID, Checksum: checksum, + Contract: &artifactmodel.ContractMetadata{ + MediaType: definition.MediaType, SchemaID: definition.SchemaID, + SchemaVersion: definition.SchemaVersion, ModuleKey: definition.ModuleKey, + }, + ExternalProvenance: &artifactmodel.ExternalProvenance{ + System: "notarius", RunID: "notarius-run-1", PipelineID: "campaign.extract", ArtifactID: definition.LaneID, + }, + }) + } + indexPath := filepath.Join(bundleRoot, "index.json") + writeAnalyzeFile(t, indexPath, `{"lanes":[]}`) + indexChecksum, err := artifacts.SHA256File(indexPath) + if err != nil { + t.Fatal(err) + } + outputs = append(outputs, manifest.ArtifactRecord{ + Kind: "notarius_index", LocalPath: indexPath, ProducerRunID: producerRunID, Checksum: indexChecksum, + }) + m.Stages["extract"] = &manifest.StageRecord{ + 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"}, + }, + } + return outputPaths +} + func setupAnalyzeEnv(t *testing.T) (*Env, *manifest.Manifest, *scriptorium.FakeRunner) { t.Helper() workspace := t.TempDir() diff --git a/internal/stage/runtime_catalog.go b/internal/stage/runtime_catalog.go new file mode 100644 index 0000000..5543162 --- /dev/null +++ b/internal/stage/runtime_catalog.go @@ -0,0 +1,24 @@ +package stage + +import ( + "gitea.maximumdirect.net/eric/narratio/internal/artifacts" + "gitea.maximumdirect.net/eric/narratio/internal/config" +) + +func configuredExtractionDefinitions(cfg *config.NotariusConfig) map[string]artifacts.ExtractionArtifactDefinition { + if cfg == nil || !cfg.Enabled || len(cfg.Outputs) == 0 { + return nil + } + definitions := make(map[string]artifacts.ExtractionArtifactDefinition, len(cfg.Outputs)) + for key, output := range cfg.Outputs { + definitions[key] = artifacts.ExtractionArtifactDefinition{ + LaneID: output.LaneID, + PipelineID: cfg.PipelineID, + MediaType: output.MediaType, + SchemaID: output.SchemaID, + SchemaVersion: output.SchemaVersion, + ModuleKey: output.ModuleKey, + } + } + return definitions +}