diff --git a/docs/config.md b/docs/config.md index 3f5db56..563e540 100644 --- a/docs/config.md +++ b/docs/config.md @@ -124,9 +124,9 @@ Pipeline fields: - `artifacts`: required for pipeline resolution. It maps artifact lane IDs to lane definitions. - `output`: optional module binding. Default module is `json`. -- `references`: optional map of extractor reference slot names to reference - paths. These bindings are defaults for artifact lanes whose extractor declares - the matching slot. +- `references`: optional map of reference slot names to reference paths. These + bindings are defaults for eligible pipeline targets that declare the matching + slot. Artifact lane fields: @@ -142,21 +142,23 @@ Artifact lane fields: against the production module catalog and fail fast for unknown or incompatible module keys. -Reference bindings are validated against extractor-declared slots during -pipeline resolution. Required slots must be bound after config defaults, -extractor binding references, lane-level compatibility bindings, and run-time -`--reference` or `--without-reference` overrides are applied. Config-relative -paths are resolved relative to the config file; CLI reference paths are resolved -relative to the current working directory. Bound files must be UTF-8 text and -are passed only to lane extractors that declare the slot. Reference media types -are inferred from file extensions, recorded as canonical base media types, and -checked only when a module declares `AcceptedMediaTypes`; unknown extensions are -recorded as `application/octet-stream`. Reference content is not written to +Reference bindings are validated against reference slots declared by eligible +chunk, extract, and normalize targets during pipeline resolution. Required slots +must be bound after config defaults, target-local references, lane-level +compatibility bindings, and extractor run-time `--reference` or +`--without-reference` overrides are applied. Config-relative paths are resolved +relative to the config file; CLI reference paths are resolved relative to the +current working directory. Materialized bound files must be UTF-8 text and are +currently passed only to lane extractors that declare the slot. Reference media +types are inferred from file extensions, recorded as canonical base media types, +and checked only when a module declares `AcceptedMediaTypes`; unknown extensions +are recorded as `application/octet-stream`. Reference content is not written to diagnostics, logs, errors, or manifests. Pipeline-level `references` are defaults. They are valid when at least one -declared lane in the pipeline has an extractor that declares the slot. During a -run, they apply only to selected lanes whose extractor declares the slot: +eligible target in the full configured pipeline declares the slot, including +chunk, extractor, and normalizer targets. During a run, they apply only to the +selected targets that declare the slot: ```yaml pipelines: @@ -191,9 +193,9 @@ pipelines: ``` `chunk.references` and `normalize.references` are accepted in object-form -bindings and preserved in the effective configuration. They are validated as -reference maps, but current reference materialization still delivers content -only to extractor bindings. +bindings. They override pipeline-level defaults for slots declared by the chunk +or normalizer module. Extractor-local references apply only to the extractor, +and normalizer-local references apply only to the normalizer. ## Module Bindings diff --git a/docs/internal/pipeline.md b/docs/internal/pipeline.md index a63e31a..1a18ae5 100644 --- a/docs/internal/pipeline.md +++ b/docs/internal/pipeline.md @@ -31,21 +31,21 @@ before execution: The CLI writes the resolved pipeline and digest to diagnostics. Pipeline profiles and artifact lanes may include reference binding maps keyed by -extractor reference slot name. During resolution, pipeline-level bindings act as -defaults for selected lanes whose extractor declares the slot, lane-level -bindings override or add lane bindings, runtime `--reference` requests override -config bindings, and runtime unbinds remove optional bindings. Flat runtime slot -names are resolved only when exactly one selected lane declares the slot; -otherwise the CLI requires `lane.slot`. Resolution validates bindings against -extractor specs and stores the bindings in target-aware resolved reference -holders. It does not read reference files or include reference bytes in source -digests. +reference slot name. During resolution, pipeline-level bindings act as defaults +for selected chunk, extractor, and normalizer targets that declare the slot; +target-local bindings override or add bindings for that target. Runtime +`--reference` requests override extractor config bindings, and runtime unbinds +remove optional extractor bindings. Flat runtime slot names are resolved only +when exactly one selected extractor lane declares the slot; otherwise the CLI +requires `lane.slot`. Resolution validates bindings against the declaring +target specs and stores the bindings in target-aware resolved reference holders. +It does not read reference files or include reference bytes in source digests. -During run preparation, resolved file references are materialized before any -LLM-backed pipeline work. Config bindings resolve relative to the config file, -CLI bindings resolve relative to the current working directory, and materialized -reference content is passed only to the matching lane extractor through -`ExtractionRequest`. Materialization accepts UTF-8 text files, computes +During run preparation, resolved extractor file references are materialized +before any LLM-backed pipeline work. Config bindings resolve relative to the +config file, CLI bindings resolve relative to the current working directory, and +materialized reference content is passed only to the matching lane extractor +through `ExtractionRequest`. Materialization accepts UTF-8 text files, computes `sha256:` content digests, records file origins, infers canonical base media types from file extensions, enforces declared byte limits, and warns for empty bound files. Media-type acceptance is checked only when a slot declares diff --git a/internal/core/config/redaction_test.go b/internal/core/config/redaction_test.go index 5154361..ad05b4f 100644 --- a/internal/core/config/redaction_test.go +++ b/internal/core/config/redaction_test.go @@ -77,16 +77,36 @@ func TestEffectiveConfigRedactedDiagnosticsPayloadRedactsAndCopies(t *testing.T) effective, err := cfg.Resolve(ResolveInput{ PipelineID: "example", Only: []string{"events"}, - Catalog: fakeCatalog(t, pipeline.ModuleSpec{ - Key: "fake/extract", - Stage: pipeline.StageExtract, - Requires: []string{"chunks"}, - Provides: []string{"artifact"}, - ReferenceSlots: []contracts.ReferenceSlot{ - {Name: "glossary"}, - {Name: "roster"}, + Catalog: fakeCatalog(t, + pipeline.ModuleSpec{ + Key: "generic", + Stage: pipeline.StageChunk, + Requires: []string{"source"}, + Provides: []string{"chunks"}, + ReferenceSlots: []contracts.ReferenceSlot{ + {Name: "scene_guide"}, + }, }, - }), + pipeline.ModuleSpec{ + Key: "fake/extract", + Stage: pipeline.StageExtract, + Requires: []string{"chunks"}, + Provides: []string{"artifact"}, + ReferenceSlots: []contracts.ReferenceSlot{ + {Name: "glossary"}, + {Name: "roster"}, + }, + }, + pipeline.ModuleSpec{ + Key: "noop", + Stage: pipeline.StageNormalize, + Requires: []string{"merged"}, + Provides: []string{"normalized"}, + ReferenceSlots: []contracts.ReferenceSlot{ + {Name: "notes"}, + }, + }, + ), }) if err != nil { t.Fatalf("Resolve: %v", err) diff --git a/internal/framework/pipeline/profile.go b/internal/framework/pipeline/profile.go index 66a7a02..574c3dd 100644 --- a/internal/framework/pipeline/profile.go +++ b/internal/framework/pipeline/profile.go @@ -145,15 +145,27 @@ func ResolvePipeline(profile PipelineProfile, options ResolveOptions, catalog Mo if len(selectedLaneIDs) == 0 { return ResolvedPipeline{}, fmt.Errorf("pipeline %q must select at least one artifact lane", pipelineID) } - if err := validatePipelineReferenceDefaults(pipelineID, profile.References, lanesByID, catalog); err != nil { + if err := validatePipelineReferenceDefaults(pipelineID, profile.References, chunkSpec, lanesByID, catalog); err != nil { return ResolvedPipeline{}, err } + chunkReferences, err := resolveReferenceTargetBindings(referenceResolutionTarget{ + PipelineID: pipelineID, + Stage: StageChunk, + Module: chunk.Module, + Slots: chunkSpec.ReferenceSlots, + PipelineReferences: profile.References, + LocalReferences: chunk.References, + Options: options, + }) + if err != nil { + return ResolvedPipeline{}, err + } resolved := ResolvedPipeline{ ID: pipelineID, Input: input, Chunk: chunk, - ChunkReferences: referenceTarget(StageChunk, "", chunk.Module, nil), + ChunkReferences: referenceTarget(StageChunk, "", chunk.Module, chunkReferences), Output: resolveBinding(profile.Output, DefaultOutputModule), } outputCapabilities := capabilities.clone() @@ -214,7 +226,16 @@ func resolveArtifactLane( return ResolvedArtifactLane{}, nil, capabilityError(pipelineID, laneID, StageExtract, lane.Extract.Module, missing) } extractReferences := mergeReferenceMaps(profile.References, lane.Extract.References) - references, err := resolveReferenceBindings(pipelineID, laneID, lane.Extract.Module, extractSpec.ReferenceSlots, pipelineReferences, extractReferences, options) + references, err := resolveReferenceTargetBindings(referenceResolutionTarget{ + PipelineID: pipelineID, + LaneID: laneID, + Stage: StageExtract, + Module: lane.Extract.Module, + Slots: extractSpec.ReferenceSlots, + PipelineReferences: pipelineReferences, + LocalReferences: extractReferences, + Options: options, + }) if err != nil { return ResolvedArtifactLane{}, nil, err } @@ -237,7 +258,20 @@ func resolveArtifactLane( if missing, ok := capabilities.missing(normalizeSpec.Requires); ok { return ResolvedArtifactLane{}, nil, capabilityError(pipelineID, laneID, StageNormalize, lane.Normalize.Module, missing) } - lane.NormalizeReferences = referenceTarget(StageNormalize, laneID, lane.Normalize.Module, nil) + normalizeReferences, err := resolveReferenceTargetBindings(referenceResolutionTarget{ + PipelineID: pipelineID, + LaneID: laneID, + Stage: StageNormalize, + Module: lane.Normalize.Module, + Slots: normalizeSpec.ReferenceSlots, + PipelineReferences: pipelineReferences, + LocalReferences: lane.Normalize.References, + Options: options, + }) + if err != nil { + return ResolvedArtifactLane{}, nil, err + } + lane.NormalizeReferences = referenceTarget(StageNormalize, laneID, lane.Normalize.Module, normalizeReferences) capabilities.add(normalizeSpec.Provides...) for _, validator := range lane.Validators { @@ -280,6 +314,7 @@ func mergeReferenceMaps(base map[string]string, override map[string]string) map[ func validatePipelineReferenceDefaults( pipelineID string, pipelineReferences map[string]string, + chunkSpec ModuleSpec, lanesByID map[string]ArtifactLaneProfile, catalog ModuleCatalog, ) error { @@ -291,7 +326,10 @@ func validatePipelineReferenceDefaults( return nil } - declaredByAnyLane := make(map[string]struct{}, len(normalizedPipelineReferences)) + declaredByAnyTarget := make(map[string]struct{}, len(normalizedPipelineReferences)) + for _, slot := range chunkSpec.ReferenceSlots { + declaredByAnyTarget[slot.Name] = struct{}{} + } for _, laneID := range sortedArtifactLaneProfileKeys(lanesByID) { laneProfile := lanesByID[laneID] extract := resolveBinding(laneProfile.Extract, "") @@ -303,29 +341,41 @@ func validatePipelineReferenceDefaults( return moduleLookupError(pipelineID, laneID, StageExtract, extract.Module, err) } for _, slot := range extractSpec.ReferenceSlots { - declaredByAnyLane[slot.Name] = struct{}{} + declaredByAnyTarget[slot.Name] = struct{}{} + } + + normalize := resolveBinding(laneProfile.Normalize, DefaultNormalizeModule) + normalizeSpec, err := normalizerSpec(catalog, normalize.Module) + if err != nil { + return moduleLookupError(pipelineID, laneID, StageNormalize, normalize.Module, err) + } + for _, slot := range normalizeSpec.ReferenceSlots { + declaredByAnyTarget[slot.Name] = struct{}{} } } for _, slotName := range sortedStringMapKeys(normalizedPipelineReferences) { - if _, ok := declaredByAnyLane[slotName]; !ok { - return fmt.Errorf("pipeline %q reference slot %q is not declared by any artifact lane", pipelineID, slotName) + if _, ok := declaredByAnyTarget[slotName]; !ok { + return fmt.Errorf("pipeline %q reference slot %q is not declared by any eligible reference target", pipelineID, slotName) } } return nil } -func resolveReferenceBindings( - pipelineID string, - laneID string, - extractorModule string, - slots []contracts.ReferenceSlot, - pipelineReferences map[string]string, - laneReferences map[string]string, - options ResolveOptions, -) ([]ReferenceBinding, error) { - slotByName := make(map[string]contracts.ReferenceSlot, len(slots)) - for _, slot := range slots { +type referenceResolutionTarget struct { + PipelineID string + LaneID string + Stage ModuleStage + Module string + Slots []contracts.ReferenceSlot + PipelineReferences map[string]string + LocalReferences map[string]string + Options ResolveOptions +} + +func resolveReferenceTargetBindings(target referenceResolutionTarget) ([]ReferenceBinding, error) { + slotByName := make(map[string]contracts.ReferenceSlot, len(target.Slots)) + for _, slot := range target.Slots { slotByName[slot.Name] = slot } @@ -334,16 +384,16 @@ func resolveReferenceBindings( slotName = strings.TrimSpace(slotName) source = strings.TrimSpace(source) if slotName == "" { - return fmt.Errorf("pipeline %q lane %q reference slot name must not be empty", pipelineID, laneID) + return fmt.Errorf("%s reference slot name must not be empty", referenceTargetErrorContext(target)) } if source == "" { - return fmt.Errorf("pipeline %q lane %q reference slot %q source must not be empty", pipelineID, laneID, slotName) + return fmt.Errorf("%s reference slot %q source must not be empty", referenceTargetErrorContext(target), slotName) } if _, ok := slotByName[slotName]; !ok { - return fmt.Errorf("pipeline %q lane %q reference slot %q is not declared by extractor %q", pipelineID, laneID, slotName, extractorModule) + return fmt.Errorf("%s reference slot %q is not declared by %s module %q", referenceTargetErrorContext(target), slotName, target.Stage, target.Module) } bindings[slotName] = ReferenceBinding{ - LaneID: laneID, + LaneID: target.LaneID, SlotName: slotName, Source: source, BindingSource: bindingSource, @@ -351,7 +401,7 @@ func resolveReferenceBindings( return nil } - normalizedPipelineReferences, err := normalizedReferenceMap(pipelineReferences, fmt.Sprintf("pipeline %q reference slot", pipelineID)) + normalizedPipelineReferences, err := normalizedReferenceMap(target.PipelineReferences, fmt.Sprintf("pipeline %q reference slot", target.PipelineID)) if err != nil { return nil, err } @@ -364,22 +414,33 @@ func resolveReferenceBindings( } } - normalizedLaneReferences, err := normalizedReferenceMap(laneReferences, fmt.Sprintf("pipeline %q lane %q reference slot", pipelineID, laneID)) + normalizedLocalReferences, err := normalizedReferenceMap(target.LocalReferences, referenceTargetSlotLabel(target)) if err != nil { return nil, err } - for _, slotName := range sortedStringMapKeys(normalizedLaneReferences) { - if err := addBinding(slotName, normalizedLaneReferences[slotName], contracts.ReferenceBindingSourceConfig); err != nil { + for _, slotName := range sortedStringMapKeys(normalizedLocalReferences) { + if err := addBinding(slotName, normalizedLocalReferences[slotName], contracts.ReferenceBindingSourceConfig); err != nil { return nil, err } } - for _, override := range options.ReferenceOverrides { + if target.Stage != StageExtract { + for _, slot := range target.Slots { + if slot.Required { + if _, ok := bindings[slot.Name]; !ok { + return nil, fmt.Errorf("%s required reference slot %q is not bound", referenceTargetErrorContext(target), slot.Name) + } + } + } + return sortedReferenceBindings(bindings), nil + } + + for _, override := range target.Options.ReferenceOverrides { optionLaneID := strings.TrimSpace(override.LaneID) if optionLaneID == "" { - return nil, fmt.Errorf("pipeline %q reference override lane id must not be empty", pipelineID) + return nil, fmt.Errorf("pipeline %q reference override lane id must not be empty", target.PipelineID) } - if optionLaneID != laneID { + if optionLaneID != target.LaneID { continue } source := override.BindingSource @@ -391,38 +452,56 @@ func resolveReferenceBindings( } } - for _, unbind := range options.ReferenceUnbinds { + for _, unbind := range target.Options.ReferenceUnbinds { optionLaneID := strings.TrimSpace(unbind.LaneID) if optionLaneID == "" { - return nil, fmt.Errorf("pipeline %q reference unbind lane id must not be empty", pipelineID) + return nil, fmt.Errorf("pipeline %q reference unbind lane id must not be empty", target.PipelineID) } - if optionLaneID != laneID { + if optionLaneID != target.LaneID { continue } slotName := strings.TrimSpace(unbind.SlotName) if slotName == "" { - return nil, fmt.Errorf("pipeline %q lane %q reference unbind slot name must not be empty", pipelineID, laneID) + return nil, fmt.Errorf("%s reference unbind slot name must not be empty", referenceTargetErrorContext(target)) } if _, ok := slotByName[slotName]; !ok { - return nil, fmt.Errorf("pipeline %q lane %q reference slot %q is not declared", pipelineID, laneID, slotName) + return nil, fmt.Errorf("%s reference slot %q is not declared", referenceTargetErrorContext(target), slotName) } delete(bindings, slotName) } - for _, slot := range slots { + for _, slot := range target.Slots { if slot.Required { if _, ok := bindings[slot.Name]; !ok { - return nil, fmt.Errorf("pipeline %q lane %q required reference slot %q is not bound", pipelineID, laneID, slot.Name) + return nil, fmt.Errorf("%s required reference slot %q is not bound", referenceTargetErrorContext(target), slot.Name) } } } + return sortedReferenceBindings(bindings), nil +} + +func sortedReferenceBindings(bindings map[string]ReferenceBinding) []ReferenceBinding { keys := sortedReferenceBindingKeys(bindings) resolved := make([]ReferenceBinding, 0, len(keys)) for _, slotName := range keys { resolved = append(resolved, bindings[slotName]) } - return resolved, nil + return resolved +} + +func referenceTargetErrorContext(target referenceResolutionTarget) string { + if target.LaneID != "" { + return fmt.Sprintf("pipeline %q lane %q %s module %q", target.PipelineID, target.LaneID, target.Stage, target.Module) + } + return fmt.Sprintf("pipeline %q %s module %q", target.PipelineID, target.Stage, target.Module) +} + +func referenceTargetSlotLabel(target referenceResolutionTarget) string { + if target.LaneID != "" { + return fmt.Sprintf("pipeline %q lane %q %s reference slot", target.PipelineID, target.LaneID, target.Stage) + } + return fmt.Sprintf("pipeline %q %s reference slot", target.PipelineID, target.Stage) } func normalizedReferenceMap(values map[string]string, keyName string) (map[string]string, error) { diff --git a/internal/framework/pipeline/profile_test.go b/internal/framework/pipeline/profile_test.go index 96e08d2..1e9b8cb 100644 --- a/internal/framework/pipeline/profile_test.go +++ b/internal/framework/pipeline/profile_test.go @@ -176,6 +176,109 @@ func TestResolvePipelineAppliesReferenceBindings(t *testing.T) { } } +func TestResolvePipelineAppliesPipelineReferenceDefaultToChunkTarget(t *testing.T) { + profile := baselineProfile() + profile.References = map[string]string{"scene_guide": "./scenes.md"} + catalog := newProfileCatalogWithOverrides(t, ModuleSpec{ + Key: "generic", + Stage: StageChunk, + Requires: []string{"source"}, + Provides: []string{"chunk"}, + ReferenceSlots: []contracts.ReferenceSlot{{Name: "scene_guide"}}, + }) + + resolved, err := ResolvePipeline(profile, ResolveOptions{}, catalog) + if err != nil { + t.Fatalf("ResolvePipeline() error = %v, want nil", err) + } + + want := []ReferenceBinding{{SlotName: "scene_guide", Source: "./scenes.md", BindingSource: contracts.ReferenceBindingSourceConfig}} + if !reflect.DeepEqual(resolved.ChunkReferences.Bindings, want) { + t.Fatalf("chunk references = %#v, want %#v", resolved.ChunkReferences.Bindings, want) + } + if refs := resolved.ArtifactLanes[0].ExtractReferences.Bindings; len(refs) != 0 { + t.Fatalf("extract references = %#v, want none", refs) + } + if refs := resolved.ArtifactLanes[0].NormalizeReferences.Bindings; len(refs) != 0 { + t.Fatalf("normalize references = %#v, want none", refs) + } +} + +func TestResolvePipelineAppliesPipelineReferenceDefaultToExtractorTarget(t *testing.T) { + profile := baselineProfile() + profile.References = map[string]string{"roster": "./roster.yml"} + catalog := newProfileCatalogWithOverrides(t, ModuleSpec{ + Key: "event-extractor", + Stage: StageExtract, + Requires: []string{"chunk"}, + Provides: []string{"candidate"}, + ReferenceSlots: []contracts.ReferenceSlot{{Name: "roster"}}, + }) + + resolved, err := ResolvePipeline(profile, ResolveOptions{}, catalog) + if err != nil { + t.Fatalf("ResolvePipeline() error = %v, want nil", err) + } + + want := []ReferenceBinding{{LaneID: "events", SlotName: "roster", Source: "./roster.yml", BindingSource: contracts.ReferenceBindingSourceConfig}} + if !reflect.DeepEqual(resolved.ArtifactLanes[0].ExtractReferences.Bindings, want) { + t.Fatalf("extract references = %#v, want %#v", resolved.ArtifactLanes[0].ExtractReferences.Bindings, want) + } + if refs := resolved.ChunkReferences.Bindings; len(refs) != 0 { + t.Fatalf("chunk references = %#v, want none", refs) + } + if refs := resolved.ArtifactLanes[0].NormalizeReferences.Bindings; len(refs) != 0 { + t.Fatalf("normalize references = %#v, want none", refs) + } +} + +func TestResolvePipelineAppliesPipelineReferenceDefaultToNormalizerTarget(t *testing.T) { + profile := baselineProfile() + profile.References = map[string]string{"normalization_notes": "./normalize.md"} + catalog := newProfileCatalogWithOverrides(t, ModuleSpec{ + Key: "noop", + Stage: StageNormalize, + Requires: []string{"merged"}, + Provides: []string{"normalized"}, + ReferenceSlots: []contracts.ReferenceSlot{{Name: "normalization_notes"}}, + }) + + resolved, err := ResolvePipeline(profile, ResolveOptions{}, catalog) + if err != nil { + t.Fatalf("ResolvePipeline() error = %v, want nil", err) + } + + want := []ReferenceBinding{{LaneID: "events", SlotName: "normalization_notes", Source: "./normalize.md", BindingSource: contracts.ReferenceBindingSourceConfig}} + if !reflect.DeepEqual(resolved.ArtifactLanes[0].NormalizeReferences.Bindings, want) { + t.Fatalf("normalize references = %#v, want %#v", resolved.ArtifactLanes[0].NormalizeReferences.Bindings, want) + } + if refs := resolved.ChunkReferences.Bindings; len(refs) != 0 { + t.Fatalf("chunk references = %#v, want none", refs) + } + if refs := resolved.ArtifactLanes[0].ExtractReferences.Bindings; len(refs) != 0 { + t.Fatalf("extract references = %#v, want none", refs) + } +} + +func TestResolvePipelineAppliesOnePipelineReferenceDefaultToMultipleTargets(t *testing.T) { + profile := baselineProfile() + profile.References = map[string]string{"context": "./context.md"} + catalog := newProfileCatalogWithOverrides(t, + ModuleSpec{Key: "generic", Stage: StageChunk, Requires: []string{"source"}, Provides: []string{"chunk"}, ReferenceSlots: []contracts.ReferenceSlot{{Name: "context"}}}, + ModuleSpec{Key: "event-extractor", Stage: StageExtract, Requires: []string{"chunk"}, Provides: []string{"candidate"}, ReferenceSlots: []contracts.ReferenceSlot{{Name: "context"}}}, + ModuleSpec{Key: "noop", Stage: StageNormalize, Requires: []string{"merged"}, Provides: []string{"normalized"}, ReferenceSlots: []contracts.ReferenceSlot{{Name: "context"}}}, + ) + + resolved, err := ResolvePipeline(profile, ResolveOptions{}, catalog) + if err != nil { + t.Fatalf("ResolvePipeline() error = %v, want nil", err) + } + + assertBindingSource(t, resolved.ChunkReferences.Bindings, "context", "./context.md") + assertBindingSource(t, resolved.ArtifactLanes[0].ExtractReferences.Bindings, "context", "./context.md") + assertBindingSource(t, resolved.ArtifactLanes[0].NormalizeReferences.Bindings, "context", "./context.md") +} + func TestResolvePipelineAllowsPipelineReferenceDeclaredOnlyByUnselectedLane(t *testing.T) { profile := multiLaneProfile() profile.References = map[string]string{"notes_context": "./notes.md"} @@ -198,6 +301,34 @@ func TestResolvePipelineAllowsPipelineReferenceDeclaredOnlyByUnselectedLane(t *t } } +func TestResolvePipelineAllowsPipelineReferenceDeclaredOnlyByUnselectedNormalizer(t *testing.T) { + profile := multiLaneProfile() + profile.References = map[string]string{"notes_context": "./notes.md"} + lane := profile.Artifacts["notes"] + lane.Normalize = Binding("note-normalizer") + profile.Artifacts["notes"] = lane + catalog := newProfileCatalogWithOverrides(t, ModuleSpec{ + Key: "note-normalizer", + Stage: StageNormalize, + Requires: []string{"merged"}, + Provides: []string{"normalized"}, + ReferenceSlots: []contracts.ReferenceSlot{ + {Name: "notes_context"}, + }, + }) + + resolved, err := ResolvePipeline(profile, ResolveOptions{Only: []string{"events"}}, catalog) + if err != nil { + t.Fatalf("ResolvePipeline() error = %v, want nil", err) + } + if refs := resolved.ArtifactLanes[0].ExtractReferences.Bindings; len(refs) != 0 { + t.Fatalf("selected extract references = %#v, want none", refs) + } + if refs := resolved.ArtifactLanes[0].NormalizeReferences.Bindings; len(refs) != 0 { + t.Fatalf("selected normalize references = %#v, want none", refs) + } +} + func TestResolvePipelineRejectsPipelineReferenceNotDeclaredByAnyLane(t *testing.T) { profile := multiLaneProfile() profile.References = map[string]string{"missing": "./missing.md"} @@ -222,6 +353,106 @@ func TestResolvePipelineRejectsUndeclaredReferenceSlot(t *testing.T) { assertErrorContains(t, err, "events", "missing", "not declared") } +func TestResolvePipelineRejectsExtractLocalReferenceDeclaredOnlyByNormalizer(t *testing.T) { + profile := baselineProfile() + lane := profile.Artifacts["events"] + lane.Extract.References = map[string]string{"normalization_notes": "./normalize.md"} + profile.Artifacts["events"] = lane + catalog := newProfileCatalogWithOverrides(t, ModuleSpec{ + Key: "noop", + Stage: StageNormalize, + Requires: []string{"merged"}, + Provides: []string{"normalized"}, + ReferenceSlots: []contracts.ReferenceSlot{{Name: "normalization_notes"}}, + }) + + _, err := ResolvePipeline(profile, ResolveOptions{}, catalog) + if err == nil { + t.Fatal("ResolvePipeline() error = nil, want error") + } + assertErrorContains(t, err, "baseline", "events", "extract", "event-extractor", "normalization_notes", "not declared") +} + +func TestResolvePipelineRejectsNormalizeLocalReferenceDeclaredOnlyByExtractor(t *testing.T) { + profile := baselineProfile() + lane := profile.Artifacts["events"] + lane.Normalize.References = map[string]string{"roster": "./roster.yml"} + profile.Artifacts["events"] = lane + catalog := newProfileCatalogWithOverrides(t, ModuleSpec{ + Key: "event-extractor", + Stage: StageExtract, + Requires: []string{"chunk"}, + Provides: []string{"candidate"}, + ReferenceSlots: []contracts.ReferenceSlot{{Name: "roster"}}, + }) + + _, err := ResolvePipeline(profile, ResolveOptions{}, catalog) + if err == nil { + t.Fatal("ResolvePipeline() error = nil, want error") + } + assertErrorContains(t, err, "baseline", "events", "normalize", "noop", "roster", "not declared") +} + +func TestResolvePipelineRequiresBoundChunkReference(t *testing.T) { + catalog := newProfileCatalogWithOverrides(t, ModuleSpec{ + Key: "generic", + Stage: StageChunk, + Requires: []string{"source"}, + Provides: []string{"chunk"}, + ReferenceSlots: []contracts.ReferenceSlot{{Name: "scene_guide", Required: true}}, + }) + + _, err := ResolvePipeline(baselineProfile(), ResolveOptions{}, catalog) + if err == nil { + t.Fatal("ResolvePipeline() error = nil, want error") + } + assertErrorContains(t, err, "baseline", "chunk", "generic", "required", "scene_guide", "not bound") +} + +func TestResolvePipelineRequiresBoundNormalizeReference(t *testing.T) { + catalog := newProfileCatalogWithOverrides(t, ModuleSpec{ + Key: "noop", + Stage: StageNormalize, + Requires: []string{"merged"}, + Provides: []string{"normalized"}, + ReferenceSlots: []contracts.ReferenceSlot{{Name: "normalization_notes", Required: true}}, + }) + + _, err := ResolvePipeline(baselineProfile(), ResolveOptions{}, catalog) + if err == nil { + t.Fatal("ResolvePipeline() error = nil, want error") + } + assertErrorContains(t, err, "baseline", "events", "normalize", "noop", "required", "normalization_notes", "not bound") +} + +func TestResolvePipelineLocalReferencesOverridePipelineDefaultsForEligibleTargets(t *testing.T) { + profile := baselineProfile() + profile.References = map[string]string{ + "context": "./shared-context.md", + "roster": "./shared-roster.yml", + "normalization_notes": "./shared-normalize.md", + } + profile.Chunk.References = map[string]string{"context": "./chunk-context.md"} + lane := profile.Artifacts["events"] + lane.Extract.References = map[string]string{"roster": "./extract-roster.yml"} + lane.Normalize.References = map[string]string{"normalization_notes": "./local-normalize.md"} + profile.Artifacts["events"] = lane + catalog := newProfileCatalogWithOverrides(t, + ModuleSpec{Key: "generic", Stage: StageChunk, Requires: []string{"source"}, Provides: []string{"chunk"}, ReferenceSlots: []contracts.ReferenceSlot{{Name: "context"}}}, + ModuleSpec{Key: "event-extractor", Stage: StageExtract, Requires: []string{"chunk"}, Provides: []string{"candidate"}, ReferenceSlots: []contracts.ReferenceSlot{{Name: "roster"}}}, + ModuleSpec{Key: "noop", Stage: StageNormalize, Requires: []string{"merged"}, Provides: []string{"normalized"}, ReferenceSlots: []contracts.ReferenceSlot{{Name: "normalization_notes"}}}, + ) + + resolved, err := ResolvePipeline(profile, ResolveOptions{}, catalog) + if err != nil { + t.Fatalf("ResolvePipeline() error = %v, want nil", err) + } + + assertBindingSource(t, resolved.ChunkReferences.Bindings, "context", "./chunk-context.md") + assertBindingSource(t, resolved.ArtifactLanes[0].ExtractReferences.Bindings, "roster", "./extract-roster.yml") + assertBindingSource(t, resolved.ArtifactLanes[0].NormalizeReferences.Bindings, "normalization_notes", "./local-normalize.md") +} + func TestResolvePipelineRequiresBoundReferenceSlotsForSelectedLanes(t *testing.T) { catalog := newProfileCatalogWithOverride(t, ModuleSpec{ Key: "event-extractor", @@ -658,6 +889,21 @@ func assertErrorContains(t *testing.T, err error, values ...string) { } } +func assertBindingSource(t *testing.T, bindings []ReferenceBinding, slotName string, source string) { + t.Helper() + + for _, binding := range bindings { + if binding.SlotName != slotName { + continue + } + if binding.Source != source { + t.Fatalf("binding %q source = %q, want %q in %#v", slotName, binding.Source, source, bindings) + } + return + } + t.Fatalf("binding %q not found in %#v", slotName, bindings) +} + func newProfileCatalog(t *testing.T) ModuleCatalog { t.Helper() @@ -669,19 +915,29 @@ func newProfileCatalog(t *testing.T) ModuleCatalog { func newProfileCatalogWithOverride(t *testing.T, override ModuleSpec) ModuleCatalog { t.Helper() + return newProfileCatalogWithOverrides(t, override) +} + +func newProfileCatalogWithOverrides(t *testing.T, overrides ...ModuleSpec) ModuleCatalog { + t.Helper() + specs := defaultProfileSpecs() - for index, spec := range specs { - if spec.Stage == override.Stage && spec.Key == override.Key { - specs[index] = override - catalog := emptyProfileCatalog() - registerProfileSpecs(t, catalog, specs...) - return catalog + for _, override := range overrides { + replaced := false + for index, spec := range specs { + if spec.Stage == override.Stage && spec.Key == override.Key { + specs[index] = override + replaced = true + break + } + } + if !replaced { + specs = append(specs, override) } } catalog := emptyProfileCatalog() registerProfileSpecs(t, catalog, specs...) - registerProfileSpecs(t, catalog, override) return catalog }