Add CLI reference binding flags

internal/cli/run.go

@@ -25,7 +25,7 @@ const defaultOutputRoot = "./notarius-output"
 
 const usage = `Usage:
   notarius help
-  notarius run <pipeline-id> --input path/to/source.json [--config path/to/config.yml] [--only lane-a,lane-b]
+  notarius run <pipeline-id> --input path/to/source.json [--config path/to/config.yml] [--only lane-a,lane-b] [--reference slot=path] [--without-reference slot]
   notarius config validate --config path/to/config.yml [--pipeline pipeline-id] [--only lane-a,lane-b]
   notarius pipelines list --config path/to/config.yml [--json]
 `
@@ -95,6 +95,10 @@ func runPipelineCommand(args []string, stdout, stderr io.Writer, opts Options) i
 	outputDir := fs.String("output-dir", "", "output directory")
 	diagnosticsDir := fs.String("diagnostics-dir", "", "diagnostics directory")
 	llmProfile := fs.String("llm-profile", "", "LLM profile override")
+	referenceFlags := stringListFlag{}
+	withoutReferenceFlags := stringListFlag{}
+	fs.Var(&referenceFlags, "reference", "reference binding, as slot=path or lane.slot=path")
+	fs.Var(&withoutReferenceFlags, "without-reference", "unbind a reference, as slot or lane.slot")
 	if err := fs.Parse(reorderRunArgs(args)); err != nil {
 		fmt.Fprintf(stderr, "notarius: %v\n", err)
 		return 2
@@ -121,6 +125,16 @@ func runPipelineCommand(args []string, stdout, stderr io.Writer, opts Options) i
 		fmt.Fprintf(stderr, "notarius: %v\n", err)
 		return 2
 	}
