Integrate extraction artifacts into analysis catalog

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, &copy, 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
 	}

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()

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
+}