From 70d733edafbb01d865775257ceb3bace9efbe6bd Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Sun, 5 Jul 2026 14:21:36 +0000 Subject: [PATCH] Resolve extraction reference bindings from config --- docs/internal/pipeline.md | 8 + internal/core/config/config.go | 13 ++ internal/core/config/effective_config.go | 28 ++- internal/core/config/file_config.go | 51 ++++- internal/core/config/file_config_test.go | 77 +++++++ internal/core/config/redaction.go | 11 +- internal/core/config/redaction_test.go | 16 +- internal/core/config/validation.go | 33 +++ internal/core/config/validation_test.go | 67 +++++++ internal/framework/pipeline/profile.go | 212 ++++++++++++++++++-- internal/framework/pipeline/profile_test.go | 143 +++++++++++++ 11 files changed, 625 insertions(+), 34 deletions(-) diff --git a/docs/internal/pipeline.md b/docs/internal/pipeline.md index 13d06c4..c9aefe8 100644 --- a/docs/internal/pipeline.md +++ b/docs/internal/pipeline.md @@ -30,6 +30,14 @@ 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, and runtime unbinds can remove bindings. +Resolution validates bindings against extractor specs and records lane-scoped +binding metadata. It does not read reference files or include reference bytes in +source digests. + ## Registries And Module Specs `pipeline.Registries` holds concrete constructors for execution. A diff --git a/internal/core/config/config.go b/internal/core/config/config.go index e0c24fb..cfc612c 100644 --- a/internal/core/config/config.go +++ b/internal/core/config/config.go @@ -73,6 +73,7 @@ func clonePipelineProfile(in pipeline.PipelineProfile) pipeline.PipelineProfile out.Input = cloneModuleBinding(in.Input) out.Chunk = cloneModuleBinding(in.Chunk) out.Output = cloneModuleBinding(in.Output) + out.References = cloneStringMap(in.References) if len(in.Artifacts) > 0 { out.Artifacts = make(map[string]pipeline.ArtifactLaneProfile, len(in.Artifacts)) for key, lane := range in.Artifacts { @@ -87,6 +88,7 @@ func cloneArtifactLaneProfile(in pipeline.ArtifactLaneProfile) pipeline.Artifact out.Extract = cloneModuleBinding(in.Extract) out.Merge = cloneModuleBinding(in.Merge) out.Normalize = cloneModuleBinding(in.Normalize) + out.References = cloneStringMap(in.References) if len(in.Validators) > 0 { out.Validators = make([]pipeline.ModuleBinding, len(in.Validators)) for i, binding := range in.Validators { @@ -96,6 +98,17 @@ func cloneArtifactLaneProfile(in pipeline.ArtifactLaneProfile) pipeline.Artifact return out } +func cloneStringMap(in map[string]string) map[string]string { + if len(in) == 0 { + return nil + } + out := make(map[string]string, len(in)) + for key, value := range in { + out[key] = value + } + return out +} + func cloneModuleBinding(in pipeline.ModuleBinding) pipeline.ModuleBinding { out := in if len(in.Options) > 0 { diff --git a/internal/core/config/effective_config.go b/internal/core/config/effective_config.go index cc959ca..dc79af5 100644 --- a/internal/core/config/effective_config.go +++ b/internal/core/config/effective_config.go @@ -14,13 +14,17 @@ type ResolveInput struct { Only []string Catalog pipeline.ModuleCatalog LLMProfileOverride string + ReferenceOverrides []pipeline.ReferenceBinding + ReferenceUnbinds []pipeline.ReferenceUnbind } type EffectiveConfig struct { - Config Config - PipelineID string - Only []string - ResolvedPipeline pipeline.ResolvedPipeline + Config Config + PipelineID string + Only []string + ReferenceOverrides []pipeline.ReferenceBinding + ReferenceUnbinds []pipeline.ReferenceUnbind + ResolvedPipeline pipeline.ResolvedPipeline } func (c Config) Resolve(input ResolveInput) (EffectiveConfig, error) { @@ -46,16 +50,22 @@ func (c Config) Resolve(input ResolveInput) (EffectiveConfig, error) { applyLLMProfileOverride(&profile, override) } - resolved, err := pipeline.ResolvePipeline(profile, pipeline.ResolveOptions{Only: input.Only}, input.Catalog) + resolved, err := pipeline.ResolvePipeline(profile, pipeline.ResolveOptions{ + Only: input.Only, + ReferenceOverrides: append([]pipeline.ReferenceBinding(nil), input.ReferenceOverrides...), + ReferenceUnbinds: append([]pipeline.ReferenceUnbind(nil), input.ReferenceUnbinds...), + }, input.Catalog) if err != nil { return EffectiveConfig{}, fmt.Errorf("resolve pipeline %q: %w", pipelineID, err) } return EffectiveConfig{ - Config: cloneConfig(c), - PipelineID: pipelineID, - Only: append([]string(nil), input.Only...), - ResolvedPipeline: resolved, + Config: cloneConfig(c), + PipelineID: pipelineID, + Only: append([]string(nil), input.Only...), + ReferenceOverrides: append([]pipeline.ReferenceBinding(nil), input.ReferenceOverrides...), + ReferenceUnbinds: append([]pipeline.ReferenceUnbind(nil), input.ReferenceUnbinds...), + ResolvedPipeline: resolved, }, nil } diff --git a/internal/core/config/file_config.go b/internal/core/config/file_config.go index 5df0d51..c61646a 100644 --- a/internal/core/config/file_config.go +++ b/internal/core/config/file_config.go @@ -35,10 +35,11 @@ type FileLLMProfile struct { } type FilePipelineProfile struct { - Input fileModuleBinding `yaml:"input"` - Chunk *fileModuleBinding `yaml:"chunk,omitempty"` - Artifacts map[string]FileArtifactLaneProfile `yaml:"artifacts,omitempty"` - Output *fileModuleBinding `yaml:"output,omitempty"` + Input fileModuleBinding `yaml:"input"` + Chunk *fileModuleBinding `yaml:"chunk,omitempty"` + Artifacts map[string]FileArtifactLaneProfile `yaml:"artifacts,omitempty"` + Output *fileModuleBinding `yaml:"output,omitempty"` + References map[string]string `yaml:"references,omitempty"` } type FileArtifactLaneProfile struct { @@ -46,6 +47,7 @@ type FileArtifactLaneProfile struct { Merge *fileModuleBinding `yaml:"merge,omitempty"` Normalize *fileModuleBinding `yaml:"normalize,omitempty"` Validators []fileModuleBinding `yaml:"validators,omitempty"` + References map[string]string `yaml:"references,omitempty"` } type FileConcurrencyConfig struct { @@ -212,6 +214,18 @@ func (c *Config) applyFileConfigWithLookup(fileCfg FileConfig, lookup func(strin if _, _, err := normalizedMapKeys(filePipeline.Artifacts, fmt.Sprintf("pipeline %q artifact lane id", pipelineID)); err != nil { return err } + if _, _, err := normalizedMapKeys(filePipeline.References, fmt.Sprintf("pipeline %q reference slot", pipelineID)); err != nil { + return err + } + for rawLaneID, fileLane := range filePipeline.Artifacts { + laneID := strings.TrimSpace(rawLaneID) + if laneID == "" { + continue + } + if _, _, err := normalizedMapKeys(fileLane.References, fmt.Sprintf("pipeline %q lane %q reference slot", pipelineID, laneID)); err != nil { + return err + } + } } for _, profileID := range profileIDs { @@ -253,9 +267,10 @@ func (c *Config) applyFileConfigWithLookup(fileCfg FileConfig, lookup func(strin return err } profile := pipeline.PipelineProfile{ - ID: pipelineID, - Input: filePipeline.Input.toPipelineBinding(), - Artifacts: make(map[string]pipeline.ArtifactLaneProfile, len(filePipeline.Artifacts)), + ID: pipelineID, + Input: filePipeline.Input.toPipelineBinding(), + Artifacts: make(map[string]pipeline.ArtifactLaneProfile, len(filePipeline.Artifacts)), + References: normalizedStringMap(filePipeline.References), } if filePipeline.Chunk != nil { profile.Chunk = filePipeline.Chunk.toPipelineBinding() @@ -266,7 +281,8 @@ func (c *Config) applyFileConfigWithLookup(fileCfg FileConfig, lookup func(strin for _, laneID := range laneIDs { fileLane := filePipeline.Artifacts[rawLaneIDs[laneID]] lane := pipeline.ArtifactLaneProfile{ - Extract: fileLane.Extract.toPipelineBinding(), + Extract: fileLane.Extract.toPipelineBinding(), + References: normalizedStringMap(fileLane.References), } if fileLane.Merge != nil { lane.Merge = fileLane.Merge.toPipelineBinding() @@ -318,6 +334,25 @@ func normalizedMapKeys[T any](values map[string]T, keyName string) ([]string, ma return keys, rawByNormalized, nil } +func normalizedStringMap(values map[string]string) map[string]string { + if len(values) == 0 { + return nil + } + out := make(map[string]string, len(values)) + keys := make([]string, 0, len(values)) + rawByNormalized := make(map[string]string, len(values)) + for rawKey := range values { + key := strings.TrimSpace(rawKey) + rawByNormalized[key] = rawKey + keys = append(keys, key) + } + sort.Strings(keys) + for _, key := range keys { + out[key] = strings.TrimSpace(values[rawByNormalized[key]]) + } + return out +} + func resolveAPIKeyEnv(envName string, lookup func(string) (string, bool)) (string, error) { name := strings.TrimSpace(envName) if name == "" { diff --git a/internal/core/config/file_config_test.go b/internal/core/config/file_config_test.go index 2009bfe..2fff9ce 100644 --- a/internal/core/config/file_config_test.go +++ b/internal/core/config/file_config_test.go @@ -144,6 +144,31 @@ pipelines: } } +func TestParseFileConfigReferenceMaps(t *testing.T) { + cfg := parseAndApplyConfig(t, ` +version: 1 +pipelines: + example: + input: fake/input + references: + " roster ": " ./shared-roster.yml " + artifacts: + events: + extract: fake/extract + references: + " lore ": " ./lore.md " +`) + + profile := cfg.Pipelines["example"] + if !reflect.DeepEqual(profile.References, map[string]string{"roster": "./shared-roster.yml"}) { + t.Fatalf("pipeline references = %#v, want trimmed map", profile.References) + } + gotLaneRefs := profile.Artifacts["events"].References + if !reflect.DeepEqual(gotLaneRefs, map[string]string{"lore": "./lore.md"}) { + t.Fatalf("lane references = %#v, want trimmed map", gotLaneRefs) + } +} + func TestParseFileConfigValidatorMixedBindingForms(t *testing.T) { cfg := parseAndApplyConfig(t, ` version: 1 @@ -297,6 +322,58 @@ pipelines: } } +func TestApplyFileConfigRejectsDuplicateTrimmedReferenceSlots(t *testing.T) { + tests := []struct { + name string + raw string + want string + }{ + { + name: "pipeline", + raw: ` +version: 1 +pipelines: + example: + input: fake/input + references: + roster: ./first.yml + " roster ": ./second.yml +`, + want: `pipeline "example" reference slot`, + }, + { + name: "lane", + raw: ` +version: 1 +pipelines: + example: + input: fake/input + artifacts: + events: + extract: fake/extract + references: + roster: ./first.yml + " roster ": ./second.yml +`, + want: `pipeline "example" lane "events" reference slot`, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + fileCfg, err := ParseFileConfigYAML([]byte(tc.raw)) + if err != nil { + t.Fatalf("ParseFileConfigYAML: %v", err) + } + cfg := Default() + err = cfg.applyFileConfigWithLookup(fileCfg, emptyLookup) + if err == nil || !strings.Contains(err.Error(), tc.want) || !strings.Contains(err.Error(), "duplicated") { + t.Fatalf("expected duplicate reference slot error, got %v", err) + } + }) + } +} + func TestApplyFileConfigAllowsRetryOnlyLLMProfile(t *testing.T) { cfg := parseAndApplyConfig(t, ` version: 1 diff --git a/internal/core/config/redaction.go b/internal/core/config/redaction.go index ee41b21..dd9494c 100644 --- a/internal/core/config/redaction.go +++ b/internal/core/config/redaction.go @@ -21,10 +21,12 @@ func (c Config) RedactedDiagnosticsPayload() any { func (e EffectiveConfig) RedactedDiagnosticsPayload() any { return EffectiveConfig{ - Config: e.Config.Redacted(), - PipelineID: e.PipelineID, - Only: append([]string(nil), e.Only...), - ResolvedPipeline: cloneResolvedPipeline(e.ResolvedPipeline), + Config: e.Config.Redacted(), + PipelineID: e.PipelineID, + Only: append([]string(nil), e.Only...), + ReferenceOverrides: append([]pipeline.ReferenceBinding(nil), e.ReferenceOverrides...), + ReferenceUnbinds: append([]pipeline.ReferenceUnbind(nil), e.ReferenceUnbinds...), + ResolvedPipeline: cloneResolvedPipeline(e.ResolvedPipeline), } } @@ -47,6 +49,7 @@ func cloneResolvedArtifactLane(in pipeline.ResolvedArtifactLane) pipeline.Resolv out.Extract = cloneModuleBinding(in.Extract) out.Merge = cloneModuleBinding(in.Merge) out.Normalize = cloneModuleBinding(in.Normalize) + out.References = append([]pipeline.ReferenceBinding(nil), in.References...) if len(in.Validators) > 0 { out.Validators = make([]pipeline.ModuleBinding, len(in.Validators)) for i, binding := range in.Validators { diff --git a/internal/core/config/redaction_test.go b/internal/core/config/redaction_test.go index 178c741..d218ac0 100644 --- a/internal/core/config/redaction_test.go +++ b/internal/core/config/redaction_test.go @@ -3,6 +3,7 @@ package config import ( "testing" + "gitea.maximumdirect.net/eric/notarius/internal/framework/contracts" "gitea.maximumdirect.net/eric/notarius/internal/framework/pipeline" ) @@ -65,12 +66,21 @@ func TestEffectiveConfigRedactedDiagnosticsPayloadRedactsAndCopies(t *testing.T) cfg.LLMProfiles[pipeline.DefaultLLMProfile] = profile lane := cfg.Pipelines["example"].Artifacts["events"] lane.Extract.Options = map[string]any{"temperature": 0.2} + lane.References = map[string]string{"roster": "./roster.yml"} cfg.Pipelines["example"].Artifacts["events"] = lane effective, err := cfg.Resolve(ResolveInput{ PipelineID: "example", Only: []string{"events"}, - Catalog: fakeCatalog(t), + Catalog: fakeCatalog(t, pipeline.ModuleSpec{ + Key: "fake/extract", + Stage: pipeline.StageExtract, + Requires: []string{"chunks"}, + Provides: []string{"artifact"}, + ReferenceSlots: []contracts.ReferenceSlot{ + {Name: "roster"}, + }, + }), }) if err != nil { t.Fatalf("Resolve: %v", err) @@ -98,4 +108,8 @@ func TestEffectiveConfigRedactedDiagnosticsPayloadRedactsAndCopies(t *testing.T) if effective.ResolvedPipeline.ArtifactLanes[0].Extract.Options["temperature"] != 0.2 { t.Fatalf("expected resolved pipeline options to be copied") } + payload.ResolvedPipeline.ArtifactLanes[0].References[0].Source = "./changed.yml" + if effective.ResolvedPipeline.ArtifactLanes[0].References[0].Source != "./roster.yml" { + t.Fatalf("expected resolved pipeline references to be copied") + } } diff --git a/internal/core/config/validation.go b/internal/core/config/validation.go index b6b2615..61d53cc 100644 --- a/internal/core/config/validation.go +++ b/internal/core/config/validation.go @@ -98,11 +98,17 @@ func validatePipelineProfiles(profiles map[string]pipeline.PipelineProfile, llmP if err := validateBindingLLMProfile(id, "", "output", profile.Output, llmProfiles); err != nil { return err } + if err := validateReferenceMap(id, "", profile.References); err != nil { + return err + } for rawLaneID, lane := range profile.Artifacts { laneID := strings.TrimSpace(rawLaneID) if laneID == "" { return fmt.Errorf("pipeline %q artifact lane id must not be empty", id) } + if err := validateReferenceMap(id, laneID, lane.References); err != nil { + return err + } if err := validateBindingLLMProfile(id, laneID, "extract", lane.Extract, llmProfiles); err != nil { return err } @@ -122,6 +128,33 @@ func validatePipelineProfiles(profiles map[string]pipeline.PipelineProfile, llmP return nil } +func validateReferenceMap(pipelineID string, laneID string, references map[string]string) error { + seen := make(map[string]struct{}, len(references)) + for rawSlotName, rawSource := range references { + slotName := strings.TrimSpace(rawSlotName) + if slotName == "" { + if laneID != "" { + return fmt.Errorf("pipeline %q lane %q reference slot name must not be empty", pipelineID, laneID) + } + return fmt.Errorf("pipeline %q reference slot name must not be empty", pipelineID) + } + if _, ok := seen[slotName]; ok { + if laneID != "" { + return fmt.Errorf("pipeline %q lane %q reference slot %q is duplicated after trimming", pipelineID, laneID, slotName) + } + return fmt.Errorf("pipeline %q reference slot %q is duplicated after trimming", pipelineID, slotName) + } + seen[slotName] = struct{}{} + if strings.TrimSpace(rawSource) == "" { + if laneID != "" { + return fmt.Errorf("pipeline %q lane %q reference slot %q source must not be empty", pipelineID, laneID, slotName) + } + return fmt.Errorf("pipeline %q reference slot %q source must not be empty", pipelineID, slotName) + } + } + return nil +} + func validateBindingLLMProfile( pipelineID string, laneID string, diff --git a/internal/core/config/validation_test.go b/internal/core/config/validation_test.go index 33a7356..1fbe0d6 100644 --- a/internal/core/config/validation_test.go +++ b/internal/core/config/validation_test.go @@ -116,6 +116,73 @@ func TestValidateRejectsInvalidDiagnosticsRetention(t *testing.T) { } } +func TestValidateRejectsInvalidReferenceMaps(t *testing.T) { + tests := []struct { + name string + mutate func(Config) Config + want []string + }{ + { + name: "empty pipeline slot", + mutate: func(cfg Config) Config { + profile := cfg.Pipelines["example"] + profile.References = map[string]string{" ": "./roster.yml"} + cfg.Pipelines["example"] = profile + return cfg + }, + want: []string{"example", "reference slot", "empty"}, + }, + { + name: "empty pipeline source", + mutate: func(cfg Config) Config { + profile := cfg.Pipelines["example"] + profile.References = map[string]string{"roster": " "} + cfg.Pipelines["example"] = profile + return cfg + }, + want: []string{"example", "roster", "source", "empty"}, + }, + { + name: "empty lane slot", + mutate: func(cfg Config) Config { + profile := cfg.Pipelines["example"] + lane := profile.Artifacts["events"] + lane.References = map[string]string{" ": "./roster.yml"} + profile.Artifacts["events"] = lane + cfg.Pipelines["example"] = profile + return cfg + }, + want: []string{"example", "events", "reference slot", "empty"}, + }, + { + name: "empty lane source", + mutate: func(cfg Config) Config { + profile := cfg.Pipelines["example"] + lane := profile.Artifacts["events"] + lane.References = map[string]string{"roster": " "} + profile.Artifacts["events"] = lane + cfg.Pipelines["example"] = profile + return cfg + }, + want: []string{"example", "events", "roster", "source", "empty"}, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + err := tc.mutate(validConfig()).Validate() + if err == nil { + t.Fatal("Validate() error = nil, want error") + } + for _, want := range tc.want { + if !strings.Contains(err.Error(), want) { + t.Fatalf("Validate() error = %q, want substring %q", err.Error(), want) + } + } + }) + } +} + func TestValidateRejectsEmptyIDs(t *testing.T) { tests := []struct { name string diff --git a/internal/framework/pipeline/profile.go b/internal/framework/pipeline/profile.go index b667e49..785b2e3 100644 --- a/internal/framework/pipeline/profile.go +++ b/internal/framework/pipeline/profile.go @@ -7,6 +7,8 @@ import ( "fmt" "sort" "strings" + + "gitea.maximumdirect.net/eric/notarius/internal/framework/contracts" ) const ( @@ -24,22 +26,38 @@ type ModuleBinding struct { } type ArtifactLaneProfile struct { - Extract ModuleBinding `json:"extract"` - Merge ModuleBinding `json:"merge,omitempty"` - Normalize ModuleBinding `json:"normalize,omitempty"` - Validators []ModuleBinding `json:"validators,omitempty"` + Extract ModuleBinding `json:"extract"` + Merge ModuleBinding `json:"merge,omitempty"` + Normalize ModuleBinding `json:"normalize,omitempty"` + Validators []ModuleBinding `json:"validators,omitempty"` + References map[string]string `json:"references,omitempty"` } type PipelineProfile struct { - ID string `json:"id"` - Input ModuleBinding `json:"input"` - Chunk ModuleBinding `json:"chunk,omitempty"` - Artifacts map[string]ArtifactLaneProfile `json:"artifacts"` - Output ModuleBinding `json:"output,omitempty"` + ID string `json:"id"` + Input ModuleBinding `json:"input"` + Chunk ModuleBinding `json:"chunk,omitempty"` + Artifacts map[string]ArtifactLaneProfile `json:"artifacts"` + Output ModuleBinding `json:"output,omitempty"` + References map[string]string `json:"references,omitempty"` } type ResolveOptions struct { - Only []string + Only []string + ReferenceOverrides []ReferenceBinding + ReferenceUnbinds []ReferenceUnbind +} + +type ReferenceBinding struct { + LaneID string `json:"lane_id,omitempty"` + SlotName string `json:"slot_name"` + Source string `json:"source"` + BindingSource string `json:"binding_source,omitempty"` +} + +type ReferenceUnbind struct { + LaneID string `json:"lane_id"` + SlotName string `json:"slot_name"` } type ResolvedArtifactLane struct { @@ -48,6 +66,7 @@ type ResolvedArtifactLane struct { Merge ModuleBinding Normalize ModuleBinding Validators []ModuleBinding + References []ReferenceBinding `json:"references,omitempty"` } type ResolvedPipeline struct { @@ -126,7 +145,7 @@ func ResolvePipeline(profile PipelineProfile, options ResolveOptions, catalog Mo for _, laneID := range selectedLaneIDs { laneProfile := lanesByID[laneID] - lane, laneCapabilities, err := resolveArtifactLane(pipelineID, laneID, laneProfile, capabilities, catalog) + lane, laneCapabilities, err := resolveArtifactLane(pipelineID, laneID, laneProfile, profile.References, options, capabilities, catalog) if err != nil { return ResolvedPipeline{}, err } @@ -150,7 +169,15 @@ func ResolvePipeline(profile PipelineProfile, options ResolveOptions, catalog Mo return resolved, nil } -func resolveArtifactLane(pipelineID, laneID string, profile ArtifactLaneProfile, inherited capabilitySet, catalog ModuleCatalog) (ResolvedArtifactLane, capabilitySet, error) { +func resolveArtifactLane( + pipelineID string, + laneID string, + profile ArtifactLaneProfile, + pipelineReferences map[string]string, + options ResolveOptions, + inherited capabilitySet, + catalog ModuleCatalog, +) (ResolvedArtifactLane, capabilitySet, error) { lane := ResolvedArtifactLane{ ID: laneID, Extract: resolveBinding(profile.Extract, ""), @@ -171,6 +198,11 @@ func resolveArtifactLane(pipelineID, laneID string, profile ArtifactLaneProfile, if missing, ok := capabilities.missing(extractSpec.Requires); ok { return ResolvedArtifactLane{}, nil, capabilityError(pipelineID, laneID, StageExtract, lane.Extract.Module, missing) } + references, err := resolveReferenceBindings(pipelineID, laneID, lane.Extract.Module, extractSpec.ReferenceSlots, pipelineReferences, profile.References, options) + if err != nil { + return ResolvedArtifactLane{}, nil, err + } + lane.References = references capabilities.add(extractSpec.Provides...) mergeSpec, err := mergerSpec(catalog, lane.Merge.Module) @@ -205,6 +237,162 @@ func resolveArtifactLane(pipelineID, laneID string, profile ArtifactLaneProfile, return lane, capabilities, 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 { + slotByName[slot.Name] = slot + } + + bindings := make(map[string]ReferenceBinding) + addBinding := func(slotName, source, bindingSource string) error { + 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) + } + if source == "" { + return fmt.Errorf("pipeline %q lane %q reference slot %q source must not be empty", pipelineID, laneID, 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) + } + bindings[slotName] = ReferenceBinding{ + LaneID: laneID, + SlotName: slotName, + Source: source, + BindingSource: bindingSource, + } + return nil + } + + normalizedPipelineReferences, err := normalizedReferenceMap(pipelineReferences, fmt.Sprintf("pipeline %q reference slot", pipelineID)) + if err != nil { + return nil, err + } + for _, slotName := range sortedStringMapKeys(normalizedPipelineReferences) { + if _, ok := slotByName[slotName]; !ok { + continue + } + if err := addBinding(slotName, normalizedPipelineReferences[slotName], contracts.ReferenceBindingSourceConfig); err != nil { + return nil, err + } + } + + normalizedLaneReferences, err := normalizedReferenceMap(laneReferences, fmt.Sprintf("pipeline %q lane %q reference slot", pipelineID, laneID)) + if err != nil { + return nil, err + } + for _, slotName := range sortedStringMapKeys(normalizedLaneReferences) { + if err := addBinding(slotName, normalizedLaneReferences[slotName], contracts.ReferenceBindingSourceConfig); err != nil { + return nil, err + } + } + + for _, override := range options.ReferenceOverrides { + optionLaneID := strings.TrimSpace(override.LaneID) + if optionLaneID == "" { + return nil, fmt.Errorf("pipeline %q reference override lane id must not be empty", pipelineID) + } + if optionLaneID != laneID { + continue + } + source := override.BindingSource + if strings.TrimSpace(source) == "" { + source = contracts.ReferenceBindingSourceCLI + } + if err := addBinding(override.SlotName, override.Source, strings.TrimSpace(source)); err != nil { + return nil, err + } + } + + for _, unbind := range options.ReferenceUnbinds { + optionLaneID := strings.TrimSpace(unbind.LaneID) + if optionLaneID == "" { + return nil, fmt.Errorf("pipeline %q reference unbind lane id must not be empty", pipelineID) + } + if optionLaneID != 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) + } + if _, ok := slotByName[slotName]; !ok { + return nil, fmt.Errorf("pipeline %q lane %q reference slot %q is not declared", pipelineID, laneID, slotName) + } + delete(bindings, slotName) + } + + for _, slot := range 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) + } + } + } + + keys := sortedReferenceBindingKeys(bindings) + resolved := make([]ReferenceBinding, 0, len(keys)) + for _, slotName := range keys { + resolved = append(resolved, bindings[slotName]) + } + return resolved, nil +} + +func normalizedReferenceMap(values map[string]string, keyName string) (map[string]string, error) { + if len(values) == 0 { + return nil, nil + } + out := make(map[string]string, len(values)) + for rawSlotName, rawSource := range values { + slotName := strings.TrimSpace(rawSlotName) + if slotName == "" { + return nil, fmt.Errorf("%s must not be empty", keyName) + } + if _, ok := out[slotName]; ok { + return nil, fmt.Errorf("%s %q is duplicated after trimming", keyName, slotName) + } + source := strings.TrimSpace(rawSource) + if source == "" { + return nil, fmt.Errorf("%s %q source must not be empty", keyName, slotName) + } + out[slotName] = source + } + return out, nil +} + +func sortedStringMapKeys(values map[string]string) []string { + if len(values) == 0 { + return nil + } + keys := make([]string, 0, len(values)) + for key := range values { + keys = append(keys, key) + } + sort.Strings(keys) + return keys +} + +func sortedReferenceBindingKeys(values map[string]ReferenceBinding) []string { + if len(values) == 0 { + return nil + } + keys := make([]string, 0, len(values)) + for key := range values { + keys = append(keys, key) + } + sort.Strings(keys) + return keys +} + func resolveBinding(binding ModuleBinding, defaultModule string) ModuleBinding { module := strings.TrimSpace(binding.Module) if module == "" { diff --git a/internal/framework/pipeline/profile_test.go b/internal/framework/pipeline/profile_test.go index 6accb0a..2fedbc1 100644 --- a/internal/framework/pipeline/profile_test.go +++ b/internal/framework/pipeline/profile_test.go @@ -3,6 +3,7 @@ package pipeline import ( "context" "encoding/json" + "errors" "reflect" "strings" "testing" @@ -125,6 +126,137 @@ func TestResolvePipelineSelectsOnlyRequestedLanes(t *testing.T) { } } +func TestResolvePipelineAppliesReferenceBindings(t *testing.T) { + profile := multiLaneProfile() + profile.References = map[string]string{ + " roster ": " ./shared-roster.yml ", + "unclaimed": "./ignored.yml", + } + lane := profile.Artifacts["events"] + lane.References = map[string]string{ + "roster": "./lane-roster.yml", + " lore ": " ./lore.md ", + } + profile.Artifacts["events"] = lane + + catalog := newProfileCatalogWithOverride(t, ModuleSpec{ + Key: "event-extractor", + Stage: StageExtract, + Requires: []string{"chunk"}, + Provides: []string{"candidate"}, + ReferenceSlots: []contracts.ReferenceSlot{ + {Name: "roster", Required: true}, + {Name: "lore"}, + }, + }) + resolved, err := ResolvePipeline(profile, ResolveOptions{Only: []string{"events", "summaries"}}, catalog) + if err != nil { + t.Fatalf("ResolvePipeline() error = %v, want nil", err) + } + + events := resolvedLane(t, resolved.ArtifactLanes, "events") + want := []ReferenceBinding{ + {LaneID: "events", SlotName: "lore", Source: "./lore.md", BindingSource: contracts.ReferenceBindingSourceConfig}, + {LaneID: "events", SlotName: "roster", Source: "./lane-roster.yml", BindingSource: contracts.ReferenceBindingSourceConfig}, + } + if !reflect.DeepEqual(events.References, want) { + t.Fatalf("events references = %#v, want %#v", events.References, want) + } + summaries := resolvedLane(t, resolved.ArtifactLanes, "summaries") + if len(summaries.References) != 0 { + t.Fatalf("summaries references = %#v, want none", summaries.References) + } +} + +func TestResolvePipelineRejectsUndeclaredReferenceSlot(t *testing.T) { + profile := baselineProfile() + lane := profile.Artifacts["events"] + lane.References = map[string]string{"missing": "./missing.yml"} + profile.Artifacts["events"] = lane + + _, err := ResolvePipeline(profile, ResolveOptions{}, newProfileCatalog(t)) + if err == nil { + t.Fatal("ResolvePipeline() error = nil, want error") + } + assertErrorContains(t, err, "events", "missing", "not declared") +} + +func TestResolvePipelineRequiresBoundReferenceSlotsForSelectedLanes(t *testing.T) { + catalog := newProfileCatalogWithOverride(t, ModuleSpec{ + Key: "event-extractor", + Stage: StageExtract, + Requires: []string{"chunk"}, + Provides: []string{"candidate"}, + ReferenceSlots: []contracts.ReferenceSlot{ + {Name: "roster", Required: true}, + }, + }) + + if _, err := ResolvePipeline(multiLaneProfile(), ResolveOptions{Only: []string{"notes"}}, catalog); err != nil { + t.Fatalf("ResolvePipeline(unselected required slot) error = %v, want nil", err) + } + + _, err := ResolvePipeline(multiLaneProfile(), ResolveOptions{Only: []string{"events"}}, catalog) + if err == nil { + t.Fatal("ResolvePipeline(selected required slot) error = nil, want error") + } + assertErrorContains(t, err, "events", "required", "roster", "not bound") +} + +func TestResolvePipelineReferenceUnbindCanLeaveRequiredSlotMissing(t *testing.T) { + profile := baselineProfile() + profile.References = map[string]string{"roster": "./roster.yml"} + catalog := newProfileCatalogWithOverride(t, ModuleSpec{ + Key: "event-extractor", + Stage: StageExtract, + Requires: []string{"chunk"}, + Provides: []string{"candidate"}, + ReferenceSlots: []contracts.ReferenceSlot{ + {Name: "roster", Required: true}, + }, + }) + + _, err := ResolvePipeline(profile, ResolveOptions{ + ReferenceUnbinds: []ReferenceUnbind{{LaneID: "events", SlotName: "roster"}}, + }, catalog) + if err == nil { + t.Fatal("ResolvePipeline() error = nil, want error") + } + assertErrorContains(t, err, "events", "required", "roster", "not bound") +} + +func TestResolvePipelineUsesReferenceSlotsFromSpecWithoutConstructingExtractor(t *testing.T) { + profile := baselineProfile() + profile.References = map[string]string{"roster": "./roster.yml"} + catalog := emptyProfileCatalog() + for _, spec := range defaultProfileSpecs() { + if spec.Key != "event-extractor" { + registerProfileSpecs(t, catalog, spec) + } + } + if err := catalog.Extractors.RegisterWithSpec(ModuleSpec{ + Key: "event-extractor", + Stage: StageExtract, + Requires: []string{"chunk"}, + Provides: []string{"candidate"}, + ReferenceSlots: []contracts.ReferenceSlot{ + {Name: "roster", Required: true}, + }, + }, func() (contracts.Extractor, error) { + return nil, errors.New("constructor should not run") + }); err != nil { + t.Fatalf("RegisterWithSpec() error = %v, want nil", err) + } + + resolved, err := ResolvePipeline(profile, ResolveOptions{}, catalog) + if err != nil { + t.Fatalf("ResolvePipeline() error = %v, want nil", err) + } + if got := resolved.ArtifactLanes[0].References[0].Source; got != "./roster.yml" { + t.Fatalf("reference source = %q, want ./roster.yml", got) + } +} + func TestResolvePipelineRejectsUnknownOnlyLane(t *testing.T) { _, err := ResolvePipeline(multiLaneProfile(), ResolveOptions{Only: []string{"missing"}}, newProfileCatalog(t)) if err == nil { @@ -463,6 +595,17 @@ func laneIDs(lanes []ResolvedArtifactLane) []string { return ids } +func resolvedLane(t *testing.T, lanes []ResolvedArtifactLane, laneID string) ResolvedArtifactLane { + t.Helper() + for _, lane := range lanes { + if lane.ID == laneID { + return lane + } + } + t.Fatalf("lane %q not found in %#v", laneID, laneIDs(lanes)) + return ResolvedArtifactLane{} +} + func assertErrorContains(t *testing.T, err error, values ...string) { t.Helper()