+	referenceRequests, err := parseReferenceFlags(referenceFlags)
+	if err != nil {
+		fmt.Fprintf(stderr, "notarius: %v\n", err)
+		return 2
+	}
+	referenceUnbindRequests, err := parseReferenceUnbindFlags(withoutReferenceFlags)
+	if err != nil {
+		fmt.Fprintf(stderr, "notarius: %v\n", err)
+		return 2
+	}
 
 	cfg, loadedConfigPath, err := loadConfig(*configPath, opts)
 	if err != nil {
@@ -155,11 +169,17 @@ func runPipelineCommand(args []string, stdout, stderr io.Writer, opts Options) i
 	if err != nil {
 		return failPipelineCommand(stderr, runDir, cfg.Diagnostics.Retention, err)
 	}
+	referenceOverrides, referenceUnbinds, err := resolveCLIReferenceRequests(cfg, pipelineID, only, catalog, referenceRequests, referenceUnbindRequests)
+	if err != nil {
+		return failPipelineCommand(stderr, runDir, cfg.Diagnostics.Retention, err)
+	}
 	effective, err := cfg.Resolve(config.ResolveInput{
 		PipelineID:         pipelineID,
 		Only:               only,
 		Catalog:            catalog,
 		LLMProfileOverride: *llmProfile,
+		ReferenceOverrides: referenceOverrides,
+		ReferenceUnbinds:   referenceUnbinds,
 	})
 	if err != nil {
 		return failPipelineCommand(stderr, runDir, cfg.Diagnostics.Retention, err)
@@ -412,7 +432,7 @@ func reorderRunArgs(args []string) []string {
 
 func runFlagTakesValue(arg string) bool {
 	switch arg {
-	case "--config", "--input", "--only", "--output-dir", "--diagnostics-dir", "--llm-profile":
+	case "--config", "--input", "--only", "--output-dir", "--diagnostics-dir", "--llm-profile", "--reference", "--without-reference":
 		return true
 	default:
 		return false
@@ -661,6 +681,255 @@ func parseOnly(raw string) ([]string, error) {
 	return result, nil
 }
 
+type stringListFlag []string
+
+func (flag *stringListFlag) String() string {
+	if flag == nil {
+		return ""
+	}
+	return strings.Join(*flag, ",")
+}
+
+func (flag *stringListFlag) Set(value string) error {
+	*flag = append(*flag, value)
+	return nil
+}
+
+type cliReferenceRequest struct {
+	LaneID   string
+	SlotName string
+	Source   string
+}
+
+type cliReferenceUnbindRequest struct {
+	LaneID   string
+	SlotName string
+}
+
+func parseReferenceFlags(values []string) ([]cliReferenceRequest, error) {
+	if len(values) == 0 {
+		return nil, nil
+	}
+	requests := make([]cliReferenceRequest, 0, len(values))
+	for _, raw := range values {
+		name, source, ok := strings.Cut(raw, "=")
+		if !ok {
+			return nil, fmt.Errorf("--reference must use slot=path or lane.slot=path")
+		}
+		if strings.TrimSpace(source) == "" {
+			return nil, fmt.Errorf("--reference path must not be empty; use --without-reference to unbind")
+		}
+		laneID, slotName, err := parseReferenceSelector(name, "--reference")
+		if err != nil {
+			return nil, err
+		}
+		requests = append(requests, cliReferenceRequest{
+			LaneID:   laneID,
+			SlotName: slotName,
+			Source:   strings.TrimSpace(source),
+		})
+	}
+	return requests, nil
+}
+
+func parseReferenceUnbindFlags(values []string) ([]cliReferenceUnbindRequest, error) {
+	if len(values) == 0 {
+		return nil, nil
+	}
+	requests := make([]cliReferenceUnbindRequest, 0, len(values))
+	for _, raw := range values {
+		if strings.Contains(raw, "=") {
+			return nil, fmt.Errorf("--without-reference must use slot or lane.slot")
+		}
+		laneID, slotName, err := parseReferenceSelector(raw, "--without-reference")
+		if err != nil {
+			return nil, err
+		}
+		requests = append(requests, cliReferenceUnbindRequest{
+			LaneID:   laneID,
+			SlotName: slotName,
+		})
+	}
+	return requests, nil
+}
+
+func parseReferenceSelector(raw string, flagName string) (string, string, error) {
+	selector := strings.TrimSpace(raw)
+	if selector == "" {
+		return "", "", fmt.Errorf("%s reference slot must not be empty", flagName)
+	}
+	if strings.Count(selector, ".") > 1 {
+		return "", "", fmt.Errorf("%s must use slot or lane.slot", flagName)
+	}
+	laneID := ""
+	slotName := selector
+	if strings.Contains(selector, ".") {
+		before, after, _ := strings.Cut(selector, ".")
+		laneID = strings.TrimSpace(before)
+		slotName = strings.TrimSpace(after)
+		if laneID == "" || slotName == "" {
+			return "", "", fmt.Errorf("%s must use non-empty lane.slot values", flagName)
+		}
+	}
+	return laneID, slotName, nil
+}
+
+func resolveCLIReferenceRequests(
+	cfg config.Config,
+	pipelineID string,
+	only []string,
+	catalog pipeline.ModuleCatalog,
+	referenceRequests []cliReferenceRequest,
+	unbindRequests []cliReferenceUnbindRequest,
+) ([]pipeline.ReferenceBinding, []pipeline.ReferenceUnbind, error) {
+	if len(referenceRequests) == 0 && len(unbindRequests) == 0 {
+		return nil, nil, nil
+	}
+
+	selected, err := selectedReferenceLanes(cfg, pipelineID, only, catalog)
+	if err != nil {
+		return nil, nil, err
+	}
+
+	overrides := make([]pipeline.ReferenceBinding, 0, len(referenceRequests))
+	for _, request := range referenceRequests {
+		laneID, err := resolveCLIReferenceLane(selected, request.LaneID, request.SlotName)
+		if err != nil {
+			return nil, nil, err
+		}
+		overrides = append(overrides, pipeline.ReferenceBinding{
+			LaneID:        laneID,
+			SlotName:      request.SlotName,
+			Source:        request.Source,
+			BindingSource: contracts.ReferenceBindingSourceCLI,
+		})
+	}
+
+	unbinds := make([]pipeline.ReferenceUnbind, 0, len(unbindRequests))
+	for _, request := range unbindRequests {
+		laneID, err := resolveCLIReferenceLane(selected, request.LaneID, request.SlotName)
+		if err != nil {
+			return nil, nil, err
+		}
+		unbinds = append(unbinds, pipeline.ReferenceUnbind{
+			LaneID:   laneID,
+			SlotName: request.SlotName,
+		})
+	}
+
+	return overrides, unbinds, nil
+}
+
+type selectedReferenceLane struct {
+	id    string
+	slots map[string]struct{}
+}
+
+func selectedReferenceLanes(cfg config.Config, pipelineID string, only []string, catalog pipeline.ModuleCatalog) ([]selectedReferenceLane, error) {
+	profile, ok := lookupCLIReferencePipeline(cfg.Pipelines, pipelineID)
+	if !ok {
+		return nil, fmt.Errorf("pipeline %q is not configured", strings.TrimSpace(pipelineID))
+	}
+
+	lanesByID := make(map[string]pipeline.ArtifactLaneProfile, len(profile.Artifacts))
+	for rawLaneID, lane := range profile.Artifacts {
+		laneID := strings.TrimSpace(rawLaneID)
+		if laneID == "" {
+			return nil, fmt.Errorf("pipeline %q artifact lane id must not be empty", strings.TrimSpace(pipelineID))
+		}
+		if _, ok := lanesByID[laneID]; ok {
+			return nil, fmt.Errorf("pipeline %q artifact lane %q is duplicated after trimming", strings.TrimSpace(pipelineID), laneID)
+		}
+		lanesByID[laneID] = lane
+	}
+
+	selectedIDs := make([]string, 0, len(lanesByID))
+	if len(only) == 0 {
+		for laneID := range lanesByID {
+			selectedIDs = append(selectedIDs, laneID)
+		}
+	} else {
+		seen := make(map[string]struct{}, len(only))
+		for _, rawLaneID := range only {
+			laneID := strings.TrimSpace(rawLaneID)
+			if laneID == "" {
+				return nil, fmt.Errorf("pipeline %q selected artifact lane id must not be empty", strings.TrimSpace(pipelineID))
+			}
+			if _, ok := lanesByID[laneID]; !ok {
+				return nil, fmt.Errorf("pipeline %q selected artifact lane %q is not declared", strings.TrimSpace(pipelineID), laneID)
+			}
+			if _, ok := seen[laneID]; !ok {
+				selectedIDs = append(selectedIDs, laneID)
+				seen[laneID] = struct{}{}
+			}
+		}
+	}
+	sort.Strings(selectedIDs)
+
+	selected := make([]selectedReferenceLane, 0, len(selectedIDs))
+	for _, laneID := range selectedIDs {
+		lane := lanesByID[laneID]
+		extractModule := strings.TrimSpace(lane.Extract.Module)
+		if extractModule == "" {
+			return nil, fmt.Errorf("pipeline %q lane %q extract module must not be empty", strings.TrimSpace(pipelineID), laneID)
+		}
+		if catalog.Extractors == nil {
+			return nil, fmt.Errorf("pipeline %q lane %q extract module %q: module %q is not registered", strings.TrimSpace(pipelineID), laneID, extractModule, extractModule)
+		}
+		spec, ok := catalog.Extractors.Spec(extractModule)
+		if !ok {
+			return nil, fmt.Errorf("pipeline %q lane %q extract module %q: module %q is not registered", strings.TrimSpace(pipelineID), laneID, extractModule, extractModule)
+		}
+		slotSet := make(map[string]struct{}, len(spec.ReferenceSlots))
+		for _, slot := range spec.ReferenceSlots {
+			slotSet[slot.Name] = struct{}{}
+		}
+		selected = append(selected, selectedReferenceLane{id: laneID, slots: slotSet})
+	}
+	return selected, nil
+}
+
+func lookupCLIReferencePipeline(profiles map[string]pipeline.PipelineProfile, pipelineID string) (pipeline.PipelineProfile, bool) {
+	pipelineID = strings.TrimSpace(pipelineID)
+	for rawID, profile := range profiles {
+		if strings.TrimSpace(rawID) == pipelineID {
+			return profile, true
+		}
+	}
+	return pipeline.PipelineProfile{}, false
+}
+
+func resolveCLIReferenceLane(selected []selectedReferenceLane, requestedLaneID string, slotName string) (string, error) {
+	slotName = strings.TrimSpace(slotName)
+	requestedLaneID = strings.TrimSpace(requestedLaneID)
+	if requestedLaneID != "" {
+		for _, lane := range selected {
+			if lane.id == requestedLaneID {
+				if _, ok := lane.slots[slotName]; !ok {
+					return "", fmt.Errorf("reference slot %q is not declared by selected lane %q", slotName, requestedLaneID)
+				}
+				return requestedLaneID, nil
+			}
+		}
+		return "", fmt.Errorf("reference lane %q is not selected", requestedLaneID)
+	}
+
+	matches := make([]string, 0, 1)
+	for _, lane := range selected {
+		if _, ok := lane.slots[slotName]; ok {
+			matches = append(matches, lane.id)
+		}
+	}
+	switch len(matches) {
+	case 0:
+		return "", fmt.Errorf("reference slot %q is not declared by any selected lane", slotName)
+	case 1:
+		return matches[0], nil
+	default:
+		return "", fmt.Errorf("reference slot %q is declared by multiple selected lanes (%s); use lane.slot", slotName, strings.Join(matches, ", "))
+	}
+}
+
 func sortedPipelineIDs(cfg config.Config) []string {
 	ids := make([]string, 0, len(cfg.Pipelines))
 	for id := range cfg.Pipelines {

internal/cli/run_test.go

@@ -8,6 +8,7 @@ import (
 	"os"
 	"path/filepath"
 	"reflect"
+	"sort"
 	"strings"
 	"testing"
 
@@ -737,6 +738,217 @@ func TestRunPipelineLLMProfileOverrideSelectsFactoryProfile(t *testing.T) {
 	}
 }
 
+func TestRunPipelineReferenceFlagBindsUnambiguousSlot(t *testing.T) {
+	configPath := writeTestConfig(t, testConfigYAML("example", "events"))
+	inputPath := filepath.Join(t.TempDir(), "missing.json")
+	diagnosticsDir := t.TempDir()
+	var stdout bytes.Buffer
+	var stderr bytes.Buffer
+
+	code := RunWithOptions([]string{
+		"run", "example",
+		"--config", configPath,
+		"--input", inputPath,
+		"--diagnostics-dir", diagnosticsDir,
+		"--reference", "roster=./roster.yml",
+	}, &stdout, &stderr, Options{
+		Catalog: fakeCatalog(t, pipeline.ModuleSpec{
+			Key:      "fake/extract",
+			Stage:    pipeline.StageExtract,
+			Requires: []string{"chunks"},
+			Provides: []string{"artifact"},
+			ReferenceSlots: []contracts.ReferenceSlot{
+				{Name: "roster"},
+			},
+		}),
+	})
+
+	if code != 1 || !strings.Contains(stderr.String(), "read input") {
+		t.Fatalf("RunWithOptions() code = %d stderr=%q, want read input failure after resolution", code, stderr.String())
+	}
+	resolved := readResolvedPipeline(t, diagnosticsDir)
+	refs := resolved.ArtifactLanes[0].References
+	want := []pipeline.ReferenceBinding{
+		{LaneID: "events", SlotName: "roster", Source: "./roster.yml", BindingSource: contracts.ReferenceBindingSourceCLI},
+	}
+	if !reflect.DeepEqual(refs, want) {
+		t.Fatalf("resolved references = %#v, want %#v", refs, want)
+	}
+}
+
+func TestRunPipelineReferenceFlagBindsLaneQualifiedSlot(t *testing.T) {
+	configPath := writeTestConfig(t, testConfigYAML("example", "events", "notes"))
+	inputPath := filepath.Join(t.TempDir(), "missing.json")
+	diagnosticsDir := t.TempDir()
+	var stdout bytes.Buffer
+	var stderr bytes.Buffer
+
+	code := RunWithOptions([]string{
+		"run", "example",
+		"--config", configPath,
+		"--input", inputPath,
+		"--only", "events,notes",
+		"--diagnostics-dir", diagnosticsDir,
+		"--reference", "notes.roster=./notes.yml",
+	}, &stdout, &stderr, Options{
+		Catalog: fakeCatalog(t, pipeline.ModuleSpec{
+			Key:      "fake/extract",
+			Stage:    pipeline.StageExtract,
+			Requires: []string{"chunks"},
+			Provides: []string{"artifact"},
+			ReferenceSlots: []contracts.ReferenceSlot{
+				{Name: "roster"},
+			},
+		}),
+	})
+
+	if code != 1 || !strings.Contains(stderr.String(), "read input") {
+		t.Fatalf("RunWithOptions() code = %d stderr=%q, want read input failure after resolution", code, stderr.String())
+	}
+	resolved := readResolvedPipeline(t, diagnosticsDir)
+	events := resolvedArtifactLane(t, resolved, "events")
+	if len(events.References) != 0 {
+		t.Fatalf("events references = %#v, want none", events.References)
+	}
+	notes := resolvedArtifactLane(t, resolved, "notes")
+	if len(notes.References) != 1 || notes.References[0].Source != "./notes.yml" {
+		t.Fatalf("notes references = %#v, want lane-qualified binding", notes.References)
+	}
+}
+
+func TestRunPipelineReferenceFlagRejectsAmbiguousFlatSlot(t *testing.T) {
+	configPath := writeTestConfig(t, testConfigYAML("example", "events", "notes"))
+	inputPath := writeSeriatimInput(t)
+	diagnosticsDir := t.TempDir()
+	var stdout bytes.Buffer
+	var stderr bytes.Buffer
+
+	code := RunWithOptions([]string{
+		"run", "example",
+		"--config", configPath,
+		"--input", inputPath,
+		"--diagnostics-dir", diagnosticsDir,
+		"--reference", "roster=./roster.yml",
+	}, &stdout, &stderr, Options{
+		Catalog: fakeCatalog(t, pipeline.ModuleSpec{
+			Key:      "fake/extract",
+			Stage:    pipeline.StageExtract,
+			Requires: []string{"chunks"},
+			Provides: []string{"artifact"},
+			ReferenceSlots: []contracts.ReferenceSlot{
+				{Name: "roster"},
+			},
+		}),
+	})
+
+	if code != 1 {
+		t.Fatalf("RunWithOptions() code = %d, want 1", code)
+	}
+	if !strings.Contains(stderr.String(), "multiple selected lanes") || !strings.Contains(stderr.String(), "lane.slot") {
+		t.Fatalf("stderr = %q, want ambiguous reference error", stderr.String())
+	}
+}
+
+func TestRunPipelineReferenceFlagsRejectMalformedValues(t *testing.T) {
+	tests := []struct {
+		name string
+		args []string
+		want string
+	}{
+		{name: "missing equals", args: []string{"--reference", "roster"}, want: "slot=path"},
+		{name: "empty path", args: []string{"--reference", "roster="}, want: "path must not be empty"},
+		{name: "empty slot", args: []string{"--reference", "=./roster.yml"}, want: "slot must not be empty"},
+		{name: "too many selector parts", args: []string{"--reference", "a.b.c=./roster.yml"}, want: "lane.slot"},
+		{name: "unbind with equals", args: []string{"--without-reference", "roster=./roster.yml"}, want: "slot or lane.slot"},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			configPath := writeTestConfig(t, testConfigYAML("example", "events"))
+			inputPath := writeSeriatimInput(t)
+			var stdout bytes.Buffer
+			var stderr bytes.Buffer
+			args := []string{"run", "example", "--config", configPath, "--input", inputPath}
+			args = append(args, test.args...)
+
+			code := RunWithOptions(args, &stdout, &stderr, Options{Catalog: fakeCatalog(t)})
+			if code != 2 {
+				t.Fatalf("RunWithOptions() code = %d, want 2", code)
+			}
+			if !strings.Contains(stderr.String(), test.want) {
+				t.Fatalf("stderr = %q, want substring %q", stderr.String(), test.want)
+			}
+		})
+	}
+}
+
+func TestRunPipelineWithoutReferenceRemovesOptionalConfigBinding(t *testing.T) {
+	configPath := writeTestConfig(t, testConfigYAMLWithReferences("example", "events", map[string]string{"roster": "./config-roster.yml"}))
+	inputPath := filepath.Join(t.TempDir(), "missing.json")
+	diagnosticsDir := t.TempDir()
+	var stdout bytes.Buffer
+	var stderr bytes.Buffer
+
+	code := RunWithOptions([]string{
+		"run", "example",
+		"--config", configPath,
+		"--input", inputPath,
+		"--diagnostics-dir", diagnosticsDir,
+		"--without-reference", "roster",
+	}, &stdout, &stderr, Options{
+		Catalog: fakeCatalog(t, pipeline.ModuleSpec{
+			Key:      "fake/extract",
+			Stage:    pipeline.StageExtract,
+			Requires: []string{"chunks"},
+			Provides: []string{"artifact"},
+			ReferenceSlots: []contracts.ReferenceSlot{
+				{Name: "roster"},
+			},
+		}),
+	})
+
+	if code != 1 || !strings.Contains(stderr.String(), "read input") {
+		t.Fatalf("RunWithOptions() code = %d stderr=%q, want read input failure after resolution", code, stderr.String())
+	}
+	resolved := readResolvedPipeline(t, diagnosticsDir)
+	if refs := resolved.ArtifactLanes[0].References; len(refs) != 0 {
+		t.Fatalf("references = %#v, want unbound optional slot", refs)
+	}
+}
+
+func TestRunPipelineWithoutReferenceFailsWhenRequiredSlotWouldBeMissing(t *testing.T) {
+	configPath := writeTestConfig(t, testConfigYAMLWithReferences("example", "events", map[string]string{"roster": "./config-roster.yml"}))
+	inputPath := writeSeriatimInput(t)
+	diagnosticsDir := t.TempDir()
+	var stdout bytes.Buffer
+	var stderr bytes.Buffer
+
+	code := RunWithOptions([]string{
+		"run", "example",
+		"--config", configPath,
+		"--input", inputPath,
+		"--diagnostics-dir", diagnosticsDir,
+		"--without-reference", "roster",
+	}, &stdout, &stderr, Options{
+		Catalog: fakeCatalog(t, pipeline.ModuleSpec{
+			Key:      "fake/extract",
+			Stage:    pipeline.StageExtract,
+			Requires: []string{"chunks"},
+			Provides: []string{"artifact"},
+			ReferenceSlots: []contracts.ReferenceSlot{
+				{Name: "roster", Required: true},
+			},
+		}),
+	})
+
+	if code != 1 {
+		t.Fatalf("RunWithOptions() code = %d, want 1", code)
+	}
+	if !strings.Contains(stderr.String(), "required reference slot") || !strings.Contains(stderr.String(), "roster") {
+		t.Fatalf("stderr = %q, want required reference error", stderr.String())
+	}
+}
+
 func TestRunPipelineWritesDurableOutputFiles(t *testing.T) {
 	diagnosticsDir := t.TempDir()
 	outputDir := t.TempDir()
@@ -1305,6 +1517,27 @@ func testConfigYAMLForPipelines(pipelines map[string][]string) string {
 	return b.String()
 }
 
+func testConfigYAMLWithReferences(pipelineID string, laneID string, references map[string]string) string {
+	var b strings.Builder
+	b.WriteString("version: 1\n")
+	b.WriteString("pipelines:\n")
+	b.WriteString("  " + pipelineID + ":\n")
+	b.WriteString("    input: fake/input\n")
+	b.WriteString("    artifacts:\n")
+	b.WriteString("      " + laneID + ":\n")
+	b.WriteString("        extract: fake/extract\n")
+	b.WriteString("        references:\n")
+	keys := make([]string, 0, len(references))
+	for key := range references {
+		keys = append(keys, key)
+	}
+	sort.Strings(keys)
+	for _, key := range keys {
+		b.WriteString("          " + key + ": " + references[key] + "\n")
+	}
+	return b.String()
+}
+
 func mvpConfigYAML(pipelineID string, extractor string) string {
 	return `version: 1
 pipelines:
@@ -1596,6 +1829,25 @@ func readJSONFile(t *testing.T, path string, out any) {
 	}
 }
 
+func readResolvedPipeline(t *testing.T, diagnosticsDir string) pipeline.ResolvedPipeline {
+	t.Helper()
+	runDir := onlyChildDir(t, diagnosticsDir)
+	var resolved pipeline.ResolvedPipeline
+	readJSONFile(t, filepath.Join(runDir, diagnostics.ArtifactResolvedPipeline), &resolved)
+	return resolved
+}
+
+func resolvedArtifactLane(t *testing.T, resolved pipeline.ResolvedPipeline, laneID string) pipeline.ResolvedArtifactLane {
+	t.Helper()
+	for _, lane := range resolved.ArtifactLanes {
+		if lane.ID == laneID {
+			return lane
+		}
+	}
+	t.Fatalf("lane %q not found in resolved pipeline", laneID)
+	return pipeline.ResolvedArtifactLane{}
+}
+
 func assertNoTemporaryFiles(t *testing.T, root string) {
 	t.Helper()
 	if err := filepath.WalkDir(root, func(path string, entry os.DirEntry, err error) error {
@@ -1611,7 +1863,7 @@ func assertNoTemporaryFiles(t *testing.T, root string) {
 	}
 }
 
-func fakeCatalog(t *testing.T) pipeline.ModuleCatalog {
+func fakeCatalog(t *testing.T, overrides ...pipeline.ModuleSpec) pipeline.ModuleCatalog {
 	t.Helper()
 	inputs := pipeline.NewInputAdapterRegistry()
 	chunkers := pipeline.NewChunkerRegistry()
@@ -1621,12 +1873,24 @@ func fakeCatalog(t *testing.T) pipeline.ModuleCatalog {
 	validators := pipeline.NewValidatorRegistry()
 	outputs := pipeline.NewOutputEncoderRegistry()
 
-	mustRegisterInput(t, inputs, pipeline.ModuleSpec{Key: "fake/input", Stage: pipeline.StageInput, Provides: []string{"source"}})
-	mustRegisterChunker(t, chunkers, pipeline.ModuleSpec{Key: "generic", Stage: pipeline.StageChunk, Requires: []string{"source"}, Provides: []string{"chunks"}})
-	mustRegisterExtractor(t, extractors, pipeline.ModuleSpec{Key: "fake/extract", Stage: pipeline.StageExtract, Requires: []string{"chunks"}, Provides: []string{"artifact"}})
-	mustRegisterMerger(t, mergers, pipeline.ModuleSpec{Key: "appendorder", Stage: pipeline.StageMerge, Requires: []string{"artifact"}, Provides: []string{"merged"}})
-	mustRegisterNormalizer(t, normalizers, pipeline.ModuleSpec{Key: "noop", Stage: pipeline.StageNormalize, Requires: []string{"merged"}, Provides: []string{"normalized"}})
-	mustRegisterOutput(t, outputs, pipeline.ModuleSpec{Key: "json", Stage: pipeline.StageOutput, Requires: []string{"normalized"}})
+	specs := map[string]pipeline.ModuleSpec{
+		"fake/input":   {Key: "fake/input", Stage: pipeline.StageInput, Provides: []string{"source"}},
+		"generic":      {Key: "generic", Stage: pipeline.StageChunk, Requires: []string{"source"}, Provides: []string{"chunks"}},
+		"fake/extract": {Key: "fake/extract", Stage: pipeline.StageExtract, Requires: []string{"chunks"}, Provides: []string{"artifact"}},
+		"appendorder":  {Key: "appendorder", Stage: pipeline.StageMerge, Requires: []string{"artifact"}, Provides: []string{"merged"}},
+		"noop":         {Key: "noop", Stage: pipeline.StageNormalize, Requires: []string{"merged"}, Provides: []string{"normalized"}},
+		"json":         {Key: "json", Stage: pipeline.StageOutput, Requires: []string{"normalized"}},
+	}
+	for _, override := range overrides {
+		specs[override.Key] = override
+	}
+
+	mustRegisterInput(t, inputs, specs["fake/input"])
+	mustRegisterChunker(t, chunkers, specs["generic"])
+	mustRegisterExtractor(t, extractors, specs["fake/extract"])
+	mustRegisterMerger(t, mergers, specs["appendorder"])
+	mustRegisterNormalizer(t, normalizers, specs["noop"])
+	mustRegisterOutput(t, outputs, specs["json"])
 
 	return pipeline.ModuleCatalog{
 		Inputs:      inputs,