eric/notarius
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add stage-local reference config bindings

Browse Source
This commit is contained in:
Eric Rakestraw
2026-07-05 16:18:00 +00:00
parent 9278797aa9
commit 51053d390d
8 changed files with 437 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
internal/core/config/config.go
View File

@@ -114,6 +114,7 @@ func cloneModuleBinding(in pipeline.ModuleBinding) pipeline.ModuleBinding {
if len(in.Options) > 0 {
out.Options = cloneOptions(in.Options)
}
out.References = cloneStringMap(in.References)
return out
}

57
internal/core/config/file_config.go
View File

@@ -99,6 +99,7 @@ type fileModuleBinding struct {
Module string
LLMProfile string
Options map[string]any
References map[string]string
}
func (b *fileModuleBinding) UnmarshalYAML(node *yaml.Node) error {
@@ -133,6 +134,12 @@ func (b *fileModuleBinding) UnmarshalYAML(node *yaml.Node) error {
return err
}
b.Options = normalizeOptions(options)
case "references":
var references map[string]string
if err := valueNode.Decode(&references); err != nil {
return err
}
b.References = references
default:
return fmt.Errorf("field %s not found in module binding", keyNode.Value)
}
@@ -148,6 +155,7 @@ func (b fileModuleBinding) toPipelineBinding() pipeline.ModuleBinding {
Module: strings.TrimSpace(b.Module),
LLMProfile: strings.TrimSpace(b.LLMProfile),
Options: cloneOptions(b.Options),
References: normalizedStringMap(b.References),
}
}
@@ -217,6 +225,19 @@ func (c *Config) applyFileConfigWithLookup(fileCfg FileConfig, lookup func(strin
if _, _, err := normalizedMapKeys(filePipeline.References, fmt.Sprintf("pipeline %q reference slot", pipelineID)); err != nil {
return err
}
if filePipeline.Chunk != nil {
if _, _, err := normalizedMapKeys(filePipeline.Chunk.References, fmt.Sprintf("pipeline %q chunk reference slot", pipelineID)); err != nil {
return err
}
}
if _, _, err := normalizedMapKeys(filePipeline.Input.References, fmt.Sprintf("pipeline %q input reference slot", pipelineID)); err != nil {
return err
}
if filePipeline.Output != nil {
if _, _, err := normalizedMapKeys(filePipeline.Output.References, fmt.Sprintf("pipeline %q output reference slot", pipelineID)); err != nil {
return err
}
}
for rawLaneID, fileLane := range filePipeline.Artifacts {
laneID := strings.TrimSpace(rawLaneID)
if laneID == "" {
@@ -225,6 +246,24 @@ func (c *Config) applyFileConfigWithLookup(fileCfg FileConfig, lookup func(strin
if _, _, err := normalizedMapKeys(fileLane.References, fmt.Sprintf("pipeline %q lane %q reference slot", pipelineID, laneID)); err != nil {
return err
}
if _, _, err := normalizedMapKeys(fileLane.Extract.References, fmt.Sprintf("pipeline %q lane %q extract reference slot", pipelineID, laneID)); err != nil {
return err
}
if fileLane.Merge != nil {
if _, _, err := normalizedMapKeys(fileLane.Merge.References, fmt.Sprintf("pipeline %q lane %q merge reference slot", pipelineID, laneID)); err != nil {
return err
}
}
if fileLane.Normalize != nil {
if _, _, err := normalizedMapKeys(fileLane.Normalize.References, fmt.Sprintf("pipeline %q lane %q normalize reference slot", pipelineID, laneID)); err != nil {
return err
}
}
for i, validator := range fileLane.Validators {
if _, _, err := normalizedMapKeys(validator.References, fmt.Sprintf("pipeline %q lane %q validator[%d] reference slot", pipelineID, laneID, i)); err != nil {
return err
}
}
}
}
@@ -280,8 +319,10 @@ func (c *Config) applyFileConfigWithLookup(fileCfg FileConfig, lookup func(strin
}
for _, laneID := range laneIDs {
fileLane := filePipeline.Artifacts[rawLaneIDs[laneID]]
extract := fileLane.Extract.toPipelineBinding()
extract.References = mergeStringMaps(normalizedStringMap(fileLane.References), extract.References)
lane := pipeline.ArtifactLaneProfile{
Extract: fileLane.Extract.toPipelineBinding(),
Extract: extract,
References: normalizedStringMap(fileLane.References),
}
if fileLane.Merge != nil {
@@ -353,6 +394,20 @@ func normalizedStringMap(values map[string]string) map[string]string {
return out
}
func mergeStringMaps(base map[string]string, override map[string]string) map[string]string {
if len(base) == 0 && len(override) == 0 {
return nil
}
out := make(map[string]string, len(base)+len(override))
for key, value := range base {
out[key] = value
}
for key, value := range override {
out[key] = value
}
return out
}
func resolveAPIKeyEnv(envName string, lookup func(string) (string, bool)) (string, error) {
name := strings.TrimSpace(envName)
if name == "" {

97
internal/core/config/file_config_test.go
View File

@@ -169,6 +169,53 @@ pipelines:
}
}
func TestParseFileConfigStageLocalReferenceMaps(t *testing.T) {
cfg := parseAndApplyConfig(t, `
version: 1
pipelines:
example:
input: fake/input
chunk:
module: generic
references:
" scene_guide ": " ./scenes.md "
artifacts:
events:
extract:
module: fake/extract
references:
" glossary ": " ./glossary.md "
" roster ": " ./extract-roster.yml "
references:
roster: ./legacy-roster.yml
lore: ./lore.md
normalize:
module: noop
references:
" normalization_notes ": " ./normalization.md "
`)
profile := cfg.Pipelines["example"]
if !reflect.DeepEqual(profile.Chunk.References, map[string]string{"scene_guide": "./scenes.md"}) {
t.Fatalf("chunk references = %#v, want trimmed map", profile.Chunk.References)
}
lane := profile.Artifacts["events"]
if !reflect.DeepEqual(lane.References, map[string]string{"lore": "./lore.md", "roster": "./legacy-roster.yml"}) {
t.Fatalf("legacy lane references = %#v, want trimmed map", lane.References)
}
wantExtract := map[string]string{
"glossary": "./glossary.md",
"lore": "./lore.md",
"roster": "./extract-roster.yml",
}
if !reflect.DeepEqual(lane.Extract.References, wantExtract) {
t.Fatalf("extract references = %#v, want legacy merged with extract override %#v", lane.Extract.References, wantExtract)
}
if !reflect.DeepEqual(lane.Normalize.References, map[string]string{"normalization_notes": "./normalization.md"}) {
t.Fatalf("normalize references = %#v, want trimmed map", lane.Normalize.References)
}
}
func TestParseFileConfigValidatorMixedBindingForms(t *testing.T) {
cfg := parseAndApplyConfig(t, `
version: 1
@@ -357,6 +404,56 @@ pipelines:
`,
want: `pipeline "example" lane "events" reference slot`,
},
{
name: "chunk",
raw: `
version: 1
pipelines:
example:
input: fake/input
chunk:
module: generic
references:
roster: ./first.yml
" roster ": ./second.yml
`,
want: `pipeline "example" chunk reference slot`,
},
{
name: "extract",
raw: `
version: 1
pipelines:
example:
input: fake/input
artifacts:
events:
extract:
module: fake/extract
references:
roster: ./first.yml
" roster ": ./second.yml
`,
want: `pipeline "example" lane "events" extract reference slot`,
},
{
name: "normalize",
raw: `
version: 1
pipelines:
example:
input: fake/input
artifacts:
events:
extract: fake/extract
normalize:
module: noop
references:
roster: ./first.yml
" roster ": ./second.yml
`,
want: `pipeline "example" lane "events" normalize reference slot`,
},
}
for _, tc := range tests {

29
internal/core/config/redaction_test.go
View File

@@ -67,7 +67,12 @@ func TestEffectiveConfigRedactedDiagnosticsPayloadRedactsAndCopies(t *testing.T)
lane := cfg.Pipelines["example"].Artifacts["events"]
lane.Extract.Options = map[string]any{"temperature": 0.2}
lane.References = map[string]string{"roster": "./roster.yml"}
lane.Extract.References = map[string]string{"glossary": "./glossary.md"}
lane.Normalize.References = map[string]string{"notes": "./normalize.md"}
cfg.Pipelines["example"].Artifacts["events"] = lane
pipelineProfile := cfg.Pipelines["example"]
pipelineProfile.Chunk.References = map[string]string{"scene_guide": "./scene.md"}
cfg.Pipelines["example"] = pipelineProfile
effective, err := cfg.Resolve(ResolveInput{
PipelineID: "example",
@@ -78,6 +83,7 @@ func TestEffectiveConfigRedactedDiagnosticsPayloadRedactsAndCopies(t *testing.T)
Requires: []string{"chunks"},
Provides: []string{"artifact"},
ReferenceSlots: []contracts.ReferenceSlot{
{Name: "glossary"},
{Name: "roster"},
},
}),
@@ -109,9 +115,21 @@ func TestEffectiveConfigRedactedDiagnosticsPayloadRedactsAndCopies(t *testing.T)
t.Fatalf("expected resolved pipeline options to be copied")
}
payload.ResolvedPipeline.ArtifactLanes[0].ExtractReferences.Bindings[0].Source = "./changed.yml"
if effective.ResolvedPipeline.ArtifactLanes[0].ExtractReferences.Bindings[0].Source != "./roster.yml" {
if referenceBindingSource(effective.ResolvedPipeline.ArtifactLanes[0].ExtractReferences.Bindings, "roster") != "./roster.yml" {
t.Fatalf("expected resolved pipeline references to be copied")
}
payload.ResolvedPipeline.Chunk.References["scene_guide"] = "./changed-scene.md"
if effective.ResolvedPipeline.Chunk.References["scene_guide"] != "./scene.md" {
t.Fatalf("expected chunk references to be copied")
}
payload.ResolvedPipeline.ArtifactLanes[0].Extract.References["glossary"] = "./changed-glossary.md"
if effective.ResolvedPipeline.ArtifactLanes[0].Extract.References["glossary"] != "./glossary.md" {
t.Fatalf("expected extract references to be copied")
}
payload.ResolvedPipeline.ArtifactLanes[0].Normalize.References["notes"] = "./changed-normalize.md"
if effective.ResolvedPipeline.ArtifactLanes[0].Normalize.References["notes"] != "./normalize.md" {
t.Fatalf("expected normalize references to be copied")
}
effective.ResolvedPipeline.ArtifactLanes[0].ExtractReferences.ReferenceSet = contracts.ReferenceSet{
Slots: map[string]contracts.ResolvedReferenceSlot{
@@ -136,3 +154,12 @@ func TestEffectiveConfigRedactedDiagnosticsPayloadRedactsAndCopies(t *testing.T)
t.Fatalf("expected materialized reference content to be copied, got %q", got)
}
}
func referenceBindingSource(bindings []pipeline.ReferenceBinding, slotName string) string {
for _, binding := range bindings {
if binding.SlotName == slotName {
return binding.Source
}
}
return ""
}

69
internal/core/config/validation.go
View File

@@ -89,13 +89,13 @@ func validatePipelineProfiles(profiles map[string]pipeline.PipelineProfile, llmP
if profile.ID != "" && strings.TrimSpace(profile.ID) != id {
return fmt.Errorf("pipeline %q profile id %q does not match map key", id, profile.ID)
}
if err := validateBindingLLMProfile(id, "", "input", profile.Input, llmProfiles); err != nil {
if err := validateBinding(id, "", "input", profile.Input, llmProfiles, false); err != nil {
return err
}
if err := validateBindingLLMProfile(id, "", "chunk", profile.Chunk, llmProfiles); err != nil {
if err := validateBinding(id, "", "chunk", profile.Chunk, llmProfiles, true); err != nil {
return err
}
if err := validateBindingLLMProfile(id, "", "output", profile.Output, llmProfiles); err != nil {
if err := validateBinding(id, "", "output", profile.Output, llmProfiles, false); err != nil {
return err
}
if err := validateReferenceMap(id, "", profile.References); err != nil {
@@ -109,17 +109,17 @@ func validatePipelineProfiles(profiles map[string]pipeline.PipelineProfile, llmP
if err := validateReferenceMap(id, laneID, lane.References); err != nil {
return err
}
if err := validateBindingLLMProfile(id, laneID, "extract", lane.Extract, llmProfiles); err != nil {
if err := validateBinding(id, laneID, "extract", lane.Extract, llmProfiles, true); err != nil {
return err
}
if err := validateBindingLLMProfile(id, laneID, "merge", lane.Merge, llmProfiles); err != nil {
if err := validateBinding(id, laneID, "merge", lane.Merge, llmProfiles, false); err != nil {
return err
}
if err := validateBindingLLMProfile(id, laneID, "normalize", lane.Normalize, llmProfiles); err != nil {
if err := validateBinding(id, laneID, "normalize", lane.Normalize, llmProfiles, true); err != nil {
return err
}
for i, validator := range lane.Validators {
if err := validateBindingLLMProfile(id, laneID, fmt.Sprintf("validator[%d]", i), validator, llmProfiles); err != nil {
if err := validateBinding(id, laneID, fmt.Sprintf("validator[%d]", i), validator, llmProfiles, false); err != nil {
return err
}
}
@@ -128,33 +128,64 @@ func validatePipelineProfiles(profiles map[string]pipeline.PipelineProfile, llmP
return nil
}
func validateBinding(
pipelineID string,
laneID string,
slot string,
binding pipeline.ModuleBinding,
profiles map[string]LLMProfile,
referencesAllowed bool,
) error {
if err := validateBindingLLMProfile(pipelineID, laneID, slot, binding, profiles); err != nil {
return err
}
if len(binding.References) == 0 {
return nil
}
if !referencesAllowed {
if laneID != "" {
return fmt.Errorf("pipeline %q lane %q %s references are not supported", pipelineID, laneID, slot)
}
return fmt.Errorf("pipeline %q %s references are not supported", pipelineID, slot)
}
return validateReferenceMapForContext(pipelineID, laneID, slot, binding.References)
}
func validateReferenceMap(pipelineID string, laneID string, references map[string]string) error {
return validateReferenceMapForContext(pipelineID, laneID, "", references)
}
func validateReferenceMapForContext(pipelineID string, laneID string, slot 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)
return fmt.Errorf("%s reference slot name must not be empty", referenceContext(pipelineID, laneID, slot))
}
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)
return fmt.Errorf("%s reference slot %q is duplicated after trimming", referenceContext(pipelineID, laneID, slot), 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 fmt.Errorf("%s reference slot %q source must not be empty", referenceContext(pipelineID, laneID, slot), slotName)
}
}
return nil
}
func referenceContext(pipelineID string, laneID string, slot string) string {
if laneID != "" && slot != "" {
return fmt.Sprintf("pipeline %q lane %q %s", pipelineID, laneID, slot)
}
if laneID != "" {
return fmt.Sprintf("pipeline %q lane %q", pipelineID, laneID)
}
if slot != "" {
return fmt.Sprintf("pipeline %q %s", pipelineID, slot)
}
return fmt.Sprintf("pipeline %q", pipelineID)
}
func validateBindingLLMProfile(
pipelineID string,
laneID string,

138
internal/core/config/validation_test.go
View File

@@ -122,6 +122,74 @@ func TestValidateRejectsInvalidReferenceMaps(t *testing.T) {
mutate func(Config) Config
want []string
}{
{
name: "empty chunk slot",
mutate: func(cfg Config) Config {
profile := cfg.Pipelines["example"]
profile.Chunk.References = map[string]string{" ": "./roster.yml"}
cfg.Pipelines["example"] = profile
return cfg
},
want: []string{"example", "chunk", "reference slot", "empty"},
},
{
name: "empty chunk source",
mutate: func(cfg Config) Config {
profile := cfg.Pipelines["example"]
profile.Chunk.References = map[string]string{"roster": " "}
cfg.Pipelines["example"] = profile
return cfg
},
want: []string{"example", "chunk", "roster", "source", "empty"},
},
{
name: "empty extract slot",
mutate: func(cfg Config) Config {
profile := cfg.Pipelines["example"]
lane := profile.Artifacts["events"]
lane.Extract.References = map[string]string{" ": "./roster.yml"}
profile.Artifacts["events"] = lane
cfg.Pipelines["example"] = profile
return cfg
},
want: []string{"example", "events", "extract", "reference slot", "empty"},
},
{
name: "empty extract source",
mutate: func(cfg Config) Config {
profile := cfg.Pipelines["example"]
lane := profile.Artifacts["events"]
lane.Extract.References = map[string]string{"roster": " "}
profile.Artifacts["events"] = lane
cfg.Pipelines["example"] = profile
return cfg
},
want: []string{"example", "events", "extract", "roster", "source", "empty"},
},
{
name: "empty normalize slot",
mutate: func(cfg Config) Config {
profile := cfg.Pipelines["example"]
lane := profile.Artifacts["events"]
lane.Normalize.References = map[string]string{" ": "./roster.yml"}
profile.Artifacts["events"] = lane
cfg.Pipelines["example"] = profile
return cfg
},
want: []string{"example", "events", "normalize", "reference slot", "empty"},
},
{
name: "empty normalize source",
mutate: func(cfg Config) Config {
profile := cfg.Pipelines["example"]
lane := profile.Artifacts["events"]
lane.Normalize.References = map[string]string{"roster": " "}
profile.Artifacts["events"] = lane
cfg.Pipelines["example"] = profile
return cfg
},
want: []string{"example", "events", "normalize", "roster", "source", "empty"},
},
{
name: "empty pipeline slot",
mutate: func(cfg Config) Config {
@@ -183,6 +251,76 @@ func TestValidateRejectsInvalidReferenceMaps(t *testing.T) {
}
}
func TestValidateRejectsReferencesOnUnsupportedBindings(t *testing.T) {
tests := []struct {
name string
mutate func(Config) Config
want []string
}{
{
name: "input",
mutate: func(cfg Config) Config {
profile := cfg.Pipelines["example"]
profile.Input.References = map[string]string{"roster": "./roster.yml"}
cfg.Pipelines["example"] = profile
return cfg
},
want: []string{"example", "input", "references", "not supported"},
},
{
name: "merge",
mutate: func(cfg Config) Config {
profile := cfg.Pipelines["example"]
lane := profile.Artifacts["events"]
lane.Merge.References = map[string]string{"roster": "./roster.yml"}
profile.Artifacts["events"] = lane
cfg.Pipelines["example"] = profile
return cfg
},
want: []string{"example", "events", "merge", "references", "not supported"},
},
{
name: "validator",
mutate: func(cfg Config) Config {
profile := cfg.Pipelines["example"]
lane := profile.Artifacts["events"]
lane.Validators = []pipeline.ModuleBinding{{
Module: "fake/validator",
References: map[string]string{"roster": "./roster.yml"},
}}
profile.Artifacts["events"] = lane
cfg.Pipelines["example"] = profile
return cfg
},
want: []string{"example", "events", "validator[0]", "references", "not supported"},
},
{
name: "output",
mutate: func(cfg Config) Config {
profile := cfg.Pipelines["example"]
profile.Output.References = map[string]string{"roster": "./roster.yml"}
cfg.Pipelines["example"] = profile
return cfg
},
want: []string{"example", "output", "references", "not supported"},
},
}
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